yantrikdb 0.23.0

Cognitive memory engine for persistent AI systems
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
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
use super::*;

// ── Issue #47: correct() semantics tightened (v0.7.20) ───────────────

#[test]
fn correct_preserves_rid_and_created_at() {
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "original text",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    let original = db.get(&rid).unwrap().unwrap();
    let original_created_at = original.created_at;
    std::thread::sleep(std::time::Duration::from_millis(10));

    // v0.9.3: text corrections are refused (CorrectionRequiresReembed);
    // the rid/created_at preservation contract is exercised via an
    // importance correction instead.
    let result = db
        .correct(&rid, None, None, Some(0.9), None, "test correction")
        .unwrap();
    assert_eq!(result.corrected_rid, rid, "rid must be preserved");
    assert_eq!(result.original_rid, rid);
    assert!(!result.original_tombstoned);
    assert_eq!(result.revision_num, 1);

    let updated = db.get(&rid).unwrap().unwrap();
    assert_eq!(updated.text, "original text", "text untouched");
    assert!((updated.importance - 0.9).abs() < 1e-9);
    assert!(
        (updated.created_at - original_created_at).abs() < 1e-9,
        "created_at must be preserved (was {}, became {})",
        original_created_at,
        updated.created_at,
    );
}

#[test]
fn correct_text_change_without_embedder_returns_no_embedder_and_no_side_effects() {
    // v0.10 Item 3: a text-changing correction must re-embed to keep the
    // retrieval vector coherent. On a raw-embedding DB with no embedder
    // attached there is nothing to embed with, so it returns NoEmbedder
    // BEFORE touching any state — no revision row, no text mutation.
    // (Metadata/scalar corrections don't re-embed and are unaffected.)
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "alice owns service A",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();

    let err = db
        .correct(
            &rid,
            Some("bob owns service B"),
            None,
            None,
            None,
            "handover",
        )
        .unwrap_err();
    assert!(
        matches!(err, crate::error::YantrikDbError::NoEmbedder),
        "wrong error: {err}"
    );

    // Zero side effects.
    let unchanged = db.get(&rid).unwrap().unwrap();
    assert_eq!(unchanged.text, "alice owns service A");
    assert!(db.history(&rid).unwrap().is_empty(), "no revision row");

    // A no-op text change (same bytes) is NOT a re-embed — it takes the
    // metadata path and succeeds even without an embedder.
    db.correct(
        &rid,
        Some("alice owns service A"),
        None,
        Some(0.9),
        None,
        "touch",
    )
    .unwrap();
    assert_eq!(db.history(&rid).unwrap().len(), 1);

    // Metadata / importance / valence corrections remain available.
    db.correct(
        &rid,
        None,
        Some(&serde_json::json!({"owner": "bob"})),
        Some(0.9),
        Some(0.1),
        "ownership metadata updated",
    )
    .unwrap();
    assert_eq!(db.history(&rid).unwrap().len(), 2);
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn correct_reembed_updates_retrieval_vector_delta_and_cold() {
    // v0.10 Item 3 / trace T4: a text-changing correction re-embeds so the
    // record is retrieved under its NEW meaning, not its old one. rid,
    // created_at, and the revision chain are preserved; the revision row
    // records the prior embedding's provenance. Verified on BOTH a
    // delta-resident record and one compacted to the cold tier.
    let db = YantrikDB::with_default(":memory:").unwrap();
    let record = |text: &str| {
        db.record_text(
            text,
            "semantic",
            0.6,
            0.0,
            604800.0,
            &empty_meta(),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap()
    };

    // Similarity of `rid` to a free-text query, via the real recall path.
    let sim_to = |rid: &str, q: &str| -> f64 {
        let emb = db.embed(q).unwrap();
        db.recall(
            &emb, 10, None, None, false, false, None, true, None, None, None, None, None, false,
            None, // event_after (#149)
            None, // event_before (#149)
        )
        .unwrap()
        .into_iter()
        .find(|r| r.rid == rid)
        .map(|r| r.scores.similarity)
        .unwrap_or(0.0)
    };

    // A distractor on the NEW topic so ranking is contested, not trivial.
    let _distractor = record("annual financial revenue and profit projections");

    // ── Case A: delta-resident record ──
    let rid = record("I love hiking in the mountains at dawn");
    let created_before = db.get(&rid).unwrap().unwrap().created_at;
    let old_topic_before = sim_to(&rid, "hiking trails and mountain trekking");
    assert!(
        old_topic_before > 0.3,
        "record embeds as its old topic: {old_topic_before}"
    );

    db.correct(
        &rid,
        Some("the quarterly financial revenue grew twenty percent"),
        None,
        None,
        None,
        "topic corrected",
    )
    .unwrap();

    // rid + created_at preserved; text updated.
    let after = db.get(&rid).unwrap().unwrap();
    assert_eq!(after.created_at, created_before, "created_at preserved");
    assert!(after.text.contains("financial revenue"));

    // New-topic query retrieves it more strongly than the old topic now,
    // and its old-topic similarity fell — the vector followed the text.
    let new_topic_after = sim_to(&rid, "financial revenue report");
    let old_topic_after = sim_to(&rid, "hiking trails and mountain trekking");
    assert!(
        new_topic_after > old_topic_after,
        "re-embedded to new topic: new={new_topic_after} old={old_topic_after}"
    );
    assert!(
        old_topic_after < old_topic_before,
        "moved away from old topic: {old_topic_after} < {old_topic_before}"
    );

    // Revision row captured the prior embedding provenance.
    let revs = db.history(&rid).unwrap();
    assert_eq!(revs.len(), 1);
    // The prior VECTOR always existed, so its hash is captured. The model
    // name is only present when the original write stamped one (reembed
    // does; plain record_text leaves it NULL) — so we assert the hash,
    // which is the load-bearing "the vector changed, here's its fingerprint"
    // provenance a replica verifies against.
    let has_hash: bool = db
        .conn()
        .query_row(
            "SELECT prior_embedding_hash IS NOT NULL \
             FROM record_revisions WHERE rid = ?1",
            rusqlite::params![rid],
            |r| r.get(0),
        )
        .unwrap();
    assert!(has_hash, "prior embedding hash captured");

    // ── Case B: cold-tier record (correction tombstones a cold vector) ──
    let cold_rid = record("the weather today is sunny and warm");
    db.search_state.load().vec_index.compact().unwrap();
    assert!(sim_to(&cold_rid, "sunny warm weather forecast") > 0.3);

    db.correct(
        &cold_rid,
        Some("the stock market closed sharply higher today"),
        None,
        None,
        None,
        "cold topic corrected",
    )
    .unwrap();

    let cold_new = sim_to(&cold_rid, "stock market closing prices");
    let cold_old = sim_to(&cold_rid, "sunny warm weather forecast");
    assert!(
        cold_new > cold_old,
        "cold-tier re-embed works: new={cold_new} old={cold_old}"
    );
}

#[test]
fn correction_epoch_toggles_and_validates() {
    // v0.10 Item 3 seqlock (sol r4/r5): the guard bumps ODD on enter, EVEN
    // on drop; correction_epoch_even() returns a stable even snapshot;
    // correction_epoch_validate() (Acquire-fenced) passes only if unchanged
    // and even. The guard borrows conn, so drop order is compiler-enforced.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let e_start = db.correction_epoch_even().unwrap();
    assert_eq!(e_start % 2, 0, "boot epoch is even");
    assert!(
        db.correction_epoch_validate(e_start),
        "even + unchanged validates"
    );
    {
        let conn = db.conn();
        let _g = db.enter_correction_epoch(&conn);
        assert!(
            !db.correction_epoch_validate(e_start),
            "odd (correction in flight) fails the reader's validation"
        );
    }
    let e_end = db.correction_epoch_even().unwrap();
    assert_eq!(e_end, e_start + 2, "one correction advances the epoch by 2");
    assert!(
        db.correction_epoch_validate(e_end),
        "post-correction even validates"
    );
    assert!(
        !db.correction_epoch_validate(e_start),
        "a stale pre-correction snapshot never re-validates"
    );
}

#[test]
fn correct_with_embedding_rejects_stale_generation() {
    // sol r8: a caller-supplied vector is pinned to the generation it was
    // embedded against. If that generation no longer matches the index's
    // (a reembed cutover raced the correction), the vector is stale-space
    // and MUST be rejected retryably — never committed — else the corrected
    // text would rank under the old vector space. Here we simulate the race
    // by passing a generation that does not match the current one.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "alice owns service A",
            "semantic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    let gen = db.search_generation();
    let new_emb = vec_seed(9.0, 8);

    // Stale generation → retryable rejection, NO mutation.
    let err = db
        .correct_with_embedding(
            &rid,
            Some("bob owns service B"),
            &new_emb,
            gen + 1,
            None,
            None,
            None,
            "handover",
        )
        .unwrap_err();
    assert!(
        matches!(
            err,
            crate::error::YantrikDbError::CorrectionDeferredDuringReembed { .. }
        ),
        "stale-generation caller embedding must be deferred, got {err:?}"
    );
    assert_eq!(
        db.get_untracked(&rid).unwrap().unwrap().text,
        "alice owns service A",
        "rejected correction must leave the text unchanged"
    );
    assert_eq!(
        db.history(&rid).unwrap().len(),
        0,
        "rejected correction must record no revision"
    );

    // Current generation → succeeds and updates the text in place.
    db.correct_with_embedding(
        &rid,
        Some("bob owns service B"),
        &new_emb,
        gen,
        None,
        None,
        None,
        "handover",
    )
    .unwrap();
    assert_eq!(
        db.get_untracked(&rid).unwrap().unwrap().text,
        "bob owns service B",
        "generation-matched caller embedding must apply in place at the same rid"
    );
    assert_eq!(
        db.history(&rid).unwrap().len(),
        1,
        "applied correction records exactly one revision"
    );
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn correct_new_text_drops_stale_entity_links() {
    // nuron finding (v0.10 Item-3 follow-up): correct(new_text) must drop
    // memory->entity links whose entity no longer appears (WORD-BOUNDARY) in
    // the corrected text — else graph expansion keeps serving the record under
    // its OLD association (why_retrieved=["graph-connected via <old-entity>"]).
    // Reproduces the Volkan/Aurelian case AND the substring-collision false-keep
    // guard: "Volkan" must drop even when the new text contains "Volkanic".
    let db = YantrikDB::with_default(":memory:").unwrap();
    let rid = db
        .record_text(
            "The Volkan salt marshes are cold",
            "semantic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    // Seed links deterministically (extraction is async + heuristic; we test the
    // DROP, not extraction): an entity present in the NEW text ("Aurelian") must
    // be kept; one absent must drop; and "Volkan" is a substring of the new
    // text's "Volkanic" — naive substring would false-KEEP it, word-boundary
    // must drop it.
    {
        let conn = db.conn();
        for e in ["Volkan", "Aurelian"] {
            conn.execute(
                "INSERT OR IGNORE INTO memory_entities (memory_rid, entity_name) VALUES (?1, ?2)",
                params![rid, e],
            )
            .unwrap();
        }
    }
    // Also seed the IN-MEMORY graph_index — this is what recall's
    // expand_entities actually reads. nuron's live verification showed a
    // durable-only drop leaves this index stale, so the corrected record keeps
    // being served under its old association. The durable-table assertion alone
    // gave a false green; this test now covers the live path.
    {
        let mut gi = db.graph_index.write();
        gi.link_memory(&rid, "Volkan");
        gi.link_memory(&rid, "Aurelian");
    }
    db.correct(
        &rid,
        Some("The Aurelian highland lakes near Volkanic rock"),
        None,
        None,
        None,
        "winter grounds moved",
    )
    .unwrap();
    let links: Vec<String> = {
        let conn = db.conn();
        let mut stmt = conn
            .prepare("SELECT entity_name FROM memory_entities WHERE memory_rid = ?1")
            .unwrap();
        stmt.query_map(params![rid], |r| r.get::<_, String>(0))
            .unwrap()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap()
    };
    assert!(
        links.contains(&"Aurelian".to_string()),
        "durable entity present in corrected text must be kept: {links:?}"
    );
    assert!(
        !links.contains(&"Volkan".to_string()),
        "durable stale entity absent from corrected text must be dropped (and NOT false-kept by 'Volkanic'): {links:?}"
    );
    // The in-memory graph_index (what recall's expand_entities reads) must ALSO
    // reflect the drop — the assertion that would have caught nuron's live
    // finding.
    let gi_entities: Vec<String> = db
        .graph_index
        .read()
        .entities_for_memory(&rid)
        .iter()
        .map(|s| s.to_lowercase())
        .collect();
    assert!(
        gi_entities.iter().any(|e| e == "aurelian"),
        "in-memory graph_index must keep the surviving entity: {gi_entities:?}"
    );
    assert!(
        !gi_entities.iter().any(|e| e == "volkan"),
        "in-memory graph_index must EVICT the stale entity (recall's expand_entities reads it): {gi_entities:?}"
    );
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn correct_correct_revision_chain_records_true_prior_state() {
    // sol finding 7: two successive text corrections must chain — the
    // second revision records the FIRST correction's state as its prior,
    // not the original (which would happen if both snapshotted the same
    // pre-loop `original`). Sequential here; the fix (re-read prior INSIDE
    // the serialized tx) is what makes the concurrent case correct too.
    let db = YantrikDB::with_default(":memory:").unwrap();
    let rid = db
        .record_text(
            "the capital is Paris",
            "semantic",
            0.6,
            0.0,
            604800.0,
            &empty_meta(),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();

    db.correct(
        &rid,
        Some("the capital is Lyon"),
        None,
        Some(0.7),
        None,
        "first fix",
    )
    .unwrap();
    db.correct(
        &rid,
        Some("the capital is Marseille"),
        None,
        Some(0.8),
        None,
        "second fix",
    )
    .unwrap();

    let hist = db.history(&rid).unwrap();
    assert_eq!(hist.len(), 2);
    // Revision 1's prior is the ORIGINAL.
    assert_eq!(hist[0].prior_text, "the capital is Paris");
    assert!((hist[0].prior_importance - 0.6).abs() < 1e-9);
    // Revision 2's prior is the FIRST correction's state — the chain fix.
    assert_eq!(hist[1].prior_text, "the capital is Lyon");
    assert!((hist[1].prior_importance - 0.7).abs() < 1e-9);
    // Both text corrections captured prior embedding provenance.
    assert!(hist[0].prior_embedding_hash.is_some());
    assert!(hist[1].prior_embedding_hash.is_some());
    // The two prior vectors differ (Paris vs Lyon embed differently).
    assert_ne!(hist[0].prior_embedding_hash, hist[1].prior_embedding_hash);

    // Current state is the last correction.
    assert_eq!(
        db.get(&rid).unwrap().unwrap().text,
        "the capital is Marseille"
    );
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn correct_then_forget_leaves_no_resurrected_vector() {
    // sol finding 3 (correct/forget shape): after a re-embedding correction
    // then a forget, the record must be GONE from recall — the correction's
    // superseding delta append must not resurface a tombstoned SQL row.
    let db = YantrikDB::with_default(":memory:").unwrap();
    let rid = db
        .record_text(
            "hiking in the alps",
            "semantic",
            0.6,
            0.0,
            604800.0,
            &empty_meta(),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    db.correct(
        &rid,
        Some("quarterly revenue growth"),
        None,
        None,
        None,
        "topic",
    )
    .unwrap();
    db.forget(&rid).unwrap();

    let q = db.embed("quarterly revenue report").unwrap();
    let hits = db
        .recall(
            &q, 10, None, None, false, false, None, true, None, None, None, None, None, false,
            None, // event_after (#149)
            None, // event_before (#149)
        )
        .unwrap();
    assert!(
        !hits.iter().any(|h| h.rid == rid),
        "forgotten record must not resurface via the correction's append"
    );
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn correction_clears_reembed_staging_columns() {
    // sol finding 4: a text correction must clear embedding_new so a
    // reembed mid-Encoding re-encodes this row (from the new text) instead
    // of promoting the stale staged vector at swap. We simulate staging
    // directly, then correct, and assert the staging is cleared.
    let db = YantrikDB::with_default(":memory:").unwrap();
    let rid = db
        .record_text(
            "old topic text",
            "semantic",
            0.6,
            0.0,
            604800.0,
            &empty_meta(),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    // Simulate a reembed having staged an embedding for the OLD text.
    db.conn()
        .execute(
            "UPDATE memories SET embedding_new = X'0011', embedding_new_model = 'stale-model' \
             WHERE rid = ?1",
            rusqlite::params![rid],
        )
        .unwrap();

    db.correct(
        &rid,
        Some("completely different new topic"),
        None,
        None,
        None,
        "fix",
    )
    .unwrap();

    let (en, enm): (Option<Vec<u8>>, Option<String>) = db
        .conn()
        .query_row(
            "SELECT embedding_new, embedding_new_model FROM memories WHERE rid = ?1",
            rusqlite::params![rid],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )
        .unwrap();
    assert!(en.is_none(), "embedding_new cleared by correction");
    assert!(enm.is_none(), "embedding_new_model cleared by correction");
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn record_batch_defers_during_reembed_instead_of_writing_a_doomed_generation() {
    // REGRESSION: record_batch never consulted the write_router. It loaded its
    // SearchState snapshot unguarded, so a reembed cutover could complete its
    // swap mid-batch and the batch would commit rows + appends stamped with an
    // `embedding_generation` that was being discarded. record() has always
    // guarded this (record.rs:105 before :138); batch silently did not, while
    // its own comment claimed every append lands on one anchored generation.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let inputs = vec![RecordInput {
        created_at: None,
        idempotency_key: None,
        text: "written during a cutover".to_string(),
        memory_type: "semantic".to_string(),
        importance: 0.5,
        valence: 0.0,
        half_life: 604800.0,
        metadata: empty_meta(),
        embedding: vec_seed(1.0, 8),
        namespace: "default".to_string(),
        certainty: 0.8,
        domain: "general".to_string(),
        source: "user".to_string(),
        emotional_state: None,
    }];

    // Simulate a reembed cutover in flight.
    db.write_router.switch_to_queueing();

    let err = db.record_batch(&inputs).unwrap_err();
    assert!(
        matches!(
            err,
            crate::error::YantrikDbError::BatchDeferredDuringReembed { count: 1 }
        ),
        "batch must defer rather than write against a doomed generation, got {err:?}"
    );
    // Retryable means exactly that: NO durable state was touched.
    let count: i64 = db
        .conn()
        .query_row("SELECT COUNT(*) FROM memories", [], |r| r.get(0))
        .unwrap();
    assert_eq!(count, 0, "a deferred batch must write nothing");

    // Once the cutover finishes the same batch succeeds verbatim.
    db.write_router.switch_to_normal();
    let rids = db.record_batch(&inputs).unwrap();
    assert_eq!(rids.len(), 1);
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn replaying_materialize_record_post_does_not_inflate_mention_count() {
    // REGRESSION: apply_materialize_record_post bumped entities.mention_count
    // unconditionally, under a comment calling the statement "idempotent". The
    // INSERT does not FAIL on conflict, but `mention_count = mention_count + 1`
    // is not idempotent — and this op really is executed more than once:
    // mark_op_applied is a completion ACK, not a pre-work claim, so N workers
    // draining the same op all do the work and only then race on
    // `WHERE applied = 0`; and an op that errors stays pending and is retried.
    // Each duplicate inflated the count, which feeds the IDF term in patterns.rs
    // and silently skewed salience.
    //
    // Resetting applied=0 and draining again is exactly the retry/duplicate-worker
    // shape, and is the only way to reach the private handler from here.
    let db = YantrikDB::with_default(":memory:").unwrap();
    db.record_text(
        "Alice and Bob shipped the Acme project",
        "semantic",
        0.5,
        0.0,
        604800.0,
        &empty_meta(),
        "default",
        0.8,
        "general",
        "user",
        None,
    )
    .unwrap();

    db.apply_pending_ops_once(100).unwrap();
    let counts_after_first: Vec<(String, i64)> = {
        let conn = db.conn();
        let mut stmt = conn
            .prepare("SELECT name, mention_count FROM entities ORDER BY name")
            .unwrap();
        let rows = stmt
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
            .unwrap()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();
        rows
    };
    assert!(
        !counts_after_first.is_empty(),
        "test needs at least one extracted entity to be meaningful"
    );

    // Simulate the duplicate execution: put the op back to pending and drain again.
    db.conn()
        .execute(
            "UPDATE oplog SET applied = 0 WHERE op_type = ?1",
            rusqlite::params![crate::engine::op_types::OP_MATERIALIZE_RECORD_POST],
        )
        .unwrap();
    db.apply_pending_ops_once(100).unwrap();

    let counts_after_replay: Vec<(String, i64)> = {
        let conn = db.conn();
        let mut stmt = conn
            .prepare("SELECT name, mention_count FROM entities ORDER BY name")
            .unwrap();
        let rows = stmt
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
            .unwrap()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();
        rows
    };
    assert_eq!(
        counts_after_first, counts_after_replay,
        "replaying materialize_record_post must not re-count mentions"
    );
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn record_batch_replicates_to_a_peer() {
    // REGRESSION: record_batch logged ONE opaque op — log_op("record_batch",
    // {count, rids}) — and replication has no "record_batch" arm, so peers hit
    // the `_ =>` forward-compat catch-all and silently dropped it. The payload
    // carried no text/embedding/scalars, so it could not have rebuilt the rows
    // even with an arm. Every memory written via the public record_batch() API
    // therefore never reached any peer, with no error anywhere.
    use crate::replication::{apply_ops, extract_ops_since};
    let leader = YantrikDB::new(":memory:", 8).unwrap();
    let follower = YantrikDB::new(":memory:", 8).unwrap();

    let mk = |text: &str, seed: f32| RecordInput {
        created_at: None,
        idempotency_key: None,
        text: text.to_string(),
        memory_type: "semantic".to_string(),
        importance: 0.6,
        valence: 0.25,
        half_life: 604800.0,
        metadata: serde_json::json!({"tag": text}),
        embedding: vec_seed(seed, 8),
        namespace: "default".to_string(),
        certainty: 0.7,
        domain: "general".to_string(),
        source: "user".to_string(),
        emotional_state: None,
    };
    let inputs = vec![mk("first batch item", 1.0), mk("second batch item", 2.0)];
    let rids = leader.record_batch(&inputs).unwrap();
    assert_eq!(rids.len(), 2);

    let ops = extract_ops_since(&leader.conn(), None, None, None, 100).unwrap();
    // One canonical "record" op per item — not one opaque batch op.
    let record_ops: Vec<_> = ops.iter().filter(|o| o.op_type == "record").collect();
    assert_eq!(
        record_ops.len(),
        2,
        "expected one canonical record op per batch item, got {:?}",
        ops.iter().map(|o| &o.op_type).collect::<Vec<_>>()
    );
    assert!(
        !ops.iter().any(|o| o.op_type == "record_batch"),
        "the unreplicable record_batch op must be gone"
    );

    apply_ops(&follower, &ops).unwrap();

    // The follower actually has the memories, with content intact.
    for (rid, input) in rids.iter().zip(inputs.iter()) {
        let got = follower
            .get(rid)
            .unwrap()
            .unwrap_or_else(|| panic!("batch item {rid} did not replicate"));
        assert_eq!(got.text, input.text);
        assert_eq!(got.source, "user");
        assert_eq!(got.namespace, "default");
        assert_eq!(got.certainty, input.certainty);
        assert_eq!(got.valence, input.valence);
        assert_eq!(got.metadata["tag"], input.text.as_str());
    }
}

#[cfg(feature = "bundled-embedder")]
#[test]
fn follower_replay_applies_corrected_embedding() {
    // sol finding 6: a re-embedding correction replicates its EXACT bytes,
    // so a same-DEK follower's stored vector + recall track the new meaning
    // (not a re-embed, not the stale old vector).
    use crate::replication::{apply_ops, extract_ops_since};
    let leader = YantrikDB::with_default(":memory:").unwrap();
    let follower = YantrikDB::with_default(":memory:").unwrap();

    let rid = leader
        .record_text(
            "sunny beach vacation",
            "semantic",
            0.6,
            0.0,
            604800.0,
            &empty_meta(),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    // Replicate the record first.
    let ops = extract_ops_since(&leader.conn(), None, None, None, 100).unwrap();
    apply_ops(&follower, &ops).unwrap();

    // Correct on the leader (re-embed), then replicate the correct op.
    leader
        .correct(
            &rid,
            Some("corporate tax filing deadline"),
            None,
            None,
            None,
            "topic",
        )
        .unwrap();
    let correct_ops: Vec<_> = extract_ops_since(&leader.conn(), None, None, None, 100)
        .unwrap()
        .into_iter()
        .filter(|o| o.op_type == "correct")
        .collect();
    assert_eq!(correct_ops.len(), 1);
    assert!(
        correct_ops[0].embedding.is_some(),
        "correct op carries exact embedding bytes"
    );
    apply_ops(&follower, &correct_ops).unwrap();

    // Follower's SQL text updated.
    let ftext: String = follower
        .conn()
        .query_row(
            "SELECT text FROM memories WHERE rid = ?1",
            rusqlite::params![rid],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(ftext, "corporate tax filing deadline");

    // Follower recall reflects the NEW meaning (vector applied, not stale).
    let q = follower.embed("corporate tax deadline").unwrap();
    let hits = follower
        .recall(
            &q, 5, None, None, false, false, None, true, None, None, None, None, None, false,
            None, // event_after (#149)
            None, // event_before (#149)
        )
        .unwrap();
    assert!(
        hits.iter().any(|h| h.rid == rid),
        "follower retrieves the record under its corrected meaning"
    );
}

#[test]
fn correct_writes_revision_row_with_prior_state() {
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "v0",
            "episodic",
            0.5,
            -0.2,
            604800.0,
            &serde_json::json!({"key": "before"}),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    // v0.9.3: metadata/importance/valence correction (text refused).
    let _ = db
        .correct(
            &rid,
            None,
            Some(&serde_json::json!({"key": "after"})),
            Some(0.7),
            Some(0.3),
            "first correction",
        )
        .unwrap();
    let history = db.history(&rid).unwrap();
    assert_eq!(history.len(), 1, "one revision expected");
    let rev = &history[0];
    assert_eq!(rev.revision_num, 1);
    assert_eq!(rev.prior_text, "v0");
    assert!((rev.prior_importance - 0.5).abs() < 1e-9);
    assert!((rev.prior_valence + 0.2).abs() < 1e-9);
    assert_eq!(rev.reason, "first correction");
    assert_eq!(
        rev.prior_metadata.get("key").and_then(|v| v.as_str()),
        Some("before")
    );
}

#[test]
fn correct_multiple_revisions_increment_revision_num() {
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "v0",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    // v0.9.3: revision-number chain exercised via importance corrections
    // (text corrections are refused pending vector-coherent correct).
    let r1 = db
        .correct(&rid, None, None, Some(0.6), None, "first")
        .unwrap();
    let r2 = db
        .correct(&rid, None, None, Some(0.7), None, "second")
        .unwrap();
    let r3 = db
        .correct(&rid, None, None, Some(0.9), None, "third")
        .unwrap();
    assert_eq!(r1.revision_num, 1);
    assert_eq!(r2.revision_num, 2);
    assert_eq!(r3.revision_num, 3);

    let history = db.history(&rid).unwrap();
    assert_eq!(history.len(), 3, "three revisions expected");
    // history returns oldest-first; prior importances chain through.
    assert!((history[0].prior_importance - 0.5).abs() < 1e-9);
    assert!((history[1].prior_importance - 0.6).abs() < 1e-9);
    assert!((history[2].prior_importance - 0.7).abs() < 1e-9);

    let final_state = db.get(&rid).unwrap().unwrap();
    assert_eq!(final_state.text, "v0", "text untouched throughout");
    assert!((final_state.importance - 0.9).abs() < 1e-9);
}

#[test]
fn correct_rejects_empty_reason() {
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "text",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();

    // Empty string.
    let err = db
        .correct(&rid, Some("new"), None, None, None, "")
        .expect_err("empty reason must be rejected");
    match err {
        crate::error::YantrikDbError::InvalidInput(msg) => {
            assert!(msg.contains("reason"), "got: {msg}");
        }
        other => panic!("expected InvalidInput, got {other:?}"),
    }

    // Whitespace-only.
    let err2 = db
        .correct(&rid, Some("new"), None, None, None, "   \t\n  ")
        .expect_err("whitespace-only reason must be rejected");
    assert!(matches!(
        err2,
        crate::error::YantrikDbError::InvalidInput(_)
    ));
}

#[test]
fn correct_rejects_no_mutation_fields() {
    // All None mutation fields == no-op correction; must be rejected.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "text",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    let err = db
        .correct(&rid, None, None, None, None, "no fields supplied")
        .expect_err("no-op correction must be rejected");
    assert!(matches!(err, crate::error::YantrikDbError::InvalidInput(_)));
}

#[test]
fn correct_preserves_inbound_graph_edges() {
    // Inbound link integrity: a graph edge pointing TO the corrected
    // rid must continue to resolve after correct(). This is the central
    // win of the v0.7.20 semantics over the v0.7.19 "tombstone + new rid"
    // approach where inbound references dangled.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid_subject = db
        .record(
            "subject memory",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    // Create an entity-graph edge that mentions rid_subject as an
    // endpoint. Using the entity-relate path (rid acts as one of two
    // entity names — the test only cares that an edge persists).
    db.relate("anchor_entity", &rid_subject, "tags", 1.0)
        .unwrap();
    let edges_before = db.get_edges(&rid_subject).unwrap();
    assert!(!edges_before.is_empty(), "edge should exist before correct");

    // Correct the memory.
    // v0.9.3: importance correction (text refused); the link-integrity
    // property under test is about rid stability, not the text.
    db.correct(
        &rid_subject,
        None,
        None,
        Some(0.9),
        None,
        "test link integrity",
    )
    .unwrap();

    // Inbound edges must still resolve (rid_subject still exists).
    let edges_after = db.get_edges(&rid_subject).unwrap();
    assert_eq!(
        edges_before.len(),
        edges_after.len(),
        "inbound edges must be preserved across correct(); \
         this is the central v0.7.20 win over the v0.7.19 tombstone semantics"
    );
}

#[test]
fn correct_history_empty_for_never_corrected_record() {
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let rid = db
        .record(
            "never corrected",
            "episodic",
            0.5,
            0.0,
            604800.0,
            &empty_meta(),
            &vec_seed(1.0, 8),
            "default",
            0.8,
            "general",
            "user",
            None,
        )
        .unwrap();
    let history = db.history(&rid).unwrap();
    assert!(
        history.is_empty(),
        "never-corrected record has no revisions"
    );
}

#[test]
fn schema_v30_fresh_install_has_record_revisions_table() {
    // Fresh install must apply v30 schema and create the
    // record_revisions table + its index. Regression guard against
    // forgetting to add the table to SCHEMA_SQL alongside the migration.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let conn = db.conn();
    let count: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master \
             WHERE type = 'table' AND name = 'record_revisions'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(
        count, 1,
        "record_revisions table must exist on fresh install"
    );

    // Schema version meta should be at least 30.
    let schema_version: String = conn
        .query_row(
            "SELECT value FROM meta WHERE key = 'schema_version'",
            [],
            |r| r.get(0),
        )
        .unwrap();
    let v: i32 = schema_version.parse().unwrap();
    assert!(
        v >= crate::base::schema::SCHEMA_VERSION,
        "schema_version must be at least {}, got {v}",
        crate::base::schema::SCHEMA_VERSION,
    );
}

#[test]
fn schema_v31_fresh_install_has_record_links_table() {
    // Issue #48: fresh install must create record_links + both covering
    // indexes. Regression guard against forgetting the table in SCHEMA_SQL
    // alongside MIGRATE_V30_TO_V31.
    let db = YantrikDB::new(":memory:", 8).unwrap();
    let conn = db.conn();
    let table: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master \
             WHERE type = 'table' AND name = 'record_links'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(table, 1, "record_links table must exist on fresh install");

    let idx: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' \
             AND name IN ('idx_record_links_source', 'idx_record_links_target')",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(idx, 2, "both record_links covering indexes must exist");

    assert!(crate::base::schema::SCHEMA_VERSION >= 31);
}

/// Event time follows the corrected TEXT. Before 2026-08-15, correcting
/// "March 15, 2024" to "April 20, 2024" left event_dates/event_time_min/max
/// saying March — metadata contradicting its own text, on the one surface
/// that rewrites text. The three keys re-derive as ONE UNIT unless this
/// correction's own metadata_merge supplies any of them (caller ownership).
#[test]
#[cfg(feature = "bundled-embedder")] // with_default needs the bundled embedder; slim CI compiles tests too
fn correct_rederives_event_time_from_new_text() {
    let db = YantrikDB::with_default(":memory:").unwrap();
    let rid = db
        .record_text(
            "the launch deadline is March 15, 2024",
            "semantic",
            0.5,
            0.0,
            604800.0,
            &serde_json::json!({}),
            "default",
            0.8,
            "work",
            "user",
            None,
        )
        .unwrap();
    let before = db.get_memory(&rid).unwrap().unwrap();
    assert_eq!(
        before
            .metadata
            .get("event_dates")
            .and_then(|v| v.as_array())
            .map(|a| a.len()),
        Some(1),
        "precondition: write-path extraction produced March"
    );

    db.correct(
        &rid,
        Some("the launch deadline is April 20, 2024"),
        None,
        None,
        None,
        "deadline moved",
    )
    .unwrap();

    let after = db.get_memory(&rid).unwrap().unwrap();
    let dates: Vec<String> = after
        .metadata
        .get("event_dates")
        .and_then(|v| v.as_array())
        .map(|a| {
            a.iter()
                .filter_map(|d| d.as_str().map(String::from))
                .collect()
        })
        .unwrap_or_default();
    assert_eq!(
        dates,
        vec!["2024-04-20".to_string()],
        "event keys must follow the corrected text, got {dates:?}"
    );
}

/// forget() must delete the DURABLE memory_entities rows, not just the
/// in-memory index — the rebuild loads durable rows with no tombstone
/// filter, so leftovers resurrected forgotten records' links on restart.
#[test]
#[cfg(feature = "bundled-embedder")] // with_default needs the bundled embedder; slim CI compiles tests too
fn forget_deletes_durable_entity_links() {
    let db = YantrikDB::with_default(":memory:").unwrap();
    // Batch path extracts entities INLINE (the single-record path is
    // async), so the durable rows exist deterministically before forget.
    let rids = db
        .record_batch(&[crate::types::RecordInput {
            created_at: None,
            idempotency_key: None,
            text: "Alice Chen is the CEO of Acme Corp".to_string(),
            memory_type: "semantic".to_string(),
            importance: 0.5,
            valence: 0.0,
            half_life: 604800.0,
            metadata: serde_json::json!({}),
            embedding: {
                let raw: Vec<f32> = (0..64).map(|i| (i as f32 + 1.0) * 0.1).collect();
                let norm: f32 = raw.iter().map(|x| x * x).sum::<f32>().sqrt();
                raw.iter().map(|x| x / norm).collect()
            },
            namespace: "default".to_string(),
            certainty: 0.8,
            domain: "people".to_string(),
            source: "user".to_string(),
            emotional_state: None,
        }])
        .unwrap();
    let rid = rids[0].clone();
    let conn = db.conn();
    let before: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM memory_entities WHERE memory_rid = ?1",
            rusqlite::params![rid],
            |r| r.get(0),
        )
        .unwrap();
    drop(conn);
    assert!(before > 0, "precondition: entity links exist");

    db.forget(&rid).unwrap();

    let conn = db.conn();
    let after: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM memory_entities WHERE memory_rid = ?1",
            rusqlite::params![rid],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(after, 0, "durable entity links must die with the record");
}