macrame-db 0.9.0

A Bitemporal Graph Ledger on libSQL · Embedded knowledge database
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
use std::path::Path;

use libsql::TransactionBehavior;

use crate::error::{Result, WriteOp};
use crate::schema::ddl::ARCHIVE_SESSION_MARKER;

/// Outcome of one archive session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArchiveReport {
    pub links_archived: usize,
    /// Concepts moved to `cold.concepts` (0.9.0, C2). Always `0` before v9,
    /// where no concept could leave the hot table at all.
    pub concepts_archived: usize,
    pub log_entries_archived: usize,
    /// Oldest `transaction_log.seq_id` still present in the hot file after the
    /// session, i.e. the new horizon (see glossary). `None` if the hot log is empty.
    pub horizon: Option<i64>,
}

/// Schema of the cold database. Deliberately trigger-free and FK-free.
///
/// **Corrected 2026-08-07.** This comment used to justify the FK-free part with
/// *"concepts are never archived (D-022)"*, which stopped being true in 0.9.0
/// when C2 added `cold.concepts` — the table declared a few lines below. The
/// reasons that survive are the other two, and they are the load-bearing ones:
/// a FK from `cold.links` to `concepts` still could not be satisfied, because
/// the cold file holds only the concepts that have gone cold and `cold.links`
/// may name any of them; and the delete guards must not exist on a file whose
/// whole purpose is to receive rows and, on rehydration, to give them back.
const COLD_SCHEMA: &[&str] = &[
    // `weight` carries the same CHECK as the hot table (T2.1, D-083). Not
    // symmetry for its own sake: the cold file is read back by `reconstruct`
    // through the same `f64` decode, so a text weight is the same panic there
    // as it is here, and a negative one is the same unsound shortest path.
    //
    // The hot table's constraint does not protect this one. Rows arrive by
    // `INSERT … SELECT` across an ATTACH, which re-checks against *this*
    // table's constraints — and a cold file may predate the hot file's rung, or
    // have been written by a version that had neither.
    //
    // `IF NOT EXISTS` means an existing cold database keeps whatever definition
    // it was created with; this constrains new cold files, and the loader guard
    // is what covers the old ones. That is the same division of labour §4.7
    // describes, and the reason the guard stays.
    r#"CREATE TABLE IF NOT EXISTS cold.links (
        source_id   TEXT NOT NULL,
        target_id   TEXT NOT NULL,
        edge_type   TEXT NOT NULL,
        valid_from  TEXT NOT NULL,
        recorded_at TEXT NOT NULL,
        valid_to    TEXT NOT NULL,
        weight      REAL NOT NULL CHECK (weight >= 0.0 AND weight < 9e999 AND typeof(weight) = 'real'),
        properties  TEXT NOT NULL,
        PRIMARY KEY (source_id, target_id, edge_type, valid_from, recorded_at)
    )"#,
    // Concepts, as of v9 (C2). Trigger-free and FK-free like `cold.links`, and
    // for the same reasons -- but note what it does NOT drop.
    //
    // **Every column crosses, `content` included.** Archival is a move, not a
    // rewrite (2.3), and a move that drops a column is a rewrite. The log
    // payload for a concept carries its `content` (4.3), so a cold concept
    // whose text had been dropped would contradict `cold.transaction_log` about
    // itself, and rehydration would return a concept the ledger never recorded:
    // empty text where the log says there was text. That is the unexplained
    // absence Doctrine V exists to prevent.
    //
    // The tension with D-116 is apparent rather than real. D-116 governs the
    // *in-memory* `NodeData` representation -- `content` is not loaded by
    // default because most readers do not want it. This is *on-disk* storage.
    // Disk carries the text; memory does not populate it until asked. Two
    // independent defaults, and conflating them would make rehydration lossy to
    // save a read nobody was performing.
    //
    // `rowid_pk` crosses as the record of what the rowid *was*. Restoring it is
    // C3's problem and not obviously safe: `concepts.rowid_pk` is a plain
    // INTEGER PRIMARY KEY, so SQLite may reuse a freed value, and archiving the
    // highest rowids can leave a later insert holding one a cold row still
    // claims. The column is carried because a move must not lose it.
    //
    // **The hazard has two exits and C3 must take one of them explicitly.**
    // Either reinstate the original `rowid_pk` when it is still free, or assign
    // a fresh one — and in the second case **update `concepts_fts`'s
    // `content_rowid` mapping to match**, because the FTS index is
    // external-content keyed on this column (4.6, D-119). A rehydration that
    // reassigns the rowid without re-pointing the index leaves the search index
    // silently describing the wrong row, which is the exact failure `rowid_pk`
    // was made explicit to prevent. Named here so C3 meets both exits rather
    // than rediscovering the FTS coupling.
    r#"CREATE TABLE IF NOT EXISTS cold.concepts (
        rowid_pk         INTEGER,
        id               TEXT NOT NULL PRIMARY KEY,
        title            TEXT NOT NULL,
        content          TEXT NOT NULL DEFAULT '',
        embedding_model  TEXT,
        valid_from       TEXT NOT NULL,
        valid_to         TEXT NOT NULL,
        recorded_at      TEXT NOT NULL,
        retired          INTEGER NOT NULL DEFAULT 0
    )"#,
    // seq_id is carried over verbatim from the hot log, so it is a plain
    // INTEGER PRIMARY KEY -- never AUTOINCREMENT, which would renumber history.
    r#"CREATE TABLE IF NOT EXISTS cold.transaction_log (
        seq_id      INTEGER PRIMARY KEY,
        table_name  TEXT NOT NULL,
        entity_id   TEXT NOT NULL,
        operation   TEXT NOT NULL,
        payload     TEXT NOT NULL,
        recorded_at TEXT NOT NULL
    )"#,
    "CREATE INDEX IF NOT EXISTS cold.idx_cold_txlog_entity ON transaction_log (entity_id)",
    "CREATE INDEX IF NOT EXISTS cold.idx_cold_txlog_time ON transaction_log (recorded_at)",
    r#"CREATE TABLE IF NOT EXISTS cold.archive_horizon (
        archived_at TEXT NOT NULL,
        cutoff      TEXT NOT NULL,
        horizon     INTEGER
    )"#,
];

