1use std::collections::{BTreeMap, BTreeSet};
47
48use rusqlite::{Connection, OptionalExtension};
49
50use crate::namespace_census::{self, NamespaceCensus, NamespaceConstraint};
51
52#[path = "namespace_move_knowledge.rs"]
53mod knowledge_move;
54#[path = "namespace_move_pack_tables.rs"]
55mod pack_tables;
56#[path = "namespace_move_routes.rs"]
57mod subject_routes;
58use knowledge_move::{move_knowledge_atoms, ordinary_atom_count};
59use pack_tables::{move_task_audit, settle_pack_tables};
60use subject_routes::routed_subjects;
61
62#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
70pub enum SubjectClass {
71 Note(String),
72 Entity(String),
73 Edge,
74 EdgeRelation(String),
75 Atom,
76 Domain,
77}
78
79impl SubjectClass {
80 pub fn parse(key: &str) -> Result<Self, MoveError> {
83 match key {
84 "edge" => return Ok(Self::Edge),
85 "atom" => return Ok(Self::Atom),
86 "domain" => return Ok(Self::Domain),
87 _ => {}
88 }
89 match key.split_once(':') {
90 Some(("note", kind)) if !kind.is_empty() => Ok(Self::Note(kind.to_string())),
91 Some(("entity", kind)) if !kind.is_empty() => Ok(Self::Entity(kind.to_string())),
92 Some(("edge", relation))
93 if khive_types::EdgeRelation::VALID_NAMES.contains(&relation) =>
94 {
95 Ok(Self::EdgeRelation(relation.to_string()))
96 }
97 _ => Err(MoveError::UnknownSubjectClass {
98 key: key.to_string(),
99 }),
100 }
101 }
102
103 pub fn render(&self) -> String {
104 match self {
105 Self::Note(kind) => format!("note:{kind}"),
106 Self::Entity(kind) => format!("entity:{kind}"),
107 Self::Edge => "edge".to_string(),
108 Self::EdgeRelation(relation) => format!("edge:{relation}"),
109 Self::Atom => "atom".to_string(),
110 Self::Domain => "domain".to_string(),
111 }
112 }
113}
114
115#[derive(Clone, Debug, PartialEq, Eq)]
117pub struct MoveRoute {
118 pub class: SubjectClass,
119 pub target: String,
120}
121
122#[derive(Clone, Debug, PartialEq, Eq)]
124pub struct MoveRequest {
125 pub source: String,
126 pub routes: Vec<MoveRoute>,
127 pub now_micros: Option<i64>,
132}
133
134impl MoveRequest {
135 pub fn new(source: impl Into<String>, routes: Vec<MoveRoute>) -> Self {
139 Self {
140 source: source.into(),
141 routes,
142 now_micros: None,
143 }
144 }
145
146 pub fn at(mut self, now_micros: i64) -> Self {
148 self.now_micros = Some(now_micros);
149 self
150 }
151
152 fn route_for(&self, class: &SubjectClass) -> Option<&MoveRoute> {
153 self.routes
154 .iter()
155 .find(|route| &route.class == class)
156 .or_else(|| match class {
157 SubjectClass::EdgeRelation(_) => self
158 .routes
159 .iter()
160 .find(|route| route.class == SubjectClass::Edge),
161 _ => None,
162 })
163 }
164
165 fn single_target(&self) -> Option<&str> {
170 let mut targets = self.routes.iter().map(|r| r.target.as_str());
171 let first = targets.next()?;
172 targets.all(|t| t == first).then_some(first)
173 }
174}
175
176#[derive(Clone, Debug, Default, PartialEq, Eq)]
183pub struct MoveCounts {
184 pub subjects: BTreeMap<String, u64>,
189 pub rows: BTreeMap<String, u64>,
191 pub left_behind: BTreeMap<String, u64>,
194 pub ann_log_appended: u64,
196 pub live_policies_left_behind: u64,
200 pub grants_in_force_left_behind: u64,
204}
205
206#[derive(Clone, Debug, PartialEq, Eq)]
208pub struct StreamMember {
209 pub note_id: String,
210 pub stream: String,
211 pub seq: i64,
212}
213
214#[derive(Clone, Debug, PartialEq, Eq)]
216pub struct Collision {
217 pub table: String,
218 pub constraint: String,
219 pub target: String,
220 pub key: String,
226}
227
228#[derive(Debug)]
230pub enum MoveError {
231 UnknownSubjectClass {
233 key: String,
234 },
235 DuplicateRoute {
238 class: String,
239 },
240 TargetIsSource {
243 class: String,
244 },
245 UnknownTable {
253 table: String,
254 rows: u64,
255 },
256 UnroutedClass {
262 class: String,
263 rows: u64,
264 },
265 StreamMembers {
273 notes: Vec<StreamMember>,
274 },
275 UnroutableVector {
278 subject_id: String,
279 table: String,
280 destinations: u64,
281 },
282 Collisions {
284 collisions: Vec<Collision>,
285 },
286 Sqlite(rusqlite::Error),
287}
288
289impl From<rusqlite::Error> for MoveError {
290 fn from(error: rusqlite::Error) -> Self {
291 Self::Sqlite(error)
292 }
293}
294
295impl std::fmt::Display for MoveError {
296 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
297 match self {
298 Self::UnknownSubjectClass { key } => write!(
299 f,
300 "no subject class named {key:?}; expected note:<kind>, entity:<kind>, edge:<relation>, edge, atom or domain"
301 ),
302 Self::DuplicateRoute { class } => {
303 write!(f, "{class} is routed more than once")
304 }
305 Self::TargetIsSource { class } => {
306 write!(f, "{class} is routed to the namespace it is already in")
307 }
308 Self::UnknownTable { table, rows } => write!(
309 f,
310 "{table} holds {rows} row(s) in the source namespace and this build has no rule \
311 for it; a table added by a later migration refuses a move rather than being \
312 left behind by one"
313 ),
314 Self::UnroutedClass { class, rows } => write!(
315 f,
316 "{class} has {rows} row(s) in the source namespace and no route; \
317 a class routed with zero rows succeeds reporting zero, an unrouted one refuses"
318 ),
319 Self::StreamMembers { notes } => {
320 write!(f, "{} note(s) belong to a stream and cannot change namespace: ", notes.len())?;
321 for (i, member) in notes.iter().enumerate() {
322 if i > 0 {
323 f.write_str(", ")?;
324 }
325 write!(f, "{} ({} seq {})", member.note_id, member.stream, member.seq)?;
326 }
327 Ok(())
328 }
329 Self::UnroutableVector { subject_id, table, destinations } => write!(
330 f,
331 "unroutable_vector: {table:?} subject {subject_id:?} has {destinations} source-subject destinations; a partitioned move requires exactly one"
332 ),
333 Self::Collisions { collisions } => {
334 write!(f, "{} collision(s): ", collisions.len())?;
335 for (i, collision) in collisions.iter().enumerate() {
336 if i > 0 {
337 f.write_str(", ")?;
338 }
339 write!(
340 f,
341 "{}.{} in {} already holds {}",
342 collision.table, collision.constraint, collision.target, collision.key
343 )?;
344 }
345 Ok(())
346 }
347 Self::Sqlite(error) => write!(f, "{error}"),
348 }
349 }
350}
351
352impl std::error::Error for MoveError {}
353
354const NAMESPACE_SCOPED_TABLES: &[&str] = &[
362 "brain_profile_snapshots",
363 "brain_event_log",
364 "proposals_open",
365];
366
367const SUBJECT_KEYED_TABLES: &[&str] = &["brain_implicit_mass", "brain_serve_ledger"];
375
376const MEMORY_VISIBILITY_RECEIPTS: &str = "memory_visibility_receipts";
379const MEMORY_VISIBILITY_FENCES: &str = "memory_visibility_fences";
380const MEMORY_VISIBILITY_EPOCHS: &str = "memory_visibility_epochs";
381
382const LEAVE_BEHIND_TABLES: &[&str] = &["comm_sender_transport", "sessions", "session_messages"];
387
388#[derive(Clone, Copy, Debug, PartialEq, Eq)]
398pub enum TableDisposition {
399 Subject,
401 Derived { trigger_maintained: bool },
418 DerivedRowidMap,
420 Appended,
424 History,
427 ConsumerWatermark,
429 RefusedBySchema,
432 NamespaceScopedAggregate,
436 SubjectKeyed { subject_column: &'static str },
438 LeaveBehind,
441}
442
443pub fn disposition(table: &namespace_census::NamespaceTable) -> Option<TableDisposition> {
450 use TableDisposition::*;
451 Some(match table.name.as_str() {
452 "notes" | "entities" | "graph_edges" | "knowledge_atoms" | "knowledge_domains" => Subject,
453
454 "knowledge_sections" | "atom_weights" => SubjectKeyed {
457 subject_column: "atom_id",
458 },
459
460 "proposals_open" => NamespaceScopedAggregate,
466
467 "fts_knowledge" | "fts_sections" => Derived {
468 trigger_maintained: true,
469 },
470 "fts_notes" | "fts_entities" => Derived {
471 trigger_maintained: false,
472 },
473 "fts_notes_rowids" | "fts_entities_rowids" => DerivedRowidMap,
474 "vector_provenance" => Derived {
476 trigger_maintained: false,
477 },
478
479 "ann_write_log" => Appended,
480 "events" | "weight_events" => History,
481 "ann_consumer_watermark" | "ann_consumer_pending" => ConsumerWatermark,
482 "note_streams" => RefusedBySchema,
483
484 "brain_profile_snapshots" | "brain_event_log" => NamespaceScopedAggregate,
485 "brain_implicit_mass" | "brain_serve_ledger" => SubjectKeyed {
486 subject_column: "target_id",
487 },
488 "memory_visibility_receipts" => SubjectKeyed {
489 subject_column: "note_id",
490 },
491 "memory_visibility_fences" => SubjectKeyed {
492 subject_column: "note_id",
493 },
494 "memory_visibility_epochs" => SubjectKeyed {
495 subject_column: "note_id",
496 },
497
498 "gtd_lifecycle_audit" => SubjectKeyed {
505 subject_column: "note_id",
506 },
507 "knowledge_eval_runs" => NamespaceScopedAggregate,
510 "exec_runs" | "exec_events" | "git_receipts" => NamespaceScopedAggregate,
516 "tool_policy" | "tool_grants" => LeaveBehind,
520 "retrieval_snapshots" => Derived {
524 trigger_maintained: false,
525 },
526 _ if LEAVE_BEHIND_TABLES.contains(&table.name.as_str()) => LeaveBehind,
527
528 _ if is_runtime_vector_table(table) => Derived {
540 trigger_maintained: false,
541 },
542
543 _ => return None,
544 })
545}
546
547fn is_runtime_vector_table(table: &namespace_census::NamespaceTable) -> bool {
553 table.name.starts_with("vec_") && table.virtual_table
554}
555
556#[derive(Debug, Default)]
559struct SourceInventory {
560 note_kinds: BTreeMap<String, u64>,
562 entity_kinds: BTreeMap<String, u64>,
563 edge_relations: BTreeMap<String, u64>,
564 atoms: u64,
565 domains: u64,
566 unknown: Vec<(String, u64)>,
569}
570
571fn count_in_namespace(conn: &Connection, table: &str, namespace: &str) -> rusqlite::Result<u64> {
572 let sql = format!(
573 "SELECT COUNT(*) FROM {} WHERE namespace = ?1",
574 namespace_census::quote_ident(table)
575 );
576 conn.query_row(&sql, [namespace], |row| row.get::<_, i64>(0))
577 .map(|n| n as u64)
578}
579
580fn kinds_in_namespace(
581 conn: &Connection,
582 table: &str,
583 namespace: &str,
584) -> rusqlite::Result<BTreeMap<String, u64>> {
585 let sql = format!(
586 "SELECT kind, COUNT(*) FROM {} WHERE namespace = ?1 GROUP BY kind",
587 namespace_census::quote_ident(table)
588 );
589 let mut stmt = conn.prepare(&sql)?;
590 let rows = stmt.query_map([namespace], |row| {
591 Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64))
592 })?;
593 let mut out = BTreeMap::new();
594 for row in rows {
595 let (kind, count) = row?;
596 out.insert(kind, count);
597 }
598 Ok(out)
599}
600
601fn edge_relations_in_namespace(
602 conn: &Connection,
603 namespace: &str,
604) -> rusqlite::Result<BTreeMap<String, u64>> {
605 conn.prepare(
606 "SELECT relation, COUNT(*) FROM graph_edges WHERE namespace = ?1 GROUP BY relation",
607 )?
608 .query_map([namespace], |row| {
609 Ok((row.get(0)?, row.get::<_, i64>(1)? as u64))
610 })?
611 .collect()
612}
613
614fn read_source(
622 conn: &Connection,
623 census: &NamespaceCensus,
624 source: &str,
625) -> Result<SourceInventory, MoveError> {
626 let mut inventory = SourceInventory {
627 note_kinds: kinds_in_namespace(conn, "notes", source)?,
628 entity_kinds: kinds_in_namespace(conn, "entities", source)?,
629 edge_relations: edge_relations_in_namespace(conn, source)?,
630 atoms: ordinary_atom_count(conn, source)?,
631 domains: count_in_namespace(conn, "knowledge_domains", source)?,
632 unknown: Vec::new(),
633 };
634
635 for table in &census.tables {
636 if disposition(table).is_some() {
637 continue;
638 }
639 let count = count_in_namespace(conn, &table.name, source)?;
640 if count > 0 {
641 inventory.unknown.push((table.name.clone(), count));
642 }
643 }
644 Ok(inventory)
645}
646
647fn stream_members(conn: &Connection, source: &str) -> rusqlite::Result<Vec<StreamMember>> {
654 let mut stmt = conn.prepare(
655 "SELECT note_id, stream, seq FROM note_streams \
656 WHERE namespace = ?1 ORDER BY stream, seq",
657 )?;
658 let rows = stmt.query_map([source], |row| {
659 Ok(StreamMember {
660 note_id: row.get(0)?,
661 stream: row.get(1)?,
662 seq: row.get(2)?,
663 })
664 })?;
665 rows.collect()
666}
667
668pub fn validate(
673 conn: &Connection,
674 census: &NamespaceCensus,
675 request: &MoveRequest,
676) -> Result<(), MoveError> {
677 let mut seen = BTreeSet::new();
678 for route in &request.routes {
679 if let SubjectClass::EdgeRelation(relation) = &route.class {
680 if !khive_types::EdgeRelation::VALID_NAMES.contains(&relation.as_str()) {
681 return Err(MoveError::UnknownSubjectClass {
682 key: route.class.render(),
683 });
684 }
685 }
686 if !seen.insert(route.class.clone()) {
687 return Err(MoveError::DuplicateRoute {
688 class: route.class.render(),
689 });
690 }
691 if route.target == request.source {
692 return Err(MoveError::TargetIsSource {
693 class: route.class.render(),
694 });
695 }
696 }
697
698 let inventory = read_source(conn, census, &request.source)?;
699
700 if let Some((table, rows)) = inventory.unknown.first() {
701 return Err(MoveError::UnknownTable {
702 table: table.clone(),
703 rows: *rows,
704 });
705 }
706
707 let unrouted = |class: SubjectClass, rows: u64| -> Result<(), MoveError> {
708 if rows > 0 && request.route_for(&class).is_none() {
709 return Err(MoveError::UnroutedClass {
710 class: class.render(),
711 rows,
712 });
713 }
714 Ok(())
715 };
716 for (kind, rows) in &inventory.note_kinds {
717 unrouted(SubjectClass::Note(kind.clone()), *rows)?;
718 }
719 for (kind, rows) in &inventory.entity_kinds {
720 unrouted(SubjectClass::Entity(kind.clone()), *rows)?;
721 }
722 for (relation, rows) in &inventory.edge_relations {
723 unrouted(SubjectClass::EdgeRelation(relation.clone()), *rows)?;
724 }
725 unrouted(SubjectClass::Atom, inventory.atoms)?;
726 unrouted(SubjectClass::Domain, inventory.domains)?;
727
728 let pinned = stream_members(conn, &request.source)?;
729 if !pinned.is_empty() {
730 return Err(MoveError::StreamMembers { notes: pinned });
731 }
732
733 validate_partitioned_vector_moves(conn, census, request)?;
734 Ok(())
735}
736
737fn validate_partitioned_vector_moves(
738 conn: &Connection,
739 census: &NamespaceCensus,
740 request: &MoveRequest,
741) -> Result<(), MoveError> {
742 if request.single_target().is_some() {
743 return Ok(());
744 }
745 let (selector, parameters) = routed_subjects(request);
746 for table in census
747 .tables
748 .iter()
749 .filter(|table| is_runtime_vector_table(table))
750 {
751 let unresolved = conn
752 .query_row(
753 &format!(
754 "WITH routed AS ({selector}) \
755 SELECT vector.subject_id, COUNT(DISTINCT routed.target) \
756 FROM {} AS vector LEFT JOIN routed ON routed.subject_id = vector.subject_id \
757 WHERE vector.namespace = ?1 GROUP BY vector.subject_id \
758 HAVING COUNT(DISTINCT routed.target) != 1 \
759 ORDER BY vector.subject_id LIMIT 1",
760 namespace_census::quote_ident(&table.name)
761 ),
762 rusqlite::params_from_iter(¶meters),
763 |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u64)),
764 )
765 .optional()?;
766 if let Some((subject_id, destinations)) = unresolved {
767 return Err(MoveError::UnroutableVector {
768 subject_id,
769 table: table.name.clone(),
770 destinations,
771 });
772 }
773 }
774 Ok(())
775}
776
777fn edge_collision_target_predicate(
778 request: &MoveRequest,
779 target: &str,
780 parameters: &mut Vec<rusqlite::types::Value>,
781) -> String {
782 let fallback = request
783 .route_for(&SubjectClass::Edge)
784 .is_some_and(|route| route.target == target);
785 let mut matching = Vec::new();
786 let mut specific = Vec::new();
787 for route in &request.routes {
788 if let SubjectClass::EdgeRelation(relation) = &route.class {
789 if route.target != target && !fallback {
790 continue;
791 }
792 parameters.push(relation.clone().into());
793 let parameter = format!("?{}", parameters.len());
794 if route.target == target {
795 matching.push(parameter.clone());
796 }
797 if fallback {
798 specific.push(parameter);
799 }
800 }
801 }
802 let mut alternatives = Vec::new();
803 if !matching.is_empty() {
804 alternatives.push(format!("source.relation IN ({})", matching.join(", ")));
805 }
806 if fallback {
807 alternatives.push(if specific.is_empty() {
808 "1".to_owned()
809 } else {
810 format!("source.relation NOT IN ({})", specific.join(", "))
811 });
812 }
813 if alternatives.is_empty() {
814 "0".to_owned()
815 } else {
816 alternatives.join(" OR ")
817 }
818}
819
820fn collisions_for(
873 conn: &Connection,
874 constraint: &NamespaceConstraint,
875 request: &MoveRequest,
876 target: &str,
877) -> rusqlite::Result<Vec<Collision>> {
878 if constraint.partial || !constraint.columns_are_nameable() {
879 return Ok(Vec::new());
880 }
881 let names: Vec<&str> = constraint
882 .columns
883 .iter()
884 .filter_map(|c| c.as_deref())
885 .collect();
886 let others: Vec<&str> = names
887 .iter()
888 .copied()
889 .filter(|c| !c.eq_ignore_ascii_case("namespace"))
890 .collect();
891 if others.len() != names.len() - 1 {
892 return Ok(Vec::new());
894 }
895 if others.is_empty() {
896 return collisions_on_namespace_alone(conn, constraint, &request.source, target);
900 }
901
902 let table = namespace_census::quote_ident(&constraint.table);
903 let join = others
904 .iter()
905 .map(|c| {
906 let q = namespace_census::quote_ident(c);
907 format!("target.{q} IS source.{q}")
911 })
912 .collect::<Vec<_>>()
913 .join(" AND ");
914 let select = others
915 .iter()
916 .map(|c| format!("source.{}", namespace_census::quote_ident(c)))
917 .collect::<Vec<_>>()
918 .join(", ");
919 let mut parameters = vec![request.source.clone().into(), target.to_owned().into()];
920 let mut sql = format!(
921 "SELECT {select} FROM {table} AS source \
922 JOIN {table} AS target ON target.namespace = ?2 AND {join} \
923 WHERE source.namespace = ?1"
924 );
925 if constraint.table == "graph_edges" && constraint.index == "idx_graph_edges_unique_triple" {
926 let predicate = edge_collision_target_predicate(request, target, &mut parameters);
927 sql.push_str(&format!(" AND ({predicate})"));
928 }
929
930 if constraint.table == "atom_weights" {
931 let atom = request
932 .route_for(&SubjectClass::Atom)
933 .is_some_and(|route| route.target == target);
934 let domain = request
935 .route_for(&SubjectClass::Domain)
936 .is_some_and(|route| route.target == target);
937 let predicate = match (atom, domain) {
938 (true, true) => "1".to_owned(),
939 (true, false) => format!("NOT {}", knowledge_move::MIRROR_EXISTS),
940 (false, true) => knowledge_move::MIRROR_EXISTS.to_owned(),
941 (false, false) => "0".to_owned(),
942 };
943 sql.push_str(&format!(
944 " AND source.atom_id IN (SELECT atom.id FROM knowledge_atoms AS atom \
945 WHERE atom.namespace = ?1 AND {predicate})"
946 ));
947 }
948
949 let mut stmt = conn.prepare(&sql)?;
950 let column_count = others.len();
951 let rows = stmt.query_map(rusqlite::params_from_iter(parameters), move |row| {
952 let mut parts = Vec::with_capacity(column_count);
953 for i in 0..column_count {
954 parts.push(match row.get_ref(i)? {
955 rusqlite::types::ValueRef::Null => "NULL".to_string(),
956 rusqlite::types::ValueRef::Integer(v) => v.to_string(),
957 rusqlite::types::ValueRef::Real(v) => v.to_string(),
958 rusqlite::types::ValueRef::Text(v) => String::from_utf8_lossy(v).into_owned(),
959 rusqlite::types::ValueRef::Blob(_) => "<blob>".to_string(),
960 });
961 }
962 Ok(parts.join(", "))
963 })?;
964
965 let mut found = Vec::new();
966 for key in rows {
967 found.push(Collision {
968 table: constraint.table.clone(),
969 constraint: constraint.index.clone(),
970 target: target.to_string(),
971 key: key?,
972 });
973 }
974 Ok(found)
975}
976
977fn collisions_on_namespace_alone(
978 conn: &Connection,
979 constraint: &NamespaceConstraint,
980 source: &str,
981 target: &str,
982) -> rusqlite::Result<Vec<Collision>> {
983 let table = namespace_census::quote_ident(&constraint.table);
984 let sql = format!(
985 "SELECT (SELECT COUNT(*) FROM {table} WHERE namespace = ?1) \
986 * (SELECT COUNT(*) FROM {table} WHERE namespace = ?2)"
987 );
988 let product: i64 = conn.query_row(&sql, [source, target], |row| row.get(0))?;
989 Ok(if product > 0 {
990 vec![Collision {
991 table: constraint.table.clone(),
992 constraint: constraint.index.clone(),
993 target: target.to_string(),
994 key: "(namespace alone)".to_string(),
995 }]
996 } else {
997 Vec::new()
998 })
999}
1000
1001struct KindedTables {
1013 base: &'static str,
1014 fts: &'static str,
1015 rowids: &'static str,
1016}
1017
1018const NOTE_TABLES: KindedTables = KindedTables {
1019 base: "notes",
1020 fts: "fts_notes",
1021 rowids: "fts_notes_rowids",
1022};
1023
1024const ENTITY_TABLES: KindedTables = KindedTables {
1025 base: "entities",
1026 fts: "fts_entities",
1027 rowids: "fts_entities_rowids",
1028};
1029
1030fn move_kinded_subject(
1031 conn: &Connection,
1032 tables: &KindedTables,
1033 source: &str,
1034 target: &str,
1035 kind: &str,
1036 rows: &mut BTreeMap<String, u64>,
1037) -> rusqlite::Result<u64> {
1038 let KindedTables { base, fts, rowids } = *tables;
1039 let statement = if base == "entities" {
1043 "UPDATE entities SET namespace = ?2, version = version + 1 \
1044 WHERE namespace = ?1 AND kind = ?3"
1045 .to_owned()
1046 } else {
1047 format!(
1048 "UPDATE {} SET namespace = ?2 WHERE namespace = ?1 AND kind = ?3",
1049 namespace_census::quote_ident(base)
1050 )
1051 };
1052 let moved = conn.execute(&statement, rusqlite::params![source, target, kind])? as u64;
1053 *rows.entry(base.to_string()).or_default() += moved;
1054
1055 let selector = format!(
1058 "SELECT id FROM {} WHERE namespace = ?2 AND kind = ?3",
1059 namespace_census::quote_ident(base)
1060 );
1061 for derived in [fts, rowids] {
1062 let n = conn.execute(
1063 &format!(
1064 "UPDATE {} SET namespace = ?2 \
1065 WHERE namespace = ?1 AND subject_id IN ({selector})",
1066 namespace_census::quote_ident(derived)
1067 ),
1068 rusqlite::params![source, target, kind],
1069 )? as u64;
1070 *rows.entry(derived.to_string()).or_default() += n;
1071 }
1072 Ok(moved)
1073}
1074
1075fn move_whole_table(
1077 conn: &Connection,
1078 table: &str,
1079 source: &str,
1080 target: &str,
1081 rows: &mut BTreeMap<String, u64>,
1082) -> rusqlite::Result<u64> {
1083 let moved = conn.execute(
1084 &format!(
1085 "UPDATE {} SET namespace = ?2 WHERE namespace = ?1",
1086 namespace_census::quote_ident(table)
1087 ),
1088 rusqlite::params![source, target],
1089 )? as u64;
1090 *rows.entry(table.to_string()).or_default() += moved;
1091 Ok(moved)
1092}
1093
1094fn move_edges(
1095 conn: &Connection,
1096 request: &MoveRequest,
1097 route: &MoveRoute,
1098 rows: &mut BTreeMap<String, u64>,
1099) -> rusqlite::Result<u64> {
1100 let mut parameters: Vec<rusqlite::types::Value> =
1101 vec![request.source.clone().into(), route.target.clone().into()];
1102 let predicate = if let SubjectClass::EdgeRelation(relation) = &route.class {
1103 parameters.push(relation.clone().into());
1104 "relation = ?3".to_owned()
1105 } else {
1106 let mut excluded = Vec::new();
1107 for specific in &request.routes {
1108 if let SubjectClass::EdgeRelation(relation) = &specific.class {
1109 parameters.push(relation.clone().into());
1110 excluded.push(format!("?{}", parameters.len()));
1111 }
1112 }
1113 if excluded.is_empty() {
1114 return move_whole_table(conn, "graph_edges", &request.source, &route.target, rows);
1115 }
1116 format!("relation NOT IN ({})", excluded.join(", "))
1117 };
1118 let moved = conn.execute(
1119 &format!("UPDATE graph_edges SET namespace = ?2 WHERE namespace = ?1 AND {predicate}"),
1120 rusqlite::params_from_iter(parameters),
1121 )? as u64;
1122 *rows.entry("graph_edges".into()).or_default() += moved;
1123 Ok(moved)
1124}
1125
1126fn move_memory_visibility(
1131 conn: &Connection,
1132 source: &str,
1133 target: &str,
1134 rows: &mut BTreeMap<String, u64>,
1135) -> rusqlite::Result<()> {
1136 let receipts = conn.execute(
1137 "INSERT INTO memory_visibility_receipts (namespace, note_id, model_count) \
1138 SELECT ?2, receipt.note_id, receipt.model_count \
1139 FROM memory_visibility_receipts AS receipt \
1140 JOIN notes AS note ON note.id = receipt.note_id \
1141 WHERE receipt.namespace = ?1 AND note.namespace = ?2",
1142 rusqlite::params![source, target],
1143 )? as u64;
1144 *rows.entry(MEMORY_VISIBILITY_RECEIPTS.into()).or_default() += receipts;
1145
1146 let fences = conn.execute(
1147 "UPDATE memory_visibility_fences SET namespace = ?2 \
1148 WHERE namespace = ?1 AND note_id IN (\
1149 SELECT note_id FROM memory_visibility_receipts WHERE namespace = ?2)",
1150 rusqlite::params![source, target],
1151 )? as u64;
1152 *rows.entry(MEMORY_VISIBILITY_FENCES.into()).or_default() += fences;
1153
1154 let removed = conn.execute(
1155 "DELETE FROM memory_visibility_receipts \
1156 WHERE namespace = ?1 AND note_id IN (\
1157 SELECT note_id FROM memory_visibility_receipts WHERE namespace = ?2)",
1158 rusqlite::params![source, target],
1159 )? as u64;
1160 debug_assert_eq!(removed, receipts);
1161 Ok(())
1162}
1163
1164fn move_vectors(
1177 conn: &Connection,
1178 table: &str,
1179 source: &str,
1180 target: Option<&str>,
1181) -> rusqlite::Result<VectorMove> {
1182 let quoted = namespace_census::quote_ident(table);
1183 let columns = "subject_id, namespace, kind, field, embedding_model, embedding";
1184
1185 conn.execute_batch("DROP TABLE IF EXISTS temp.namespace_move_vectors")?;
1186 let staged = if let Some(target) = target {
1187 conn.execute(
1188 &format!(
1189 "CREATE TEMP TABLE namespace_move_vectors AS \
1190 SELECT {columns}, ?2 AS target FROM {quoted} WHERE namespace = ?1"
1191 ),
1192 rusqlite::params![source, target],
1193 )
1194 } else {
1195 conn.execute(
1196 &format!(
1197 "CREATE TEMP TABLE namespace_move_vectors AS \
1198 SELECT vector.subject_id, vector.namespace, vector.kind, vector.field, \
1199 vector.embedding_model, vector.embedding, routed.target \
1200 FROM {quoted} AS vector JOIN (\
1201 SELECT DISTINCT subject_id, target FROM temp.namespace_move_subject_targets\
1202 ) AS routed ON routed.subject_id = vector.subject_id \
1203 WHERE vector.namespace = ?1"
1204 ),
1205 [source],
1206 )
1207 };
1208 staged?;
1211 let staged_rows: i64 = conn.query_row(
1212 "SELECT COUNT(*) FROM temp.namespace_move_vectors",
1213 [],
1214 |r| r.get(0),
1215 )?;
1216
1217 conn.execute(
1218 &format!("DELETE FROM {quoted} WHERE namespace = ?1"),
1219 [source],
1220 )?;
1221 let inserted = conn.execute(
1222 &format!(
1223 "INSERT INTO {quoted} ({columns}) \
1224 SELECT subject_id, target, kind, field, embedding_model, embedding \
1225 FROM temp.namespace_move_vectors"
1226 ),
1227 [],
1228 )? as u64;
1229 debug_assert_eq!(
1230 inserted, staged_rows as u64,
1231 "every staged vector is re-inserted or the move is losing embeddings"
1232 );
1233
1234 let has_provenance: bool = conn.query_row(
1240 "SELECT EXISTS(SELECT 1 FROM sqlite_master \
1241 WHERE type = 'table' AND name = 'vector_provenance')",
1242 [],
1243 |row| row.get(0),
1244 )?;
1245 if has_provenance {
1246 let model_key = table
1247 .strip_prefix("vec_")
1248 .expect("runtime vector tables use the vec_ prefix");
1249 conn.execute(
1250 "DELETE FROM vector_provenance \
1251 WHERE model_key = ?1 \
1252 AND subject_id IN (SELECT subject_id FROM temp.namespace_move_vectors)",
1253 [model_key],
1254 )?;
1255 }
1256
1257 let deleted = conn.execute(
1266 "INSERT INTO ann_write_log (namespace, embedding_model, kind, field, subject_id, op) \
1267 SELECT ?1, embedding_model, kind, field, subject_id, 'delete' \
1268 FROM temp.namespace_move_vectors",
1269 [source],
1270 )? as u64;
1271 let upserts = conn
1272 .prepare(
1273 "INSERT INTO ann_write_log \
1274 (namespace, embedding_model, kind, field, subject_id, op) \
1275 SELECT target, embedding_model, kind, field, subject_id, 'upsert' \
1276 FROM temp.namespace_move_vectors \
1277 RETURNING seq, subject_id, embedding_model, kind, field",
1278 )?
1279 .query_map([], |row| {
1280 Ok((
1281 row.get::<_, i64>(0)?,
1282 row.get::<_, String>(1)?,
1283 row.get::<_, String>(2)?,
1284 row.get::<_, String>(3)?,
1285 row.get::<_, String>(4)?,
1286 ))
1287 })?
1288 .collect::<rusqlite::Result<Vec<_>>>()?;
1289 for (seq, subject, model, kind, field) in &upserts {
1290 if kind == "note" && field == "note.content" {
1291 refresh_moved_memory_fence(conn, source, subject, model, *seq)?;
1292 }
1293 }
1294 let appended = deleted + upserts.len() as u64;
1295
1296 conn.execute_batch("DROP TABLE temp.namespace_move_vectors")?;
1297
1298 Ok(VectorMove {
1299 moved: inserted,
1300 ann_appended: appended,
1301 })
1302}
1303
1304fn refresh_moved_memory_fence(
1307 conn: &Connection,
1308 source: &str,
1309 subject: &str,
1310 model: &str,
1311 seq: i64,
1312) -> rusqlite::Result<()> {
1313 conn.execute(
1314 "UPDATE memory_visibility_fences SET ann_write_log_seq = ?4 \
1315 WHERE namespace = ?1 AND note_id = ?2 AND model = ?3",
1316 rusqlite::params![source, subject, model, seq],
1317 )?;
1318 Ok(())
1319}
1320
1321struct VectorMove {
1336 moved: u64,
1337 ann_appended: u64,
1338}
1339
1340pub fn move_namespace(conn: &Connection, request: &MoveRequest) -> Result<MoveCounts, MoveError> {
1348 let census = namespace_census::census(conn)?;
1349 validate(conn, &census, request)?;
1350
1351 let mut targets: BTreeSet<&str> = BTreeSet::new();
1354 for route in &request.routes {
1355 targets.insert(route.target.as_str());
1356 }
1357 let mut collisions = Vec::new();
1358 for constraint in namespace_census::reachable_constraints(&census) {
1359 for target in &targets {
1360 collisions.extend(collisions_for(conn, constraint, request, target)?);
1361 }
1362 }
1363 if !collisions.is_empty() {
1364 return Err(MoveError::Collisions { collisions });
1365 }
1366
1367 let mut counts = MoveCounts::default();
1368 let source = request.source.as_str();
1369
1370 if request.single_target().is_none() {
1373 let (selector, parameters) = routed_subjects(request);
1374 conn.execute_batch("DROP TABLE IF EXISTS temp.namespace_move_subject_targets")?;
1375 conn.execute(
1376 &format!("CREATE TEMP TABLE namespace_move_subject_targets AS {selector}"),
1377 rusqlite::params_from_iter(parameters),
1378 )?;
1379 }
1380
1381 move_task_audit(conn, &census, request, &mut counts.rows)?;
1384
1385 for route in &request.routes {
1386 let target = route.target.as_str();
1387 let moved = match &route.class {
1388 SubjectClass::Note(kind) => {
1389 move_kinded_subject(conn, &NOTE_TABLES, source, target, kind, &mut counts.rows)?
1390 }
1391 SubjectClass::Entity(kind) => {
1392 move_kinded_subject(conn, &ENTITY_TABLES, source, target, kind, &mut counts.rows)?
1393 }
1394 SubjectClass::Edge | SubjectClass::EdgeRelation(_) => {
1395 move_edges(conn, request, route, &mut counts.rows)?
1396 }
1397 SubjectClass::Atom => {
1398 move_knowledge_atoms(conn, &census, request, target, false, &mut counts.rows)?
1399 }
1400 SubjectClass::Domain => {
1401 move_knowledge_atoms(conn, &census, request, target, true, &mut counts.rows)?;
1403 move_whole_table(conn, "knowledge_domains", source, target, &mut counts.rows)?
1404 }
1405 };
1406 counts.subjects.insert(route.class.render(), moved);
1407 }
1408
1409 if let Some(target) = request.single_target() {
1412 move_whole_table(conn, "knowledge_sections", source, target, &mut counts.rows)?;
1413 }
1414
1415 for table in &census.tables {
1419 if !is_runtime_vector_table(table) {
1420 continue;
1421 }
1422 let moved = move_vectors(conn, &table.name, source, request.single_target())?;
1423 *counts.rows.entry(table.name.clone()).or_default() += moved.moved;
1424 counts.ann_log_appended += moved.ann_appended;
1425 }
1426 if request.single_target().is_none() {
1427 conn.execute_batch("DROP TABLE temp.namespace_move_subject_targets")?;
1428 }
1429 for table in ["vector_provenance", "atom_weights", "weight_events"] {
1430 if census
1431 .tables
1432 .iter()
1433 .any(|candidate| candidate.name == table)
1434 {
1435 let left = count_in_namespace(conn, table, source)?;
1436 if left > 0 {
1437 counts.left_behind.insert(table.into(), left);
1438 }
1439 }
1440 }
1441
1442 for table in SUBJECT_KEYED_TABLES {
1446 for target in &targets {
1447 let moved = conn.execute(
1448 &format!(
1449 "UPDATE {} SET namespace = ?2 WHERE namespace = ?1 AND target_id IN (\
1450 SELECT id FROM notes WHERE namespace = ?2 \
1451 UNION ALL SELECT id FROM entities WHERE namespace = ?2 \
1452 UNION ALL SELECT id FROM knowledge_atoms WHERE namespace = ?2)",
1453 namespace_census::quote_ident(table)
1454 ),
1455 rusqlite::params![source, target],
1456 )? as u64;
1457 *counts.rows.entry((*table).to_string()).or_default() += moved;
1458 }
1459 let left = count_in_namespace(conn, table, source)?;
1460 if left > 0 {
1461 counts.left_behind.insert((*table).to_string(), left);
1462 }
1463 }
1464
1465 for target in &targets {
1467 let moved = conn.execute(
1468 "UPDATE memory_visibility_epochs SET namespace = ?2 \
1469 WHERE namespace = ?1 AND note_id IN (SELECT id FROM notes WHERE namespace = ?2)",
1470 rusqlite::params![source, target],
1471 )? as u64;
1472 *counts
1473 .rows
1474 .entry(MEMORY_VISIBILITY_EPOCHS.into())
1475 .or_default() += moved;
1476 }
1477 let left = count_in_namespace(conn, MEMORY_VISIBILITY_EPOCHS, source)?;
1478 if left > 0 {
1479 counts
1480 .left_behind
1481 .insert(MEMORY_VISIBILITY_EPOCHS.into(), left);
1482 }
1483
1484 for target in &targets {
1488 move_memory_visibility(conn, source, target, &mut counts.rows)?;
1489 }
1490 for table in [MEMORY_VISIBILITY_RECEIPTS, MEMORY_VISIBILITY_FENCES] {
1491 let left = count_in_namespace(conn, table, source)?;
1492 if left > 0 {
1493 counts.left_behind.insert((*table).to_string(), left);
1494 }
1495 }
1496
1497 for table in LEAVE_BEHIND_TABLES {
1498 let left = count_in_namespace(conn, table, source)?;
1499 if left > 0 {
1500 counts.left_behind.insert((*table).to_string(), left);
1501 }
1502 }
1503
1504 for table in NAMESPACE_SCOPED_TABLES {
1508 match request.single_target() {
1509 Some(target) => {
1510 move_whole_table(conn, table, source, target, &mut counts.rows)?;
1511 }
1512 None => {
1513 let left = count_in_namespace(conn, table, source)?;
1514 if left > 0 {
1515 counts.left_behind.insert((*table).to_string(), left);
1516 }
1517 }
1518 }
1519 }
1520
1521 settle_pack_tables(conn, &census, request, &mut counts)?;
1522
1523 Ok(counts)
1524}
1525
1526#[cfg(test)]
1527mod tests {
1528 use super::*;
1529 use crate::migrations::run_migrations_for_test as run_migrations;
1530 use rusqlite::Connection;
1531
1532 pub(super) fn migrated() -> Connection {
1533 let mut conn = Connection::open_in_memory().expect("open");
1534 run_migrations(&mut conn).expect("migrate");
1535 conn
1536 }
1537
1538 pub(super) fn seed_note(conn: &Connection, id: &str, namespace: &str, kind: &str) {
1548 conn.execute(
1549 "INSERT INTO notes (id, namespace, kind, name, content, created_at, updated_at) \
1550 VALUES (?1, ?2, ?3, 'a name', 'some content', 1, 1)",
1551 rusqlite::params![id, namespace, kind],
1552 )
1553 .expect("seed note");
1554 }
1555
1556 pub(super) fn route(key: &str, target: &str) -> MoveRoute {
1557 MoveRoute {
1558 class: SubjectClass::parse(key).expect("route key"),
1559 target: target.to_string(),
1560 }
1561 }
1562
1563 #[test]
1568 fn a_routed_class_with_no_rows_succeeds_reporting_zero() {
1569 let conn = migrated();
1570 let request = MoveRequest::new(
1571 "empty-source",
1572 vec![route("note:observation", "target"), route("atom", "target")],
1573 );
1574 let counts = move_namespace(&conn, &request).expect("a backend with nothing routed here");
1575 assert_eq!(counts.subjects.get("note:observation"), Some(&0));
1576 assert_eq!(counts.subjects.get("atom"), Some(&0));
1577 assert_eq!(counts.rows.get(MEMORY_VISIBILITY_RECEIPTS), Some(&0));
1578 assert_eq!(counts.rows.get(MEMORY_VISIBILITY_FENCES), Some(&0));
1579 }
1580
1581 #[test]
1582 fn a_memory_visibility_receipt_follows_its_note_and_reports_count() {
1583 let conn = migrated();
1584 conn.pragma_update(None, "foreign_keys", "ON").unwrap();
1585 seed_note(&conn, "n1", "source", "observation");
1586 conn.execute(
1587 "INSERT INTO memory_visibility_receipts (namespace, note_id, model_count) \
1588 VALUES ('source', 'n1', 0)",
1589 [],
1590 )
1591 .unwrap();
1592
1593 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
1594 let counts = move_namespace(&conn, &request).expect("the receipt follows its note");
1595 assert_eq!(counts.subjects.get("note:observation"), Some(&1));
1596 assert_eq!(counts.rows.get(MEMORY_VISIBILITY_RECEIPTS), Some(&1));
1597 assert_eq!(counts.rows.get(MEMORY_VISIBILITY_FENCES), Some(&0));
1598 let stored: (String, i64) = conn
1599 .query_row(
1600 "SELECT namespace, model_count FROM memory_visibility_receipts \
1601 WHERE note_id = 'n1'",
1602 [],
1603 |row| Ok((row.get(0)?, row.get(1)?)),
1604 )
1605 .unwrap();
1606 assert_eq!(stored, ("target".into(), 0));
1607 let source_rows: i64 = conn
1608 .query_row(
1609 "SELECT COUNT(*) FROM memory_visibility_receipts WHERE namespace = 'source'",
1610 [],
1611 |row| row.get(0),
1612 )
1613 .unwrap();
1614 assert_eq!(source_rows, 0);
1615 }
1616
1617 #[test]
1618 fn memory_visibility_fences_follow_their_note_with_foreign_keys_enabled() {
1619 let conn = migrated();
1620 conn.pragma_update(None, "foreign_keys", "ON").unwrap();
1621 seed_note(&conn, "n1", "source", "observation");
1622 seed_note(&conn, "n2", "source", "decision");
1623 conn.execute(
1624 "INSERT INTO memory_visibility_receipts (namespace, note_id, model_count) \
1625 VALUES ('source', 'n1', 2), ('source', 'n2', 0)",
1626 [],
1627 )
1628 .unwrap();
1629 conn.execute(
1630 "INSERT INTO memory_visibility_fences \
1631 (namespace, note_id, model, ann_write_log_seq) VALUES \
1632 ('source', 'n1', 'model-a', 10), ('source', 'n1', 'model-b', 11)",
1633 [],
1634 )
1635 .unwrap();
1636
1637 let request = MoveRequest::new(
1638 "source",
1639 vec![
1640 route("note:observation", "target-a"),
1641 route("note:decision", "target-b"),
1642 ],
1643 );
1644 let counts = move_namespace(&conn, &request).expect("both receipts follow their notes");
1645 assert_eq!(counts.rows.get(MEMORY_VISIBILITY_RECEIPTS), Some(&2));
1646 assert_eq!(counts.rows.get(MEMORY_VISIBILITY_FENCES), Some(&2));
1647 let receipt_places: Vec<(String, String)> = conn
1648 .prepare("SELECT note_id, namespace FROM memory_visibility_receipts ORDER BY note_id")
1649 .unwrap()
1650 .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
1651 .unwrap()
1652 .collect::<rusqlite::Result<_>>()
1653 .unwrap();
1654 assert_eq!(
1655 receipt_places,
1656 vec![
1657 ("n1".into(), "target-a".into()),
1658 ("n2".into(), "target-b".into())
1659 ]
1660 );
1661 let fences: Vec<(String, String, i64)> = conn
1662 .prepare(
1663 "SELECT namespace, model, ann_write_log_seq \
1664 FROM memory_visibility_fences ORDER BY model",
1665 )
1666 .unwrap()
1667 .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
1668 .unwrap()
1669 .collect::<rusqlite::Result<_>>()
1670 .unwrap();
1671 assert_eq!(
1672 fences,
1673 vec![
1674 ("target-a".into(), "model-a".into(), 10),
1675 ("target-a".into(), "model-b".into(), 11)
1676 ]
1677 );
1678 let foreign_key_errors: i64 = conn
1679 .query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
1680 row.get(0)
1681 })
1682 .unwrap();
1683 assert_eq!(foreign_key_errors, 0);
1684 }
1685
1686 #[cfg(feature = "vectors")]
1687 #[test]
1688 fn moved_visibility_fences_use_each_exact_upsert_and_preserve_unrelated_receipts() {
1689 crate::extension::ensure_extensions_loaded();
1690 let mut conn = migrated();
1691 conn.pragma_update(None, "foreign_keys", "ON").unwrap();
1692 for (id, namespace) in [
1693 ("moved", "source"),
1694 ("zero", "source"),
1695 ("unfenced", "source"),
1696 ("existing", "target"),
1697 ] {
1698 seed_note(&conn, id, namespace, "memory");
1699 }
1700 conn.execute_batch(
1701 "INSERT INTO memory_visibility_receipts (namespace, note_id, model_count) VALUES \
1702 ('source', 'moved', 2), ('source', 'zero', 0), ('target', 'existing', 1); \
1703 INSERT INTO memory_visibility_fences (namespace, note_id, model, ann_write_log_seq) VALUES \
1704 ('source', 'moved', 'model-a', 17), ('source', 'moved', 'model-b', 18), \
1705 ('target', 'existing', 'model-a', 99);",
1706 ).unwrap();
1707 for (table, model) in [("vec_model_a", "model-a"), ("vec_model_b", "model-b")] {
1708 conn.execute_batch(&format!(
1709 "CREATE VIRTUAL TABLE {table} USING vec0(\
1710 subject_id TEXT PRIMARY KEY, namespace TEXT NOT NULL, \
1711 kind TEXT NOT NULL, field TEXT NOT NULL, embedding_model TEXT NOT NULL, \
1712 embedding float[2] distance_metric=cosine)"
1713 ))
1714 .unwrap();
1715 conn.execute(&format!(
1716 "INSERT INTO {table} (subject_id, namespace, kind, field, embedding_model, embedding) \
1717 VALUES ('moved', 'source', 'note', 'note.content', ?1, '[0.1, 0.2]')"
1718 ), [model]).unwrap();
1719 }
1720 conn.execute_batch(
1721 "INSERT INTO vec_model_a (subject_id, namespace, kind, field, embedding_model, embedding) VALUES \
1722 ('unfenced', 'source', 'note', 'note.content', 'model-a', '[0.1, 0.2]'), \
1723 ('existing', 'target', 'note', 'note.content', 'model-a', '[0.1, 0.2]'); \
1724 INSERT INTO ann_write_log (seq, namespace, embedding_model, kind, field, subject_id, op) \
1725 VALUES (100, 'target', 'model-a', 'note', 'note.content', 'existing', 'upsert');",
1726 ).unwrap();
1727 let transaction = conn.transaction().unwrap();
1728 let counts = move_namespace(
1729 &transaction,
1730 &MoveRequest::new("source", vec![route("note:memory", "target")]),
1731 )
1732 .expect("move memory receipts and both vector models");
1733 assert_eq!(counts.ann_log_appended, 6);
1734 for model in ["model-a", "model-b"] {
1735 let fence: i64 = transaction
1736 .query_row(
1737 "SELECT ann_write_log_seq FROM memory_visibility_fences \
1738 WHERE namespace = 'target' AND note_id = 'moved' AND model = ?1",
1739 [model],
1740 |row| row.get(0),
1741 )
1742 .unwrap();
1743 let upsert: i64 = transaction
1744 .query_row(
1745 "SELECT seq FROM ann_write_log WHERE namespace = 'target' \
1746 AND subject_id = 'moved' AND embedding_model = ?1 AND op = 'upsert'",
1747 [model],
1748 |row| row.get(0),
1749 )
1750 .unwrap();
1751 assert!(upsert > 100);
1752 assert_eq!(fence, upsert, "each model owes its own destination upsert");
1753 }
1754 let untouched: i64 = transaction
1755 .query_row(
1756 "SELECT ann_write_log_seq FROM memory_visibility_fences \
1757 WHERE namespace = 'target' AND note_id = 'existing' AND model = 'model-a'",
1758 [],
1759 |row| row.get(0),
1760 )
1761 .unwrap();
1762 assert_eq!(untouched, 99);
1763 let zero_count: i64 = transaction
1764 .query_row(
1765 "SELECT model_count FROM memory_visibility_receipts \
1766 WHERE namespace = 'target' AND note_id = 'zero'",
1767 [],
1768 |row| row.get(0),
1769 )
1770 .unwrap();
1771 assert_eq!(zero_count, 0);
1772 let invented: i64 = transaction.query_row(
1773 "SELECT COUNT(*) FROM memory_visibility_fences WHERE note_id IN ('zero', 'unfenced')",
1774 [], |row| row.get(0),
1775 ).unwrap();
1776 assert_eq!(invented, 0);
1777 transaction.commit().unwrap();
1778 }
1779
1780 #[test]
1784 fn a_class_with_rows_and_no_route_refuses_and_says_how_many() {
1785 let conn = migrated();
1786 seed_note(&conn, "n1", "source", "observation");
1787 seed_note(&conn, "n2", "source", "decision");
1788
1789 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
1790 let error = move_namespace(&conn, &request).expect_err("decision notes are unrouted");
1791 match error {
1792 MoveError::UnroutedClass { class, rows } => {
1793 assert_eq!(class, "note:decision");
1794 assert_eq!(rows, 1);
1795 }
1796 other => panic!("expected an unrouted class, got {other}"),
1797 }
1798
1799 let still_here: i64 = conn
1800 .query_row(
1801 "SELECT COUNT(*) FROM notes WHERE namespace = 'source'",
1802 [],
1803 |r| r.get(0),
1804 )
1805 .expect("count");
1806 assert_eq!(still_here, 2, "a refusal writes nothing");
1807 }
1808
1809 #[test]
1814 fn a_namespace_table_this_build_has_no_rule_for_refuses_the_move() {
1815 let conn = migrated();
1816 seed_note(&conn, "n1", "source", "observation");
1817 conn.execute_batch(
1818 "CREATE TABLE later_migration_added_this (\
1819 id TEXT PRIMARY KEY, namespace TEXT NOT NULL);\
1820 INSERT INTO later_migration_added_this VALUES ('x', 'source');",
1821 )
1822 .expect("a migration lands");
1823
1824 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
1825 let error = move_namespace(&conn, &request).expect_err("an unknown table refuses");
1826 match error {
1827 MoveError::UnknownTable { table, rows } => {
1828 assert_eq!(table, "later_migration_added_this");
1829 assert_eq!(rows, 1);
1830 }
1831 other => panic!("expected an unknown table, got {other}"),
1832 }
1833 }
1834
1835 #[test]
1847 fn a_table_named_like_a_vector_table_but_not_one_refuses_by_name() {
1848 let conn = migrated();
1849 seed_note(&conn, "n1", "source", "observation");
1850 conn.execute_batch(
1851 "CREATE TABLE vec_audit (\
1852 id TEXT PRIMARY KEY, namespace TEXT NOT NULL);\
1853 INSERT INTO vec_audit VALUES ('x', 'source');",
1854 )
1855 .expect("a migration lands a table whose name starts with the prefix");
1856
1857 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
1858 let error = move_namespace(&conn, &request).expect_err("the prefix is not enough");
1859 match error {
1860 MoveError::UnknownTable { table, rows } => {
1861 assert_eq!(table, "vec_audit");
1862 assert_eq!(rows, 1);
1863 }
1864 other => panic!("expected an unknown table, got {other}"),
1865 }
1866 }
1867
1868 #[test]
1872 fn an_unknown_table_holding_nothing_here_does_not_refuse() {
1873 let conn = migrated();
1874 seed_note(&conn, "n1", "source", "observation");
1875 conn.execute_batch(
1876 "CREATE TABLE later_migration_added_this (\
1877 id TEXT PRIMARY KEY, namespace TEXT NOT NULL);\
1878 INSERT INTO later_migration_added_this VALUES ('x', 'somewhere-else');",
1879 )
1880 .expect("a migration lands");
1881
1882 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
1883 let counts = move_namespace(&conn, &request).expect("nothing of ours is in that table");
1884 assert_eq!(counts.subjects.get("note:observation"), Some(&1));
1885 }
1886
1887 #[test]
1888 fn a_route_to_the_namespace_it_is_already_in_refuses() {
1889 let conn = migrated();
1890 let request = MoveRequest::new("source", vec![route("atom", "source")]);
1891 match move_namespace(&conn, &request).expect_err("a no-op written as an instruction") {
1892 MoveError::TargetIsSource { class } => assert_eq!(class, "atom"),
1893 other => panic!("expected target-is-source, got {other}"),
1894 }
1895 }
1896
1897 #[test]
1898 fn the_same_class_routed_twice_refuses_rather_than_picking_one() {
1899 let conn = migrated();
1900 let request = MoveRequest::new(
1901 "source",
1902 vec![route("atom", "one"), route("atom", "another")],
1903 );
1904 match move_namespace(&conn, &request).expect_err("two targets, no rule to choose") {
1905 MoveError::DuplicateRoute { class } => assert_eq!(class, "atom"),
1906 other => panic!("expected a duplicate route, got {other}"),
1907 }
1908 }
1909
1910 #[test]
1911 fn an_unknown_route_key_names_what_it_was_given() {
1912 match SubjectClass::parse("notes:observation").expect_err("plural is a typo") {
1913 MoveError::UnknownSubjectClass { key } => assert_eq!(key, "notes:observation"),
1914 other => panic!("expected an unknown class, got {other}"),
1915 }
1916 assert_eq!(
1917 SubjectClass::parse("note:observation").expect("singular"),
1918 SubjectClass::Note("observation".into())
1919 );
1920 }
1921
1922 #[test]
1927 fn a_partitioning_move_reports_the_aggregates_it_leaves_behind() {
1928 let conn = migrated();
1929 seed_note(&conn, "n1", "source", "observation");
1930 seed_note(&conn, "n2", "source", "decision");
1931 conn.execute(
1932 "INSERT INTO brain_profile_snapshots (profile_id, namespace, snapshot_json, updated_at) \
1933 VALUES ('p', 'source', '{}', 1)",
1934 [],
1935 )
1936 .expect("seed a snapshot");
1937
1938 let request = MoveRequest::new(
1939 "source",
1940 vec![
1941 route("note:observation", "one"),
1942 route("note:decision", "another"),
1943 ],
1944 );
1945 let counts = move_namespace(&conn, &request).expect("a partitioning move");
1946 assert_eq!(counts.left_behind.get("brain_profile_snapshots"), Some(&1));
1947
1948 let stayed: i64 = conn
1949 .query_row(
1950 "SELECT COUNT(*) FROM brain_profile_snapshots WHERE namespace = 'source'",
1951 [],
1952 |r| r.get(0),
1953 )
1954 .expect("count");
1955 assert_eq!(stayed, 1);
1956 }
1957
1958 #[test]
1961 fn a_total_single_target_move_carries_the_aggregates() {
1962 let conn = migrated();
1963 seed_note(&conn, "n1", "source", "observation");
1964 conn.execute(
1965 "INSERT INTO brain_profile_snapshots (profile_id, namespace, snapshot_json, updated_at) \
1966 VALUES ('p', 'source', '{}', 1)",
1967 [],
1968 )
1969 .expect("seed a snapshot");
1970
1971 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
1972 let counts = move_namespace(&conn, &request).expect("a total move");
1973 assert!(counts.left_behind.is_empty(), "{:?}", counts.left_behind);
1974
1975 let moved: i64 = conn
1976 .query_row(
1977 "SELECT COUNT(*) FROM brain_profile_snapshots WHERE namespace = 'target'",
1978 [],
1979 |r| r.get(0),
1980 )
1981 .expect("count");
1982 assert_eq!(moved, 1);
1983 }
1984 #[cfg(feature = "vectors")]
1997 #[test]
1998 fn a_vector_move_tells_the_source_side_to_drop_what_left() {
1999 crate::extension::ensure_extensions_loaded();
2002 let conn = migrated();
2003 seed_note(&conn, "n1", "source", "observation");
2004 conn.execute_batch(
2005 "CREATE VIRTUAL TABLE vec_test_model USING vec0(\
2006 subject_id TEXT PRIMARY KEY, \
2007 namespace TEXT NOT NULL, \
2008 kind TEXT NOT NULL, \
2009 field TEXT NOT NULL, \
2010 embedding_model TEXT NOT NULL, \
2011 embedding float[4] distance_metric=cosine\
2012 )",
2013 )
2014 .expect("the vector table an embedding model creates at runtime");
2015 conn.execute(
2016 "INSERT INTO vec_test_model \
2017 (subject_id, namespace, kind, field, embedding_model, embedding) \
2018 VALUES ('n1', 'source', 'observation', 'content', 'test-model', \
2019 '[0.1, 0.2, 0.3, 0.4]')",
2020 [],
2021 )
2022 .expect("seed a vector");
2023
2024 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
2025 let counts = move_namespace(&conn, &request).expect("a total move");
2026
2027 assert_eq!(
2028 counts.rows.get("vec_test_model"),
2029 Some(&1),
2030 "the vector itself moved"
2031 );
2032 let left_in_source: i64 = conn
2033 .query_row(
2034 "SELECT COUNT(*) FROM vec_test_model WHERE namespace = 'source'",
2035 [],
2036 |r| r.get(0),
2037 )
2038 .expect("count");
2039 assert_eq!(left_in_source, 0);
2040
2041 assert_eq!(counts.ann_log_appended, 2);
2044 let dropped_from_source: i64 = conn
2045 .query_row(
2046 "SELECT COUNT(*) FROM ann_write_log \
2047 WHERE namespace = 'source' AND op = 'delete' \
2048 AND subject_id = 'n1' AND embedding_model = 'test-model'",
2049 [],
2050 |r| r.get(0),
2051 )
2052 .expect("count");
2053 assert_eq!(
2054 dropped_from_source, 1,
2055 "the source consumer is never told to drop the vector, so its index \
2056 keeps answering with a subject that has left the namespace"
2057 );
2058 let taken_by_target: i64 = conn
2059 .query_row(
2060 "SELECT COUNT(*) FROM ann_write_log \
2061 WHERE namespace = 'target' AND op = 'upsert' \
2062 AND subject_id = 'n1' AND embedding_model = 'test-model'",
2063 [],
2064 |r| r.get(0),
2065 )
2066 .expect("count");
2067 assert_eq!(taken_by_target, 1);
2068 }
2069
2070 #[cfg(feature = "vectors")]
2071 #[tokio::test]
2072 async fn vector_provenance_namespace_move_clears_sidecar() {
2073 use std::sync::Arc;
2074
2075 use khive_storage::VectorStore;
2076
2077 use crate::pool::{ConnectionPool, PoolConfig};
2078 use crate::stores::vectors::SqliteVecStore;
2079
2080 crate::extension::ensure_extensions_loaded();
2081 let dir = tempfile::tempdir().expect("tempdir");
2082 let path = dir.path().join("namespace-move-provenance.db");
2083 let mut conn = Connection::open(&path).expect("open database");
2084 run_migrations(&mut conn).expect("migrate database");
2085 let subject_id = uuid::Uuid::new_v4();
2086 let subject = subject_id.to_string();
2087 seed_note(&conn, &subject, "source", "observation");
2088 conn.execute_batch(
2089 "CREATE VIRTUAL TABLE vec_test_model USING vec0(\
2090 subject_id TEXT PRIMARY KEY, namespace TEXT NOT NULL, \
2091 kind TEXT NOT NULL, field TEXT NOT NULL, \
2092 embedding_model TEXT NOT NULL, embedding float[2] distance_metric=cosine)",
2093 )
2094 .unwrap();
2095 conn.execute(
2096 "INSERT INTO vec_test_model \
2097 (subject_id, namespace, kind, field, embedding_model, embedding) \
2098 VALUES (?1, 'source', 'observation', 'content', 'test-model', '[0.1, 0.2]')",
2099 [&subject],
2100 )
2101 .unwrap();
2102 let before_blob: Vec<u8> = conn
2103 .query_row(
2104 "SELECT embedding FROM vec_test_model WHERE subject_id = ?1",
2105 [&subject],
2106 |row| row.get(0),
2107 )
2108 .unwrap();
2109 let digest = blake3::hash(&before_blob).to_hex().to_string();
2110 conn.execute(
2111 "INSERT INTO vector_provenance \
2112 (model_key, subject_id, namespace, embedding_digest, text_fingerprint, updated_at) \
2113 VALUES ('test_model', ?1, 'source', ?2, ?3, '2026-09-25T12:34:56Z')",
2114 rusqlite::params![&subject, &digest, "a".repeat(64)],
2115 )
2116 .unwrap();
2117
2118 let stale_target_id = uuid::Uuid::new_v4();
2121 let stale_target_subject = stale_target_id.to_string();
2122 seed_note(&conn, &stale_target_subject, "source", "observation");
2123 conn.execute(
2124 "INSERT INTO vec_test_model \
2125 (subject_id, namespace, kind, field, embedding_model, embedding) \
2126 VALUES (?1, 'source', 'observation', 'content', 'test-model', '[0.1, 0.2]')",
2127 [&stale_target_subject],
2128 )
2129 .unwrap();
2130 conn.execute(
2131 "INSERT INTO vector_provenance \
2132 (model_key, subject_id, namespace, embedding_digest, text_fingerprint, updated_at) \
2133 VALUES ('test_model', ?1, 'target', ?2, ?3, '2026-09-25T12:34:56Z')",
2134 rusqlite::params![&stale_target_subject, &digest, "b".repeat(64)],
2135 )
2136 .unwrap();
2137
2138 conn.execute_batch("BEGIN IMMEDIATE").unwrap();
2139 move_namespace(
2140 &conn,
2141 &MoveRequest::new("source", vec![route("note:observation", "target")]),
2142 )
2143 .unwrap();
2144 conn.execute_batch("COMMIT").unwrap();
2145 let sidecars: i64 = conn
2146 .query_row(
2147 "SELECT COUNT(*) FROM vector_provenance \
2148 WHERE model_key = 'test_model' AND subject_id IN (?1, ?2)",
2149 rusqlite::params![&subject, &stale_target_subject],
2150 |row| row.get(0),
2151 )
2152 .unwrap();
2153 assert_eq!(
2154 sidecars, 0,
2155 "a move must invalidate even an equal-BLOB sidecar"
2156 );
2157 let after_blob: Vec<u8> = conn
2158 .query_row(
2159 "SELECT embedding FROM vec_test_model \
2160 WHERE subject_id = ?1 AND namespace = 'target'",
2161 [&subject],
2162 |row| row.get(0),
2163 )
2164 .unwrap();
2165 assert_eq!(after_blob, before_blob, "the vector itself must survive");
2166 drop(conn);
2167
2168 let pool = Arc::new(
2169 ConnectionPool::new(PoolConfig {
2170 path: Some(path),
2171 write_queue_enabled: Some(false),
2172 ..PoolConfig::for_test()
2173 })
2174 .expect("reopen moved database"),
2175 );
2176 let vectors = SqliteVecStore::new(
2177 pool,
2178 true,
2179 "test_model".into(),
2180 "test-model".into(),
2181 2,
2182 "target".into(),
2183 )
2184 .expect("open target vector store");
2185 let observed = vectors
2186 .provenance(subject_id)
2187 .await
2188 .expect("read moved vector")
2189 .expect("moved vector remains present");
2190 assert_eq!(observed.text_fingerprint, None);
2191 assert_eq!(observed.updated_at, None);
2192 let stale_target = vectors
2193 .provenance(stale_target_id)
2194 .await
2195 .expect("read formerly stale target vector")
2196 .expect("second moved vector remains present");
2197 assert_eq!(stale_target.text_fingerprint, None);
2198 assert_eq!(stale_target.updated_at, None);
2199 }
2200
2201 #[cfg(feature = "vectors")]
2212 #[test]
2213 fn a_vector_the_target_already_held_is_not_in_the_instructions() {
2214 crate::extension::ensure_extensions_loaded();
2215 let conn = migrated();
2216 seed_note(&conn, "n1", "source", "observation");
2217 conn.execute_batch(
2218 "CREATE VIRTUAL TABLE vec_test_model USING vec0(\
2219 subject_id TEXT PRIMARY KEY, \
2220 namespace TEXT NOT NULL, \
2221 kind TEXT NOT NULL, \
2222 field TEXT NOT NULL, \
2223 embedding_model TEXT NOT NULL, \
2224 embedding float[4] distance_metric=cosine\
2225 )",
2226 )
2227 .expect("the vector table an embedding model creates at runtime");
2228 conn.execute(
2229 "INSERT INTO vec_test_model \
2230 (subject_id, namespace, kind, field, embedding_model, embedding) \
2231 VALUES ('n1', 'source', 'observation', 'content', 'test-model', \
2232 '[0.1, 0.2, 0.3, 0.4]')",
2233 [],
2234 )
2235 .expect("the vector that moves");
2236 conn.execute(
2237 "INSERT INTO vec_test_model \
2238 (subject_id, namespace, kind, field, embedding_model, embedding) \
2239 VALUES ('already-there', 'target', 'observation', 'content', 'test-model', \
2240 '[0.5, 0.6, 0.7, 0.8]')",
2241 [],
2242 )
2243 .expect("a vector the target already holds");
2244
2245 let request = MoveRequest::new("source", vec![route("note:observation", "target")]);
2246 let counts = move_namespace(&conn, &request).expect("a total move");
2247
2248 assert_eq!(
2249 counts.ann_log_appended, 2,
2250 "two instructions are owed for the one vector that moved"
2251 );
2252 let about_the_resident: i64 = conn
2253 .query_row(
2254 "SELECT COUNT(*) FROM ann_write_log WHERE subject_id = 'already-there'",
2255 [],
2256 |r| r.get(0),
2257 )
2258 .expect("count");
2259 assert_eq!(
2260 about_the_resident, 0,
2261 "a vector that did not move is told nothing, and is certainly not \
2262 dropped from a namespace it was never in"
2263 );
2264 let about_the_mover: i64 = conn
2267 .query_row(
2268 "SELECT COUNT(*) FROM ann_write_log WHERE subject_id = 'n1'",
2269 [],
2270 |r| r.get(0),
2271 )
2272 .expect("count");
2273 assert_eq!(about_the_mover, 2);
2274 }
2275 #[test]
2287 fn two_routes_to_one_target_report_a_shared_collision_once() {
2288 let conn = migrated();
2289 seed_note(&conn, "n1", "source", "observation");
2294 seed_note(&conn, "n2", "source", "insight");
2295 for (id, namespace) in [("a1", "source"), ("a2", "target")] {
2296 conn.execute(
2297 "INSERT INTO knowledge_atoms \
2298 (id, namespace, slug, name, created_at, updated_at) \
2299 VALUES (?1, ?2, 'shared-slug', 'an atom', 1, 1)",
2300 rusqlite::params![id, namespace],
2301 )
2302 .expect("seed an atom on each side of the move");
2303 }
2304
2305 let request = MoveRequest::new(
2306 "source",
2307 vec![
2308 route("note:observation", "target"),
2309 route("note:insight", "target"),
2310 route("atom", "target"),
2311 ],
2312 );
2313 let error = move_namespace(&conn, &request).expect_err("the pre-flight refuses");
2314 let MoveError::Collisions { collisions } = error else {
2315 panic!("expected a named collision list, got {error:?}");
2316 };
2317
2318 assert_eq!(
2319 collisions.len(),
2320 1,
2321 "three routes share one target, so the one blocking row is reported \
2322 once: {collisions:?}"
2323 );
2324 assert_eq!(collisions[0].table, "knowledge_atoms");
2325 assert_eq!(collisions[0].constraint, "idx_knowledge_atoms_ns_slug");
2326 assert_eq!(collisions[0].key, "shared-slug");
2327 }
2328}
2329
2330#[cfg(test)]
2331#[test]
2332fn issue2673_namespace_move_advances_entity_version_without_changing_timestamp() {
2333 let mut conn = Connection::open_in_memory().unwrap();
2334 crate::migrations::run_migrations(&mut conn).unwrap();
2335 conn.execute("INSERT INTO entities(id,namespace,kind,name,created_at,updated_at) VALUES('versioned','source','concept','moved',7,7)", []).unwrap();
2336 conn.execute(
2337 "INSERT INTO notes(id,namespace,kind,content,created_at,updated_at) \
2338 VALUES('note-moved','source','observation','moved',11,11), \
2339 ('note-control','unrelated','observation','unchanged',13,13)",
2340 [],
2341 )
2342 .unwrap();
2343 let request = MoveRequest::new(
2344 "source",
2345 vec![
2346 MoveRoute {
2347 class: SubjectClass::Entity("concept".into()),
2348 target: "target".into(),
2349 },
2350 MoveRoute {
2351 class: SubjectClass::Note("observation".into()),
2352 target: "target".into(),
2353 },
2354 ],
2355 );
2356 let tx = conn.transaction().unwrap();
2357 let moved = move_namespace(&tx, &request).unwrap();
2358 assert_eq!(moved.subjects.get("entity:concept"), Some(&1));
2359 assert_eq!(moved.subjects.get("note:observation"), Some(&1));
2360 tx.commit().unwrap();
2361 let stored = conn
2362 .query_row(
2363 "SELECT namespace,updated_at,version FROM entities WHERE id='versioned'",
2364 [],
2365 |row| {
2366 Ok((
2367 row.get::<_, String>(0)?,
2368 row.get::<_, i64>(1)?,
2369 row.get::<_, i64>(2)?,
2370 ))
2371 },
2372 )
2373 .unwrap();
2374 assert_eq!(stored, ("target".into(), 7, 2));
2375 for (id, namespace, timestamp, version) in [
2376 ("note-moved", "target", 11_i64, 2_i64),
2377 ("note-control", "unrelated", 13_i64, 1_i64),
2378 ] {
2379 let stored = conn
2380 .query_row(
2381 "SELECT namespace,updated_at,version FROM notes WHERE id=?1",
2382 [id],
2383 |row| {
2384 Ok((
2385 row.get::<_, String>(0)?,
2386 row.get::<_, i64>(1)?,
2387 row.get::<_, i64>(2)?,
2388 ))
2389 },
2390 )
2391 .unwrap();
2392 assert_eq!(stored, (namespace.into(), timestamp, version));
2393 }
2394
2395 let tx = conn.transaction().unwrap();
2396 let repeated = move_namespace(&tx, &request).unwrap();
2397 assert_eq!(repeated.subjects.get("entity:concept"), Some(&0));
2398 assert_eq!(repeated.subjects.get("note:observation"), Some(&0));
2399 tx.commit().unwrap();
2400 for (sql, expected) in [
2401 ("SELECT version FROM entities WHERE id='versioned'", 2_i64),
2402 ("SELECT version FROM notes WHERE id='note-moved'", 2_i64),
2403 ("SELECT version FROM notes WHERE id='note-control'", 1_i64),
2404 ] {
2405 assert_eq!(
2406 conn.query_row(sql, [], |row| row.get::<_, i64>(0)).unwrap(),
2407 expected
2408 );
2409 }
2410}
2411
2412#[cfg(all(test, feature = "vectors"))]
2413#[path = "namespace_move_partition_tests.rs"]
2414mod partition_tests;
2415
2416#[cfg(test)]
2417#[path = "namespace_move_edge_tests.rs"]
2418mod edge_tests;
2419
2420#[cfg(test)]
2421#[path = "namespace_move_pack_tables_tests.rs"]
2422mod pack_table_tests;
2423
2424#[cfg(all(test, feature = "vectors"))]
2425#[path = "namespace_move_orphan_section_tests.rs"]
2426mod orphan_section_tests;