Skip to main content

macrame/temporal/
replay.rs

1use serde::{Deserialize, Serialize};
2use std::collections::{HashMap, HashSet};
3use std::path::{Path, PathBuf};
4
5use crate::error::{DbError, Result};
6use crate::temporal::as_of::NodeAttributes;
7
8/// One lineage's belief about one edge, at the instant a fold asked (§15.2).
9///
10/// # Why this is a struct and was a five-tuple until 0.14.5
11///
12/// The tuple had nowhere to put `branch_id`, and that was not a cosmetic
13/// shortfall: [D-216](../../docs/architecture/s13-decision-register.md) widened
14/// the four SQL folds to partition by `(table_name, entity_id, branch_id)` so
15/// two lineages' beliefs about one edge would stay two rows, and then the
16/// composition immediately downstream re-collapsed them, because `edge_key`
17/// composed `source|target|type|valid_from` and the map it fed had one slot per
18/// edge key. The widened partition was handing two rows to a container that
19/// could not hold two. That is [D-221](../../docs/architecture/s13-decision-register.md#d-221),
20/// and this type is its fix.
21///
22/// A struct rather than a six-tuple because the next field to arrive should be
23/// additive, which is why it is also `#[non_exhaustive]` — the same call
24/// [D-207](../../docs/architecture/s13-decision-register.md#d-207) made for
25/// `DbError`, one release earlier, for the same reason. Construct these by
26/// reading a [`MaterializedState`]; the crate is the only writer.
27///
28/// **Ordered by the tuple order of its fields**, so a `Vec<EdgeBelief>` sorts to
29/// a canonical form and two reconstructions of the same instant are *equal*
30/// rather than merely equivalent — a property the snapshot suite compares on.
31///
32/// # Constructing one
33///
34/// `#[non_exhaustive]` means no crate but this one may write the literal, and
35/// [`save_snapshot`](crate::temporal::save_snapshot) is public and takes a
36/// `MaterializedState` — so without a constructor the attribute would not make
37/// the next field additive, it would make a public function uncallable. Use
38/// [`EdgeBelief::new`], which takes the five fields that were the tuple and
39/// defaults the sixth to the trunk, with [`EdgeBelief::on_branch`] for the
40/// rest. That is [`EdgeAssertion::new`](crate::graph::EdgeAssertion::new)'s
41/// shape, deliberately: the two are the same fact travelling in opposite
42/// directions and should not need two idioms.
43#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
44#[non_exhaustive]
45pub struct EdgeBelief {
46    pub source_id: String,
47    pub target_id: String,
48    pub edge_type: String,
49    pub valid_from: String,
50    pub valid_to: String,
51    /// The lineage that holds this belief (0.14.5, D-221).
52    ///
53    /// `#[serde(default = "default_branch")]` so the field is additive at the
54    /// bincode level. It is belt and braces — the snapshot container refuses any
55    /// file whose format version is not this build's, and 0.14.5 bumps it
56    /// precisely so a state written without this field gets a named refusal
57    /// rather than a deserialisation error — but a default that is *right* costs
58    /// nothing and `'main'` is what every pre-v12 row actually carried.
59    #[serde(default = "default_branch")]
60    pub branch_id: String,
61}
62
63fn default_branch() -> String {
64    crate::schema::ddl::MAIN_BRANCH.to_string()
65}
66
67impl EdgeBelief {
68    /// A belief held by the trunk. Use [`Self::on_branch`] for any other.
69    ///
70    /// Five arguments rather than six because `main` is what every belief
71    /// written before 0.14.5 carried, so a caller porting a five-tuple wraps it
72    /// and is correct rather than being asked a question the old shape could
73    /// not have answered.
74    pub fn new(
75        source_id: impl Into<String>,
76        target_id: impl Into<String>,
77        edge_type: impl Into<String>,
78        valid_from: impl Into<String>,
79        valid_to: impl Into<String>,
80    ) -> Self {
81        Self {
82            source_id: source_id.into(),
83            target_id: target_id.into(),
84            edge_type: edge_type.into(),
85            valid_from: valid_from.into(),
86            valid_to: valid_to.into(),
87            branch_id: default_branch(),
88        }
89    }
90
91    /// The lineage holding this belief.
92    ///
93    /// Unchecked against `branches`, because this type is a value and not a
94    /// write: a `MaterializedState` naming a lineage the register has never
95    /// heard of is a snapshot that will disagree with the log, which
96    /// `verify_snapshot_chain` is there to report.
97    pub fn on_branch(mut self, branch_id: impl Into<String>) -> Self {
98        self.branch_id = branch_id.into();
99        self
100    }
101
102    /// The log `entity_id` this belief was folded under.
103    ///
104    /// Must match `trg_links_log_insert`'s
105    /// `source_id || '|' || target_id || '|' || edge_type || '|' || valid_from`
106    /// exactly, or a delta row will fail to replace the snapshot row it
107    /// supersedes. Safe because ULIDs are Crockford base32 and edge types are
108    /// `[A-Z0-9]+`, so `|` cannot occur inside a component (§4.3).
109    ///
110    /// **This is not a unique key across lineages** and must not be used as one
111    /// — see [`Self::belief_key`], which is.
112    pub fn entity_id(&self) -> String {
113        format!(
114            "{}|{}|{}|{}",
115            self.source_id, self.target_id, self.edge_type, self.valid_from
116        )
117    }
118
119    /// What identifies this belief: the edge key **and** the lineage holding it.
120    pub fn belief_key(&self) -> String {
121        format!("{}|{}", self.entity_id(), self.branch_id)
122    }
123}
124
125/// Full materialized state reconstructed from transaction_log replay (§5.5).
126#[derive(Debug, Clone, Serialize, Deserialize)]
127pub struct MaterializedState {
128    pub seq_anchor: i64,
129    pub timestamp: String,
130    pub concepts: HashMap<String, NodeAttributes>,
131    /// Every lineage's belief, each labelled with the lineage holding it.
132    ///
133    /// **Not resolved to one lineage's view**, and deliberately: `reconstruct`
134    /// asks a whole-ledger question — *what did the ledger hold at `ts`* — and
135    /// the ledger held both. Resolving here would require an ancestry, which
136    /// requires a connection this type does not have, and would silently answer
137    /// a narrower question than the one asked. A caller wanting one lineage's
138    /// view uses `graph::TraversalBuilder::on_branch` or
139    /// `temporal::query_as_of_edges_on`, which resolve against the register
140    /// ([D-220](../../docs/architecture/s13-decision-register.md#d-220)).
141    pub edges: Vec<EdgeBelief>,
142    /// **Nothing had been recorded yet at `timestamp`** (0.8.0, B5, D-121).
143    ///
144    /// An empty state has two meanings and a caller can act differently on
145    /// them. *Everything was retired by then* is a fact about the data;
146    /// *the ledger had not started* is a fact about the question. Both come
147    /// back as zero concepts and zero edges, so the difference has to be
148    /// carried rather than inferred.
149    ///
150    /// Set only when the log was verified **intact** — see
151    /// `hot_log_reach`. If rows had been archived away, `ts` below the hot
152    /// floor is not "before history", it is "the history is in the other file",
153    /// and that path raises instead of answering.
154    ///
155    /// `#[serde(default)]` so the field is additive: a snapshot written without
156    /// it deserialises with `false`, which is the right answer for any state
157    /// that had rows to fold. Old snapshots cannot actually reach this code —
158    /// the container carries `SCHEMA_VERSION` and v8 refused every v7 file
159    /// (D-043) — but the tolerance costs nothing and the next field to arrive
160    /// may not land in a release that bumps the schema.
161    #[serde(default)]
162    pub predates_recorded_history: bool,
163}
164
165impl MaterializedState {
166    /// The state before any log row has been applied.
167    fn empty(ts: &str) -> Self {
168        Self {
169            seq_anchor: 0,
170            timestamp: ts.to_string(),
171            concepts: HashMap::new(),
172            edges: Vec::new(),
173            predates_recorded_history: false,
174        }
175    }
176}
177
178/// The newest log payload shape this build writes and the highest it can read.
179///
180/// Kept beside the folds because they are the only readers, and bumped in step
181/// with the `json_object('v', …)` literals in `schema::ddl` — a test asserts the
182/// two agree, since nothing else would notice them drifting apart.
183pub(crate) const PAYLOAD_VERSION: u8 = 2;
184
185/// Every fold partitions on `(table_name, entity_id)`, never `entity_id` alone.
186///
187/// The two namespaces are not disjoint and nothing makes them so. A link's
188/// `entity_id` is the synthetic `source|target|type|valid_from`; a concept's is
189/// whatever the caller passed, unvalidated (defect AD). Partitioning on the id
190/// alone therefore lets a concept and a link contend for one window, and
191/// `ROW_NUMBER() = 1` hands the whole partition to whichever has the greater
192/// `seq_id` — so the loser vanishes from the reconstruction while sitting
193/// plainly in both `concepts` and `transaction_log`. Silent, and on the read
194/// path the ledger exists to make trustworthy.
195///
196/// Validating identifiers would make the collision unreachable and is the
197/// durable fix; this makes it harmless regardless, which is the property worth
198/// having at the fold. `table_name` leads the partition because the log is
199/// already indexed on `entity_id` and the discriminator is two values wide.
200const HOT_FOLD: &str = r#"
201    SELECT seq_id, table_name, entity_id, operation, payload, branch_id
202    FROM (
203        SELECT seq_id, table_name, entity_id, operation, payload, branch_id,
204               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id, branch_id ORDER BY seq_id DESC) as rn
205        FROM transaction_log
206        WHERE recorded_at <= ?1
207    ) WHERE rn = 1
208"#;
209
210/// Fold over hot and cold together (§5.5, D-026). Requires `cold` to be ATTACHed.
211///
212/// The hot entry wins for entities present in both files because its `seq_id` is
213/// greater — the same last-writer-wins rule as snapshot composition.
214fn cold_fold(cold_lineage: ColdLineage) -> String {
215    format!(
216        r#"
217    SELECT seq_id, table_name, entity_id, operation, payload, branch_id
218    FROM (
219        SELECT seq_id, table_name, entity_id, operation, payload, branch_id,
220               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id, branch_id ORDER BY seq_id DESC) as rn
221        FROM (
222            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at, branch_id FROM main.transaction_log
223            UNION ALL
224            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at, {cold} FROM cold.transaction_log
225        ) WHERE recorded_at <= ?1
226    ) WHERE rn = 1
227"#,
228        cold = cold_lineage.projection()
229    )
230}
231
232/// Fold over the hot log *above a snapshot anchor* (§5.5, D-049).
233///
234/// `seq_id > ?2` is an inequality, and deliberately so: `AUTOINCREMENT` leaves
235/// gaps whenever a transaction rolls back, so successor arithmetic
236/// (`seq_id = :anchor + 1`) would stop at the first gap and silently truncate
237/// the delta. This is the first anchored fold in the crate, which makes it the
238/// first code D-024's rule has ever bound — before this the rule was vacuous,
239/// not satisfied.
240const ANCHORED_HOT_FOLD: &str = r#"
241    SELECT seq_id, table_name, entity_id, operation, payload, branch_id
242    FROM (
243        SELECT seq_id, table_name, entity_id, operation, payload, branch_id,
244               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id, branch_id ORDER BY seq_id DESC) as rn
245        FROM transaction_log
246        WHERE recorded_at <= ?1 AND seq_id > ?2
247    ) WHERE rn = 1
248"#;
249
250/// Fold over hot **and cold** above a snapshot anchor (§5.5, 0.5.5).
251///
252/// The union is what lets composition survive an archive. Rows keep their
253/// `seq_id` when they move to cold — the cold schema declares a plain `INTEGER
254/// PRIMARY KEY` precisely so history is not renumbered — so `seq_id > ?2`
255/// partitions the two files consistently and last-writer-wins across them by the
256/// same rule the unanchored folds use.
257fn anchored_cold_fold(cold_lineage: ColdLineage) -> String {
258    format!(
259        r#"
260    SELECT seq_id, table_name, entity_id, operation, payload, branch_id
261    FROM (
262        SELECT seq_id, table_name, entity_id, operation, payload, branch_id,
263               ROW_NUMBER() OVER (PARTITION BY table_name, entity_id, branch_id ORDER BY seq_id DESC) as rn
264        FROM (
265            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at, branch_id FROM main.transaction_log
266            UNION ALL
267            SELECT seq_id, table_name, entity_id, operation, payload, recorded_at, {cold} FROM cold.transaction_log
268        ) WHERE recorded_at <= ?1 AND seq_id > ?2
269    ) WHERE rn = 1
270"#,
271        cold = cold_lineage.projection()
272    )
273}
274
275/// Whether an attached cold file predates the lineage column (§15.2, v12).
276///
277/// Cold files are **read-only media as far as the read path is concerned**.
278/// They get moved (D-026), they can sit on a share, and a fold that upgraded
279/// one in order to read it would be a write on a path callers have every reason
280/// to believe is a read. So the shape is detected and tolerated, never
281/// corrected: the archive *writer* upgrades, and only inside its own
282/// transaction.
283///
284/// Detection is column presence rather than a version stamp, because a cold
285/// file carries no version anyone can trust — it is a file that has been moved.
286#[derive(Debug, Clone, Copy, PartialEq, Eq)]
287enum ColdLineage {
288    /// v12 or later: the file stamps its own rows.
289    Stamped,
290    /// Pre-v12: every row in it was written when only the trunk existed.
291    PreV12,
292}
293
294impl ColdLineage {
295    fn projection(self) -> &'static str {
296        match self {
297            ColdLineage::Stamped => "branch_id",
298            // A literal, not a default: rows written before lineage existed
299            // *were* trunk rows, and saying so is a fact about them rather than
300            // a fallback.
301            ColdLineage::PreV12 => "'main' AS branch_id",
302        }
303    }
304}
305
306/// Ask the attached cold file whether it carries `transaction_log.branch_id`.
307///
308/// Returns [`ColdLineage::PreV12`] when the pragma cannot be read at all. That
309/// is the conservative direction: a fold that guesses "stamped" against a v11
310/// file fails with `no such column`, while a fold that guesses "pre-v12"
311/// against a v12 file reads rows it can still fold — it would mislabel a
312/// branch's rows as trunk, which is why the guess is never made when the pragma
313/// answers.
314async fn cold_lineage(conn: &libsql::Connection) -> ColdLineage {
315    let Ok(mut rows) = conn
316        .query("PRAGMA cold.table_info(transaction_log)", ())
317        .await
318    else {
319        return ColdLineage::PreV12;
320    };
321    while let Ok(Some(row)) = rows.next().await {
322        if row.get::<String>(1).is_ok_and(|name| name == "branch_id") {
323            return ColdLineage::Stamped;
324        }
325    }
326    ColdLineage::PreV12
327}
328
329/// The winning log rows for one fold, before they are applied to a base state.
330///
331/// Absence and disappearance are different facts, and a merge is where the
332/// difference starts to matter. A full fold from nothing can treat "this entity
333/// went away" and "there is no row for it" identically — both end as absence.
334/// Composed onto a snapshot they are opposites: a disappearance must *remove*
335/// the entity the snapshot carries, and skipping it leaves the snapshot's stale
336/// row standing as though nothing had happened. So they are collected rather
337/// than dropped, and the full fold applies them to an empty base, which keeps
338/// one code path for both cases (D-049).
339///
340/// **There is one such set, not two (D-072).** It used to carry `edges_gone`
341/// beside `concepts_gone`, and both were populated only from the `'D'` branch of
342/// [`fold_delta`] — so when that branch became an error, `edges_gone` was left
343/// reachable by nothing. Closing one unreachable path by opening another is not
344/// a fix, so it went too.
345///
346/// The asymmetry is real and worth stating, because "concepts can vanish and
347/// edges cannot" looks like an oversight until you follow it:
348///
349/// * A **concept** disappears by being *retired*, which writes a `'U'` row whose
350///   payload has `retired = 1`. That is a genuine removal from a composed state
351///   and `concepts_gone` carries it.
352/// * An **edge** never disappears. It is retired by asserting a successor over
353///   the same interval key — same `source|target|type|valid_from`, later
354///   `recorded_at` — so the log row is an `'I'` under the *same* `entity_id`, and
355///   last-writer-wins in [`Self::apply_to`] replaces the tuple in place. There is
356///   nothing to remove because nothing left; the interval simply closed.
357///
358/// That is Doctrine III showing through: an edge assertion is immutable and
359/// superseded, never deleted.
360#[derive(Default)]
361struct Delta {
362    concepts: HashMap<String, NodeAttributes>,
363    /// Keyed by `entity_id` **and** `branch_id` — see [`EdgeBelief::belief_key`].
364    ///
365    /// `entity_id` alone was the collapse [D-221](../../docs/architecture/s13-decision-register.md#d-221)
366    /// records: it is the edge key, shared across lineages by design, so an
367    /// ancestor's assertion and a descendant's correction landed in one slot.
368    edges: HashMap<String, EdgeBelief>,
369    /// Concepts retired as of the fold's instant. See the type's note for why
370    /// there is no edge equivalent.
371    concepts_gone: HashSet<String>,
372    max_seq: i64,
373}
374
375/// Release a `cold` handle left attached by an earlier call (§5.5, D-044).
376///
377/// Both ATTACH sites pair with an unconditional DETACH on the way out, so in
378/// the normal course this finds nothing and the statement fails harmlessly with
379/// "no such database: cold". It exists for the case the pairing cannot cover: a
380/// panic unwinding between the two, which skips the DETACH no matter which exit
381/// path the `Result` would have taken.
382///
383/// A `Drop` guard is the reflex here and does not work — `execute` is `async`,
384/// and a `Drop` impl cannot await, so it would build a future, discard it, and
385/// leave the handle attached while looking like it had cleaned up. Recovering
386/// on the way *in* needs no destructor, works regardless of how the handle
387/// leaked, and turns permanent poisoning of the connection into one failed
388/// statement nobody sees.
389pub(crate) async fn detach_stale_cold(conn: &libsql::Connection) {
390    let _ = conn.execute("DETACH DATABASE cold", ()).await;
391}
392
393/// Reconstruct database state as believed at past instant `ts` using window-function log fold (§5.5, D-026).
394///
395/// When `ts` predates the hot log's horizon the cold database is ATTACHed for
396/// exactly one fold and DETACHed unconditionally on the way out, error paths
397/// included. ATTACH is not transactional and survives ROLLBACK, so a handle
398/// leaked by an early return would make every later `reconstruct` *and* every
399/// later `archive` fail with "database cold is already in use" — one corrupt
400/// payload would permanently poison the connection. This is the same failure
401/// mode `archive()` carries a note about, and the two now share a shape.
402/// Snapshot composition (§5.5, D-049) applies when `snapshots_dir` holds a
403/// snapshot at or before `ts` and no archive database exists — see
404/// `snapshot_anchor` for why archiving disables it. Otherwise the fold runs
405/// from genesis, which is correct and costs what the whole log costs.
406pub async fn reconstruct(
407    conn: &libsql::Connection,
408    ts: &str,
409    archive_path: Option<&Path>,
410    snapshots_dir: Option<&Path>,
411) -> Result<MaterializedState> {
412    match hot_log_reach(conn, ts, archive_path).await? {
413        HotLogReach::Covers => {
414            if let Some(base) = snapshot_anchor(snapshots_dir, ts).await {
415                let anchor = base.seq_anchor;
416                let delta =
417                    fold_delta(conn, ANCHORED_HOT_FOLD, libsql::params![ts, anchor]).await?;
418                return Ok(delta.apply_to(base, ts));
419            }
420            return fold(conn, ts, HOT_FOLD).await;
421        }
422        HotLogReach::PredatesRecordedHistory => {
423            // Nothing had been recorded by `ts`, and nothing has been removed
424            // from the log, so there is no history anywhere to go looking for.
425            // The empty state is the answer, flagged so a caller can tell it
426            // from a state that is empty because everything was retired.
427            let mut state = MaterializedState::empty(ts);
428            state.predates_recorded_history = true;
429            return Ok(state);
430        }
431        HotLogReach::NeedsArchive => {}
432    }
433
434    // The delta lives in the cold archive database. Both ways of failing to
435    // reach it carry `archive_hint`, which is the message the rejected hot-side
436    // marker was wanted for — see that function for why no marker is needed.
437    //
438    // **Computed inside the error arms, not before them.** `NeedsArchive` is the
439    // ordinary path to a cold fold and usually succeeds; an eager hint would put
440    // an extra query on it for a string almost every caller discards. An
441    // injection probe caught this — `a_failed_cold_reconstruct_still_detaches`
442    // reached the hint on a run that raised nothing from here.
443    let archive = match archive_path {
444        Some(p) => p,
445        None => {
446            return Err(DbError::ReplayCorrupt {
447                seq: 0,
448                reason: format!(
449                    "state at {ts} predates the hot log and no archive path was given; {}",
450                    archive_hint(conn).await
451                ),
452            })
453        }
454    };
455    if !archive.exists() {
456        return Err(DbError::ReplayCorrupt {
457            seq: 0,
458            reason: format!(
459                "archive database file {archive:?} does not exist; {}",
460                archive_hint(conn).await
461            ),
462        });
463    }
464
465    detach_stale_cold(conn).await;
466
467    // Bound, not interpolated: a path is caller data, and hand-rolled quote
468    // doubling is a worse version of what the driver already does correctly.
469    conn.execute(
470        "ATTACH DATABASE ?1 AS cold",
471        libsql::params![archive.to_string_lossy().as_ref()],
472    )
473    .await?;
474
475    // Asked once, after the ATTACH and before either fold, because both arms
476    // need it and the answer cannot change while we hold the handle.
477    let cold_shape = cold_lineage(conn).await;
478
479    // Composition works across the archive boundary because the anchored fold
480    // unions both files; before 0.5.5 it was refused here rather than made to
481    // work, and the refusal was the only thing keeping the answer right.
482    let result = match snapshot_anchor(snapshots_dir, ts).await {
483        Some(base) => {
484            let anchor = base.seq_anchor;
485            fold_delta(
486                conn,
487                &anchored_cold_fold(cold_shape),
488                libsql::params![ts, anchor],
489            )
490            .await
491            .map(|delta| delta.apply_to(base, ts))
492        }
493        None => fold(conn, ts, &cold_fold(cold_shape)).await,
494    };
495
496    // Unconditional: see the ATTACH note above.
497    if let Err(e) = conn.execute("DETACH DATABASE cold", ()).await {
498        tracing::warn!("reconstruct: failed to DETACH cold database: {e}");
499    }
500
501    result
502}
503
504/// Fold from genesis and compare against the composed answer (§5.5, T5.3,
505/// D-092).
506///
507/// # The problem this exists for
508///
509/// [`crate::temporal::save_snapshot`] is written by `write_final`, which calls
510/// [`reconstruct`] — and `reconstruct` composes onto the *previous* snapshot
511/// whenever one is usable. So snapshot *n* is derived from snapshot *n−1*, and
512/// there is no periodic full fold anywhere in the chain. An error introduced at
513/// any link is copied forward indefinitely, and every subsequent read agrees
514/// with it, because they are all reading the same descendant.
515///
516/// The project's own open item names the difficulty honestly: a full fold is
517/// exactly the cost snapshots exist to avoid, so this cannot run on every read.
518/// It is a **scheduling** problem, and this function is the thing to schedule.
519///
520/// # It reports; it does not repair
521///
522/// Deliberate, and not merely conservative. Under [Doctrine VI] a snapshot is
523/// derivative and disposable, so the repair is *delete the snapshots* — one
524/// line, available to the caller, and correct without this function's help.
525/// What the caller cannot get for themselves is the knowledge that the chain
526/// diverged, and silently rewriting the file would destroy the only evidence of
527/// a bug in composition. A divergence here is not a corrupt database; it is a
528/// wrong **cache**, and it means composition has a defect worth finding.
529///
530/// # Cost
531///
532/// One fold from genesis over the whole log, plus one composed reconstruction.
533/// That is the expensive path by construction — see [`crate::Database::
534/// verify_snapshot_chain`] for the handle-level entry point and the note on
535/// when to run it.
536///
537/// [Doctrine VI]: ../../../docs/architecture/s0-s3-foundations.md#doctrine-vi
538pub async fn verify_snapshot_chain(
539    conn: &libsql::Connection,
540    ts: &str,
541    archive_path: Option<&Path>,
542    snapshots_dir: &Path,
543) -> Result<ChainCheck> {
544    // The composed answer: what every reader gets today.
545    let composed = reconstruct(conn, ts, archive_path, Some(snapshots_dir)).await?;
546    // The authority: the same instant, with the snapshot directory withheld, so
547    // `snapshot_anchor` finds nothing and the fold runs from genesis. Passing
548    // `None` is what makes this an independent computation rather than a second
549    // call to the thing under test.
550    let folded = reconstruct(conn, ts, archive_path, None).await?;
551    Ok(ChainCheck::compare(ts, &composed, &folded))
552}
553
554/// The result of a [`verify_snapshot_chain`] cross-check.
555///
556/// Carries the disagreements rather than a bool, because "the chain diverged" is
557/// not actionable and "these three concepts differ, and this edge is present in
558/// one and not the other" is. Bounded — see [`ChainCheck::SAMPLE_LIMIT`] — since
559/// a chain that went wrong early can disagree about every row, and a report that
560/// is the size of the database is one nobody reads.
561#[derive(Debug, Clone)]
562pub struct ChainCheck {
563    pub timestamp: String,
564    /// `seq_anchor` of the composed answer and of the genesis fold. These
565    /// **may legitimately differ**: the composed answer anchors at the snapshot
566    /// it started from plus its delta, and the fold anchors at the newest row it
567    /// saw. Reported for diagnosis, never compared.
568    pub composed_anchor: i64,
569    pub folded_anchor: i64,
570    pub composed_concepts: usize,
571    pub folded_concepts: usize,
572    pub composed_edges: usize,
573    pub folded_edges: usize,
574    /// Concept ids present in one and not the other, or whose attributes differ.
575    pub concept_disagreements: Vec<String>,
576    /// Edge keys present in one and not the other.
577    pub edge_disagreements: Vec<String>,
578    /// True when either list was truncated at [`ChainCheck::SAMPLE_LIMIT`].
579    pub truncated: bool,
580}
581
582impl ChainCheck {
583    /// How many disagreements of each kind to carry.
584    pub const SAMPLE_LIMIT: usize = 32;
585
586    pub fn diverged(&self) -> bool {
587        !self.concept_disagreements.is_empty() || !self.edge_disagreements.is_empty()
588    }
589
590    fn compare(ts: &str, composed: &MaterializedState, folded: &MaterializedState) -> Self {
591        let mut concept_disagreements = Vec::new();
592        let mut truncated = false;
593
594        let mut ids: Vec<&String> = composed.concepts.keys().collect();
595        ids.extend(folded.concepts.keys());
596        ids.sort_unstable();
597        ids.dedup();
598        for id in ids {
599            let a = composed.concepts.get(id);
600            let b = folded.concepts.get(id);
601            let same = match (a, b) {
602                (Some(a), Some(b)) => {
603                    a.title == b.title
604                        && a.content == b.content
605                        && a.embedding_model == b.embedding_model
606                }
607                (None, None) => true,
608                _ => false,
609            };
610            if !same {
611                if concept_disagreements.len() < Self::SAMPLE_LIMIT {
612                    concept_disagreements.push(id.clone());
613                } else {
614                    truncated = true;
615                }
616            }
617        }
618
619        // Edges are a `Vec` of tuples with no declared order, so the comparison
620        // is on the set. Comparing the vectors directly would report a
621        // divergence for a reordering, which is not one — and that false
622        // positive is worse than useless here, because the whole point of this
623        // check is that a report means "go and find the bug".
624        // `valid_to` is in the key as well as the identity, because a
625        // divergence in *what* the two paths believe is exactly what this
626        // reports — two rows agreeing on the edge and the lineage and
627        // disagreeing on the interval are a disagreement, not one row.
628        let key = |e: &EdgeBelief| format!("{}|{}", e.belief_key(), e.valid_to);
629        let ca: HashSet<String> = composed.edges.iter().map(key).collect();
630        let fa: HashSet<String> = folded.edges.iter().map(key).collect();
631        let mut edge_disagreements: Vec<String> = ca.symmetric_difference(&fa).cloned().collect();
632        edge_disagreements.sort_unstable();
633        if edge_disagreements.len() > Self::SAMPLE_LIMIT {
634            edge_disagreements.truncate(Self::SAMPLE_LIMIT);
635            truncated = true;
636        }
637
638        Self {
639            timestamp: ts.to_string(),
640            composed_anchor: composed.seq_anchor,
641            folded_anchor: folded.seq_anchor,
642            composed_concepts: composed.concepts.len(),
643            folded_concepts: folded.concepts.len(),
644            composed_edges: ca.len(),
645            folded_edges: fa.len(),
646            concept_disagreements,
647            edge_disagreements,
648            truncated,
649        }
650    }
651}
652
653impl std::fmt::Display for ChainCheck {
654    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
655        if !self.diverged() {
656            return write!(
657                f,
658                "snapshot chain agrees with a genesis fold at {}: {} concepts, {} edges",
659                self.timestamp, self.folded_concepts, self.folded_edges
660            );
661        }
662        write!(
663            f,
664            "snapshot chain DIVERGED at {}: composed {} concepts / {} edges, \
665             genesis fold {} concepts / {} edges; {} concept and {} edge \
666             disagreements{}. The snapshots are a wrong cache, not a corrupt \
667             ledger — deleting the snapshot directory restores correctness and \
668             loses only speed (Doctrine VI). concepts: {:?} edges: {:?}",
669            self.timestamp,
670            self.composed_concepts,
671            self.composed_edges,
672            self.folded_concepts,
673            self.folded_edges,
674            self.concept_disagreements.len(),
675            self.edge_disagreements.len(),
676            if self.truncated { " (truncated)" } else { "" },
677            self.concept_disagreements,
678            self.edge_disagreements,
679        )
680    }
681}
682
683/// The newest usable snapshot at or before `ts`, or `None` to fold from genesis.
684///
685/// **Composition used to be disabled once an archive database existed, and as of
686/// 0.5.5 it is not.** The reason for the refusal was real: `LOG_ARCHIVABLE`
687/// (§5.7) removes superseded rows scattered through the sequence, so a row above
688/// the anchor and at or before `ts` could be in cold while a newer row for the
689/// same entity — recorded *after* `ts`, invisible to the fold — kept it out of
690/// the hot log. The delta missed it and the snapshot answered with a stale
691/// value. The fix is the one that note named: the cold log is now in the delta,
692/// via [`ANCHORED_COLD_FOLD`], so the archived row is visible again and there is
693/// nothing left to refuse.
694///
695/// Selection loads candidates newest-first and stops at the first whose
696/// timestamp is at or before `ts`, so the common case — `reconstruct(now)` —
697/// reads exactly one file. A snapshot this build cannot read
698/// ([`DbError::SnapshotIncompatible`], D-043) is skipped, not raised: an
699/// incompatible snapshot is an ordinary consequence of upgrading, and the whole
700/// point of distinguishing it from corruption is that the answer is to carry on
701/// without it.
702///
703/// # It runs on a blocking thread, and a lost one costs speed only (0.13.11, W8.1, D-184)
704///
705/// The scan is a directory listing plus one or more full
706/// [`load_snapshot`](super::snapshot::load_snapshot) calls — decompression and
707/// bincode over the whole state, on a worker that has other tasks waiting. The
708/// *whole scan* is offloaded rather than each file, because the loop is
709/// sequential by construction (it stops at the first usable file) and a hop per
710/// candidate would add scheduling to a path whose common case reads exactly one.
711///
712/// A [`tokio::task::JoinError`] means the loader panicked, and the answer is the
713/// same one this function already gives for every other kind of unusable file:
714/// `None`, and fold from genesis. That is not leniency, it is what a snapshot
715/// *is* — derivative and disposable under [Doctrine VI], so the cost of ignoring
716/// one is a slower reconstruction and never a wrong one. It is also a real
717/// improvement over the previous arrangement: inline, a panic in the loader
718/// unwound through [`reconstruct`] and took the caller's task with it, which
719/// meant a single corrupt file could stop a process that had a correct answer
720/// available the whole time. W8.4 fuzzes for exactly those panics; this is what
721/// happens to the ones it has not found yet.
722///
723/// [Doctrine VI]: ../../../docs/architecture/s0-s3-foundations.md#doctrine-vi
724async fn snapshot_anchor(snapshots_dir: Option<&Path>, ts: &str) -> Option<MaterializedState> {
725    let dir = snapshots_dir?.to_path_buf();
726    let ts = ts.to_string();
727    match tokio::task::spawn_blocking(move || newest_usable_snapshot(&dir, &ts)).await {
728        Ok(found) => found,
729        Err(e) => {
730            tracing::warn!("the snapshot scan did not finish ({e}); folding from genesis");
731            None
732        }
733    }
734}
735
736/// The blocking half of [`snapshot_anchor`]: read the directory, load
737/// newest-first, stop at the first snapshot at or before `ts`.
738fn newest_usable_snapshot(dir: &Path, ts: &str) -> Option<MaterializedState> {
739    let mut candidates: Vec<(i64, PathBuf)> = std::fs::read_dir(dir)
740        .ok()?
741        .flatten()
742        .map(|e| e.path())
743        .filter_map(|p| super::snapshot::seq_from_filename(&p).map(|s| (s, p)))
744        .collect();
745    candidates.sort_by_key(|(seq, _)| std::cmp::Reverse(*seq));
746
747    for (_, path) in candidates {
748        match super::snapshot::load_snapshot(&path) {
749            // Sound as a string comparison because every timestamp is the
750            // canonical fixed width (D-029).
751            Ok(state) if state.timestamp.as_str() <= ts => return Some(state),
752            Ok(_) => continue,
753            Err(DbError::SnapshotIncompatible { reason, .. }) => {
754                tracing::warn!("skipping snapshot {path:?}: {reason}");
755                continue;
756            }
757            Err(e) => {
758                tracing::warn!("skipping unreadable snapshot {path:?}: {e}");
759                continue;
760            }
761        }
762    }
763    None
764}
765
766/// Where the answer for `ts` lives.
767///
768/// Three cases, not two (0.8.0, B5, D-121). This used to be a `bool`, and the
769/// missing third case is the whole of B5: *below the log's floor* was folded in
770/// with *the delta is elsewhere*, so a question about a time before the ledger
771/// started came back as [`DbError::ReplayCorrupt`] — the class meaning the
772/// ledger is damaged — naming an archive file the caller had never created.
773enum HotLogReach {
774    /// The hot log holds everything needed at `ts`. Fold it.
775    Covers,
776    /// Nothing had been recorded by `ts`, and nothing has ever been removed
777    /// from the log, so no other file could hold it either. The empty state is
778    /// the correct answer, not a failure to find one.
779    PredatesRecordedHistory,
780    /// The delta is in the cold archive. If it cannot be reached, that is an
781    /// error and stays one.
782    NeedsArchive,
783}
784
785/// Whether the hot log alone can answer for `ts` — a *completeness* test.
786///
787/// **This replaces a reach test that was not one (0.5.5).** The previous version
788/// asked `MIN(recorded_at) <= ts`: whether the hot log stretches back far enough
789/// to contain `ts`. That is a different question from whether it still contains
790/// everything needed to answer at `ts`, and `LOG_ARCHIVABLE` (§5.7) is exactly
791/// what pulls the two apart — it removes *superseded* rows, scattered through
792/// the sequence rather than forming a prefix. One entity archived and another
793/// not is enough: the unarchived one keeps `MIN` pointing before the cutoff
794/// while the archived one's winning row is gone, and the fold silently returns a
795/// state missing an entity. Measured, not theorised — see
796/// `reconstructing_before_the_archive_cutoff_keeps_every_entity`.
797///
798/// The sound test rests on the one guarantee the archive does make: **the newest
799/// row per entity is never archivable**, because archivability requires a later
800/// row to exist. So if `ts` is at or after the newest hot stamp, every entity's
801/// winning row at `ts` is its newest row overall, and every such row is hot.
802/// That covers `reconstruct(now)` — the common case, and the case §5.7 designed
803/// `LOG_ARCHIVABLE` around — and nothing else.
804///
805/// Anything earlier goes to the cold file. That is more ATTACHes than the old
806/// rule performed, and the trade is not close: the old rule was cheaper because
807/// it was answering a question nobody asked.
808///
809/// With no archive database in play the reach test *is* the completeness test —
810/// nothing has been removed, so the hot log is the whole log — and it is kept,
811/// because it is also what distinguishes "before recorded history" from "the
812/// cold file is missing" (D-026).
813async fn hot_log_reach(
814    conn: &libsql::Connection,
815    ts: &str,
816    archive_path: Option<&Path>,
817) -> Result<HotLogReach> {
818    let row = conn
819        .query(
820            "SELECT MIN(recorded_at), MAX(recorded_at) FROM transaction_log",
821            (),
822        )
823        .await?
824        .next()
825        .await?;
826    let (min_recorded_at, max_recorded_at): (Option<String>, Option<String>) = match row {
827        Some(r) => (r.get(0).ok(), r.get(1).ok()),
828        None => (None, None),
829    };
830
831    // Sound as string comparisons because every recorded_at is the canonical
832    // fixed width (D-029).
833    if archive_path.is_some_and(|p| p.exists()) {
834        // An empty hot log beside an archive is the fully-archived case and
835        // covers nothing. It cannot arise from `archive()` itself — the newest
836        // row per entity always stays — but answering "covered" here would make
837        // such a file reconstruct to the empty state with no error at all.
838        return Ok(match max_recorded_at {
839            Some(max_ts) if max_ts.as_str() <= ts => HotLogReach::Covers,
840            _ => HotLogReach::NeedsArchive,
841        });
842    }
843
844    match min_recorded_at {
845        Some(min_ts) if min_ts.as_str() <= ts => Ok(HotLogReach::Covers),
846        // No log at all: a genuinely empty database, and the empty state has
847        // always been the answer here.
848        None => Ok(HotLogReach::PredatesRecordedHistory),
849        // `ts` is below the hot log's floor, and there is no archive file to
850        // consult. Which of the two meanings that has is decided by whether
851        // anything was ever removed from the log — see `hot_log_is_intact`.
852        Some(_) => Ok(if hot_log_is_intact(conn).await? {
853            HotLogReach::PredatesRecordedHistory
854        } else {
855            HotLogReach::NeedsArchive
856        }),
857    }
858}
859
860/// What the caller needs to know when the cold delta cannot be reached —
861/// **assembled from the hot file alone** (0.9.0, C4).
862///
863/// # This is the message the hot-side marker was wanted for
864///
865/// [D-121](../../docs/architecture/s13-decision-register.md) rejected a hot-side
866/// marker recording *archived at* and *horizon*, then left the door open: 0.9.0
867/// was to adopt it "only if it wants the richer message". C4 asked for the
868/// message and found the marker cannot supply it, because the proposed message —
869/// *"this database was archived on X; pass the archive path"* — is **weaker**
870/// than what the hot log already carries:
871///
872/// * *how many rows went* is `MAX(seq_id) - COUNT(*)`, exact for the reason
873///   [`hot_log_is_intact`] gives;
874/// * *how far back the hot file still reaches* is `MIN(seq_id)` and its
875///   `recorded_at` — which is the fact that actually tells a caller whether the
876///   archive is worth fetching, and which a marker's archive **timestamp** does
877///   not give them;
878/// * *that archiving happened at all* is the one bit [`hot_log_is_intact`]
879///   already answers.
880///
881/// The only datum a marker would add is the wall-clock instant of the last
882/// archive run, and no branch and no caller needs it. So the marker is refused
883/// outright rather than deferred again: under
884/// [D-036](../../docs/architecture/s13-decision-register.md) a hot-table addition
885/// lands pre-1.0 or not at all, and a table whose whole content is a timestamp
886/// used in one error string is not worth a rung.
887///
888/// # There is no "nothing was archived" case, and that was settled by injection
889///
890/// This first carried a branch for `removed == 0`, on the reasoning that the
891/// `NeedsArchive` arm is reachable without any archiving. That reasoning was
892/// **wrong about where the cost lands and right about the branch**, and only a
893/// probe told the two apart: replacing the branch body with a panic showed it
894/// firing from `a_failed_cold_reconstruct_still_detaches`, a test that raises
895/// nothing from here — because the hint was being computed *before* the two
896/// arms that use it, on every cold fold. Made lazy, the probe went quiet across
897/// all 27 targets.
898///
899/// So the branch was dead at the use sites: both arms require
900/// [`hot_log_is_intact`] to have returned false, or an archive file to have
901/// existed when `hot_log_reach` looked and to have gone by the time this did.
902/// Rows really were removed in every case that gets here, and the message may
903/// say so without qualification. Deleted rather than kept as a defensive
904/// fallback, for the reason `delete_guarded` records about
905/// `classify_archive_violation`: unreachable code that looks reasonable is
906/// harder to remove later than now.
907///
908/// Best-effort by construction: this runs on the error path, where a second
909/// failure must not replace the diagnosis with its own. A query that does not
910/// answer yields a hint that says so, and the caller still gets the error it came
911/// for.
912async fn archive_hint(conn: &libsql::Connection) -> String {
913    // `COUNT(*)` always returns a row, so `None` here means the query itself
914    // failed and there is nothing to say beyond that.
915    let row = match conn
916        .query(
917            "SELECT COUNT(*), MIN(seq_id), MAX(seq_id), MIN(recorded_at) FROM transaction_log",
918            (),
919        )
920        .await
921    {
922        Ok(mut rows) => rows.next().await.ok().flatten(),
923        Err(_) => None,
924    };
925
926    let Some(row) = row else {
927        return "the hot log could not be inspected for an archive horizon".into();
928    };
929    let count: i64 = row.get(0).unwrap_or(0);
930    if count == 0 {
931        return "the hot log is empty".into();
932    }
933    let min: i64 = row.get(1).unwrap_or(0);
934    let max: i64 = row.get(2).unwrap_or(0);
935    let floor: String = row.get(3).unwrap_or_default();
936    let removed = max - count;
937
938    format!(
939        "{removed} log rows have been archived out of this database; the hot log \
940         now begins at seq_id {min} ({floor})"
941    )
942}
943
944/// Was any row ever removed from `transaction_log`? — answered exactly, from
945/// the hot file alone (0.8.0, B5, D-121).
946///
947/// # Why this question needs answering at all
948///
949/// With `ts` below the hot log's floor and no archive file present, the state
950/// on disk is consistent with two very different histories: **nothing was ever
951/// archived**, in which case the hot log is the whole log and the answer to
952/// *what was believed at `ts`* is "nothing yet"; or **rows were archived and
953/// the cold file is gone**, in which case the answer is unknowable and saying
954/// "nothing" would be inventing one. Before this, the two were conflated and
955/// both raised — which made an ordinary question about a young database report
956/// the ledger as damaged.
957///
958/// # Why `seq_id` settles it, with no marker and no schema change
959///
960/// `transaction_log.seq_id` is `INTEGER PRIMARY KEY AUTOINCREMENT`, so values
961/// are allocated 1, 2, 3, … and **never reused**. A rolled-back transaction
962/// leaves no gap — `sqlite_sequence` rolls back with it, which
963/// [D-049](../../docs/architecture/s13-decision-register.md) established by
964/// measurement after assuming the opposite. So the only thing that can perturb
965/// the sequence is deletion, and `trg_txlog_guard_delete` confines deletion to
966/// an archive session.
967///
968/// Therefore: if nothing was removed, the ids are exactly `1..=MAX` and
969/// `COUNT(*) == MAX(seq_id)` with `MIN(seq_id) == 1`. And conversely — this is
970/// the half that makes it a proof rather than a heuristic — those two equalities
971/// force the set of `COUNT` distinct ids inside `[1, MAX]` to be all of it, so
972/// nothing is missing. The test is exact in both directions, not merely
973/// suggestive.
974///
975/// **It does not depend on the archive removing a contiguous block**, which it
976/// does not: `archive()` removes *superseded* rows scattered through the
977/// sequence. Scattered removal leaves interior gaps, which fails the count
978/// equality; removal from the front raises `MIN` above 1. Removal from the end
979/// cannot happen, because the newest row per entity is never archivable.
980///
981/// # What it deliberately does not claim
982///
983/// Nothing about *when* the archiving happened or *what* went, which is what
984/// the rejected hot-side marker would have carried. It answers one bit, and one
985/// bit is what the branch above needs.
986async fn hot_log_is_intact(conn: &libsql::Connection) -> Result<bool> {
987    let row = conn
988        .query(
989            "SELECT COUNT(*), MIN(seq_id), MAX(seq_id) FROM transaction_log",
990            (),
991        )
992        .await?
993        .next()
994        .await?;
995    let Some(row) = row else {
996        return Ok(true);
997    };
998    let count: i64 = row.get(0).unwrap_or(0);
999    if count == 0 {
1000        return Ok(true);
1001    }
1002    let min: i64 = row.get(1).unwrap_or(0);
1003    let max: i64 = row.get(2).unwrap_or(0);
1004    Ok(min == 1 && count == max)
1005}
1006
1007/// Whether a connection alone can fold `transaction_log` at `ts` (W7.1, D-174).
1008///
1009/// The completeness question [`hot_log_reach`] answers, minus the archive file
1010/// it does not have. Both callers take a `Connection`, so when the hot log is
1011/// short they have nowhere to go and must refuse rather than fold what is left:
1012/// [`crate::graph::TraversalBuilder::as_of_recorded`] folds for topology, and
1013/// [`crate::temporal::hydrate_attributes`] folds for the text (0.13.16, W9.1,
1014/// [D-189](../../docs/architecture/s13-decision-register.md#d-189)). The second
1015/// was folding without asking, which is what §3.2 was.
1016///
1017/// **One bit, and the conservative one.** `hot_log_is_intact` says whether
1018/// anything was ever removed, not whether *this* instant survived the removal.
1019/// The archive cutoff is not recorded hot-side — that is the marker D-132
1020/// refused — so an archived database refuses every instant here, including ones
1021/// a fold would have got right. `ts` is taken anyway rather than dropped from the
1022/// signature, because the refusal names it and because a cutoff-aware version
1023/// would need it.
1024pub(crate) async fn hot_log_answers_for(conn: &libsql::Connection, _ts: &str) -> Result<bool> {
1025    hot_log_is_intact(conn).await
1026}
1027
1028/// Run one fold query from nothing — the unanchored path.
1029async fn fold(conn: &libsql::Connection, ts: &str, query: &str) -> Result<MaterializedState> {
1030    let delta = fold_delta(conn, query, libsql::params![ts]).await?;
1031    Ok(delta.apply_to(MaterializedState::empty(ts), ts))
1032}
1033
1034/// Run one fold query and collect the winning rows, deletions included.
1035async fn fold_delta(
1036    conn: &libsql::Connection,
1037    query: &str,
1038    params: impl libsql::params::IntoParams,
1039) -> Result<Delta> {
1040    let mut rows = conn.query(query, params).await?;
1041    let mut d = Delta::default();
1042    let (concepts, edges, max_seq) = (&mut d.concepts, &mut d.edges, &mut d.max_seq);
1043
1044    while let Some(row) = rows.next().await? {
1045        let seq_id: i64 = row.get(0)?;
1046        let table_name: String = row.get(1)?;
1047        let _entity_id: String = row.get(2)?;
1048        let op: String = row.get(3)?;
1049        let payload_str: String = row.get(4)?;
1050        // Projected by all four folds since 0.14.5. They have partitioned on it
1051        // since D-216; what was missing was carrying it out of the query, which
1052        // is why the correct partition produced a collapsed result anyway.
1053        let branch_id: String = row.get(5)?;
1054
1055        if seq_id > *max_seq {
1056            *max_seq = seq_id;
1057        }
1058
1059        // A `'D'` row is corruption, not a tombstone (D-072).
1060        //
1061        // Doctrine V permits no physical delete outside an archive session, and
1062        // the archive *moves* rows rather than logging their removal — so no
1063        // trigger in the schema writes a `'D'`, and no code path in the crate
1064        // can produce one. This arm used to treat it as a tombstone, which read
1065        // as a claim that deletions are recorded and reconstructible. They are
1066        // not. Refusing here makes the doctrine enforced at the fold rather than
1067        // assumed by it, and is the same call D-060 made for overlap: the layer
1068        // that can notice should.
1069        //
1070        // Retirement is unaffected and is the mechanism that actually removes a
1071        // concept from a composed state — see the `retired != 0` branch below,
1072        // which is where `concepts_gone` is populated in practice.
1073        if op == "D" {
1074            return Err(DbError::ReplayCorrupt {
1075                seq: seq_id,
1076                reason: format!(
1077                    "transaction_log carries a 'D' operation for {table_name} \
1078                     entity {_entity_id:?}; Doctrine V permits no physical delete \
1079                     outside an archive session, and the archive logs none. This \
1080                     row was not written by this crate."
1081                ),
1082            });
1083        }
1084
1085        let payload: serde_json::Value =
1086            serde_json::from_str(&payload_str).map_err(|e| DbError::ReplayCorrupt {
1087                seq: seq_id,
1088                reason: format!("Failed to parse payload JSON: {e}"),
1089            })?;
1090
1091        // v1 and v2 differ by one added field, so v1 folds by reading it as
1092        // absent — which is what `Option` already means here. A future shape
1093        // that *removes* or *retypes* a field would not be able to share this
1094        // path, and would want a match on `v` rather than a ceiling.
1095        let v = payload.get("v").and_then(|v| v.as_u64()).unwrap_or(1);
1096        if v > PAYLOAD_VERSION as u64 {
1097            return Err(DbError::PayloadVersion {
1098                got: v as u8,
1099                max: PAYLOAD_VERSION,
1100            });
1101        }
1102
1103        if table_name == "concepts" {
1104            let id = _entity_id;
1105            let retired = payload.get("retired").and_then(|r| r.as_i64()).unwrap_or(0);
1106            if retired == 0 {
1107                let title = payload
1108                    .get("title")
1109                    .and_then(|s| s.as_str())
1110                    .unwrap_or("")
1111                    .to_string();
1112                let content = payload
1113                    .get("content")
1114                    .and_then(|s| s.as_str())
1115                    .unwrap_or("")
1116                    .to_string();
1117                let embedding_model = payload
1118                    .get("embedding_model")
1119                    .and_then(|s| s.as_str())
1120                    .map(|s| s.to_string());
1121                concepts.insert(
1122                    id.clone(),
1123                    NodeAttributes {
1124                        id,
1125                        title,
1126                        content,
1127                        embedding_model,
1128                    },
1129                );
1130            } else {
1131                // Retirement is the application axis (§4.1), and a reconstruction
1132                // shows what was visible. Onto a snapshot that means removing
1133                // the concept, not declining to add it.
1134                d.concepts_gone.insert(id);
1135            }
1136        } else if table_name == "links" {
1137            let src = payload
1138                .get("source_id")
1139                .and_then(|s| s.as_str())
1140                .unwrap_or("")
1141                .to_string();
1142            let tgt = payload
1143                .get("target_id")
1144                .and_then(|s| s.as_str())
1145                .unwrap_or("")
1146                .to_string();
1147            let edge_type = payload
1148                .get("edge_type")
1149                .and_then(|s| s.as_str())
1150                .unwrap_or("")
1151                .to_string();
1152            let vf = payload
1153                .get("valid_from")
1154                .and_then(|s| s.as_str())
1155                .unwrap_or("")
1156                .to_string();
1157            let vt = payload
1158                .get("valid_to")
1159                .and_then(|s| s.as_str())
1160                .unwrap_or("")
1161                .to_string();
1162            let belief = EdgeBelief {
1163                source_id: src,
1164                target_id: tgt,
1165                edge_type,
1166                valid_from: vf,
1167                valid_to: vt,
1168                branch_id,
1169            };
1170            edges.insert(belief.belief_key(), belief);
1171        }
1172    }
1173
1174    Ok(d)
1175}
1176
1177impl Delta {
1178    /// Compose onto `base` under last-writer-wins by `seq_id` (§5.5).
1179    ///
1180    /// The delta is by construction newer than the base — it is the fold of
1181    /// everything above the base's anchor — so every row it carries wins, and
1182    /// every retirement it carries removes. This is the same rule
1183    /// `trg_links_current_sync`'s upsert applies and the same rule the cold
1184    /// fold applies; that the three agree is asserted by test rather than by
1185    /// this comment (§8).
1186    fn apply_to(self, base: MaterializedState, ts: &str) -> MaterializedState {
1187        let mut concepts = base.concepts;
1188        let mut edges: HashMap<String, EdgeBelief> = base
1189            .edges
1190            .into_iter()
1191            .map(|e| (e.belief_key(), e))
1192            .collect();
1193
1194        for id in self.concepts_gone {
1195            concepts.remove(&id);
1196        }
1197        // No edge equivalent: an edge is superseded in place under the same
1198        // `entity_id`, never removed — see [`Delta`] (D-072).
1199        concepts.extend(self.concepts);
1200        edges.extend(self.edges);
1201
1202        // Sorted so the result is a function of the state and not of hash
1203        // iteration order — `reconstruct` is compared against itself by the
1204        // property suite, and two runs must be equal, not merely equivalent.
1205        let mut edges: Vec<_> = edges.into_values().collect();
1206        edges.sort();
1207
1208        MaterializedState {
1209            seq_anchor: self.max_seq.max(base.seq_anchor),
1210            timestamp: ts.to_string(),
1211            concepts,
1212            edges,
1213            // A delta was applied, so there was history to fold. `reconstruct`
1214            // sets the flag on the one path that never gets here.
1215            predates_recorded_history: false,
1216        }
1217    }
1218}