/// A links assertion is archivable when it is older than the cutoff AND it is
/// either superseded by a later assertion for the same interval key, or it is
/// the current belief for an interval that closed before the cutoff.
///
/// This keeps every row that `links_current` still projects (Doctrine VI: the
/// materialization must stay rebuildable from `links`) while moving exactly the
/// "closed intervals, superseded history" the §2 diagram assigns to the cold file.
const LINKS_ARCHIVABLE: &str = r#"
    recorded_at < :cutoff AND (
        EXISTS (
            SELECT 1 FROM links newer
            WHERE newer.source_id   = links.source_id
              AND newer.target_id   = links.target_id
              AND newer.edge_type   = links.edge_type
              AND newer.valid_from  = links.valid_from
              AND newer.recorded_at > links.recorded_at
        )
        OR (valid_to <> '9999-12-31T23:59:59.999999Z' AND valid_to <= :cutoff)
    )
"#;

/// A log entry is archivable when it is older than the cutoff and a later entry
/// exists for the same entity, i.e. it is superseded. The newest entry per
/// entity always stays hot so that `reconstruct(now)` never needs the cold file.
const LOG_ARCHIVABLE: &str = r#"
    recorded_at < :cutoff AND EXISTS (
        SELECT 1 FROM transaction_log newer
        WHERE newer.entity_id = transaction_log.entity_id
          AND newer.seq_id    > transaction_log.seq_id
    )
"#;

