rto_graph/store.rs
1//! SQLite-backed graph store.
2
3use std::collections::BTreeSet;
4use std::path::Path;
5
6use rusqlite::{Connection, OptionalExtension, params};
7
8use crate::findings::{self, AnalysisRun, Finding, FindingsApplied, FindingsLayer};
9use crate::media::{self, MediaFilter, MediaKind, MediaRecord, MediaWrite, ProducerSummary};
10use crate::memory::{
11 self, CacheEntry, CacheStats, CacheSweep, CacheWrite, MemoryError, MemoryFilter,
12 MemoryForgotten, MemoryListing, MemoryRecord, MemoryWrite, Recall, RecallOptions,
13};
14use crate::migrations;
15use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
16use crate::provenance::Provenance;
17
18/// Errors raised by the store.
19#[derive(Debug, thiserror::Error)]
20pub enum StoreError {
21 /// Underlying `SQLite` failure.
22 #[error("sqlite error: {0}")]
23 Sqlite(#[from] rusqlite::Error),
24 /// A node's `meta` could not be (de)serialized as JSON.
25 #[error("json error: {0}")]
26 Json(#[from] serde_json::Error),
27 /// An edge referenced a node key that does not exist in the store.
28 #[error("unknown node key: {0}")]
29 UnknownNode(String),
30 /// An edge violated the provenance/confidence invariant.
31 #[error("invalid edge: {0}")]
32 InvalidEdge(String),
33 /// A stored value could not be interpreted (database corruption).
34 #[error("corrupt store: {0}")]
35 Corrupt(String),
36}
37
38/// A store written by a build **newer** than this one: it records migrations
39/// this binary has never heard of. Produced by [`Store::schema_ahead`].
40///
41/// # Why this is its own type, and not a `StoreError` variant
42///
43/// Opening such a store is not an error and reading its *schema* is sound —
44/// migrations are additive in effect, so every column an older build selects is
45/// still there (each `DROP TABLE` in `crate::migrations` is a table rebuild
46/// that re-selects every prior column). What is *not* sound is **rewriting** the
47/// graph: this build would re-extract every file with its older extractor and
48/// replace the newer build's content with worse content, silently. So the
49/// refusal belongs on the write paths, and the carrier of that refusal has no
50/// business in the type every `Store` call returns.
51///
52/// # Columns are not the whole story, since migration 14
53///
54/// "Additive in effect" was a claim about *schema*, and every migration up to 13
55/// kept it about values too. Migration 14 does not: it widened the provenance
56/// vocabulary, so an older build reading a row written at `external-authored`
57/// gets [`StoreError::Corrupt`] from `Provenance::from_token`, which knows only
58/// its own six-minus-three. That is a real limit of the read guarantee and is
59/// recorded here rather than left to be discovered.
60///
61/// It is not left silent, either. Such a store necessarily records migration 14,
62/// so this type reports it as written by a newer Roteiro *before* any row is
63/// read, and [`StoreError::Corrupt`]'s message names the same cause. The
64/// alternative — teaching `from_token` a tolerant fallback — was rejected: every
65/// fallback value is one of the six, and each choice silently re-tiers a fact.
66///
67/// The second reason is semver, and it is the binding one. `StoreError` is a
68/// public enum on a published 1.x crate and is not `#[non_exhaustive]`, so
69/// adding a variant would stop a downstream exhaustive `match` from compiling —
70/// a breaking change, which in this workspace means a major bump of all seven
71/// crates. A new type is purely additive. Its fields are private and read
72/// through accessors for the same reason: adding one later stays non-breaking.
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub struct SchemaAhead {
75 store: u32,
76 build: u32,
77 unknown: Vec<u32>,
78}
79
80impl SchemaAhead {
81 /// The highest schema version this store records — necessarily one this
82 /// build does not know.
83 #[must_use]
84 pub fn store_version(&self) -> u32 {
85 self.store
86 }
87
88 /// The highest schema version this build knows how to apply.
89 #[must_use]
90 pub fn build_version(&self) -> u32 {
91 self.build
92 }
93
94 /// Every recorded version this build has never heard of, ascending. Never
95 /// empty. More than one means the store ran several unknown migrations, not
96 /// that anything is inconsistent.
97 #[must_use]
98 pub fn unknown_versions(&self) -> &[u32] {
99 &self.unknown
100 }
101}
102
103impl std::fmt::Display for SchemaAhead {
104 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105 let unknown = self
106 .unknown
107 .iter()
108 .map(u32::to_string)
109 .collect::<Vec<_>>()
110 .join(", ");
111 // Both versions, and what to do about it. A bare "schema mismatch" would
112 // leave the reader with nothing to act on: the useful facts are *which*
113 // side is behind (this binary) and that the fix is upgrading it, not
114 // deleting the store.
115 write!(
116 f,
117 "this graph store was written by a newer Roteiro: it is at schema \
118 version {store}, and this build knows only up to {build} \
119 (unrecognised migration(s): {unknown}). Refusing to rewrite the \
120 graph — this build would re-extract every file with its older \
121 extractor and replace the newer build's graph with worse content, \
122 silently. Upgrade the `roteiro` binary to one that knows schema \
123 version {store} or later, then retry. Reading this store is \
124 unaffected and needs no upgrade.",
125 store = self.store,
126 build = self.build,
127 )
128 }
129}
130
131impl std::error::Error for SchemaAhead {}
132
133/// A summary of applying/re-applying import layers (see
134/// [`Store::apply_import_layer`] and [`Store::reapply_imports`]).
135#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
136pub struct ImportApplied {
137 /// Number of import layers processed.
138 pub layers: usize,
139 /// Import nodes upserted (across all layers).
140 pub nodes: usize,
141 /// Import edges applied — both endpoints resolved. A duplicate of an
142 /// already-present edge is a harmless no-op but still counted as applied.
143 pub edges_applied: usize,
144 /// Import edges **pruned**: an endpoint was absent (a cross-reference to code
145 /// that no longer exists), so the edge was dropped from the persisted layer
146 /// rather than kept as stale data.
147 pub edges_pruned: usize,
148 /// Import **nodes removed**: a node this `src_ref` used to contribute and no
149 /// longer does, which nothing else in the store still refers to. See
150 /// [`Store::apply_import_layer`] for the rule and why a withdrawn concept
151 /// has to go.
152 pub nodes_removed: usize,
153}
154
155/// Qualified node columns for `SELECT`s that alias the `nodes` table as `n`.
156const 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";
157
158/// `SELECT` prefix that yields an [`Edge`] row (endpoints resolved back to keys).
159const EDGE_SELECT: &str = "SELECT ns.key AS src, nd.key AS dst, e.kind, e.provenance, \
160 e.confidence, e.src_ref \
161 FROM edges e JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst";
162
163/// A Roteiro graph store backed by a single `SQLite` database.
164pub struct Store {
165 conn: Connection,
166}
167
168impl Store {
169 /// Open (creating if absent) a store at `path` and apply pending migrations.
170 ///
171 /// # Errors
172 /// Returns [`StoreError::Sqlite`] if the database cannot be opened or a
173 /// migration fails.
174 pub fn open(path: &Path) -> Result<Self, StoreError> {
175 let conn = Connection::open(path)?;
176 Self::from_conn(conn)
177 }
178
179 /// Open an in-memory store (tests, previews).
180 ///
181 /// # Errors
182 /// Returns [`StoreError::Sqlite`] if a migration fails.
183 pub fn open_in_memory() -> Result<Self, StoreError> {
184 let conn = Connection::open_in_memory()?;
185 Self::from_conn(conn)
186 }
187
188 fn from_conn(mut conn: Connection) -> Result<Self, StoreError> {
189 conn.execute_batch("PRAGMA foreign_keys = ON;")?;
190 // Wait for a concurrent writer instead of failing with `database is
191 // locked`. Matters for workspace `serve` (ADR-0008), where a long-lived
192 // server reads a project's graph while that repo's own `roteiro sync`
193 // commits an update to the same file — and for any two Roteiro processes
194 // over one repository, which is ordinary rather than exotic: an editor's
195 // MCP client and a terminal, or several clients each spawning their own
196 // stdio server.
197 conn.busy_timeout(std::time::Duration::from_secs(5))?;
198
199 // **The timeout alone does not deliver that**, which is what this pair of
200 // settings is for. Both are needed and neither substitutes for the other.
201 //
202 // `Immediate` takes the write lock when the transaction *opens*, rather
203 // than upgrading to it on the first write. rusqlite's default is the lazy
204 // one — `TransactionBehavior::Deferred` — and under it two writers that roteiro:ignore
205 // each read before they write, which every upsert here does, both hold a
206 // shared lock and then both ask to upgrade.
207 // SQLite refuses that immediately with `SQLITE_BUSY` and **does not invoke
208 // the busy handler**, because waiting could only deadlock. So the five
209 // second timeout above was never reached in the case it looks like it
210 // covers: one of the two writers died at once. Taking the lock up front
211 // turns the deadlock into a queue, which is what the timeout can wait on.
212 conn.set_transaction_behavior(rusqlite::TransactionBehavior::Immediate);
213
214 // WAL is the other half: under the default rollback journal a writer
215 // excludes readers for the whole transaction, so a [`crate::sync`] blocks every
216 // query in a running `serve`. In WAL, readers never block writers and
217 // writers never block readers, so only writer-against-writer waits.
218 //
219 // Tolerated rather than required, and **silently** — which is the one place
220 // silence is right here. WAL needs shared memory and is refused on most
221 // network filesystems; `PRAGMA journal_mode` reports the mode actually in
222 // force rather than failing. A store that cannot take WAL keeps exactly
223 // the behaviour it had before this change, so there is no new failure to
224 // report — only an improvement that did not apply. `:memory:` answers
225 // `memory` for the same reason and is equally fine.
226 //
227 // The result is discarded rather than checked because nothing here can act
228 // on it. `PRAGMA journal_mode` answers for anyone who needs to know which
229 // of the two a given store got.
230 let _: String = conn.query_row("PRAGMA journal_mode = WAL", [], |r| r.get(0))?;
231
232 migrations::apply(&mut conn)?;
233 Ok(Self { conn })
234 }
235
236 /// The schema version this store has been migrated to: the highest `v` for
237 /// which every migration `1..=v` is recorded as applied.
238 ///
239 /// **Not the maximum recorded version.** Migrations are selected by set
240 /// membership rather than `> MAX(version)` (see `crate::migrations`), so a
241 /// store *can* hold a gap — one written by a build that knew a higher
242 /// migration but not a lower one. For a store recorded as `1..11, 13`, the
243 /// maximum is 13 while none of migration 12's schema is present; reporting 13
244 /// would be a higher number than the truth. The contiguous prefix reports 11,
245 /// which is the version the store can actually be relied on to provide.
246 ///
247 /// The gap is transient in practice — the next [`Store::open`] repairs it —
248 /// but this must not answer wrongly in the window where it exists, which is
249 /// exactly the window in which someone is debugging.
250 ///
251 /// # Errors
252 /// Returns [`StoreError::Sqlite`] on query failure.
253 pub fn schema_version(&self) -> Result<u32, StoreError> {
254 Ok(migrations::store_version(&self.conn)?)
255 }
256
257 /// The highest schema version **this build** knows how to apply — the
258 /// binary's half of the comparison [`Store::schema_ahead`] makes.
259 #[must_use]
260 pub fn build_schema_version() -> u32 {
261 migrations::latest_version()
262 }
263
264 /// `Some` when this store was written by a **newer** build: it records
265 /// migrations this binary has never heard of. `None` — the ordinary case —
266 /// when the store is level with this build or behind it.
267 ///
268 /// Call this before **rewriting** a graph (sync, reconcile, rebuild). Do not
269 /// call it to gate reads: an older binary's reads are sound, which is the
270 /// whole reason this is not checked in [`Store::open`] (issue #342).
271 ///
272 /// # Which version this compares, and why it is not [`Store::schema_version`]
273 ///
274 /// It compares the **set** of recorded versions against the migrations this
275 /// build carries, so it answers *has anything newer than me written here?*
276 /// [`Store::schema_version`] answers a different question — the highest
277 /// **gap-free** version, a floor describing what a reader may rely on — and
278 /// using it here would let the bug through: a store recorded `1..=13, 15`,
279 /// read by a build that knows 13, has a gap-free version of 13, so
280 /// `schema_version() > build` is false while migration 15's schema sits in
281 /// the file and a newer build plainly assembled the graph.
282 ///
283 /// Conflating the two situations would also misreport both, so it does not.
284 /// A **gap** — a store missing a lower migration this build knows — is not
285 /// a store from the future, and [`Store::open`] has already repaired it by
286 /// applying the missing migration before this can be called. What is left is
287 /// only ever a version no build of this vintage could have written.
288 ///
289 /// # Errors
290 /// Returns [`StoreError::Sqlite`] on query failure.
291 pub fn schema_ahead(&self) -> Result<Option<SchemaAhead>, StoreError> {
292 let unknown = migrations::versions_ahead_of_build(&self.conn)?;
293 let Some(&store) = unknown.last() else {
294 return Ok(None);
295 };
296 Ok(Some(SchemaAhead {
297 store,
298 build: Self::build_schema_version(),
299 unknown,
300 }))
301 }
302
303 /// Number of nodes currently in the store.
304 ///
305 /// # Errors
306 /// Returns [`StoreError::Sqlite`] on query failure.
307 pub fn node_count(&self) -> Result<u64, StoreError> {
308 let n: i64 = self
309 .conn
310 .query_row("SELECT COUNT(*) FROM nodes", [], |r| r.get(0))?;
311 Ok(u64::try_from(n).unwrap_or(0))
312 }
313
314 /// Number of `file` nodes currently in the store — the count `SyncReport`
315 /// means by `blobs_total`.
316 ///
317 /// Exists so a **no-op** sync reports the same quantity the reconciling
318 /// paths do. Those derive `blobs_total` from the assembled graph
319 /// (`file_count`), deliberately, so that a path the repository excluded is
320 /// not counted; a no-op branch that answered with the size of its input blob
321 /// list instead would claim excluded files were in the graph, and say so only
322 /// on the runs that did no work.
323 ///
324 /// # Errors
325 /// Returns [`StoreError::Sqlite`] on query failure.
326 pub fn file_node_count(&self) -> Result<usize, StoreError> {
327 let n: i64 = self.conn.query_row(
328 "SELECT COUNT(*) FROM nodes WHERE kind = ?1",
329 [crate::NodeKind::File.as_str()],
330 |r| r.get(0),
331 )?;
332 Ok(usize::try_from(n).unwrap_or(0))
333 }
334
335 /// Number of edges currently in the store.
336 ///
337 /// # Errors
338 /// Returns [`StoreError::Sqlite`] on query failure.
339 pub fn edge_count(&self) -> Result<u64, StoreError> {
340 let n: i64 = self
341 .conn
342 .query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))?;
343 Ok(u64::try_from(n).unwrap_or(0))
344 }
345
346 /// Insert or update a node, keyed by its natural [`Node::key`].
347 ///
348 /// # Errors
349 /// Returns [`StoreError::Json`] if `meta` cannot be serialized, or
350 /// [`StoreError::Sqlite`] on write failure.
351 pub fn upsert_node(&self, node: &Node) -> Result<(), StoreError> {
352 upsert_node(&self.conn, node)
353 }
354
355 /// Insert an edge. Both endpoints must already resolve to nodes.
356 ///
357 /// # Errors
358 /// Returns [`StoreError::InvalidEdge`] if the provenance/confidence
359 /// invariant is violated, [`StoreError::UnknownNode`] if an endpoint key is
360 /// absent, or [`StoreError::Sqlite`] on write failure.
361 pub fn insert_edge(&self, edge: &Edge) -> Result<(), StoreError> {
362 insert_edge(&self.conn, edge)
363 }
364
365 /// Apply a fact set atomically: all nodes are upserted, then all edges are
366 /// inserted, in a single transaction. On any error nothing is committed.
367 ///
368 /// # Errors
369 /// Returns the first error encountered (see [`Store::upsert_node`] and
370 /// [`Store::insert_edge`]); the transaction is rolled back.
371 pub fn apply_factset(&mut self, facts: &FactSet) -> Result<(), StoreError> {
372 let tx = self.conn.transaction()?;
373 for node in &facts.nodes {
374 upsert_node(&tx, node)?;
375 }
376 for edge in &facts.edges {
377 insert_edge(&tx, edge)?;
378 }
379 tx.commit()?;
380 Ok(())
381 }
382
383 /// The `HEAD` tree id recorded at the last successful [`Store::rebuild`], if
384 /// any. Used by the sync engine to detect an unchanged tree.
385 ///
386 /// # Errors
387 /// Returns [`StoreError::Sqlite`] on query failure.
388 pub fn sync_state(&self) -> Result<Option<String>, StoreError> {
389 Ok(self
390 .conn
391 .query_row("SELECT tree FROM sync_state WHERE id = 0", [], |r| r.get(0))
392 .optional()?)
393 }
394
395 /// The extractor environment recorded with the last committed [`crate::sync`],
396 /// `None` if unset (a legacy row, or the last sync was a worktree/index
397 /// preview). The incremental committed [`crate::sync`] compares this to the current
398 /// env and falls back to a full re-extraction when they differ.
399 ///
400 /// # Errors
401 /// Returns [`StoreError::Sqlite`] on query failure.
402 pub fn sync_env(&self) -> Result<Option<String>, StoreError> {
403 Ok(self
404 .conn
405 .query_row("SELECT env FROM sync_state WHERE id = 0", [], |r| r.get(0))
406 .optional()?
407 .flatten())
408 }
409
410 /// Record the extractor environment for the current synced tree. Called by a
411 /// committed [`crate::sync`] right after it writes the tree, so a later sync can decide
412 /// whether the incremental fast path is sound. A no-op if no tree is recorded.
413 ///
414 /// # Errors
415 /// Returns [`StoreError::Sqlite`] on write failure.
416 pub fn set_sync_env(&self, env: &str) -> Result<(), StoreError> {
417 self.conn
418 .execute("UPDATE sync_state SET env = ?1 WHERE id = 0", [env])?;
419 Ok(())
420 }
421
422 /// Which working tree this graph was assembled from, or `None` when the store
423 /// has never been synced or predates the column (issue #330).
424 ///
425 /// `graph.db` is an assembled view of **one** tree, so a store that came to
426 /// describe a different tree — restored from a backup, copied along with a
427 /// `.git` directory, or reached after a layout change — would otherwise
428 /// answer confidently about the wrong one: [`crate::sync`] reporting "up to date"
429 /// against a state id belonging to someone else's tree, `check` validating a
430 /// tree nobody is looking at. `None` reads as "unknown" and is *adopted*
431 /// rather than treated as a mismatch, so an existing store is never rebuilt
432 /// merely for predating the stamp.
433 ///
434 /// # Errors
435 /// Returns [`StoreError::Sqlite`] on query failure.
436 pub fn synced_worktree(&self) -> Result<Option<String>, StoreError> {
437 Ok(self
438 .conn
439 .query_row("SELECT worktree FROM sync_state WHERE id = 0", [], |r| {
440 r.get(0)
441 })
442 .optional()?
443 .flatten())
444 }
445
446 /// Stamp this graph as assembled from `worktree`. Called by every sync entry
447 /// point right after it writes the tree state, so the two always agree about
448 /// whose tree the state describes. A no-op if no tree is recorded.
449 ///
450 /// Unlike [`Store::set_sync_env`]'s value, this survives a tree write: it
451 /// identifies the *store*, not the tree, and a new commit does not move the
452 /// graph to a different working tree.
453 ///
454 /// # Errors
455 /// Returns [`StoreError::Sqlite`] on write failure.
456 pub fn set_synced_worktree(&self, worktree: &str) -> Result<(), StoreError> {
457 self.conn.execute(
458 "UPDATE sync_state SET worktree = ?1 WHERE id = 0",
459 [worktree],
460 )?;
461 Ok(())
462 }
463
464 /// Atomically replace the entire graph with `facts`, recording `tree` as the
465 /// synced state (or clearing it when `tree` is `None`). All existing nodes
466 /// and edges are deleted first, so the store reflects exactly the given fact
467 /// set.
468 ///
469 /// Passing `None` records *no* synced tree — distinct from an empty string —
470 /// so [`Store::sync_state`] returns `None` and a later [`crate::sync`] will not
471 /// spuriously short-circuit.
472 ///
473 /// # Errors
474 /// Returns the first error encountered (see [`Store::apply_factset`]); on any
475 /// error nothing is committed.
476 pub fn rebuild(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
477 let tx = self.conn.transaction()?;
478 tx.execute("DELETE FROM edges", [])?;
479 tx.execute("DELETE FROM nodes", [])?;
480 for node in &facts.nodes {
481 upsert_node(&tx, node)?;
482 }
483 for edge in &facts.edges {
484 insert_edge(&tx, edge)?;
485 }
486 write_sync_state(&tx, tree)?;
487 tx.commit()?;
488 Ok(())
489 }
490
491 /// Bring the store to exactly `facts` (as [`Store::rebuild`] does) but writing
492 /// only what **differs** instead of wiping and reinserting the whole graph —
493 /// the git-style "write only the delta". Unchanged node rows (which carry the
494 /// heavy JSON `meta`) and unchanged edge rows are left untouched; only removed
495 /// rows are deleted and new/changed rows written. The final state — nodes,
496 /// edges, and `sync_state` — is identical to `rebuild(facts, tree)`.
497 ///
498 /// Leaving unchanged edges in place means their row ids do not match a cold
499 /// rebuild's — which is safe *because* every edge query is content-ordered
500 /// (`(src, dst, kind, provenance)`, see [`Store::all_edges`]), never by row
501 /// id. So an incrementally reconciled store and a fresh rebuild return every
502 /// query identically; the delta is invisible above the storage layer.
503 ///
504 /// # Errors
505 /// Returns [`StoreError`] on a query failure; the transaction is rolled back.
506 pub fn reconcile(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
507 let current_nodes = self.all_nodes()?;
508 let current_edges = self.all_edges()?;
509 let cur_by_key: std::collections::HashMap<&str, &Node> =
510 current_nodes.iter().map(|n| (n.key.as_str(), n)).collect();
511 let new_keys: std::collections::HashSet<&str> =
512 facts.nodes.iter().map(|n| n.key.as_str()).collect();
513
514 // Edge identity is the full tuple, so a changed `confidence`/`src_ref`
515 // counts as remove-old + add-new — keeping the result identical to a
516 // wholesale rebuild, not the store's insert-time `DO NOTHING` semantics.
517 let new_edge_ids: std::collections::HashSet<EdgeId> =
518 facts.edges.iter().map(edge_identity).collect();
519 let cur_edge_ids: std::collections::HashSet<EdgeId> =
520 current_edges.iter().map(edge_identity).collect();
521
522 let tx = self.conn.transaction()?;
523 // 1. Delete removed edges first, so any node they reference can then be
524 // dropped (edges are FK-constrained on node ids, with no cascade). A
525 // removed node's edges are all removals, so they are gone before step 2.
526 for edge in ¤t_edges {
527 if !new_edge_ids.contains(&edge_identity(edge)) {
528 delete_edge(&tx, edge)?;
529 }
530 }
531 // 2. Drop nodes that no longer exist.
532 for old in ¤t_nodes {
533 if !new_keys.contains(old.key.as_str()) {
534 tx.execute("DELETE FROM nodes WHERE key = ?1", [&old.key])?;
535 }
536 }
537 // 3. Upsert only the nodes that are new or whose content changed (an upsert
538 // keeps the row id, so unchanged edges stay valid).
539 for node in &facts.nodes {
540 if cur_by_key
541 .get(node.key.as_str())
542 .is_none_or(|cur| *cur != node)
543 {
544 upsert_node(&tx, node)?;
545 }
546 }
547 // 4. Insert only the added edges (their endpoints now all exist).
548 for edge in &facts.edges {
549 if !cur_edge_ids.contains(&edge_identity(edge)) {
550 insert_edge(&tx, edge)?;
551 }
552 }
553 write_sync_state(&tx, tree)?;
554 tx.commit()?;
555 Ok(())
556 }
557
558 /// Fetch a node by its natural key.
559 ///
560 /// # Errors
561 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
562 /// [`StoreError::Corrupt`] if a stored value cannot be decoded.
563 pub fn get_node(&self, key: &str) -> Result<Option<Node>, StoreError> {
564 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.key = ?1");
565 let mut stmt = self.conn.prepare(&sql)?;
566 let mut rows = stmt.query([key])?;
567 match rows.next()? {
568 Some(row) => Ok(Some(row_to_node(row)?)),
569 None => Ok(None),
570 }
571 }
572
573 /// Every node key in the store, ordered. Useful for whole-graph exports.
574 ///
575 /// # Errors
576 /// Returns [`StoreError::Sqlite`] on query failure.
577 pub fn all_keys(&self) -> Result<Vec<String>, StoreError> {
578 let mut stmt = self.conn.prepare("SELECT key FROM nodes ORDER BY key")?;
579 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
580 let mut out = Vec::new();
581 for row in rows {
582 out.push(row?);
583 }
584 Ok(out)
585 }
586
587 /// Dump the entire graph as a single [`FactSet`], with nodes and edges in a
588 /// deterministic order — suitable for a portable, content-stable artifact.
589 ///
590 /// # Errors
591 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
592 /// [`StoreError::Corrupt`] on decode failure.
593 pub fn export_factset(&self) -> Result<FactSet, StoreError> {
594 let node_sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
595 let mut node_stmt = self.conn.prepare(&node_sql)?;
596 let mut node_rows = node_stmt.query([])?;
597 let nodes = collect_nodes(&mut node_rows)?;
598
599 // Order edges by their resolved endpoint keys (not row id) so the dump is
600 // stable regardless of insertion order.
601 let edge_sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
602 let mut edge_stmt = self.conn.prepare(&edge_sql)?;
603 let mut edge_rows = edge_stmt.query([])?;
604 let edges = collect_edges(&mut edge_rows)?;
605
606 Ok(FactSet { nodes, edges })
607 }
608
609 /// All nodes of a given kind.
610 ///
611 /// # Errors
612 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
613 /// [`StoreError::Corrupt`] on decode failure.
614 pub fn nodes_by_kind(&self, kind: &NodeKind) -> Result<Vec<Node>, StoreError> {
615 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.kind = ?1 ORDER BY n.key");
616 let mut stmt = self.conn.prepare(&sql)?;
617 let mut rows = stmt.query([kind.as_str()])?;
618 collect_nodes(&mut rows)
619 }
620
621 /// Nodes of a given kind whose `name` equals `name_lower` **case-insensitively**,
622 /// ordered by key. Narrows a lookup at the SQL layer — using the `kind` index and
623 /// filtering `name` in-query — so only matching rows are decoded, never every
624 /// node of that kind. Used by the cross-repo follow bridge to fetch just the
625 /// candidate struct(s) for a config section rather than scanning all structs.
626 ///
627 /// # Errors
628 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
629 /// [`StoreError::Corrupt`] on decode failure.
630 pub fn nodes_by_kind_named(
631 &self,
632 kind: &NodeKind,
633 name_lower: &str,
634 ) -> Result<Vec<Node>, StoreError> {
635 let sql = format!(
636 "SELECT {NODE_COLS} FROM nodes n \
637 WHERE n.kind = ?1 AND lower(n.name) = ?2 ORDER BY n.key"
638 );
639 let mut stmt = self.conn.prepare(&sql)?;
640 let mut rows = stmt.query([kind.as_str(), name_lower])?;
641 collect_nodes(&mut rows)
642 }
643
644 /// Every `config_key` node's flattened setting (ADR-0009), read back out of
645 /// the graph as [`crate::ConfigKey`]s — the graph-native source the cross-repo
646 /// link matcher (`roteiro links --infer`) consumes, so it never re-parses
647 /// config files. Ordered by node key (deterministic). A node missing the
648 /// `key`/`path` a well-formed `config_key` carries is skipped defensively.
649 ///
650 /// # Errors
651 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
652 /// [`StoreError::Corrupt`] on decode failure.
653 pub fn config_keys(&self) -> Result<Vec<crate::ConfigKey>, StoreError> {
654 let nodes = self.nodes_by_kind(&NodeKind::Other(crate::config_keys::KIND.to_owned()))?;
655 let mut out = Vec::with_capacity(nodes.len());
656 for n in &nodes {
657 let key = n.meta.get("key").and_then(serde_json::Value::as_str);
658 // A `config_key` node carries a `value` in `meta` when it has a real
659 // setting (file-derived keys always do, even an empty string). A
660 // struct-derived key (`meta.source = "struct"`) omits it — its value is
661 // *unknown*, not empty — so record that absence explicitly rather than
662 // defaulting it to `""`, which would false-match in value agreement.
663 let value = n.meta.get("value").and_then(serde_json::Value::as_str);
664 if let (Some(key), Some(file)) = (key, n.path.as_deref()) {
665 out.push(crate::ConfigKey {
666 file: file.to_owned(),
667 key: key.to_owned(),
668 value: value.unwrap_or_default().to_owned(),
669 value_known: value.is_some(),
670 });
671 }
672 }
673 Ok(out)
674 }
675
676 /// Every node whose source `path` is `path`, ordered by key — the file node
677 /// plus the symbols and markers defined in it. Used to scope a change to the
678 /// graph (e.g. `roteiro review`).
679 ///
680 /// # Errors
681 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
682 /// [`StoreError::Corrupt`] on decode failure.
683 pub fn nodes_by_path(&self, path: &str) -> Result<Vec<Node>, StoreError> {
684 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.path = ?1 ORDER BY n.key");
685 let mut stmt = self.conn.prepare(&sql)?;
686 let mut rows = stmt.query([path])?;
687 collect_nodes(&mut rows)
688 }
689
690 /// Every node produced by a given layer, ordered by key. The incremental
691 /// [`crate::sync`] loads the `Derived` layer to reconstruct the extraction graph
692 /// without re-reading every blob.
693 ///
694 /// # Errors
695 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
696 /// [`StoreError::Corrupt`] on decode failure.
697 pub fn nodes_by_provenance(&self, provenance: Provenance) -> Result<Vec<Node>, StoreError> {
698 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.provenance = ?1 ORDER BY n.key");
699 let mut stmt = self.conn.prepare(&sql)?;
700 let mut rows = stmt.query([provenance.as_str()])?;
701 collect_nodes(&mut rows)
702 }
703
704 /// Every node in the store, ordered by key. Unlike [`Store::export_factset`]
705 /// this decodes no edges, so it is cheap for node-only scans (e.g. search).
706 ///
707 /// # Errors
708 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
709 /// [`StoreError::Corrupt`] on decode failure.
710 pub fn all_nodes(&self) -> Result<Vec<Node>, StoreError> {
711 let sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
712 let mut stmt = self.conn.prepare(&sql)?;
713 let mut rows = stmt.query([])?;
714 collect_nodes(&mut rows)
715 }
716
717 /// Every edge in the store, with endpoints resolved to their node keys. Used
718 /// by [`Store::reconcile`] to diff the edge set.
719 ///
720 /// Ordered by the edge's **content** — `(src key, dst key, kind, provenance)`,
721 /// the table's unique tuple — not by row id. This makes the order a function
722 /// of the *graph*, not of insertion history, so an incrementally
723 /// [`reconcile`](Store::reconcile)d store and a cold [`rebuild`](Store::rebuild)
724 /// return edges identically. (The same reason node scans order by `key`.)
725 ///
726 /// # Errors
727 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
728 pub fn all_edges(&self) -> Result<Vec<Edge>, StoreError> {
729 let sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
730 let mut stmt = self.conn.prepare(&sql)?;
731 let mut rows = stmt.query([])?;
732 collect_edges(&mut rows)
733 }
734
735 /// Edges whose source is the node with the given key, in content order
736 /// (`(dst key, kind, provenance)` — `src` is fixed). Content-ordered rather
737 /// than by row id so the result is history-independent; see [`Store::all_edges`].
738 ///
739 /// # Errors
740 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
741 pub fn edges_from(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
742 let sql = format!("{EDGE_SELECT} WHERE ns.key = ?1 ORDER BY nd.key, e.kind, e.provenance");
743 let mut stmt = self.conn.prepare(&sql)?;
744 let mut rows = stmt.query([key])?;
745 collect_edges(&mut rows)
746 }
747
748 /// Edges whose destination is the node with the given key, in content order
749 /// (`(src key, kind, provenance)` — `dst` is fixed). See [`Store::all_edges`].
750 ///
751 /// # Errors
752 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
753 pub fn edges_to(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
754 let sql = format!("{EDGE_SELECT} WHERE nd.key = ?1 ORDER BY ns.key, e.kind, e.provenance");
755 let mut stmt = self.conn.prepare(&sql)?;
756 let mut rows = stmt.query([key])?;
757 collect_edges(&mut rows)
758 }
759
760 /// All edges with the given provenance, in content order
761 /// (`(src key, dst key, kind)` — `provenance` is fixed). See [`Store::all_edges`].
762 ///
763 /// # Errors
764 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
765 pub fn edges_by_provenance(&self, provenance: Provenance) -> Result<Vec<Edge>, StoreError> {
766 let sql = format!("{EDGE_SELECT} WHERE e.provenance = ?1 ORDER BY ns.key, nd.key, e.kind");
767 let mut stmt = self.conn.prepare(&sql)?;
768 let mut rows = stmt.query([provenance.as_str()])?;
769 collect_edges(&mut rows)
770 }
771
772 /// Delete all edges with the given provenance, returning how many were
773 /// removed. Used to re-derive a whole provenance class authoritatively (e.g.
774 /// `inferred` edges when re-running inference with different parameters).
775 ///
776 /// # Errors
777 /// Returns [`StoreError::Sqlite`] on write failure.
778 pub fn delete_edges_by_provenance(&self, provenance: Provenance) -> Result<u64, StoreError> {
779 let n = self.conn.execute(
780 "DELETE FROM edges WHERE provenance = ?1",
781 [provenance.as_str()],
782 )?;
783 Ok(u64::try_from(n).unwrap_or(0))
784 }
785
786 /// Delete all edges carrying the given `src_ref`, returning how many were
787 /// removed. Lets one producer of `inferred` edges (e.g. the embedding layer,
788 /// or a Graphify import) re-derive its own edges authoritatively without
789 /// touching edges another producer contributed.
790 ///
791 /// # Errors
792 /// Returns [`StoreError::Sqlite`] on write failure.
793 pub fn delete_edges_by_src_ref(&self, src_ref: &str) -> Result<u64, StoreError> {
794 let n = self
795 .conn
796 .execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
797 Ok(u64::try_from(n).unwrap_or(0))
798 }
799
800 /// Apply an import layer to the live graph **and** persist it durably under
801 /// `src_ref`, validating as it goes: this ref's prior edges are cleared
802 /// (an authoritative re-import), the layer's nodes are upserted, and each
803 /// edge is applied only if both endpoints resolve. Dangling edges — cross-
804 /// references to code that is not present — are dropped, and only the
805 /// validated (trimmed) layer is persisted, so stale data is never stored.
806 ///
807 /// This is the "validate on import" half; [`Store::reapply_imports`] is the
808 /// "validate on sync" half, re-checking layers against the rebuilt graph.
809 ///
810 /// # Removal propagates, because a re-run that only adds is not idempotent
811 ///
812 /// Re-importing cannot **duplicate**: nodes are `INSERT … ON CONFLICT(key) DO
813 /// UPDATE`, and this ref's edges are deleted before being re-applied. What is
814 /// not free is the other direction. A node the source has since *withdrawn*
815 /// used to survive as an orphan carrying the layer's provenance, because only
816 /// edges were pruned by `src_ref` — so a re-run **looked** idempotent while
817 /// quietly accumulating deleted facts, which is harder to notice than a
818 /// duplicate precisely because nothing appears twice.
819 ///
820 /// So a node this ref previously contributed and no longer does is removed,
821 /// under a rule deliberately narrower than "this ref owned it":
822 ///
823 /// 1. it must not appear in **any other persisted import layer** — two layers
824 /// can name one node on purpose, and `import:links` and
825 /// `import:links/authored` both contribute the same `extref:` placeholder;
826 /// 2. **nothing may still be attached to it** — a surviving edge means some
827 /// other producer still relates it to the graph, and a node's removal must
828 /// not silently take a relationship it did not own with it.
829 ///
830 /// Both are checked against the live store rather than assumed from the key,
831 /// so this stays correct for a layer whose namespace it knows nothing about.
832 /// The node's cached context row goes with it — a bundle for a node that no
833 /// longer exists can only be served as a stale answer. Episodic
834 /// [`crate::MemoryKind`] records anchored to it do **not**: ADR-0013 makes
835 /// memory the one tier with no generating function, so dropping it is data
836 /// loss, and its anchor already reports the node as gone.
837 ///
838 /// # Errors
839 /// Returns [`StoreError::Json`] if `facts` cannot be (de)serialized,
840 /// [`StoreError::InvalidEdge`] on a malformed edge, or [`StoreError::Sqlite`]
841 /// on write failure.
842 pub fn apply_import_layer(
843 &mut self,
844 src_ref: &str,
845 facts: &FactSet,
846 ) -> Result<ImportApplied, StoreError> {
847 // Read before the row is overwritten: what this ref contributed last time
848 // is the only record of what it may now be withdrawing.
849 let prior = self.import_layer_node_keys(src_ref)?;
850 let tx = self.conn.transaction()?;
851 // Authoritative re-import: drop this ref's prior edges from the live graph.
852 tx.execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
853 for node in &facts.nodes {
854 upsert_node(&tx, node)?;
855 }
856 let (kept, applied) = apply_edges_pruning(&tx, &facts.edges)?;
857 let trimmed = FactSet {
858 nodes: facts.nodes.clone(),
859 edges: kept,
860 };
861 put_import_row(&tx, src_ref, &trimmed)?;
862 // After the edges are gone and the new nodes are in, so the "still
863 // attached" test sees the graph as it will be rather than as it was.
864 let nodes_removed = remove_withdrawn_nodes(&tx, src_ref, &prior, &facts.nodes)?;
865 tx.commit()?;
866 Ok(ImportApplied {
867 layers: 1,
868 nodes: facts.nodes.len(),
869 nodes_removed,
870 ..applied
871 })
872 }
873
874 /// The node keys the persisted layer for `src_ref` currently contributes, or
875 /// an empty set when there is no such layer.
876 ///
877 /// # Errors
878 /// Returns [`StoreError::Sqlite`] on query failure or [`StoreError::Json`] if
879 /// the stored layer cannot be decoded.
880 fn import_layer_node_keys(&self, src_ref: &str) -> Result<BTreeSet<String>, StoreError> {
881 let json: Option<String> = self
882 .conn
883 .query_row(
884 "SELECT facts FROM imports WHERE src_ref = ?1",
885 [src_ref],
886 |r| r.get(0),
887 )
888 .optional()?;
889 let Some(json) = json else {
890 return Ok(BTreeSet::new());
891 };
892 let facts: FactSet = serde_json::from_str(&json)?;
893 Ok(facts.nodes.into_iter().map(|n| n.key).collect())
894 }
895
896 /// Remove the persisted import layer for `src_ref`, returning whether one
897 /// existed. Does not remove edges already in the live graph (use
898 /// [`Store::delete_edges_by_src_ref`] for that).
899 ///
900 /// # Errors
901 /// Returns [`StoreError::Sqlite`] on write failure.
902 pub fn delete_import(&self, src_ref: &str) -> Result<bool, StoreError> {
903 let n = self
904 .conn
905 .execute("DELETE FROM imports WHERE src_ref = ?1", [src_ref])?;
906 Ok(n > 0)
907 }
908
909 /// The `src_ref`s of all persisted import layers, ordered.
910 ///
911 /// # Errors
912 /// Returns [`StoreError::Sqlite`] on query failure.
913 pub fn import_refs(&self) -> Result<Vec<String>, StoreError> {
914 let mut stmt = self
915 .conn
916 .prepare("SELECT src_ref FROM imports ORDER BY src_ref")?;
917 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
918 let mut out = Vec::new();
919 for row in rows {
920 out.push(row?);
921 }
922 Ok(out)
923 }
924
925 /// Re-apply every persisted import layer on top of the current graph and
926 /// **re-validate** it: all import nodes are upserted first (so cross-layer
927 /// and self references resolve), then each edge is applied; an edge whose
928 /// endpoint is now absent — a cross-reference to code a sync removed — is
929 /// pruned from the persisted layer, not merely skipped. So the durable store
930 /// keeps only still-correct data. Idempotent; safe to run after each rebuild.
931 ///
932 /// # Errors
933 /// Returns [`StoreError::Json`] if a stored layer cannot be (de)serialized,
934 /// or [`StoreError::Sqlite`] on write failure.
935 pub fn reapply_imports(&mut self) -> Result<ImportApplied, StoreError> {
936 let layers = self.load_import_layers()?;
937 let tx = self.conn.transaction()?;
938 // Pass 1: upsert every layer's nodes so intra-import edges resolve
939 // regardless of which layer defines the endpoint.
940 for (_, facts) in &layers {
941 for node in &facts.nodes {
942 upsert_node(&tx, node)?;
943 }
944 }
945 // Pass 2: apply edges, pruning (and rewriting) any that dangle.
946 let mut applied = ImportApplied {
947 layers: layers.len(),
948 ..ImportApplied::default()
949 };
950 for (src_ref, facts) in &layers {
951 applied.nodes += facts.nodes.len();
952 let (kept, counts) = apply_edges_pruning(&tx, &facts.edges)?;
953 applied.edges_applied += counts.edges_applied;
954 applied.edges_pruned += counts.edges_pruned;
955 if kept.len() != facts.edges.len() {
956 let trimmed = FactSet {
957 nodes: facts.nodes.clone(),
958 edges: kept,
959 };
960 put_import_row(&tx, src_ref, &trimmed)?;
961 }
962 }
963 tx.commit()?;
964 Ok(applied)
965 }
966
967 /// Load and decode every persisted import layer as `(src_ref, FactSet)`, in
968 /// `src_ref` order.
969 fn load_import_layers(&self) -> Result<Vec<(String, FactSet)>, StoreError> {
970 let mut stmt = self
971 .conn
972 .prepare("SELECT src_ref, facts FROM imports ORDER BY src_ref")?;
973 let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
974 let mut out = Vec::new();
975 for row in rows {
976 let (src_ref, json) = row?;
977 let mut facts: FactSet = serde_json::from_str(&json)?;
978 // An import-layer node is *never* derived (derivation is [`crate::sync`]'s job),
979 // so a `Derived` tag here is always wrong. It arises two ways, both
980 // repaired the same: a legacy layer persisted before nodes carried
981 // provenance (the field is absent → serde defaults `Derived`), or —
982 // anomalously — a layer that stored an explicit `"provenance":"derived"`
983 // (a producer/data bug). We deliberately repair *both* rather than only
984 // the absent case: leaving an explicit-`derived` import node in place
985 // would let a layer-scoped [`crate::sync`] treat it as derived and delete it —
986 // the exact corruption this guards against — so repair is the safe
987 // recovery, not silent masking. Idempotent, runs on every reapply
988 // (old stores self-heal), and a no-op for correctly-tagged fresh imports.
989 for node in &mut facts.nodes {
990 if node.provenance == Provenance::Derived {
991 node.provenance = import_node_provenance(&node.key);
992 }
993 }
994 out.push((src_ref, facts));
995 }
996 Ok(out)
997 }
998
999 /// Neighbouring nodes reachable from `key` in the given direction. Returns
1000 /// an empty vector if the node does not exist.
1001 ///
1002 /// # Errors
1003 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
1004 /// [`StoreError::Corrupt`] on failure.
1005 pub fn neighbors(&self, key: &str, dir: Direction) -> Result<Vec<Node>, StoreError> {
1006 let out = format!(
1007 "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.dst \
1008 JOIN nodes s ON s.id = e.src WHERE s.key = ?1"
1009 );
1010 let inc = format!(
1011 "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.src \
1012 JOIN nodes d ON d.id = e.dst WHERE d.key = ?1"
1013 );
1014 // Order by output column 1 (the node key) so results are deterministic
1015 // across SQLite versions/plans. Positional ordering avoids both the
1016 // ambiguity of a bare `key` (present in every joined table) and the fact
1017 // that a table-qualified name cannot be used after the `Both` UNION.
1018 let sql = match dir {
1019 Direction::Outgoing => format!("{out} ORDER BY 1"),
1020 Direction::Incoming => format!("{inc} ORDER BY 1"),
1021 Direction::Both => format!("{out} UNION {inc} ORDER BY 1"),
1022 };
1023 let mut stmt = self.conn.prepare(&sql)?;
1024 let mut rows = stmt.query([key])?;
1025 collect_nodes(&mut rows)
1026 }
1027
1028 /// Fetch the cached context bundle for `key` as `(fingerprint, json)`, if
1029 /// present. The caller compares the fingerprint to the node's current one to
1030 /// decide whether the entry is fresh (see [`crate::context`]).
1031 ///
1032 /// # Errors
1033 /// Returns [`StoreError::Sqlite`] on query failure.
1034 pub fn context_cache_get(&self, key: &str) -> Result<Option<(String, String)>, StoreError> {
1035 let row = self
1036 .conn
1037 .query_row(
1038 "SELECT fingerprint, json FROM node_context WHERE key = ?1",
1039 [key],
1040 |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)),
1041 )
1042 .optional()?;
1043 Ok(row)
1044 }
1045
1046 /// Fetch just the cached fingerprint for `key`, without reading the (larger)
1047 /// JSON payload — for a cheap freshness check.
1048 ///
1049 /// # Errors
1050 /// Returns [`StoreError::Sqlite`] on query failure.
1051 pub fn context_cache_fingerprint(&self, key: &str) -> Result<Option<String>, StoreError> {
1052 let fp = self
1053 .conn
1054 .query_row(
1055 "SELECT fingerprint FROM node_context WHERE key = ?1",
1056 [key],
1057 |r| r.get::<_, String>(0),
1058 )
1059 .optional()?;
1060 Ok(fp)
1061 }
1062
1063 /// Store (or replace) the cached context bundle for `key`.
1064 ///
1065 /// # Errors
1066 /// Returns [`StoreError::Sqlite`] on write failure.
1067 pub fn context_cache_put(
1068 &self,
1069 key: &str,
1070 fingerprint: &str,
1071 json: &str,
1072 ) -> Result<(), StoreError> {
1073 self.conn.execute(
1074 "INSERT INTO node_context (key, fingerprint, json) VALUES (?1, ?2, ?3)
1075 ON CONFLICT(key) DO UPDATE SET
1076 fingerprint = excluded.fingerprint, json = excluded.json",
1077 [key, fingerprint, json],
1078 )?;
1079 Ok(())
1080 }
1081
1082 /// Delete the cached context entry for `key`, returning whether one existed.
1083 ///
1084 /// # Errors
1085 /// Returns [`StoreError::Sqlite`] on write failure.
1086 pub fn context_cache_delete(&self, key: &str) -> Result<bool, StoreError> {
1087 let n = self
1088 .conn
1089 .execute("DELETE FROM node_context WHERE key = ?1", [key])?;
1090 Ok(n > 0)
1091 }
1092
1093 /// Every key with a cached context entry, ordered. Used to prune entries for
1094 /// nodes that no longer exist.
1095 ///
1096 /// # Errors
1097 /// Returns [`StoreError::Sqlite`] on query failure.
1098 pub fn context_cache_keys(&self) -> Result<Vec<String>, StoreError> {
1099 let mut stmt = self
1100 .conn
1101 .prepare("SELECT key FROM node_context ORDER BY key")?;
1102 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
1103 let mut out = Vec::new();
1104 for row in rows {
1105 out.push(row?);
1106 }
1107 Ok(out)
1108 }
1109
1110 // --- Analyzer findings (ADR-0012). A separate artifact store: these methods
1111 // touch `analysis_runs`/`findings` only, never `nodes`/`edges`, so
1112 // `export_factset` — and the published `GraphArtifact` — stays a pure
1113 // function of the tree no matter what an analyzer reports. ---
1114
1115 /// The recorded answer to the OKF trust question for `peer`, if there is one
1116 /// (#706 phase 2).
1117 ///
1118 /// # Errors
1119 /// Returns [`StoreError::Sqlite`] on query failure.
1120 pub fn okf_consent(&self, peer: &str) -> Result<Option<crate::OkfConsent>, StoreError> {
1121 let row = self
1122 .conn
1123 .query_row(
1124 "SELECT peer, decision, root, screen_classes, decided_at
1125 FROM okf_consent WHERE peer = ?1",
1126 [peer],
1127 |r| {
1128 Ok((
1129 r.get::<_, String>(0)?,
1130 r.get::<_, String>(1)?,
1131 r.get::<_, String>(2)?,
1132 r.get::<_, String>(3)?,
1133 r.get::<_, String>(4)?,
1134 ))
1135 },
1136 )
1137 .optional()?;
1138 let Some((peer, decision, root, screen_classes, decided_at)) = row else {
1139 return Ok(None);
1140 };
1141 // A token the `CHECK` constraint permits but this build does not know is
1142 // a store written by a newer Roteiro. Reading it as "no answer" is the
1143 // safe direction: it asks again rather than acting on a decision whose
1144 // meaning this build cannot state.
1145 let Some(decision) = crate::OkfDecision::from_token(&decision) else {
1146 return Ok(None);
1147 };
1148 Ok(Some(crate::OkfConsent {
1149 peer,
1150 decision,
1151 root,
1152 screen_classes,
1153 decided_at,
1154 }))
1155 }
1156
1157 /// Whether the recorded answer for `peer` still covers the bundle now at
1158 /// `root` screening as `classes`, and why not when it does not.
1159 ///
1160 /// The two invalidators, and the one deliberately absent, are argued in
1161 /// [`crate::ConsentState::Lapsed`].
1162 ///
1163 /// # Errors
1164 /// Returns [`StoreError::Sqlite`] on query failure.
1165 pub fn okf_consent_holds(
1166 &self,
1167 peer: &str,
1168 root: &str,
1169 classes: &str,
1170 ) -> Result<crate::ConsentState, StoreError> {
1171 let Some(record) = self.okf_consent(peer)? else {
1172 return Ok(crate::ConsentState::Unasked);
1173 };
1174 if record.root != root {
1175 return Ok(crate::ConsentState::Moved { was: record.root });
1176 }
1177 if crate::screen_regressed(&record.screen_classes, classes) {
1178 return Ok(crate::ConsentState::Lapsed {
1179 was: record.screen_classes,
1180 now: classes.to_owned(),
1181 });
1182 }
1183 Ok(crate::ConsentState::Holds(record.decision))
1184 }
1185
1186 /// Record (or replace) the answer for one peer.
1187 ///
1188 /// Replacing rather than appending is deliberate: this is a *current
1189 /// decision*, not a history. ADR-0019's ledger appends because it records
1190 /// acts that happened; a consent record answers "what is it now", and two
1191 /// rows for one peer would need a tie-break nobody would get right.
1192 ///
1193 /// # Errors
1194 /// Returns [`StoreError::Sqlite`] on write failure.
1195 pub fn put_okf_consent(&self, record: &crate::OkfConsent) -> Result<(), StoreError> {
1196 self.conn.execute(
1197 "INSERT INTO okf_consent (peer, decision, root, screen_classes, decided_at)
1198 VALUES (?1, ?2, ?3, ?4, ?5)
1199 ON CONFLICT(peer) DO UPDATE SET
1200 decision = excluded.decision,
1201 root = excluded.root,
1202 screen_classes = excluded.screen_classes,
1203 decided_at = excluded.decided_at",
1204 [
1205 record.peer.as_str(),
1206 record.decision.as_str(),
1207 record.root.as_str(),
1208 record.screen_classes.as_str(),
1209 record.decided_at.as_str(),
1210 ],
1211 )?;
1212 Ok(())
1213 }
1214
1215 /// Every recorded OKF answer, in peer order — for `roteiro import --from okf`
1216 /// reporting and for tests.
1217 ///
1218 /// # Errors
1219 /// Returns [`StoreError::Sqlite`] on query failure.
1220 pub fn okf_consents(&self) -> Result<Vec<crate::OkfConsent>, StoreError> {
1221 let mut stmt = self.conn.prepare(
1222 "SELECT peer, decision, root, screen_classes, decided_at
1223 FROM okf_consent ORDER BY peer",
1224 )?;
1225 let rows = stmt.query_map([], |r| {
1226 Ok((
1227 r.get::<_, String>(0)?,
1228 r.get::<_, String>(1)?,
1229 r.get::<_, String>(2)?,
1230 r.get::<_, String>(3)?,
1231 r.get::<_, String>(4)?,
1232 ))
1233 })?;
1234 let mut out = Vec::new();
1235 for row in rows {
1236 let (peer, decision, root, screen_classes, decided_at) = row?;
1237 if let Some(decision) = crate::OkfDecision::from_token(&decision) {
1238 out.push(crate::OkfConsent {
1239 peer,
1240 decision,
1241 root,
1242 screen_classes,
1243 decided_at,
1244 });
1245 }
1246 }
1247 Ok(out)
1248 }
1249
1250 /// Replace the findings layer `run.layer` **wholesale**, atomically: the
1251 /// previous run for that layer and every finding row it owned are deleted,
1252 /// then this run and its findings are written. A finding that has since been
1253 /// fixed therefore *disappears* instead of lingering, and re-ingesting an
1254 /// unchanged report is idempotent — the store ends up with the same rows and
1255 /// no growth.
1256 ///
1257 /// The owned-record cleanup is explicit rather than inherited. The import
1258 /// path ([`Store::apply_import_layer`]) deletes a layer's edges but leaves its
1259 /// obsolete *nodes* behind; copying that shape here would silently orphan
1260 /// findings, so the previous run's rows are deleted by hand and counted in
1261 /// [`FindingsApplied::removed`]. The schema's `ON DELETE CASCADE` is kept as
1262 /// defence in depth, not as the mechanism.
1263 ///
1264 /// `findings` must carry distinct [`FindingKey`](crate::FindingKey)s: a
1265 /// duplicate identity is a producer bug and is rejected by the unique index
1266 /// *inside* the transaction, so nothing is committed. Callers that parse
1267 /// untrusted reports should reject duplicates earlier, with a better message.
1268 ///
1269 /// # Errors
1270 /// Returns [`StoreError::Json`] if the run's command policy or a finding's
1271 /// `meta` cannot be serialized, or [`StoreError::Sqlite`] on write failure. On
1272 /// any error the transaction is rolled back and the previous layer survives
1273 /// intact.
1274 pub fn replace_findings_layer(
1275 &mut self,
1276 run: &AnalysisRun,
1277 findings: &[Finding],
1278 ) -> Result<FindingsApplied, StoreError> {
1279 let tx = self.conn.transaction()?;
1280 let applied = findings::replace_layer(&tx, run, findings)?;
1281 tx.commit()?;
1282 Ok(applied)
1283 }
1284
1285 /// Delete a findings layer and every finding row it owns, returning how many
1286 /// findings went with it, or `None` if the layer was not live.
1287 ///
1288 /// # Errors
1289 /// Returns [`StoreError::Sqlite`] on write failure; nothing is committed on
1290 /// error.
1291 pub fn delete_findings_layer(&mut self, layer: &str) -> Result<Option<usize>, StoreError> {
1292 let tx = self.conn.transaction()?;
1293 let removed = findings::delete_layer(&tx, layer)?;
1294 tx.commit()?;
1295 Ok(removed)
1296 }
1297
1298 /// Every live findings layer with its findings, ordered by layer key and, in
1299 /// each layer, by finding key. Pass `analyzer` to narrow to one analyzer.
1300 ///
1301 /// # Errors
1302 /// Returns [`StoreError::Sqlite`] on query failure, [`StoreError::Json`] if a
1303 /// stored policy or `meta` cannot be decoded, or [`StoreError::Corrupt`] on an
1304 /// unrecognised stored token.
1305 pub fn findings_layers(
1306 &self,
1307 analyzer: Option<&str>,
1308 ) -> Result<Vec<FindingsLayer>, StoreError> {
1309 findings::layers(&self.conn, analyzer)
1310 }
1311
1312 /// Number of findings currently stored, across every layer.
1313 ///
1314 /// # Errors
1315 /// Returns [`StoreError::Sqlite`] on query failure.
1316 pub fn finding_count(&self) -> Result<u64, StoreError> {
1317 findings::count_findings(&self.conn)
1318 }
1319
1320 /// Number of live analysis runs — one per findings layer.
1321 ///
1322 /// # Errors
1323 /// Returns [`StoreError::Sqlite`] on query failure.
1324 pub fn analysis_run_count(&self) -> Result<u64, StoreError> {
1325 findings::count_runs(&self.conn)
1326 }
1327
1328 // --- Generated media content (ADR-0015). A separate artifact store, on the
1329 // same terms as findings above: these methods touch `media_content` only,
1330 // never `nodes`/`edges`, so `export_factset` stays a pure function of the
1331 // tree across a `media build`, and generated text can never reach the
1332 // `authored` relevance boost in `search`. ---
1333
1334 /// Write one generated-content record, returning `true` if a row was written.
1335 ///
1336 /// Keyed by `(blob_id, producer)`. A record for that exact pair already
1337 /// present is **left alone** and `false` is returned — which is what makes
1338 /// `media build` incremental and a second run free. A *different* producer is
1339 /// a new row beside it, never an overwrite: that is the point of keying on the
1340 /// producer identity, so a better model's description can be compared with the
1341 /// one it replaces and a distrusted producer can be dropped wholesale.
1342 /// [`MediaWrite::replace`] (only `media build --force`) is the one path that
1343 /// overwrites, and only for the identical producer.
1344 ///
1345 /// # Errors
1346 /// Returns [`StoreError::Sqlite`] on a write failure; the transaction is
1347 /// rolled back.
1348 pub fn record_media_content(&mut self, write: &MediaWrite<'_>) -> Result<bool, StoreError> {
1349 let tx = self.conn.transaction()?;
1350 let written = media::record(&tx, write)?;
1351 tx.commit()?;
1352 Ok(written)
1353 }
1354
1355 /// Whether a record already exists for exactly this `(blob, producer)`.
1356 ///
1357 /// # Errors
1358 /// Returns [`StoreError::Sqlite`] on query failure.
1359 pub fn has_media_record(&self, blob_id: &str, producer: &str) -> Result<bool, StoreError> {
1360 media::exists(&self.conn, blob_id, producer)
1361 }
1362
1363 /// Stored records matching `filter`, ordered by `(producer, blob id)`.
1364 ///
1365 /// # Errors
1366 /// Returns [`StoreError::Sqlite`] on query failure, or
1367 /// [`StoreError::Corrupt`] if a row carries an unknown modality token.
1368 pub fn media_records(&self, filter: &MediaFilter<'_>) -> Result<Vec<MediaRecord>, StoreError> {
1369 media::records(&self.conn, filter)
1370 }
1371
1372 /// Discard records — all of them, or only those written by `producer`.
1373 /// Returns how many rows went.
1374 ///
1375 /// Nothing in the graph is touched: dropping a model you no longer trust must
1376 /// not cost you a re-sync.
1377 ///
1378 /// # Errors
1379 /// Returns [`StoreError::Sqlite`] on a write failure.
1380 pub fn clear_media_content(&mut self, producer: Option<&str>) -> Result<usize, StoreError> {
1381 let tx = self.conn.transaction()?;
1382 let removed = media::delete(&tx, producer)?;
1383 tx.commit()?;
1384 Ok(removed)
1385 }
1386
1387 /// Number of generated-content records currently stored.
1388 ///
1389 /// # Errors
1390 /// Returns [`StoreError::Sqlite`] on query failure.
1391 pub fn media_content_count(&self) -> Result<u64, StoreError> {
1392 media::count(&self.conn)
1393 }
1394
1395 /// One summary per producer that owns records, ordered by producer id.
1396 ///
1397 /// # Errors
1398 /// Returns [`StoreError::Sqlite`] on query failure, or
1399 /// [`StoreError::Corrupt`] if a row carries an unknown modality token.
1400 pub fn media_producer_summaries(&self) -> Result<Vec<ProducerSummary>, StoreError> {
1401 media::producer_summaries(&self.conn)
1402 }
1403
1404 /// The blob ids that have at least one record carrying **generated text**
1405 /// for `kind`. A blob the pre-generation gate refused is not described, so it
1406 /// is not here — see [`Store::gated_media_blobs`].
1407 ///
1408 /// # Errors
1409 /// Returns [`StoreError::Sqlite`] on query failure.
1410 pub fn described_media_blobs(
1411 &self,
1412 kind: MediaKind,
1413 ) -> Result<std::collections::BTreeSet<String>, StoreError> {
1414 media::described_blobs(&self.conn, kind)
1415 }
1416
1417 /// The blob ids the [pre-generation gate](crate::media::gate) refused for
1418 /// `kind` — silent clips, blank images — each of which has a record naming
1419 /// the value that refused it.
1420 ///
1421 /// # Errors
1422 /// Returns [`StoreError::Sqlite`] on query failure.
1423 pub fn gated_media_blobs(
1424 &self,
1425 kind: MediaKind,
1426 ) -> Result<std::collections::BTreeSet<String>, StoreError> {
1427 media::gated_blobs(&self.conn, kind)
1428 }
1429
1430 /// Records whose blob is no longer anywhere in `present`. Exposed so a caller
1431 /// can see what a tree change has orphaned; nothing deletes them implicitly,
1432 /// because a record is expensive to reproduce and a blob can return.
1433 ///
1434 /// # Errors
1435 /// Returns [`StoreError::Sqlite`] on query failure.
1436 pub fn orphan_media_records(
1437 &self,
1438 present: &std::collections::BTreeSet<String>,
1439 ) -> Result<Vec<MediaRecord>, StoreError> {
1440 Ok(self
1441 .media_records(&MediaFilter::default())?
1442 .into_iter()
1443 .filter(|r| !present.contains(&r.blob_id))
1444 .collect())
1445 }
1446
1447 // --- Episodic agent memory (ADR-0013). A separate artifact store, on the
1448 // same terms as findings and media above: these methods write `agent_memory`
1449 // and nothing else — they *read* `nodes` to capture and check an anchor, and
1450 // never write one. So `export_factset` stays a pure function of the tree
1451 // across every memory write, and nothing an agent remembers can reach the
1452 // `authored` relevance boost in `search`.
1453 //
1454 // These rows are not touched by `rebuild`, following the `imports` precedent:
1455 // what has no generating function must not be destroyed by a re-derivation.
1456 // ---
1457
1458 /// Record one memory, returning its id — the monotonic generation it was
1459 /// written at.
1460 ///
1461 /// The anchor's blob and path, and the `sync_state` tree witness, are
1462 /// captured **here, from the graph**, so a caller cannot record evidence the
1463 /// node never carried. An anchor key naming no node is accepted and reads
1464 /// back as [`crate::AnchorState::Vanished`].
1465 ///
1466 /// [`MemoryWrite::supersedes`] records the supersession explicitly, in the
1467 /// same transaction as the successor: either both land or neither does.
1468 ///
1469 /// # Errors
1470 /// Returns [`MemoryError::InvalidScope`] / [`MemoryError::InvalidBody`] /
1471 /// [`MemoryError::InvalidConfidence`] for a record that could not be recalled,
1472 /// [`MemoryError::NotFound`] or [`MemoryError::AlreadySuperseded`] for a bad
1473 /// supersession target, or [`MemoryError::Store`] on a write failure — in
1474 /// every case with nothing committed.
1475 pub fn record_memory(&mut self, write: &MemoryWrite<'_>) -> Result<i64, MemoryError> {
1476 let tx = self.conn.transaction().map_err(StoreError::from)?;
1477 let id = memory::record(&tx, write)?;
1478 tx.commit().map_err(StoreError::from)?;
1479 Ok(id)
1480 }
1481
1482 /// Memory records matching `filter`, newest generation first.
1483 ///
1484 /// Live records only unless [`MemoryFilter::include_superseded`] is set: a
1485 /// superseded record drops out immediately and regardless of age, because the
1486 /// test is a recorded pointer and not a clock.
1487 ///
1488 /// Each record's [`crate::AnchorState`] is computed against the **current**
1489 /// graph on every call and is never stored.
1490 ///
1491 /// # Errors
1492 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1493 /// if a row carries an unknown kind token.
1494 pub fn memory_records(
1495 &self,
1496 filter: &MemoryFilter<'_>,
1497 ) -> Result<Vec<MemoryRecord>, StoreError> {
1498 memory::records(&self.conn, filter)
1499 }
1500
1501 /// One memory record by id, or `None` if there is no such record.
1502 ///
1503 /// # Errors
1504 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1505 /// if the row carries an unknown kind token.
1506 pub fn memory_record(&self, id: i64) -> Result<Option<MemoryRecord>, StoreError> {
1507 memory::get(&self.conn, id)
1508 }
1509
1510 /// A [`MemoryListing`]: the records matching `filter`, plus the whole store's
1511 /// live and superseded counts, so an empty result is legible as *nothing
1512 /// matched* rather than *nothing is stored*.
1513 ///
1514 /// # Errors
1515 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1516 /// if a row carries an unknown kind token.
1517 pub fn memory_listing(&self, filter: &MemoryFilter<'_>) -> Result<MemoryListing, StoreError> {
1518 let (live, superseded) = memory::counts(&self.conn)?;
1519 Ok(MemoryListing {
1520 schema: memory::MEMORY_SCHEMA,
1521 records: self.memory_records(filter)?,
1522 live,
1523 superseded,
1524 })
1525 }
1526
1527 /// **The only way a memory record is ever removed.** Deletes it, and returns
1528 /// `None` if there was no such record.
1529 ///
1530 /// Episodic memory is unbounded and never auto-evicted — no sweep, no TTL, no
1531 /// capacity bound reaches this table — so an explicit call is the whole
1532 /// reclamation story. It is also the privacy story: memory has no redaction
1533 /// chokepoint, so a record that captured a token or a customer name is
1534 /// removed by asking.
1535 ///
1536 /// Anything the deleted record had superseded becomes **live again** and is
1537 /// named in [`MemoryForgotten::restored`]: leaving it superseded would hide it
1538 /// on the authority of a record that no longer exists.
1539 ///
1540 /// # Errors
1541 /// Returns [`StoreError::Sqlite`] on a write failure; the transaction is
1542 /// rolled back.
1543 pub fn forget_memory(&mut self, id: i64) -> Result<Option<MemoryForgotten>, StoreError> {
1544 let tx = self.conn.transaction()?;
1545 let forgotten = memory::forget(&tx, id)?;
1546 tx.commit()?;
1547 Ok(forgotten)
1548 }
1549
1550 /// How many memory records are stored, as `(live, superseded)`.
1551 ///
1552 /// # Errors
1553 /// Returns [`StoreError::Sqlite`] on query failure.
1554 pub fn memory_counts(&self) -> Result<(u64, u64), StoreError> {
1555 memory::counts(&self.conn)
1556 }
1557
1558 /// **Ranked recall**: the live records that match `opts`, scored
1559 /// `base_confidence × anchor_penalty × decay(age)` and ordered best first.
1560 ///
1561 /// Every term is computed **here, at retrieval time, and written to no
1562 /// column** (ADR-0013). A stored score that decayed would rewrite the store on
1563 /// every read and would be wrong in between, making recall depend on when you
1564 /// last looked. Three consequences follow, and each is a promise:
1565 ///
1566 /// - **This call mutates nothing.** Recall over an unchanged store and an
1567 /// unchanged tree is idempotent, which is what makes
1568 /// [`crate::Decay::None`] byte-identical across runs.
1569 /// - **A superseded record is never returned**, immediately and regardless of
1570 /// age: the test is a recorded pointer, not a clock.
1571 /// - **A record whose anchor no longer resolves is still returned**, demoted
1572 /// and labelled. Drift ranks it down; nothing deletes it.
1573 ///
1574 /// # Errors
1575 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1576 /// if a row carries an unknown kind token.
1577 pub fn recall_memory(&self, opts: &RecallOptions<'_>) -> Result<Recall, StoreError> {
1578 let (live, superseded) = memory::counts(&self.conn)?;
1579 Ok(Recall {
1580 schema: memory::RECALL_SCHEMA,
1581 generation: memory::generation(&self.conn)?,
1582 decay: opts.decay,
1583 reproducible: opts.decay.is_reproducible(),
1584 results: memory::recall(&self.conn, opts)?,
1585 live,
1586 superseded,
1587 })
1588 }
1589
1590 // --- The bounded cache tier (ADR-0013, Tier 2). The *opposite* rules to the
1591 // episodic tier above, because it holds the opposite kind of knowledge:
1592 // everything here is re-derivable, so eviction costs cycles and never
1593 // information. Nothing in this section can reach `agent_memory` — it has no
1594 // `bytes`, no `last_used` and no `hits` for a capacity policy to grip. ---
1595
1596 /// Write (or replace) one cache entry.
1597 ///
1598 /// The payload size is computed here and the anchor's blob is captured from
1599 /// the graph here, so the sweep can order and total the tier without reading
1600 /// it, and a caller cannot record evidence a node never carried.
1601 ///
1602 /// # Errors
1603 /// Returns [`StoreError::Sqlite`] on write failure.
1604 pub fn agent_cache_put(&self, write: &CacheWrite<'_>) -> Result<(), StoreError> {
1605 memory::cache_put(&self.conn, write)
1606 }
1607
1608 /// Read one cache entry back, **recording the access** — `hits` increments and
1609 /// `last_used` advances.
1610 ///
1611 /// This is the one read in the memory store that writes, and what it writes is
1612 /// the cache's own bookkeeping: those two columns exist to be moved by exactly
1613 /// this, and a hit counter nothing increments is a column that lies. It
1614 /// touches nothing outside `agent_cache`, so [`Store::recall_memory`] — the
1615 /// read whose reproducibility is promised — stays free of it.
1616 ///
1617 /// # Errors
1618 /// Returns [`StoreError::Sqlite`] on query failure.
1619 pub fn agent_cache_get(&self, key: &str) -> Result<Option<CacheEntry>, StoreError> {
1620 memory::cache_get(&self.conn, key)
1621 }
1622
1623 /// Every cache entry, ordered by key, **without** recording an access:
1624 /// inspecting a cache is not using it.
1625 ///
1626 /// # Errors
1627 /// Returns [`StoreError::Sqlite`] on query failure.
1628 pub fn agent_cache_entries(&self) -> Result<Vec<CacheEntry>, StoreError> {
1629 memory::cache_entries(&self.conn)
1630 }
1631
1632 /// Delete one cache entry, returning whether there was one.
1633 ///
1634 /// # Errors
1635 /// Returns [`StoreError::Sqlite`] on write failure.
1636 pub fn agent_cache_forget(&self, key: &str) -> Result<bool, StoreError> {
1637 memory::cache_forget(&self.conn, key)
1638 }
1639
1640 /// What the cache tier holds, against `budget_bytes`.
1641 ///
1642 /// # Errors
1643 /// Returns [`StoreError::Sqlite`] on query failure.
1644 pub fn agent_cache_stats(&self, budget_bytes: u64) -> Result<CacheStats, StoreError> {
1645 memory::cache_stats(&self.conn, budget_bytes)
1646 }
1647
1648 /// **Sweep the cache tier down to `budget_bytes`**, evicting oldest-first on
1649 /// `(anchor_valid ASC, last_used ASC)`, and advance the generation.
1650 ///
1651 /// Called at the maintenance seam (beside `refresh_contexts`) and **never on
1652 /// the read path**, so an ordinary query never mutates the store. Three things
1653 /// are never evicted: anything episodic — structurally, there is no column to
1654 /// grip it by; an entry written in the current generation whose anchor still
1655 /// applies, which is the session's own work; and the most-recently-used entry,
1656 /// always, even if it alone exceeds the budget.
1657 ///
1658 /// That last pair means a sweep can legitimately finish still over budget.
1659 /// [`CacheSweep::over_budget`] reports it rather than leaving a bound that
1660 /// silently failed to bind.
1661 ///
1662 /// # Errors
1663 /// Returns [`StoreError::Sqlite`] on failure; the transaction is rolled back.
1664 pub fn sweep_agent_cache(&mut self, budget_bytes: u64) -> Result<CacheSweep, StoreError> {
1665 let tx = self.conn.transaction()?;
1666 let swept = memory::cache_sweep(&tx, budget_bytes)?;
1667 tx.commit()?;
1668 Ok(swept)
1669 }
1670
1671 /// Findings whose owning run no longer exists. Always `0` in a healthy store;
1672 /// exposed so layer replacement can be asserted to clean up its own records
1673 /// rather than orphaning them.
1674 ///
1675 /// # Errors
1676 /// Returns [`StoreError::Sqlite`] on query failure.
1677 pub fn orphan_finding_count(&self) -> Result<u64, StoreError> {
1678 findings::count_orphan_findings(&self.conn)
1679 }
1680}
1681
1682// --- Free helpers operating on a `Connection` (a `Transaction` derefs to one) ---
1683
1684fn node_row_id(conn: &Connection, key: &str) -> rusqlite::Result<Option<i64>> {
1685 conn.query_row("SELECT id FROM nodes WHERE key = ?1", [key], |r| r.get(0))
1686 .optional()
1687}
1688
1689/// Record (or clear) the last-synced `HEAD` tree id. Shared by `rebuild` and
1690/// `reconcile` so both leave identical `sync_state`.
1691fn write_sync_state(conn: &Connection, tree: Option<&str>) -> Result<(), StoreError> {
1692 match tree {
1693 // Clear `env` on every tree write: it is only valid for the tree a
1694 // committed [`crate::sync`] set it against, and that sync re-records it (via
1695 // `set_sync_env`) immediately after. So a worktree/index sync, or any
1696 // path that does not re-set it, leaves `env` NULL — reading as "unknown"
1697 // and forcing the safe full re-extraction next time.
1698 Some(tree) => conn.execute(
1699 "INSERT INTO sync_state (id, tree) VALUES (0, ?1)
1700 ON CONFLICT(id) DO UPDATE SET tree = excluded.tree, env = NULL",
1701 [tree],
1702 )?,
1703 None => conn.execute("DELETE FROM sync_state WHERE id = 0", [])?,
1704 };
1705 Ok(())
1706}
1707
1708fn upsert_node(conn: &Connection, node: &Node) -> Result<(), StoreError> {
1709 let meta = serde_json::to_string(&node.meta)?;
1710 let (span_start, span_end) = match node.span {
1711 Some(s) => (Some(i64::from(s.start)), Some(i64::from(s.end))),
1712 None => (None, None),
1713 };
1714 conn.execute(
1715 "INSERT INTO nodes (key, kind, name, path, lang, blob_hash, span_start, span_end, provenance, meta)
1716 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
1717 ON CONFLICT(key) DO UPDATE SET
1718 kind = excluded.kind, name = excluded.name, path = excluded.path,
1719 lang = excluded.lang, blob_hash = excluded.blob_hash,
1720 span_start = excluded.span_start, span_end = excluded.span_end,
1721 provenance = excluded.provenance, meta = excluded.meta",
1722 params![
1723 node.key,
1724 node.kind.as_str(),
1725 node.name,
1726 node.path,
1727 node.lang,
1728 node.blob_hash,
1729 span_start,
1730 span_end,
1731 node.provenance.as_str(),
1732 meta,
1733 ],
1734 )?;
1735 Ok(())
1736}
1737
1738fn insert_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
1739 validate_edge(edge)?;
1740 let src_id =
1741 node_row_id(conn, &edge.src)?.ok_or_else(|| StoreError::UnknownNode(edge.src.clone()))?;
1742 let dst_id =
1743 node_row_id(conn, &edge.dst)?.ok_or_else(|| StoreError::UnknownNode(edge.dst.clone()))?;
1744 insert_edge_row(conn, edge, src_id, dst_id)
1745}
1746
1747/// Apply `edge` only if both endpoints already resolve to nodes, returning
1748/// whether it was **applied** (both endpoints resolved; a duplicate of an
1749/// existing edge is a harmless no-op via `ON CONFLICT DO NOTHING` but still
1750/// reports `true`). A missing endpoint returns `false` rather than erroring —
1751/// the caller prunes such dangling cross-references from the import layer.
1752fn insert_edge_if_present(conn: &Connection, edge: &Edge) -> Result<bool, StoreError> {
1753 validate_edge(edge)?;
1754 let (Some(src_id), Some(dst_id)) =
1755 (node_row_id(conn, &edge.src)?, node_row_id(conn, &edge.dst)?)
1756 else {
1757 return Ok(false);
1758 };
1759 insert_edge_row(conn, edge, src_id, dst_id)?;
1760 Ok(true)
1761}
1762
1763/// Apply `edges`, keeping those whose endpoints resolve and pruning the rest.
1764/// Returns the kept edges plus the applied/pruned counts (in an [`ImportApplied`]
1765/// whose `layers`/`nodes` are left zero for the caller to fill).
1766fn apply_edges_pruning(
1767 conn: &Connection,
1768 edges: &[Edge],
1769) -> Result<(Vec<Edge>, ImportApplied), StoreError> {
1770 let mut kept = Vec::with_capacity(edges.len());
1771 let mut counts = ImportApplied::default();
1772 for edge in edges {
1773 if insert_edge_if_present(conn, edge)? {
1774 kept.push(edge.clone());
1775 counts.edges_applied += 1;
1776 } else {
1777 counts.edges_pruned += 1;
1778 }
1779 }
1780 Ok((kept, counts))
1781}
1782
1783/// Delete the nodes `src_ref` has withdrawn since its last apply, returning how
1784/// many went.
1785///
1786/// `prior` is the key set the persisted layer held before this apply; `current`
1787/// is what it holds now. The difference is what the source stopped asserting —
1788/// and the two guards below are what keep "stopped asserting" from becoming
1789/// "deleted somebody else's node".
1790///
1791/// Both guards read the **live store**, not the key's namespace. A namespace rule
1792/// would have to be re-derived for every layer that ever exists, and would be
1793/// wrong the first time two producers shared a prefix; a query is right by
1794/// construction for a layer this function has never heard of.
1795fn remove_withdrawn_nodes(
1796 conn: &Connection,
1797 src_ref: &str,
1798 prior: &BTreeSet<String>,
1799 current: &[Node],
1800) -> Result<usize, StoreError> {
1801 let held: BTreeSet<&str> = current.iter().map(|n| n.key.as_str()).collect();
1802 let withdrawn: Vec<&str> = prior
1803 .iter()
1804 .map(String::as_str)
1805 .filter(|k| !held.contains(k))
1806 .collect();
1807 if withdrawn.is_empty() {
1808 return Ok(0);
1809 }
1810
1811 // Guard 1: every other persisted layer's node keys. Two layers naming one
1812 // node is a supported arrangement, not an anomaly — `import:links` and
1813 // `import:links/authored` share every `extref:` placeholder between them.
1814 let mut claimed: BTreeSet<String> = BTreeSet::new();
1815 {
1816 let mut stmt = conn.prepare("SELECT facts FROM imports WHERE src_ref <> ?1")?;
1817 let rows = stmt.query_map([src_ref], |r| r.get::<_, String>(0))?;
1818 for row in rows {
1819 let facts: FactSet = serde_json::from_str(&row?)?;
1820 claimed.extend(facts.nodes.into_iter().map(|n| n.key));
1821 }
1822 }
1823
1824 let mut removed = 0usize;
1825 for key in withdrawn {
1826 if claimed.contains(key) {
1827 continue;
1828 }
1829 // Guard 2: anything still attached. A surviving edge belongs to some
1830 // other producer, and removing the node would take that relationship
1831 // with it — a deletion nobody asked for, disguised as a cleanup. It
1832 // would also violate the foreign key, so this is correctness twice over.
1833 //
1834 // `EXISTS`, not `COUNT(*) > 0`. The question is "is there one", and a
1835 // count says "tally them all, then ask whether the tally was zero" —
1836 // which is both a different sentence and more work: with the `OR dst`
1837 // arm, a well-connected node in this repository would scan on the order
1838 // of 1,600 edge rows where the probe stops at the first hit.
1839 let attached: bool = conn.query_row(
1840 "SELECT EXISTS (SELECT 1 FROM edges \
1841 WHERE src = (SELECT id FROM nodes WHERE key = ?1) \
1842 OR dst = (SELECT id FROM nodes WHERE key = ?1))",
1843 [key],
1844 |r| r.get(0),
1845 )?;
1846 if attached {
1847 continue;
1848 }
1849 let gone = conn.execute("DELETE FROM nodes WHERE key = ?1", [key])?;
1850 if gone > 0 {
1851 // A cached context bundle for a node that no longer exists can only
1852 // ever be served as a stale answer, so it goes with it.
1853 conn.execute("DELETE FROM node_context WHERE key = ?1", [key])?;
1854 removed += 1;
1855 }
1856 }
1857 Ok(removed)
1858}
1859
1860/// Upsert a persisted import layer row. Free helper so it can run inside the same
1861/// transaction as an apply/prune pass.
1862fn put_import_row(conn: &Connection, src_ref: &str, facts: &FactSet) -> Result<(), StoreError> {
1863 let json = serde_json::to_string(facts)?;
1864 conn.execute(
1865 "INSERT INTO imports (src_ref, facts) VALUES (?1, ?2)
1866 ON CONFLICT(src_ref) DO UPDATE SET facts = excluded.facts, imported_at = datetime('now')",
1867 params![src_ref, json],
1868 )?;
1869 Ok(())
1870}
1871
1872/// The provenance/confidence invariant guard shared by the strict and tolerant
1873/// edge inserts.
1874fn validate_edge(edge: &Edge) -> Result<(), StoreError> {
1875 if edge.is_valid() {
1876 Ok(())
1877 } else {
1878 Err(StoreError::InvalidEdge(format!(
1879 "confidence must be present iff provenance is inferred (src={}, dst={})",
1880 edge.src, edge.dst
1881 )))
1882 }
1883}
1884
1885/// Insert an edge row given already-resolved endpoint ids. Edges are a set: a
1886/// duplicate `(src, dst, kind, provenance)` is a no-op via `ON CONFLICT … DO
1887/// NOTHING`, so re-applying a fact set never accumulates duplicates. Other
1888/// constraint violations (guarded in Rust above) still surface.
1889fn insert_edge_row(
1890 conn: &Connection,
1891 edge: &Edge,
1892 src_id: i64,
1893 dst_id: i64,
1894) -> Result<(), StoreError> {
1895 conn.execute(
1896 "INSERT INTO edges (src, dst, kind, provenance, confidence, src_ref)
1897 VALUES (?1, ?2, ?3, ?4, ?5, ?6)
1898 ON CONFLICT(src, dst, kind, provenance) DO NOTHING",
1899 params![
1900 src_id,
1901 dst_id,
1902 edge.kind.as_str(),
1903 edge.provenance.as_str(),
1904 edge.confidence,
1905 edge.src_ref,
1906 ],
1907 )?;
1908 Ok(())
1909}
1910
1911/// A hashable identity for an edge over **all** its fields — used by
1912/// [`Store::reconcile`] to diff the edge set. A tuple (not a delimiter-joined
1913/// string) so no field value can be confused with a separator: node keys embed
1914/// git paths, which may legally contain any byte (including control characters),
1915/// so a joined string could collapse distinct edges to one identity and drop an
1916/// edge. Confidence is compared by its exact bit pattern (`f64::to_bits`, wrapped
1917/// in `Option` so `None` and `Some(_)` stay distinct), the only non-`Eq` field.
1918fn edge_identity(edge: &Edge) -> EdgeId {
1919 (
1920 edge.src.clone(),
1921 edge.dst.clone(),
1922 edge.kind.as_str().to_owned(),
1923 edge.provenance.as_str().to_owned(),
1924 edge.confidence.map(f64::to_bits),
1925 edge.src_ref.clone(),
1926 )
1927}
1928
1929/// The tuple form of an edge's full-field identity (see [`edge_identity`]):
1930/// `(src, dst, kind, provenance, confidence-bits, src_ref)`.
1931type EdgeId = (String, String, String, String, Option<u64>, Option<String>);
1932
1933/// Delete the edge row identified by `(src, dst, kind, provenance)` — the table's
1934/// unique key — resolving the endpoint node keys to ids. A no-op if absent.
1935fn delete_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
1936 conn.execute(
1937 "DELETE FROM edges
1938 WHERE src = (SELECT id FROM nodes WHERE key = ?1)
1939 AND dst = (SELECT id FROM nodes WHERE key = ?2)
1940 AND kind = ?3 AND provenance = ?4",
1941 params![
1942 edge.src,
1943 edge.dst,
1944 edge.kind.as_str(),
1945 edge.provenance.as_str()
1946 ],
1947 )?;
1948 Ok(())
1949}
1950
1951fn collect_nodes(rows: &mut rusqlite::Rows) -> Result<Vec<Node>, StoreError> {
1952 let mut out = Vec::new();
1953 while let Some(row) = rows.next()? {
1954 out.push(row_to_node(row)?);
1955 }
1956 Ok(out)
1957}
1958
1959fn collect_edges(rows: &mut rusqlite::Rows) -> Result<Vec<Edge>, StoreError> {
1960 let mut out = Vec::new();
1961 while let Some(row) = rows.next()? {
1962 out.push(row_to_edge(row)?);
1963 }
1964 Ok(out)
1965}
1966
1967fn row_to_node(row: &rusqlite::Row) -> Result<Node, StoreError> {
1968 let kind: String = row.get("kind")?;
1969 let span_start: Option<i64> = row.get("span_start")?;
1970 let span_end: Option<i64> = row.get("span_end")?;
1971 let span = match (span_start, span_end) {
1972 (Some(s), Some(e)) => Some(Span::new(to_u32(s)?, to_u32(e)?)),
1973 _ => None,
1974 };
1975 let meta: String = row.get("meta")?;
1976 let provenance: String = row.get("provenance")?;
1977 let provenance = Provenance::from_token(&provenance)
1978 .ok_or_else(|| StoreError::Corrupt(unknown_provenance("node", &provenance)))?;
1979 Ok(Node {
1980 key: row.get("key")?,
1981 kind: NodeKind::from_token(&kind),
1982 name: row.get("name")?,
1983 path: row.get("path")?,
1984 lang: row.get("lang")?,
1985 blob_hash: row.get("blob_hash")?,
1986 span,
1987 provenance,
1988 meta: serde_json::from_str(&meta)?,
1989 })
1990}
1991
1992fn row_to_edge(row: &rusqlite::Row) -> Result<Edge, StoreError> {
1993 let kind: String = row.get("kind")?;
1994 let provenance: String = row.get("provenance")?;
1995 let provenance = Provenance::from_token(&provenance)
1996 .ok_or_else(|| StoreError::Corrupt(unknown_provenance("edge", &provenance)))?;
1997 Ok(Edge {
1998 src: row.get("src")?,
1999 dst: row.get("dst")?,
2000 kind: EdgeKind::from_token(&kind),
2001 provenance,
2002 confidence: row.get("confidence")?,
2003 src_ref: row.get("src_ref")?,
2004 })
2005}
2006
2007fn to_u32(v: i64) -> Result<u32, StoreError> {
2008 u32::try_from(v).map_err(|_| StoreError::Corrupt(format!("span offset out of range: {v}")))
2009}
2010
2011/// The message for a stored provenance token this build cannot parse.
2012///
2013/// It names **both** causes, because until migration 14 there was only one and
2014/// the old wording said so: "e.g. a corrupt database row". That migration widened
2015/// the vocabulary, so an unrecognised token now most often means the store was
2016/// written by a newer Roteiro — [`Store::schema_ahead`] says so first and more
2017/// precisely, but a reader who reaches this message deserves not to be sent
2018/// looking for corruption that is not there.
2019fn unknown_provenance(what: &str, token: &str) -> String {
2020 format!(
2021 "unknown {what} provenance `{token}`: expected one of {}. Either this row is corrupt, \
2022 or this store was written by a newer Roteiro that knows a provenance this build does \
2023 not — check `roteiro --version` against the build that last wrote the graph.",
2024 Provenance::tokens().join(", ")
2025 )
2026}
2027
2028/// The true provenance of an import-layer node, from its key namespace: Graphify
2029/// nodes (`graphify:`) are [`Provenance::Inferred`]; an OKF concept (`okf:`) is
2030/// [`Provenance::ExternalInferred`]; every other import node (lat, …) is
2031/// [`Provenance::Authored`]. Import-layer nodes are never derived, so this is
2032/// used to repair a legacy `Derived` tag on load (see `load_import_layers`).
2033fn import_node_provenance(key: &str) -> Provenance {
2034 if key.starts_with("graphify:") {
2035 Provenance::Inferred
2036 } else if key.starts_with("okf:") {
2037 // An OKF concept is a peer's, and this is a *repair* path with no
2038 // frontmatter left to consult — so it takes the lowest claim the peer
2039 // could have made. Falling through to `Authored` below would assert that
2040 // this graph human-authored another repository's concept, which is the
2041 // exact laundering the `external-*` tier exists to refuse, and it would
2042 // arrive through an error-recovery path where nobody would look for it.
2043 //
2044 // A **filled placeholder** deliberately does not match: its key is
2045 // `extref:`, it is a node this repo created, and its own provenance is
2046 // whichever layer wrote it last (see `rto_graph::links`).
2047 Provenance::ExternalInferred
2048 } else {
2049 Provenance::Authored
2050 }
2051}
2052
2053#[cfg(test)]
2054mod tests {
2055 use super::Store;
2056 use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
2057 use crate::provenance::Provenance;
2058
2059 /// A fresh directory for a file-backed store, cleared on the way out of this
2060 /// function rather than trusted to be absent.
2061 ///
2062 /// Keyed by process id **and** a monotonic counter, matching
2063 /// `rto-render`'s `okf_inspect` tests and the CLI tests' `bundle` helper —
2064 /// whose own note explains why, and which I should have followed first time:
2065 /// *"uniqueness must not depend on everyone remembering to pick a distinct
2066 /// name."*
2067 ///
2068 /// The pre-clean is the half that matters most here, and is what review
2069 /// caught. A store is not one file: a failed run leaves `graph.db` beside its
2070 /// `-wal` and `-shm`, and reusing that state would have the concurrency test
2071 /// counting rows a previous run inserted — failing, eventually, for a reason
2072 /// that has nothing to do with locking. A test that can fail for the wrong
2073 /// reason is worse than no test, because the next person debugs the wrong
2074 /// thing.
2075 fn scratch(tag: &str) -> std::path::PathBuf {
2076 static SEQ: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
2077 let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2078 let root =
2079 std::env::temp_dir().join(format!("rto-store-{}-{seq}-{tag}", std::process::id()));
2080 let _ = std::fs::remove_dir_all(&root);
2081 root
2082 }
2083
2084 fn sample_node(key: &str) -> Node {
2085 Node {
2086 key: key.to_owned(),
2087 kind: NodeKind::Fn,
2088 name: "sample".to_owned(),
2089 path: Some("src/lib.rs".to_owned()),
2090 lang: Some("rust".to_owned()),
2091 blob_hash: Some("deadbeef".to_owned()),
2092 span: Some(Span::new(10, 42)),
2093 provenance: Provenance::Derived,
2094 meta: serde_json::json!({"vis": "pub"}),
2095 }
2096 }
2097
2098 /// Two Roteiro processes over one repository is ordinary rather than exotic —
2099 /// an editor's MCP client and a terminal, a long-lived `serve` and a [`crate::sync`],
2100 /// or several clients each spawning their own stdio server. Both writing must
2101 /// **queue**, not kill one of them.
2102 ///
2103 /// Reproduces the shape every upsert in this file has: *read, then write,
2104 /// inside one transaction.* That is precisely the case the five-second
2105 /// `busy_timeout` never covered — SQLite refuses a lock upgrade outright
2106 /// rather than waiting on it, because waiting could deadlock, so the handler
2107 /// is never invoked and the second writer dies at once. See `from_conn`.
2108 ///
2109 /// Drives the connection directly rather than a public method because the
2110 /// first writer has to still be *holding* the lock when the second arrives,
2111 /// and no public API lets a caller pause mid-transaction.
2112 #[test]
2113 fn a_second_writer_waits_for_the_first_instead_of_failing() {
2114 let dir = scratch("lock");
2115 std::fs::create_dir_all(&dir).expect("temp dir");
2116 let path = dir.join("graph.db");
2117
2118 let mut first = Store::open(&path).expect("first store opens");
2119 first
2120 .conn
2121 .execute_batch("CREATE TABLE IF NOT EXISTS probe(v INTEGER);")
2122 .expect("probe table");
2123 let mut second = Store::open(&path).expect("second store opens");
2124
2125 // Signalled once the first writer is inside its transaction *and* holding
2126 // the write lock, so the second arrives during the contention rather than
2127 // racing to it.
2128 //
2129 // A channel rather than a `Barrier`, because `Barrier::wait` cannot fail:
2130 // if the writer panicked before reaching it — a failing `expect`, a
2131 // migration error — the main thread would wait for ever and the test would
2132 // hang CI instead of failing it. Dropping the `Sender` on a panicking
2133 // thread disconnects the channel, so `recv_timeout` returns *immediately*
2134 // in that case, and the timeout covers the merely-pathological one.
2135 // Raised in review of this change.
2136 let (signal, holding) = std::sync::mpsc::channel::<()>();
2137
2138 let writer = std::thread::spawn(move || {
2139 let tx = first.conn.transaction().expect("first writer opens");
2140 let _: i64 = tx
2141 .query_row("SELECT count(*) FROM probe", [], |r| r.get(0))
2142 .expect("first writer reads");
2143 tx.execute("INSERT INTO probe(v) VALUES (1)", [])
2144 .expect("first writer writes");
2145 // Ignored deliberately: if the main thread has already gone (its own
2146 // assertion failed), there is nobody to tell and nothing to do.
2147 let _ = signal.send(());
2148 std::thread::sleep(std::time::Duration::from_millis(250));
2149 tx.commit().expect("first writer commits");
2150 });
2151
2152 if holding
2153 .recv_timeout(std::time::Duration::from_secs(10))
2154 .is_err()
2155 {
2156 // Re-raise the writer's own panic, which says far more than "timed
2157 // out" would; only if it did not panic is the timeout itself the news.
2158 writer.join().expect("first writer reached the lock");
2159 panic!("first writer neither signalled nor panicked");
2160 }
2161 let tx = second
2162 .conn
2163 .transaction()
2164 .expect("a second writer must wait for the first, not be refused");
2165 let _: i64 = tx
2166 .query_row("SELECT count(*) FROM probe", [], |r| r.get(0))
2167 .expect("second writer reads");
2168 tx.execute("INSERT INTO probe(v) VALUES (2)", [])
2169 .expect("second writer writes rather than failing `database is locked`");
2170 tx.commit().expect("second writer commits");
2171
2172 writer.join().expect("first writer finished");
2173
2174 let seen: i64 = second
2175 .conn
2176 .query_row("SELECT count(*) FROM probe", [], |r| r.get(0))
2177 .expect("count");
2178 assert_eq!(seen, 2, "both writers' rows survive — neither was dropped");
2179
2180 drop(second);
2181 std::fs::remove_dir_all(&dir).ok();
2182 }
2183
2184 /// A file-backed store takes WAL wherever the filesystem allows it, so a
2185 /// reader is not shut out for the whole of a writer's transaction.
2186 ///
2187 /// Asserted **against what a plain connection gets on this same filesystem**
2188 /// rather than against the literal `wal`, because `from_conn` treats WAL as
2189 /// tolerated rather than required: it is refused on most network filesystems,
2190 /// and a hard assertion would fail there for a store behaving exactly as
2191 /// designed. Raised in review — the first version of this test contradicted
2192 /// the tolerance documented by the code it tests.
2193 ///
2194 /// Still not vacuous, which is what an "assert the two agree" test has to
2195 /// earn: the probe is a raw `Connection` that asks for WAL itself, so a
2196 /// `from_conn` that stopped asking would leave the probe reporting `wal` and
2197 /// the store reporting something else, and the two would disagree.
2198 #[test]
2199 fn a_file_backed_store_takes_whatever_wal_this_filesystem_allows() {
2200 let dir = scratch("wal");
2201 std::fs::create_dir_all(&dir).expect("temp dir");
2202
2203 let available: String = {
2204 let probe = super::Connection::open(dir.join("probe.db")).expect("probe opens");
2205 probe
2206 .query_row("PRAGMA journal_mode = WAL", [], |r| r.get(0))
2207 .expect("probe journal mode")
2208 };
2209
2210 let store = Store::open(&dir.join("graph.db")).expect("store opens");
2211 let mode: String = store
2212 .conn
2213 .query_row("PRAGMA journal_mode", [], |r| r.get(0))
2214 .expect("journal mode");
2215
2216 assert_eq!(
2217 mode.to_ascii_lowercase(),
2218 available.to_ascii_lowercase(),
2219 "a store must reach the same journal mode a plain connection does here"
2220 );
2221
2222 drop(store);
2223 std::fs::remove_dir_all(&dir).ok();
2224 }
2225
2226 /// And an in-memory store still opens, which is the tolerance path: WAL is
2227 /// meaningless there, `PRAGMA journal_mode` answers `memory`, and that must
2228 /// not be an error — the same branch a network filesystem takes.
2229 #[test]
2230 fn an_in_memory_store_opens_without_wal() {
2231 let store = Store::open_in_memory().expect("in-memory store opens");
2232 let mode: String = store
2233 .conn
2234 .query_row("PRAGMA journal_mode", [], |r| r.get(0))
2235 .expect("journal mode");
2236 assert_ne!(mode.to_ascii_lowercase(), "wal");
2237 }
2238
2239 #[test]
2240 fn reconcile_matches_a_full_rebuild() {
2241 // reconcile must leave the store identical to a fresh rebuild, across an
2242 // add, a remove, a content change, and edge churn.
2243 let node = |k: &str, name: &str| {
2244 let mut n = sample_node(k);
2245 n.name = name.to_owned();
2246 n
2247 };
2248 let edge =
2249 |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
2250 // An inferred edge carries confidence and a src_ref — exercise both so the
2251 // equivalence claim covers every edge field, not just derived calls.
2252 let inferred = |src: &str, dst: &str, conf: f64| {
2253 let mut e = Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, conf);
2254 e.src_ref = Some("import:demo".to_owned());
2255 e
2256 };
2257
2258 let facts1 = FactSet {
2259 nodes: vec![node("a", "A"), node("b", "B"), node("c", "C")],
2260 edges: vec![edge("a", "b"), edge("b", "c"), inferred("a", "c", 0.7)],
2261 };
2262 // b changes (name), c is removed, d is added; edge b->c drops, a->d added,
2263 // and the inferred edge's confidence changes.
2264 let facts2 = FactSet {
2265 nodes: vec![node("a", "A"), node("b", "B2"), node("d", "D")],
2266 edges: vec![edge("a", "b"), edge("a", "d"), inferred("a", "d", 0.9)],
2267 };
2268
2269 // Path 1: rebuild facts1, then reconcile to facts2.
2270 let mut reconciled = Store::open_in_memory().expect("open");
2271 reconciled.rebuild(&facts1, Some("t1")).expect("rebuild");
2272 reconciled
2273 .reconcile(&facts2, Some("t2"))
2274 .expect("reconcile");
2275
2276 // Path 2: a fresh full rebuild of facts2.
2277 let mut rebuilt = Store::open_in_memory().expect("open");
2278 rebuilt.rebuild(&facts2, Some("t2")).expect("rebuild");
2279
2280 let canon = |fs: FactSet| {
2281 let mut nodes = fs.nodes;
2282 nodes.sort_by(|a, b| a.key.cmp(&b.key));
2283 let mut edges: Vec<String> = fs
2284 .edges
2285 .iter()
2286 .map(|e| {
2287 format!(
2288 "{}\0{}\0{}\0{}\0{:?}\0{:?}",
2289 e.kind.as_str(),
2290 e.src,
2291 e.dst,
2292 e.provenance.as_str(),
2293 e.confidence,
2294 e.src_ref
2295 )
2296 })
2297 .collect();
2298 edges.sort();
2299 (nodes, edges)
2300 };
2301 assert_eq!(
2302 canon(reconciled.export_factset().expect("export")),
2303 canon(rebuilt.export_factset().expect("export")),
2304 "reconcile must match a full rebuild",
2305 );
2306 assert_eq!(
2307 reconciled.sync_state().expect("state").as_deref(),
2308 Some("t2")
2309 );
2310 }
2311
2312 /// The worktree stamp (issue #330) identifies the *store*, not the tree: an
2313 /// unstamped store reads as "unknown" and is adopted rather than rebuilt, and
2314 /// a new tree state must not clear the stamp the way it clears `env`.
2315 #[test]
2316 fn the_worktree_stamp_survives_tree_writes_and_starts_unknown() {
2317 let mut store = Store::open_in_memory().expect("open");
2318 let facts = FactSet::new().with_node(sample_node("sym:a"));
2319
2320 // Never synced ⇒ no stamp. A legacy store reads the same way, which is
2321 // why "unknown" must mean adopt, not mismatch.
2322 assert_eq!(store.synced_worktree().expect("stamp"), None);
2323
2324 store.reconcile(&facts, Some("t1")).expect("reconcile");
2325 assert_eq!(
2326 store.synced_worktree().expect("stamp"),
2327 None,
2328 "writing a tree does not invent a stamp; the sync engine records it"
2329 );
2330
2331 store.set_synced_worktree("/w/one").expect("stamp");
2332 store.set_sync_env("env-1").expect("env");
2333 assert_eq!(
2334 store.synced_worktree().expect("stamp").as_deref(),
2335 Some("/w/one")
2336 );
2337
2338 // A later tree write clears `env` (it is only valid for one tree) but must
2339 // KEEP the stamp: a new commit does not move the graph to another tree.
2340 store.reconcile(&facts, Some("t2")).expect("reconcile");
2341 assert_eq!(store.sync_env().expect("env"), None, "env is tree-scoped");
2342 assert_eq!(
2343 store.synced_worktree().expect("stamp").as_deref(),
2344 Some("/w/one"),
2345 "the stamp identifies the store, not the tree, so it must survive"
2346 );
2347
2348 // Re-stamping replaces it (the store was adopted by another tree).
2349 store.set_synced_worktree("/w/two").expect("restamp");
2350 assert_eq!(
2351 store.synced_worktree().expect("stamp").as_deref(),
2352 Some("/w/two")
2353 );
2354
2355 // Clearing the synced state clears the stamp with it: a store with no
2356 // recorded tree describes no working tree either.
2357 store.rebuild(&facts, None).expect("rebuild unstated");
2358 assert_eq!(store.sync_state().expect("state"), None);
2359 assert_eq!(store.synced_worktree().expect("stamp"), None);
2360 }
2361
2362 #[test]
2363 fn reconcile_writes_only_the_edge_delta() {
2364 // An unchanged edge must keep its row (proving reconcile does not wipe and
2365 // reinsert the whole edge set); a removed edge's row goes; a new edge's row
2366 // appears. Row identity is the SQLite `rowid` — stable unless deleted.
2367 let n = |k: &str| sample_node(k);
2368 let e =
2369 |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
2370
2371 let mut store = Store::open_in_memory().expect("open");
2372 store
2373 .rebuild(
2374 &FactSet {
2375 nodes: vec![n("a"), n("b"), n("c")],
2376 edges: vec![e("a", "b"), e("b", "c")],
2377 },
2378 None,
2379 )
2380 .expect("rebuild");
2381
2382 // Map (src_key, dst_key) → rowid via the private connection.
2383 let rowids = |store: &Store| -> std::collections::HashMap<(String, String), i64> {
2384 let mut stmt = store
2385 .conn
2386 .prepare(
2387 "SELECT ns.key, nd.key, e.rowid FROM edges e \
2388 JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst",
2389 )
2390 .expect("prepare");
2391 stmt.query_map([], |r| {
2392 Ok((
2393 (r.get::<_, String>(0)?, r.get::<_, String>(1)?),
2394 r.get::<_, i64>(2)?,
2395 ))
2396 })
2397 .expect("query")
2398 .map(Result::unwrap)
2399 .collect()
2400 };
2401
2402 let before = rowids(&store);
2403 let ab_rowid = before[&("a".to_owned(), "b".to_owned())];
2404
2405 // Keep a->b, drop b->c, add a->c.
2406 store
2407 .reconcile(
2408 &FactSet {
2409 nodes: vec![n("a"), n("b"), n("c")],
2410 edges: vec![e("a", "b"), e("a", "c")],
2411 },
2412 None,
2413 )
2414 .expect("reconcile");
2415
2416 let after = rowids(&store);
2417 assert_eq!(
2418 after.get(&("a".to_owned(), "b".to_owned())),
2419 Some(&ab_rowid),
2420 "the unchanged edge keeps its row (not rewritten)"
2421 );
2422 assert!(
2423 !after.contains_key(&("b".to_owned(), "c".to_owned())),
2424 "the removed edge's row is gone"
2425 );
2426 assert!(
2427 after.contains_key(&("a".to_owned(), "c".to_owned())),
2428 "the added edge has a new row"
2429 );
2430 }
2431
2432 #[test]
2433 fn reconcile_is_history_independent_for_edge_queries() {
2434 // The whole point of the edge delta: it must be invisible above storage.
2435 // A store reached by rebuild(f1)+reconcile(f2) has different edge row ids
2436 // than a cold rebuild at f2, yet every edge query must return byte-for-byte
2437 // the same result — order included — because the queries are content-ordered.
2438 let n = |k: &str| sample_node(k);
2439 let d =
2440 |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
2441 let inf = |src: &str, dst: &str, c: f64| {
2442 Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, c)
2443 };
2444 let f1 = FactSet {
2445 nodes: vec![n("a"), n("b"), n("c")],
2446 edges: vec![d("a", "b"), d("b", "c"), inf("a", "c", 0.7)],
2447 };
2448 let f2 = FactSet {
2449 nodes: vec![n("a"), n("b"), n("d")],
2450 edges: vec![d("a", "b"), d("a", "d"), inf("a", "d", 0.9)],
2451 };
2452
2453 let mut incremental = Store::open_in_memory().expect("open");
2454 incremental.rebuild(&f1, None).expect("rebuild");
2455 incremental.reconcile(&f2, None).expect("reconcile");
2456 let mut cold = Store::open_in_memory().expect("open");
2457 cold.rebuild(&f2, None).expect("rebuild");
2458
2459 // Project to all fields so the comparison covers order *and* content.
2460 let proj = |es: Vec<Edge>| -> Vec<String> {
2461 es.into_iter()
2462 .map(|e| {
2463 format!(
2464 "{}|{}|{}|{}|{:?}|{:?}",
2465 e.src,
2466 e.dst,
2467 e.kind.as_str(),
2468 e.provenance.as_str(),
2469 e.confidence,
2470 e.src_ref
2471 )
2472 })
2473 .collect()
2474 };
2475
2476 for key in ["a", "b", "d"] {
2477 assert_eq!(
2478 proj(incremental.edges_from(key).expect("from")),
2479 proj(cold.edges_from(key).expect("from")),
2480 "edges_from({key}) must match a cold rebuild"
2481 );
2482 assert_eq!(
2483 proj(incremental.edges_to(key).expect("to")),
2484 proj(cold.edges_to(key).expect("to")),
2485 "edges_to({key}) must match a cold rebuild"
2486 );
2487 }
2488 assert_eq!(
2489 proj(incremental.all_edges().expect("all")),
2490 proj(cold.all_edges().expect("all")),
2491 "all_edges must match a cold rebuild"
2492 );
2493 for p in [Provenance::Derived, Provenance::Inferred] {
2494 assert_eq!(
2495 proj(incremental.edges_by_provenance(p).expect("prov")),
2496 proj(cold.edges_by_provenance(p).expect("prov")),
2497 "edges_by_provenance({}) must match a cold rebuild",
2498 p.as_str()
2499 );
2500 }
2501 }
2502
2503 #[test]
2504 fn reconcile_updates_confidence_on_an_unchanged_tuple() {
2505 // A change to *only* an edge's confidence — same (src, dst, kind,
2506 // provenance) — must still be applied. Edge identity includes confidence,
2507 // so it is delete+add (matching a full rebuild), not the insert-time
2508 // `DO NOTHING` that would leave the stale confidence in place.
2509 let n = |k: &str| sample_node(k);
2510 let inf = |c: f64| {
2511 let mut e = Edge::inferred("a".to_owned(), "b".to_owned(), EdgeKind::Related, c);
2512 e.src_ref = Some("import:demo".to_owned());
2513 e
2514 };
2515
2516 let mut store = Store::open_in_memory().expect("open");
2517 store
2518 .rebuild(
2519 &FactSet {
2520 nodes: vec![n("a"), n("b")],
2521 edges: vec![inf(0.5)],
2522 },
2523 None,
2524 )
2525 .expect("rebuild");
2526 store
2527 .reconcile(
2528 &FactSet {
2529 nodes: vec![n("a"), n("b")],
2530 edges: vec![inf(0.9)],
2531 },
2532 None,
2533 )
2534 .expect("reconcile");
2535
2536 let edges = store.edges_from("a").expect("edges");
2537 assert_eq!(edges.len(), 1);
2538 assert_eq!(
2539 edges[0].confidence,
2540 Some(0.9),
2541 "confidence updated, not left stale"
2542 );
2543 }
2544
2545 #[test]
2546 fn edge_identity_does_not_collide_across_field_boundaries() {
2547 // Node keys embed git paths, which may contain any byte — including the
2548 // unit separator (`\x1f`). A delimiter-joined identity would map these two
2549 // distinct edges to the same string (`a\x1fb\x1fc\x1f…`); the tuple identity
2550 // must keep them apart, or reconcile would drop one edge as a "duplicate".
2551 let e1 = super::edge_identity(&Edge::derived(
2552 "a\u{1f}b".to_owned(),
2553 "c".to_owned(),
2554 EdgeKind::Calls,
2555 ));
2556 let e2 = super::edge_identity(&Edge::derived(
2557 "a".to_owned(),
2558 "b\u{1f}c".to_owned(),
2559 EdgeKind::Calls,
2560 ));
2561 assert_ne!(
2562 e1, e2,
2563 "control chars in a key must not collapse identities"
2564 );
2565
2566 // A confidence-only difference (same tuple otherwise) also stays distinct,
2567 // and `None` (derived) never equals `Some(0.0)`.
2568 let derived = super::edge_identity(&Edge::derived(
2569 "a".to_owned(),
2570 "b".to_owned(),
2571 EdgeKind::Related,
2572 ));
2573 let inferred0 = super::edge_identity(&Edge::inferred(
2574 "a".to_owned(),
2575 "b".to_owned(),
2576 EdgeKind::Related,
2577 0.0,
2578 ));
2579 assert_ne!(derived, inferred0, "None vs Some(0.0) confidence differ");
2580 }
2581
2582 #[test]
2583 fn open_in_memory_applies_schema() {
2584 let store = Store::open_in_memory().expect("open");
2585 assert_eq!(store.node_count().expect("count"), 0);
2586 // Written against `latest_version()` rather than the literal that stood
2587 // here (8 for analyzer findings, then 9, 10, 11 as the media and
2588 // agent-memory tables landed). The literal was defended as making someone
2589 // confirm a new migration is meant to apply on open — but it could not do
2590 // that job: `apply` applies every migration not already recorded, so
2591 // "applies on open" is not a per-migration choice there is anything to
2592 // confirm. What the literal actually asserted was the value of a shared
2593 // constant, which every future migration then has to come here and edit,
2594 // in a file it otherwise has no business in — the brittleness #329
2595 // replaced with a property test elsewhere. This is the idiom
2596 // `migrations::tests::a_later_migration_is_additive_on_a_populated_store`
2597 // already uses, for the same reason.
2598 //
2599 // Since migrations are selected by set membership, the property is now
2600 // strictly stronger than "the biggest recorded number is N":
2601 // `schema_version` is the highest **gap-free** version, so equality with
2602 // `latest_version` also rules out a store that skipped one and stamped a
2603 // higher one. A fresh store cannot exhibit that gap, so
2604 // `reopening_repairs_a_skipped_migration` exercises it directly.
2605 assert_eq!(
2606 store.schema_version().expect("version"),
2607 crate::migrations::latest_version(),
2608 "opening a store must apply every known migration, with no gaps"
2609 );
2610 }
2611
2612 /// Reopening a store that is **missing a lower-numbered migration while
2613 /// carrying a higher one** repairs it — through the real public API, which
2614 /// is where it matters to callers.
2615 ///
2616 /// This is not a hypothetical: it is the shape `main` itself carries.
2617 /// `feat/stage25-memory-recall` merged migration **13** while migration
2618 /// **12** was still on this branch, so every store opened by a current-`main`
2619 /// build — including this repository's own `.git/roteiro/graph.db` and each
2620 /// linked worktree's — records `1..11, 13`. Under the old
2621 /// `version > MAX(recorded)` rule, `12 > 13` is false and migration 12 would
2622 /// never be applied to any of them, permanently: the store opens cleanly,
2623 /// reports a schema it does not have, and `synced_worktree()` fails with
2624 /// `no such column: worktree` — the issue #330 tree stamp, silently absent.
2625 ///
2626 /// The two migrations are deliberately named by number rather than derived
2627 /// from `latest_version()`. Which one is missing and which is present is the
2628 /// specific historical fact being tested; arithmetic on the newest version
2629 /// would quietly re-aim the test at whatever lands next and stop covering
2630 /// this.
2631 #[test]
2632 fn reopening_repairs_a_skipped_migration() {
2633 /// The migration this branch adds — absent from a current-`main` store.
2634 const SKIPPED: u32 = 12;
2635 /// The migration `main` added in parallel — present, and stamped higher.
2636 const AHEAD: u32 = 13;
2637
2638 let path = std::env::temp_dir().join(format!(
2639 "roteiro-migration-gap-{}-{:?}.db",
2640 std::process::id(),
2641 std::thread::current().id()
2642 ));
2643 std::fs::remove_file(&path).ok();
2644
2645 {
2646 let mut store = Store::open(&path).expect("open");
2647 // Reconcile rather than a bare upsert: it records a synced tree, so
2648 // `sync_state` has the row the worktree stamp updates. Without one,
2649 // `set_synced_worktree` is a documented no-op and would prove nothing.
2650 let facts = FactSet::new().with_node(sample_node("kept"));
2651 store
2652 .reconcile(&facts, Some("t1"))
2653 .expect("seed graph and sync state");
2654 }
2655 // Rewind to exactly what a current-`main` build leaves behind: migration
2656 // 12's column and record gone, 13's schema and record intact.
2657 {
2658 let conn = rusqlite::Connection::open(&path).expect("reopen raw");
2659 conn.execute_batch("ALTER TABLE sync_state DROP COLUMN worktree;")
2660 .expect("undo migration 12");
2661 conn.execute(
2662 "DELETE FROM schema_migrations WHERE version = ?1",
2663 [SKIPPED],
2664 )
2665 .expect("unrecord 12");
2666 // Sanity: the store really is in the damaged shape, 13 and all.
2667 let recorded: i64 = conn
2668 .query_row(
2669 "SELECT COUNT(*) FROM schema_migrations WHERE version = ?1",
2670 [AHEAD],
2671 |r| r.get(0),
2672 )
2673 .expect("count 13");
2674 assert_eq!(recorded, 1, "migration {AHEAD} must still be recorded");
2675 }
2676 // Migration 13's own state, captured before the repair so it can be shown
2677 // untouched afterwards — a repair that healed the gap by disturbing the
2678 // other branch's tables would be no better than the bug.
2679 let clock_before = cache_clock(&path);
2680
2681 // Before the repair the store under-reports rather than over-reports: it
2682 // claims the last gap-free version, never the higher stamped one.
2683 {
2684 let damaged = rusqlite::Connection::open(&path).expect("reopen raw");
2685 assert_eq!(
2686 crate::migrations::store_version(&damaged).expect("version"),
2687 SKIPPED - 1,
2688 "a gapped store must report the version it actually provides, \
2689 not the maximum recorded ({AHEAD})"
2690 );
2691 }
2692
2693 // Reopening through the public API repairs the gap…
2694 let store = Store::open(&path).expect("reopen");
2695 assert_eq!(
2696 store.schema_version().expect("version"),
2697 crate::migrations::latest_version(),
2698 "1..={AHEAD} is contiguous once the skipped migration is applied"
2699 );
2700 // …migration 12's schema is really back, which is what the caller
2701 // depends on — recorded-but-absent would be the same bug in a new
2702 // costume…
2703 assert!(
2704 store.synced_worktree().is_ok(),
2705 "the repaired migration's column must exist, not merely be recorded"
2706 );
2707 store
2708 .set_synced_worktree("/w/repaired")
2709 .expect("the stamp must be writable, not just readable");
2710 assert_eq!(
2711 store.synced_worktree().expect("stamp").as_deref(),
2712 Some("/w/repaired")
2713 );
2714 // …migration 13's tables are untouched, clock row and all…
2715 assert_eq!(
2716 cache_clock(&path),
2717 clock_before,
2718 "the other branch's seeded `agent_cache_clock` row must survive the \
2719 repair unchanged"
2720 );
2721 {
2722 let conn = rusqlite::Connection::open(&path).expect("reopen raw");
2723 let rows: i64 = conn
2724 .query_row("SELECT COUNT(*) FROM agent_cache", [], |r| r.get(0))
2725 .expect("agent_cache must still exist");
2726 assert_eq!(rows, 0, "`agent_cache` is present and empty, not recreated");
2727 }
2728 // …and no data was lost.
2729 assert!(store.get_node("kept").expect("get").is_some());
2730 assert_eq!(store.node_count().expect("count"), 1);
2731
2732 std::fs::remove_file(&path).expect("cleanup");
2733 }
2734
2735 /// Migration 13's single `agent_cache_clock` row as `(ticks, generation)`,
2736 /// read raw so this test does not depend on the recall API another branch
2737 /// owns.
2738 fn cache_clock(path: &std::path::Path) -> (i64, i64) {
2739 let conn = rusqlite::Connection::open(path).expect("open raw");
2740 conn.query_row(
2741 "SELECT ticks, generation FROM agent_cache_clock WHERE id = 0",
2742 [],
2743 |r| Ok((r.get(0)?, r.get(1)?)),
2744 )
2745 .expect("the seeded clock row must exist")
2746 }
2747
2748 /// Record `version` in `store`'s migration table as though a build that knew
2749 /// that migration had run it — the only way to manufacture a store from the
2750 /// future without a time machine.
2751 ///
2752 /// Only the *record* is written, not any schema the migration would have
2753 /// added. That is exactly the position an older binary is in: it can see the
2754 /// stamp and cannot see, or reason about, the shape.
2755 fn stamp_version(store: &Store, version: u32) {
2756 store
2757 .conn
2758 .execute(
2759 "INSERT INTO schema_migrations (version) VALUES (?1)",
2760 [version],
2761 )
2762 .expect("stamp a version this build does not know");
2763 }
2764
2765 #[test]
2766 fn a_store_this_build_wrote_is_not_ahead_of_it() {
2767 let store = Store::open_in_memory().expect("open");
2768 assert_eq!(
2769 store.schema_ahead().expect("schema_ahead"),
2770 None,
2771 "a store this very build just migrated cannot be ahead of it"
2772 );
2773 }
2774
2775 /// A store carrying migrations this build has never heard of is reported
2776 /// ahead, and the report names **both** sides plus what to do about it.
2777 ///
2778 /// Versions are `latest_version() + n` rather than literals: the property is
2779 /// "beyond what this build knows", and a literal would silently stop testing
2780 /// that the day a real migration reached the same number.
2781 #[test]
2782 fn a_store_stamped_beyond_this_build_is_reported_ahead() {
2783 let build = Store::build_schema_version();
2784 let store = Store::open_in_memory().expect("open");
2785 stamp_version(&store, build + 1);
2786 stamp_version(&store, build + 2);
2787
2788 let ahead = store
2789 .schema_ahead()
2790 .expect("schema_ahead")
2791 .expect("a store two migrations beyond this build must be reported");
2792 assert_eq!(ahead.build_version(), build);
2793 assert_eq!(
2794 ahead.store_version(),
2795 build + 2,
2796 "the store's version is the highest it records, not the first \
2797 unknown one"
2798 );
2799 assert_eq!(ahead.unknown_versions(), [build + 1, build + 2]);
2800
2801 // The message has to be actionable, not merely correct: both versions by
2802 // number, and the fix (upgrade the binary — never "delete the store").
2803 let message = ahead.to_string();
2804 for expected in [
2805 &format!("{}", build + 2),
2806 &format!("{build}"),
2807 &"pgrade".to_owned(),
2808 ] {
2809 assert!(
2810 message.contains(expected.as_str()),
2811 "the refusal must name `{expected}`: {message}"
2812 );
2813 }
2814 }
2815
2816 /// A **gap** — a store missing a lower migration while carrying a higher one
2817 /// this build knows — is not a store from the future, and must not be
2818 /// reported as one. Conflating them would name a version the store is not
2819 /// at, in a message telling the reader to upgrade a binary that is already
2820 /// new enough.
2821 ///
2822 /// This is the shape `main` itself carried (see
2823 /// `reopening_repairs_a_skipped_migration`, which proves `Store::open`
2824 /// repairs it), so it is a live case, not a hypothetical.
2825 ///
2826 /// The second half is the one that matters: a store that is *both* gapped
2827 /// and from the future must be described only by the half the reader can act
2828 /// on. Naming the gap would tell someone to upgrade a binary that is already
2829 /// new enough for it.
2830 #[test]
2831 fn a_gap_below_this_build_is_not_a_store_from_the_future() {
2832 let build = Store::build_schema_version();
2833
2834 // A gap and nothing else: behind this build, so never ahead of it.
2835 let gapped = Store::open_in_memory().expect("open");
2836 gapped
2837 .conn
2838 .execute("DELETE FROM schema_migrations WHERE version = ?1", [build])
2839 .expect("open a gap");
2840 assert!(
2841 gapped.schema_version().expect("version") < build,
2842 "the gapped store must under-report, or this test proves nothing"
2843 );
2844 assert_eq!(
2845 gapped.schema_ahead().expect("schema_ahead"),
2846 None,
2847 "a store *missing* a migration is behind this build, never ahead"
2848 );
2849
2850 // A gap *and* a version from the future: reported ahead, and reported
2851 // only in terms of the future version.
2852 stamp_version(&gapped, build + 1);
2853 let ahead = gapped
2854 .schema_ahead()
2855 .expect("schema_ahead")
2856 .expect("the unknown migration is still unknown, gap or no gap");
2857 assert_eq!(ahead.unknown_versions(), [build + 1]);
2858 assert_eq!(
2859 ahead.store_version(),
2860 build + 1,
2861 "the gap is a separate condition and must not colour this report"
2862 );
2863 }
2864
2865 /// The reason the guard compares the recorded **set** and not
2866 /// [`Store::schema_version`].
2867 ///
2868 /// `schema_version` is the highest *gap-free* version — a floor for readers.
2869 /// A store recorded `1..=build, build + 2` therefore reports `build`, and a
2870 /// `schema_version() > build` test sees a perfectly ordinary store, while
2871 /// migration `build + 2`'s schema is in the file and a newer binary plainly
2872 /// assembled the graph. That is issue #342 surviving the check written to
2873 /// catch it, so it is asserted directly rather than trusted.
2874 #[test]
2875 fn the_gap_free_version_alone_would_miss_a_store_from_the_future() {
2876 let build = Store::build_schema_version();
2877 let store = Store::open_in_memory().expect("open");
2878 stamp_version(&store, build + 2); // note: `build + 1` deliberately absent
2879
2880 assert_eq!(
2881 store.schema_version().expect("version"),
2882 build,
2883 "the gap-free version stops at the last contiguous migration, so it \
2884 cannot see the one above the hole"
2885 );
2886 let ahead = store
2887 .schema_ahead()
2888 .expect("schema_ahead")
2889 .expect("the set difference sees what the contiguity walk cannot");
2890 assert_eq!(ahead.store_version(), build + 2);
2891 assert_eq!(ahead.unknown_versions(), [build + 2]);
2892 }
2893
2894 #[test]
2895 fn upsert_and_get_round_trips_all_fields() {
2896 let store = Store::open_in_memory().expect("open");
2897 let node = sample_node("sym:rust:src/lib.rs#sample");
2898 store.upsert_node(&node).expect("upsert");
2899 let got = store.get_node(&node.key).expect("get").expect("present");
2900 assert_eq!(got, node);
2901 }
2902
2903 #[test]
2904 fn upsert_updates_in_place() {
2905 let store = Store::open_in_memory().expect("open");
2906 let mut node = sample_node("k");
2907 store.upsert_node(&node).expect("insert");
2908 node.name = "renamed".to_owned();
2909 node.kind = NodeKind::Struct;
2910 store.upsert_node(&node).expect("update");
2911 assert_eq!(store.node_count().expect("count"), 1);
2912 let got = store.get_node("k").expect("get").expect("present");
2913 assert_eq!(got.name, "renamed");
2914 assert_eq!(got.kind, NodeKind::Struct);
2915 }
2916
2917 #[test]
2918 fn edge_with_unknown_endpoint_is_rejected() {
2919 let store = Store::open_in_memory().expect("open");
2920 store
2921 .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
2922 .expect("a");
2923 let edge = Edge::derived("a", "missing", EdgeKind::Calls);
2924 let err = store.insert_edge(&edge).expect_err("should reject");
2925 assert!(matches!(err, super::StoreError::UnknownNode(k) if k == "missing"));
2926 }
2927
2928 #[test]
2929 fn inferred_edge_requires_confidence() {
2930 let store = Store::open_in_memory().expect("open");
2931 store
2932 .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
2933 .expect("a");
2934 store
2935 .upsert_node(&Node::new("b", NodeKind::Fn, "b"))
2936 .expect("b");
2937 // Hand-build an inferred edge with no confidence to violate the invariant.
2938 let bad = Edge {
2939 src: "a".to_owned(),
2940 dst: "b".to_owned(),
2941 kind: EdgeKind::References,
2942 provenance: Provenance::Inferred,
2943 confidence: None,
2944 src_ref: None,
2945 };
2946 assert!(matches!(
2947 store.insert_edge(&bad).expect_err("reject"),
2948 super::StoreError::InvalidEdge(_)
2949 ));
2950 }
2951
2952 #[test]
2953 fn apply_factset_is_atomic() {
2954 let mut store = Store::open_in_memory().expect("open");
2955 // Second edge references a missing node, so the whole set must roll back.
2956 let facts = FactSet::new()
2957 .with_node(Node::new("a", NodeKind::Fn, "a"))
2958 .with_node(Node::new("b", NodeKind::Fn, "b"))
2959 .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
2960 .with_edge(Edge::derived("a", "ghost", EdgeKind::Calls));
2961 assert!(store.apply_factset(&facts).is_err());
2962 assert_eq!(store.node_count().expect("count"), 0, "rolled back");
2963 assert_eq!(store.edge_count().expect("count"), 0, "rolled back");
2964 }
2965
2966 #[test]
2967 fn neighbors_and_provenance_queries() {
2968 let mut store = Store::open_in_memory().expect("open");
2969 let facts = FactSet::new()
2970 .with_node(Node::new("a", NodeKind::Fn, "a"))
2971 .with_node(Node::new("b", NodeKind::Fn, "b"))
2972 .with_node(Node::new("c", NodeKind::Fn, "c"))
2973 .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
2974 .with_edge(Edge::inferred("a", "c", EdgeKind::References, 0.5));
2975 store.apply_factset(&facts).expect("apply");
2976
2977 let out = store.neighbors("a", Direction::Outgoing).expect("out");
2978 let mut keys: Vec<_> = out.iter().map(|n| n.key.clone()).collect();
2979 keys.sort();
2980 assert_eq!(keys, ["b", "c"]);
2981
2982 assert!(
2983 store
2984 .neighbors("b", Direction::Outgoing)
2985 .expect("b out")
2986 .is_empty()
2987 );
2988 assert_eq!(
2989 store
2990 .neighbors("b", Direction::Incoming)
2991 .expect("b in")
2992 .len(),
2993 1
2994 );
2995
2996 let inferred = store
2997 .edges_by_provenance(Provenance::Inferred)
2998 .expect("inf");
2999 assert_eq!(inferred.len(), 1);
3000 assert_eq!(inferred[0].confidence, Some(0.5));
3001 }
3002
3003 #[test]
3004 fn neighbors_of_absent_node_is_empty() {
3005 let store = Store::open_in_memory().expect("open");
3006 assert!(
3007 store
3008 .neighbors("nope", Direction::Both)
3009 .expect("q")
3010 .is_empty()
3011 );
3012 }
3013
3014 #[test]
3015 fn get_missing_node_is_none() {
3016 let store = Store::open_in_memory().expect("open");
3017 assert!(store.get_node("absent").expect("get").is_none());
3018 }
3019
3020 #[test]
3021 fn nodes_by_kind_and_edges_to() {
3022 let mut store = Store::open_in_memory().expect("open");
3023 let facts = FactSet::new()
3024 .with_node(Node::new("f1", NodeKind::Fn, "f1"))
3025 .with_node(Node::new("f2", NodeKind::Fn, "f2"))
3026 .with_node(Node::new("s1", NodeKind::Struct, "s1"))
3027 .with_edge(Edge::derived("f1", "s1", EdgeKind::References))
3028 .with_edge(Edge::derived("f2", "s1", EdgeKind::References));
3029 store.apply_factset(&facts).expect("apply");
3030
3031 let fns = store.nodes_by_kind(&NodeKind::Fn).expect("fns");
3032 assert_eq!(
3033 fns.iter().map(|n| n.key.as_str()).collect::<Vec<_>>(),
3034 ["f1", "f2"]
3035 );
3036 assert!(
3037 store
3038 .nodes_by_kind(&NodeKind::Enum)
3039 .expect("enums")
3040 .is_empty()
3041 );
3042
3043 let into_s1 = store.edges_to("s1").expect("edges_to");
3044 assert_eq!(into_s1.len(), 2);
3045 assert!(into_s1.iter().all(|e| e.dst == "s1"));
3046 }
3047
3048 #[test]
3049 fn open_persists_across_reopen() {
3050 let path =
3051 std::env::temp_dir().join(format!("roteiro-open-test-{}.db", std::process::id()));
3052 std::fs::remove_file(&path).ok();
3053 {
3054 let store = Store::open(&path).expect("open");
3055 store
3056 .upsert_node(&sample_node("persisted"))
3057 .expect("upsert");
3058 }
3059 {
3060 let store = Store::open(&path).expect("reopen");
3061 assert_eq!(store.node_count().expect("count"), 1);
3062 // The subject here is that a *reopen* is at the same schema as a
3063 // fresh open — not what number that happens to be. See
3064 // `open_in_memory_applies_schema` on why the literal went.
3065 assert_eq!(
3066 store.schema_version().expect("version"),
3067 crate::migrations::latest_version(),
3068 "reopening applies any migration added since the file was written"
3069 );
3070 assert!(store.get_node("persisted").expect("get").is_some());
3071 }
3072 std::fs::remove_file(&path).expect("cleanup");
3073 }
3074
3075 fn graphify_layer(dst: &str) -> FactSet {
3076 FactSet::new()
3077 .with_node(Node::new("graphify:doc1", NodeKind::Doc, "Doc 1"))
3078 .with_edge({
3079 let mut e = Edge::inferred("graphify:doc1", dst, EdgeKind::References, 0.9);
3080 e.src_ref = Some("import:graphify".to_owned());
3081 e
3082 })
3083 }
3084
3085 /// A **withdrawn** concept goes, and a node another layer also asserts stays.
3086 ///
3087 /// This is the half of idempotence that was missing. Re-importing could never
3088 /// duplicate — nodes upsert by key, edges are deleted by `src_ref` first — so
3089 /// a re-run always *looked* idempotent, while a node the source had since
3090 /// deleted survived for ever as an orphan carrying the layer's provenance.
3091 /// Nothing appeared twice, which is exactly why nobody would notice.
3092 ///
3093 /// Three nodes, and each one is a different rule:
3094 ///
3095 /// * `okf:acme/gone.md` — withdrawn, and only this layer's. It must go.
3096 /// * `okf:acme/kept.md` — still asserted. Untouched.
3097 /// * `extref:acme::shared` — withdrawn *by this layer*, and asserted by
3098 /// another. It must survive, because two layers naming one node is a
3099 /// supported arrangement rather than an anomaly: `import:links` and
3100 /// `import:links/authored` share every placeholder between them.
3101 #[test]
3102 fn re_importing_a_smaller_bundle_removes_what_was_withdrawn_and_nothing_else() {
3103 let mut store = Store::open_in_memory().expect("open");
3104
3105 let okf_node = |key: &str| {
3106 Node::new(key, NodeKind::Doc, key).with_provenance(Provenance::ExternalAuthored)
3107 };
3108 // Another producer, which also asserts the shared placeholder.
3109 let other = FactSet::new().with_node(okf_node("extref:acme::shared"));
3110 store
3111 .apply_import_layer("import:links", &other)
3112 .expect("other layer");
3113
3114 let first = FactSet::new()
3115 .with_node(okf_node("okf:acme/gone.md"))
3116 .with_node(okf_node("okf:acme/kept.md"))
3117 .with_node(okf_node("extref:acme::shared"));
3118 let applied = store
3119 .apply_import_layer("import:okf/acme", &first)
3120 .expect("first import");
3121 assert_eq!(
3122 applied.nodes_removed, 0,
3123 "nothing is withdrawn on a first run"
3124 );
3125 assert_eq!(store.node_count().expect("count"), 3);
3126
3127 // The peer deletes one concept and stops asserting the placeholder.
3128 let smaller = FactSet::new().with_node(okf_node("okf:acme/kept.md"));
3129 let applied = store
3130 .apply_import_layer("import:okf/acme", &smaller)
3131 .expect("re-import");
3132
3133 // Which nodes, before how many. The count alone catches both directions
3134 // of failure but names only one of them: removing the shared placeholder
3135 // as well would fail a `nodes_removed == 1` assertion with a message
3136 // about a concept surviving, which is the opposite of what went wrong.
3137 assert!(
3138 store
3139 .get_node("extref:acme::shared")
3140 .expect("get")
3141 .is_some(),
3142 "a node another layer also asserts must not be removed by this one"
3143 );
3144 assert!(
3145 store.get_node("okf:acme/kept.md").expect("get").is_some(),
3146 "a concept still asserted must survive"
3147 );
3148 assert!(
3149 store.get_node("okf:acme/gone.md").expect("get").is_none(),
3150 "a concept the peer deleted must not survive a re-import"
3151 );
3152 assert_eq!(
3153 applied.nodes_removed, 1,
3154 "the withdrawn concept must be removed, and only it"
3155 );
3156
3157 // And it is genuinely idempotent now: re-running the same smaller
3158 // bundle removes nothing further.
3159 let again = store
3160 .apply_import_layer("import:okf/acme", &smaller)
3161 .expect("third run");
3162 assert_eq!(again.nodes_removed, 0);
3163 assert_eq!(store.node_count().expect("count"), 2);
3164 }
3165
3166 /// A withdrawn node that something is **still attached to** survives, in
3167 /// either direction.
3168 ///
3169 /// The layer stopped asserting the node, but another producer's edge still
3170 /// relates it to the graph. Removing it would take that relationship with it
3171 /// — a deletion nobody asked for, wearing a cleanup's clothes — and would
3172 /// violate the foreign key on the way out.
3173 ///
3174 /// Both ends are exercised on purpose: the probe is an `OR` over `src` and
3175 /// `dst`, so a fixture attaching the node only as a destination leaves the
3176 /// other arm free to be deleted with nothing going red.
3177 #[test]
3178 fn a_withdrawn_node_another_producer_still_points_at_is_kept() {
3179 let mut store = Store::open_in_memory().expect("open");
3180 let derived = FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a.rs"));
3181 store.rebuild(&derived, Some("tree1")).expect("rebuild");
3182
3183 // Two withdrawn nodes, one for each **direction** an edge can attach.
3184 // The probe is an `OR` over `src` and `dst`, and a test that only ever
3185 // put the node on one end would leave the other arm unguarded: deleting
3186 // `src = …` from the query changed no test until this second node
3187 // existed, which is how the gap was found.
3188 let inbound = Node::new("okf:acme/pointed-at.md", NodeKind::Doc, "pointed at")
3189 .with_provenance(Provenance::ExternalDerived);
3190 let outbound = Node::new("okf:acme/points-from.md", NodeKind::Doc, "points from")
3191 .with_provenance(Provenance::ExternalDerived);
3192 let first = FactSet::new().with_node(inbound).with_node(outbound);
3193 store
3194 .apply_import_layer("import:okf/acme", &first)
3195 .expect("first");
3196
3197 // A different layer relates the derived file to one imported concept,
3198 // and the other imported concept to the derived file.
3199 let linking = FactSet::new()
3200 .with_edge({
3201 let mut e = Edge::inferred(
3202 "file:a.rs",
3203 "okf:acme/pointed-at.md",
3204 EdgeKind::References,
3205 0.9,
3206 );
3207 e.src_ref = Some("import:links".to_owned());
3208 e
3209 })
3210 .with_edge({
3211 let mut e = Edge::inferred(
3212 "okf:acme/points-from.md",
3213 "file:a.rs",
3214 EdgeKind::References,
3215 0.9,
3216 );
3217 e.src_ref = Some("import:links".to_owned());
3218 e
3219 });
3220 store
3221 .apply_import_layer("import:links", &linking)
3222 .expect("link layer");
3223
3224 let applied = store
3225 .apply_import_layer("import:okf/acme", &FactSet::new())
3226 .expect("withdraw everything");
3227 assert!(
3228 store
3229 .get_node("okf:acme/pointed-at.md")
3230 .expect("get")
3231 .is_some(),
3232 "a node another producer's edge points *at* must not be removed"
3233 );
3234 assert!(
3235 store
3236 .get_node("okf:acme/points-from.md")
3237 .expect("get")
3238 .is_some(),
3239 "nor one another producer's edge points *from*"
3240 );
3241 assert_eq!(applied.nodes_removed, 0);
3242 }
3243
3244 /// Removal takes the node's cached context with it. A bundle assembled for a
3245 /// node that no longer exists can only ever be served as a stale answer, and
3246 /// `node_context` has no foreign key to clean it up for us.
3247 #[test]
3248 fn removing_a_withdrawn_node_drops_its_cached_context() {
3249 let mut store = Store::open_in_memory().expect("open");
3250 let first = FactSet::new().with_node(
3251 Node::new("okf:acme/gone.md", NodeKind::Doc, "gone")
3252 .with_provenance(Provenance::ExternalInferred),
3253 );
3254 store
3255 .apply_import_layer("import:okf/acme", &first)
3256 .expect("first");
3257 store
3258 .context_cache_put("okf:acme/gone.md", "fp1", "{}")
3259 .expect("cache");
3260 assert_eq!(
3261 store.context_cache_keys().expect("keys"),
3262 vec!["okf:acme/gone.md".to_owned()]
3263 );
3264
3265 store
3266 .apply_import_layer("import:okf/acme", &FactSet::new())
3267 .expect("withdraw");
3268 assert!(
3269 store.context_cache_keys().expect("keys").is_empty(),
3270 "a cached bundle for a deleted node is a stale answer waiting to be served"
3271 );
3272 }
3273
3274 /// A persisted import layer is re-applied after a `rebuild` wipes the graph,
3275 /// so imported facts survive a code-changing sync.
3276 #[test]
3277 fn imports_survive_rebuild() {
3278 let mut store = Store::open_in_memory().expect("open");
3279 let derived = FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a.rs"));
3280 store.rebuild(&derived, Some("tree1")).expect("rebuild");
3281
3282 // apply_import_layer applies to the live graph and persists in one step.
3283 let applied = store
3284 .apply_import_layer("import:graphify", &graphify_layer("file:a.rs"))
3285 .expect("apply import");
3286 assert_eq!(applied.edges_applied, 1);
3287 assert_eq!(applied.edges_pruned, 0);
3288 assert_eq!(store.import_refs().expect("refs"), vec!["import:graphify"]);
3289
3290 // Simulate a code-changing sync: the derived graph is rebuilt (still has
3291 // file:a.rs), which drops the imported doc + edge from the live graph.
3292 store.rebuild(&derived, Some("tree2")).expect("rebuild2");
3293 assert!(store.get_node("graphify:doc1").expect("get").is_none());
3294
3295 // Re-applying imports restores them; nothing is pruned (target present).
3296 let applied = store.reapply_imports().expect("reapply");
3297 assert_eq!(applied.layers, 1);
3298 assert_eq!(applied.nodes, 1);
3299 assert_eq!(applied.edges_applied, 1);
3300 assert_eq!(applied.edges_pruned, 0);
3301 assert!(store.get_node("graphify:doc1").expect("get").is_some());
3302 assert_eq!(store.edges_from("graphify:doc1").expect("edges").len(), 1);
3303 }
3304
3305 /// When a sync removes an edge's target (e.g. a deleted file), re-applying
3306 /// **prunes** that stale cross-reference from the persisted layer — it is not
3307 /// kept and retried forever. The import node itself is preserved.
3308 #[test]
3309 fn reapply_prunes_stale_cross_references() {
3310 let mut store = Store::open_in_memory().expect("open");
3311 let derived = FactSet::new().with_node(Node::new("file:gone.rs", NodeKind::File, "g"));
3312 store.rebuild(&derived, Some("t1")).expect("rebuild");
3313 store
3314 .apply_import_layer("import:graphify", &graphify_layer("file:gone.rs"))
3315 .expect("import");
3316
3317 // Code-changing sync: file:gone.rs is deleted from the derived graph.
3318 store
3319 .rebuild(&FactSet::new(), Some("t2"))
3320 .expect("rebuild2");
3321 let applied = store.reapply_imports().expect("reapply");
3322 assert_eq!(applied.nodes, 1);
3323 assert_eq!(applied.edges_applied, 0);
3324 assert_eq!(
3325 applied.edges_pruned, 1,
3326 "the edge to the deleted file is pruned"
3327 );
3328 assert!(store.get_node("graphify:doc1").expect("get").is_some());
3329
3330 // The prune is durable: a second reapply finds nothing left to prune,
3331 // proving the stale edge was removed from the persisted layer.
3332 let again = store.reapply_imports().expect("reapply2");
3333 assert_eq!(again.edges_applied, 0);
3334 assert_eq!(again.edges_pruned, 0, "already pruned; not retried");
3335 }
3336
3337 /// `apply_import_layer` validates on import: a dangling edge in the incoming
3338 /// layer is dropped and never persisted.
3339 #[test]
3340 fn apply_import_layer_prunes_on_import() {
3341 let mut store = Store::open_in_memory().expect("open");
3342 let present = || FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a"));
3343 store.rebuild(&present(), Some("t")).expect("rebuild");
3344
3345 let layer = graphify_layer("file:a.rs").with_edge({
3346 // Points at a file that does not exist → pruned on import.
3347 let mut e = Edge::inferred("graphify:doc1", "file:ghost.rs", EdgeKind::References, 0.9);
3348 e.src_ref = Some("import:graphify".to_owned());
3349 e
3350 });
3351 let applied = store
3352 .apply_import_layer("import:graphify", &layer)
3353 .expect("import");
3354 assert_eq!(applied.edges_applied, 1);
3355 assert_eq!(applied.edges_pruned, 1);
3356
3357 // A rebuild + reapply confirms only the valid edge was persisted.
3358 store.rebuild(&present(), Some("t2")).expect("rebuild2");
3359 let re = store.reapply_imports().expect("reapply");
3360 assert_eq!(re.edges_applied, 1);
3361 assert_eq!(re.edges_pruned, 0, "ghost edge was not persisted");
3362 }
3363
3364 #[test]
3365 fn legacy_import_layer_nodes_are_retagged_non_derived() {
3366 // A layer persisted before nodes carried provenance: its node objects have
3367 // no `provenance` field, so serde defaults them to Derived. On reapply the
3368 // store must repair them — a Graphify node to Inferred, a lat node to
3369 // Authored — so a later derived-only sync never mistakes them for derived.
3370 let mut store = Store::open_in_memory().expect("open");
3371 let legacy = r#"{"nodes":[
3372 {"key":"graphify:doc1","kind":"doc","name":"d","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null},
3373 {"key":"lat:lat.md/a.md","kind":"doc","name":"a","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null}
3374 ],"edges":[]}"#;
3375 store
3376 .conn
3377 .execute(
3378 "INSERT INTO imports (src_ref, facts) VALUES ('import:legacy', ?1)",
3379 [legacy],
3380 )
3381 .expect("seed legacy import row");
3382
3383 store.reapply_imports().expect("reapply");
3384
3385 let g = store
3386 .get_node("graphify:doc1")
3387 .expect("get")
3388 .expect("graphify node");
3389 assert_eq!(
3390 g.provenance,
3391 Provenance::Inferred,
3392 "graphify import node repaired to inferred"
3393 );
3394 let l = store
3395 .get_node("lat:lat.md/a.md")
3396 .expect("get")
3397 .expect("lat node");
3398 assert_eq!(
3399 l.provenance,
3400 Provenance::Authored,
3401 "lat import node repaired to authored"
3402 );
3403 }
3404
3405 /// `apply_import_layer` replaces the layer for a ref; `delete_import` removes.
3406 #[test]
3407 fn apply_import_replaces_and_delete_removes() {
3408 let mut store = Store::open_in_memory().expect("open");
3409 let a = FactSet::new().with_node(Node::new("graphify:x", NodeKind::Doc, "x"));
3410 let b = FactSet::new().with_node(Node::new("graphify:y", NodeKind::Doc, "y"));
3411 store.apply_import_layer("import:graphify", &a).expect("a");
3412 store.apply_import_layer("import:graphify", &b).expect("b");
3413 assert_eq!(
3414 store.import_refs().expect("refs").len(),
3415 1,
3416 "same ref replaced"
3417 );
3418 assert!(store.delete_import("import:graphify").expect("del"));
3419 assert!(store.import_refs().expect("refs").is_empty());
3420 assert!(!store.delete_import("import:graphify").expect("del again"));
3421 }
3422}