/// A concept is archivable when it is `retired`, both its clocks are behind the
/// cutoff, **and no surviving row of hot `links` mentions it in either
/// direction** (C1, D-128).
///
/// # Why reachability, and not a closed interval
///
/// A link assertion has a closed interval, so [`LINKS_ARCHIVABLE`] can ask
/// whether the interval ended. A concept is an *entity*, and has no closed
/// state: `retired = 1` says belief in it stopped, which is not the same claim
/// as "nothing points at it any more". The two `links` foreign keys are what
/// make that difference matter — archiving a concept physically removes its row,
/// and a surviving hot link naming it would leave the key unsatisfiable.
/// `ON DELETE CASCADE` is not the way out, because the rows it would cascade
/// onto are ledger rows.
///
/// So concept archival is **strictly downstream of link archival**: a concept
/// becomes eligible only once every edge mentioning it has itself gone cold.
/// Inside a session this predicate is therefore evaluated *after* the `links`
/// delete and never before it, and the same question asked before and after one
/// session legitimately gives two different answers. That is a property of the
/// predicate, not a race.
///
/// # The other two foreign keys, and why they are not clauses here
///
/// `concepts` also has inbound keys from `analytics_annotations` and from every
/// registered `embeddings_*` table ([`crate::schema::migrations`] lists all
/// four). Neither appears above, and the distinction is the point: those hold
/// **derived** rows. Doctrine VII makes an embedding an artifact of a model
/// applied to content, and an annotation is the output of an algorithm that read
/// `concepts` in the first place. A derived row is removed and recomputed; a
/// ledger row is neither. Making archivability wait on a recomputable artifact
/// would answer "not yet" forever for any concept that had ever been embedded.
///
/// # Both clocks, because one of them is not enough
///
/// The specification for this predicate named `valid_to` alone.
/// `recorded_at < :cutoff` is here as well, mirroring [`LINKS_ARCHIVABLE`]:
/// a concept retired with its `valid_to` behind the cutoff but *recorded* at or
/// after it is a fact the session is not meant to touch yet, and archiving it
/// would send the concept cold while the log entries describing it stayed hot.
/// That is the same two-clock mismatch the `links_current` compensation carried
/// until Wave 4.5 (see [`archive_session`]), reached from the other side.
/// Doctrine II: two clocks, never mixed.
///
/// The open sentinel needs no clause of its own — `9999-12-31T23:59:59.999999Z`
/// sorts above every canonical stamp (D-029), so a concept whose validity is
/// still open fails `valid_to < :cutoff` for any cutoff a caller can pass.
const CONCEPTS_ARCHIVABLE: &str = r#"
    retired = 1
    AND recorded_at < :cutoff
    AND valid_to    < :cutoff
    AND NOT EXISTS (
        SELECT 1 FROM links
        WHERE links.source_id = concepts.id
           OR links.target_id = concepts.id
    )
"#;

/// The ids of every concept that `CONCEPTS_ARCHIVABLE` admits at `cutoff`, in
/// `id` order.
///
/// **Read-only, and deliberately available before anything can act on it.**
/// Concept archival is the one operation in this crate a caller cannot undo
/// without a cold file to hand, so the predicate that decides it is observable
/// on its own rather than only as a count in a report after the fact.
///
/// The answer is a function of the hot state *now*. Archiving links first will
/// generally enlarge it — that is the downstream relationship
/// `CONCEPTS_ARCHIVABLE` describes, not an inconsistency — so a caller
/// planning a session should ask after the link archive, not before it.
pub async fn archivable_concepts(conn: &libsql::Connection, cutoff: &str) -> Result<Vec<String>> {
    let mut rows = conn
        .query(
            &format!("SELECT id FROM concepts WHERE {CONCEPTS_ARCHIVABLE} ORDER BY id"),
            libsql::named_params! {":cutoff": cutoff},
        )
        .await?;

    let mut ids = Vec::new();
    while let Some(row) = rows.next().await? {
        ids.push(row.get::<String>(0)?);
    }
    Ok(ids)
}

/// Move closed edge intervals and superseded log rows older than `cutoff` into
/// the cold database at `archive_path` (§5.7, D-012, D-022).
///
/// The whole session is one `BEGIN IMMEDIATE … COMMIT` transaction (D-012):
/// copy-then-delete must be atomic, or a crash between the phases duplicates or
/// loses rows. The archive-session marker that unlocks the delete guards
/// (D-008 revised) is created as the first statement of that transaction and
/// dropped as the last, so it never exists as committed state — commit drops
/// it, rollback discards it, and there is no crash path that leaves the guards
/// disarmed.
///
/// ATTACH is issued outside the transaction and DETACH is issued unconditionally
/// on the way out, including on error: ATTACH is not transactional and survives
/// ROLLBACK, so a leaked handle would make every later archive or cold-DB
/// reconstruct fail with "database cold is already in use".
/// `archived_at` is **when the session ran**; `cutoff` is the boundary it used.
///
/// Both go into `cold.archive_horizon`, and until Wave 4.5 both columns were
/// written with the cutoff — so the table recorded that every archive had run at
/// the instant it was archiving *up to*, which is the one time it certainly did
/// not run. The two are different clocks (Doctrine II) and the row exists to
/// carry both: the cutoff says what was moved, `archived_at` says when the
/// decision was taken, and only the second can answer "how stale is this cold
/// file". The column was there, correctly named, holding the wrong value.
pub async fn archive(
    conn: &libsql::Connection,
    cutoff: &str,
    archived_at: &str,
    archive_path: &Path,
) -> Result<ArchiveReport> {
    crate::temporal::replay::detach_stale_cold(conn).await;

    // ATTACH creates the cold file if it does not exist.
    conn.execute(
        "ATTACH DATABASE ?1 AS cold",
        libsql::params![archive_path.to_string_lossy().as_ref()],
    )
    .await?;

    let result = archive_session(conn, cutoff, archived_at).await;

    // Unconditional: see the DETACH note above.
    if let Err(e) = conn.execute("DETACH DATABASE cold", ()).await {
        tracing::warn!("archive: failed to DETACH cold database: {e}");
    }

    result
}

/// `conn` is passed alongside `tx` only so [`delete_guarded`] can hand it to
/// `classify`, which queries on the error path. Both name the same connection.
async fn archive_session(
    conn: &libsql::Connection,
    cutoff: &str,
    archived_at: &str,
) -> Result<ArchiveReport> {
    for ddl in COLD_SCHEMA {
        conn.execute(ddl, ()).await?;
    }

    let tx = conn
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .await?;

    // --- archive session opens: the delete guards are now satisfied ---
    tx.execute(&format!("CREATE TABLE {ARCHIVE_SESSION_MARKER} (x)"), ())
        .await?;

    let links_archived = tx
        .execute(
            &format!(
                "INSERT OR IGNORE INTO cold.links
                     (source_id, target_id, edge_type, valid_from, recorded_at,
                      valid_to, weight, properties)
                 SELECT source_id, target_id, edge_type, valid_from, recorded_at,
                        valid_to, weight, properties
                 FROM links WHERE {LINKS_ARCHIVABLE}"
            ),
            libsql::named_params! {":cutoff": cutoff},
        )
        .await? as usize;

    let links_deleted = delete_guarded(
        &tx,
        conn,
        &format!("DELETE FROM links WHERE {LINKS_ARCHIVABLE}"),
        cutoff,
        "links",
    )
    .await?;

    // links_current is derivative (Doctrine VI) and must equal the latest-belief
    // projection of whatever remains in links, or audit_current() reports drift
    // the moment an archive runs. Re-derive it rather than trying to describe
    // the deletion's shadow: this used to be a hand-written
    // `DELETE FROM links_current WHERE valid_to <= :cutoff`, which filters on
    // *valid* time while LINKS_ARCHIVABLE also requires `recorded_at < :cutoff`.
    // A row closed at the cutoff but recorded at or after it therefore survived
    // in links and was deleted from links_current — permanent drift no later
    // audit could explain, from a compensation that had quietly stopped being
    // the image of the thing it compensated for. Doctrine II: two clocks, never
    // mixed. Deriving from the definition cannot drift from the definition.
    //
    // **Skipped when the DELETE removed nothing (T1.1, D-080).** `links_current`
    // is a function of `links`, so if `links` did not change its projection did
    // not either, and there is no drift for a rebuild to repair. This was
    // harmless while `archive()` was called once against a whole backlog,
    // because the one session always had work. It stops being harmless the
    // moment the caller windows: `rebuild_within` costs O(surviving `links`)
    // regardless of how much the session archived (D-077), so without this a run
    // of twenty windows over a quiet stretch of history pays twenty full
    // reprojections to delete nothing — and windowing makes the archive slower
    // in total than not windowing. `log_entries_archived` deliberately does not
    // enter into it: archiving the transaction log cannot change `links`.
    if links_deleted > 0 {
        crate::integrity::rebuild::rebuild_within(&tx, crate::integrity::rebuild::Verify::No)
            .await?;
    }

    // Concepts, and **only now** — after the `links` delete, never before it
    // ([D-128](../../docs/architecture/s13-decision-register.md)). A concept is
    // archivable when nothing in hot `links` names it, so evaluating the
    // predicate before the edges have gone cold archives strictly less than the
    // session is entitled to. This ordering is the whole content of "concept
    // archival is downstream of link archival".
    let concepts_archived = archive_concepts(&tx, conn, cutoff).await?;

    let log_entries_archived = tx
        .execute(
            &format!(
                "INSERT OR IGNORE INTO cold.transaction_log
                     (seq_id, table_name, entity_id, operation, payload, recorded_at)
                 SELECT seq_id, table_name, entity_id, operation, payload, recorded_at
                 FROM transaction_log WHERE {LOG_ARCHIVABLE}"
            ),
            libsql::named_params! {":cutoff": cutoff},
        )
        .await? as usize;

    delete_guarded(
        &tx,
        conn,
        &format!("DELETE FROM transaction_log WHERE {LOG_ARCHIVABLE}"),
        cutoff,
        "transaction_log",
    )
    .await?;

    // Record the new horizon in the cold file so a pre-horizon reconstruct can
    // tell "archived" from "never existed" (glossary; R14).
    let horizon: Option<i64> = tx
        .query("SELECT MIN(seq_id) FROM transaction_log", ())
        .await?
        .next()
        .await?
        .and_then(|row| row.get(0).ok());

    tx.execute(
        "INSERT INTO cold.archive_horizon (archived_at, cutoff, horizon) VALUES (?1, ?2, ?3)",
        libsql::params![archived_at, cutoff, horizon],
    )
    .await?;

    // --- archive session closes: the guards re-arm before COMMIT ---
    tx.execute(&format!("DROP TABLE {ARCHIVE_SESSION_MARKER}"), ())
        .await?;

    tx.commit().await?;

    Ok(ArchiveReport {
        links_archived,
        concepts_archived,
        log_entries_archived,
        horizon,
    })
}

/// Move every concept [`CONCEPTS_ARCHIVABLE`] admits into `cold.concepts`, and
/// dispose of its derived rows (C2).
///
/// # The partition, which is the decision this function encodes
///
/// **Entity data crosses; derivative data does not.** The concept row itself is
/// moved column for column — a move that drops a column is a rewrite, and
/// [Doctrine V] does not permit an absence the ledger cannot explain. Its
/// `analytics_annotations` and `embeddings_*` rows are *deleted* rather than
/// moved, because [Doctrine VII] makes both recomputable from the content that
/// did cross. Carrying them would also be unimplementable for the vectors:
/// `F32_BLOB` and DiskANN are libSQL-specific and the cold file is a plain
/// database opened through `ATTACH`.
///
/// The disposal is not incidental to the move — it is what makes the move
/// legal. `concepts` has four inbound foreign keys, and the two derived ones
/// would refuse the `DELETE` outright.
///
/// # Why the deletes are not logged, and why that is right
///
/// `concepts` carries log triggers on insert and update but **not** on delete —
/// there was no delete path to log while the guard was unconditional, and there
/// deliberately still is not. Archival mints no transaction-time facts: the
/// concept is in the cold file, the log entries describing it are either still
/// hot or in `cold.transaction_log`, and nothing about what was believed, or
/// when, has changed. A log entry here would assert that something happened to
/// the concept at archive time, which is exactly the lie [Doctrine III] forbids.
async fn archive_concepts(
    tx: &libsql::Transaction,
    conn: &libsql::Connection,
    cutoff: &str,
) -> Result<usize> {
    let moved = tx
        .execute(
            &format!(
                "INSERT OR IGNORE INTO cold.concepts
                     (rowid_pk, id, title, content, embedding_model,
                      valid_from, valid_to, recorded_at, retired)
                 SELECT rowid_pk, id, title, content, embedding_model,
                        valid_from, valid_to, recorded_at, retired
                 FROM concepts WHERE {CONCEPTS_ARCHIVABLE}"
            ),
            libsql::named_params! {":cutoff": cutoff},
        )
        .await? as usize;

    if moved == 0 {
        return Ok(0);
    }

    // The derived rows, before the concept they hang off. `embeddings_*` is
    // enumerated from the catalogue rather than from a list, because the set is
    // whatever `register_model` has created on *this* database and a hard-coded
    // list would silently miss a model the caller added.
    let mut derived: Vec<String> = vec!["analytics_annotations".to_string()];
    let mut rows = tx
        .query(
            "SELECT name FROM sqlite_master WHERE type = 'table' \
             AND name LIKE 'embeddings\\_%' ESCAPE '\\'",
            (),
        )
        .await?;
    while let Some(row) = rows.next().await? {
        derived.push(row.get::<String>(0)?);
    }
    drop(rows);

    for table in &derived {
        tx.execute(
            &format!(
                "DELETE FROM {table} WHERE concept_id IN \
                 (SELECT id FROM cold.concepts)"
            ),
            (),
        )
        .await?;
    }

    // `trg_concepts_fts_delete` fires on this and keeps the search index
    // correct — the capability v8 installed inert and this rung made reachable.
    let deleted = delete_guarded(
        tx,
        conn,
        &format!("DELETE FROM concepts WHERE {CONCEPTS_ARCHIVABLE}"),
        cutoff,
        "concepts",
    )
    .await? as usize;

    debug_assert_eq!(
        moved, deleted,
        "the predicate selected a different set for the copy than for the delete"
    );

    Ok(deleted)
}

/// Outcome of one rehydration (0.9.0, C3).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RehydrateReport {
    /// Concepts moved back into the hot table.
    pub concepts_rehydrated: usize,
    /// Of those, how many could **not** keep their original `rowid_pk` because
    /// something else had claimed it while they were cold, and were reassigned
    /// with the FTS index re-pointed to match.
    ///
    /// Reported rather than hidden because it is the one way a rehydrated
    /// concept differs from the row that was archived, and a caller comparing
    /// rowids across the boundary should be able to see that it happened.
    pub rowids_reassigned: usize,
}

/// Move concepts back from the cold file into the hot tables (§2.3, C3).
///
/// # Rehydration is a move back, not a write
///
/// It mints no transaction-time facts and is invisible to both clocks. The
/// concept's log entries were never removed, so the ledger already says
/// everything true about it; writing a fresh `'I'` would assert the concept was
/// *learned* at rehydration time, and — because the fold takes the highest
/// `seq_id` per entity — would additionally outrank any later `'U'` that retired
/// it. See [`crate::schema::ddl::CREATE_CONCEPTS_LOG_INSERT`], which is
/// marker-gated at v10 for exactly this reason. The whole operation therefore
/// runs inside a declared archive session, which is what suppresses the trigger.
///
/// # `rowid_pk`: reinstate, or reassign and re-point the index
///
/// The common case has no collision — the rowid was freed by archival and
/// nothing has claimed it since — and reinstating is the clean move-back with no
/// side effects at all. When something *has* taken it, the fallback is a fresh
/// `rowid_pk` plus an FTS correction: `concepts_fts` is external-content keyed
/// on `rowid_pk` ([D-119]), so a reassignment without re-pointing leaves the
/// index describing the wrong row, silently. Both exits are taken here rather
/// than one being assumed, and [`RehydrateReport::rowids_reassigned`] reports
/// which was used.
pub async fn rehydrate(
    conn: &libsql::Connection,
    ids: &[&str],
    archive_path: &Path,
) -> Result<RehydrateReport> {
    if ids.is_empty() {
        return Ok(RehydrateReport {
            concepts_rehydrated: 0,
            rowids_reassigned: 0,
        });
    }

    crate::temporal::replay::detach_stale_cold(conn).await;
    conn.execute(
        "ATTACH DATABASE ?1 AS cold",
        libsql::params![archive_path.to_string_lossy().as_ref()],
    )
    .await?;

    let result = rehydrate_session(conn, ids).await;

    if let Err(e) = conn.execute("DETACH DATABASE cold", ()).await {
        tracing::warn!("rehydrate: failed to DETACH cold database: {e}");
    }
    result
}

async fn rehydrate_session(conn: &libsql::Connection, ids: &[&str]) -> Result<RehydrateReport> {
    let tx = conn
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .await?;

    // The session opens for the same reason the archive's does, plus one more:
    // it is what stops `trg_concepts_log_insert` from firing (v10).
    tx.execute(&format!("CREATE TABLE {ARCHIVE_SESSION_MARKER} (x)"), ())
        .await?;

    let mut rehydrated = 0usize;
    let mut reassigned = 0usize;

    for id in ids {
        let Some(row) = tx
            .query(
                "SELECT rowid_pk, id, title, content, embedding_model, \
                 valid_from, valid_to, recorded_at, retired \
                 FROM cold.concepts WHERE id = ?1",
                libsql::params![*id],
            )
            .await?
            .next()
            .await?
        else {
            continue;
        };

        let old_rowid: i64 = row.get(0)?;
        let title: String = row.get(2)?;
        let content: String = row.get(3)?;
        let model: Option<String> = row.get(4)?;
        let valid_from: String = row.get(5)?;
        let valid_to: String = row.get(6)?;
        let recorded_at: String = row.get(7)?;
        let retired: i64 = row.get(8)?;

        let taken: i64 = tx
            .query(
                "SELECT COUNT(*) FROM concepts WHERE rowid_pk = ?1",
                libsql::params![old_rowid],
            )
            .await?
            .next()
            .await?
            .expect("COUNT(*) always returns a row")
            .get(0)?;

        if taken == 0 {
            // The clean move back: same row, same rowid, no side effects.
            tx.execute(
                "INSERT INTO concepts (rowid_pk, id, title, content, embedding_model, \
                 valid_from, valid_to, recorded_at, retired) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                libsql::params![
                    old_rowid,
                    *id,
                    title.clone(),
                    content.clone(),
                    model,
                    valid_from,
                    valid_to,
                    recorded_at,
                    retired
                ],
            )
            .await?;
        } else {
            // Something claimed the rowid while this concept was cold. Take a
            // fresh one, then correct the index: `concepts_fts` is
            // external-content keyed on `rowid_pk`, and its insert trigger will
            // have written an entry at the *new* rowid — what has to be undone
            // is the stale entry still sitting at the old one, which the archive
            // could not remove because the row it described had already gone.
            tx.execute(
                "INSERT INTO concepts (id, title, content, embedding_model, \
                 valid_from, valid_to, recorded_at, retired) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                libsql::params![
                    *id,
                    title.clone(),
                    content.clone(),
                    model,
                    valid_from,
                    valid_to,
                    recorded_at,
                    retired
                ],
            )
            .await?;
            tx.execute(
                "INSERT INTO concepts_fts (concepts_fts, rowid, title, content) \
                 VALUES ('delete', ?1, ?2, ?3)",
                libsql::params![old_rowid, title, content],
            )
            .await?;
            reassigned += 1;
        }

        tx.execute(
            "DELETE FROM cold.concepts WHERE id = ?1",
            libsql::params![*id],
        )
        .await?;
        rehydrated += 1;
    }

    tx.execute(&format!("DROP TABLE {ARCHIVE_SESSION_MARKER}"), ())
        .await?;
    tx.commit().await?;

    Ok(RehydrateReport {
        concepts_rehydrated: rehydrated,
        rowids_reassigned: reassigned,
    })
}

/// Run one of the archive's `DELETE`s, naming the table if a guard refuses it.
///
/// **This is what closes defect AC, and the shape of the fix is the point.**
/// There used to be a second classifier here — `classify_archive_violation` —
/// which was defined, delegated correctly to [`crate::error::abort_kind`], and
/// called from nowhere, so `DbError::ArchiveViolation` was unreachable by any
/// code path in the crate. It was recorded as defect H, marked Fixed by a commit
/// that made the *body* delegate rather than making the function *called*, and
/// so survived its own repair. It is deleted rather than wired up, because
/// [`crate::error::classify`] with [`WriteOp::Delete`] already did exactly what
/// it did: the defect was one classifier too many, not one too few.
///
/// A guard firing here means the marker table is absent or was dropped early —
/// the session's invariant broken from inside. That is worth a typed error
/// naming the table rather than a raw engine message naming a trigger.
async fn delete_guarded(
    tx: &libsql::Transaction,
    conn: &libsql::Connection,
    sql: &str,
    cutoff: &str,
    table: &str,
) -> Result<u64> {
    match tx
        .execute(sql, libsql::named_params! {":cutoff": cutoff})
        .await
    {
        Ok(n) => Ok(n),
        Err(e) => Err(crate::error::classify(conn, e, WriteOp::Delete { table }).await),
    }
}