Skip to main content

silk/
graph.rs

1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use crate::clock::LamportClock;
4use crate::entry::{Entry, GraphOp, Hash, Value};
5use crate::ontology::Ontology;
6
7/// A materialized node in the graph.
8#[derive(Debug, Clone, PartialEq)]
9pub struct Node {
10    pub node_id: String,
11    pub node_type: String,
12    pub subtype: Option<String>,
13    pub label: String,
14    pub properties: BTreeMap<String, Value>,
15    /// Per-property clocks for LWW conflict resolution.
16    /// Each property key tracks the clock of its last write, so
17    /// concurrent updates to different properties don't conflict.
18    pub property_clocks: HashMap<String, LamportClock>,
19    /// Clock of the entry that last modified this node.
20    /// Used for add-wins semantics and label LWW.
21    pub last_clock: LamportClock,
22    /// Clock of the most recent AddNode for this node.
23    /// Used for add-wins semantics: remove only wins if its clock
24    /// is strictly greater than last_add_clock.
25    pub last_add_clock: LamportClock,
26    /// Whether this node has been tombstoned (removed).
27    pub tombstoned: bool,
28}
29
30/// A materialized edge in the graph.
31#[derive(Debug, Clone, PartialEq)]
32pub struct Edge {
33    pub edge_id: String,
34    pub edge_type: String,
35    pub source_id: String,
36    pub target_id: String,
37    pub properties: BTreeMap<String, Value>,
38    /// Per-property clocks for LWW conflict resolution.
39    pub property_clocks: HashMap<String, LamportClock>,
40    pub last_clock: LamportClock,
41    /// Clock of the most recent AddEdge for this edge.
42    pub last_add_clock: LamportClock,
43    pub tombstoned: bool,
44}
45
46/// Materialized graph — derived from the op log.
47///
48/// Provides fast queries without replaying the full log.
49/// Updated incrementally as new entries arrive, or rebuilt
50/// from scratch by replaying the entire op log.
51///
52/// CRDT semantics:
53/// - **Add-wins** for topology (concurrent add + remove → node/edge exists)
54/// - **LWW** (Last-Writer-Wins) per property key (highest Lamport clock wins)
55/// - **Tombstones** for deletes (mark as deleted, don't physically remove)
56pub struct MaterializedGraph {
57    /// node_id → Node
58    pub nodes: HashMap<String, Node>,
59    /// edge_id → Edge
60    pub edges: HashMap<String, Edge>,
61    /// node_id → set of outgoing edge_ids
62    pub outgoing: HashMap<String, HashSet<String>>,
63    /// node_id → set of incoming edge_ids
64    pub incoming: HashMap<String, HashSet<String>>,
65    /// node_type → set of node_ids (type index)
66    pub by_type: HashMap<String, HashSet<String>>,
67    /// The ontology (for validation during materialization)
68    pub ontology: Ontology,
69    /// R-02: entries that failed ontology validation during apply().
70    /// These entries exist in the oplog (for CRDT convergence) but are
71    /// invisible in the materialized graph. Grow-only within a single
72    /// materialization pass. Cleared and rebuilt on `rebuild()` — this
73    /// allows previously-quarantined entries to be re-evaluated when the
74    /// ontology evolves (e.g., after ExtendOntology arrives via sync).
75    pub quarantined: HashSet<Hash>,
76}
77
78impl MaterializedGraph {
79    /// Create an empty materialized graph with the given ontology.
80    pub fn new(ontology: Ontology) -> Self {
81        Self {
82            nodes: HashMap::new(),
83            edges: HashMap::new(),
84            outgoing: HashMap::new(),
85            incoming: HashMap::new(),
86            by_type: HashMap::new(),
87            ontology,
88            quarantined: HashSet::new(),
89        }
90    }
91
92    /// Apply a single entry to the graph (incremental materialization).
93    ///
94    /// R-02: Validates AddNode/AddEdge payloads against the ontology.
95    /// Invalid entries are quarantined (added to `self.quarantined`) and
96    /// skipped for materialization. They remain in the oplog for CRDT
97    /// convergence — quarantine is a graph-layer concern, not an oplog concern.
98    pub fn apply(&mut self, entry: &Entry) {
99        self.apply_entry(entry, false)
100    }
101
102    /// Core of `apply`. `trusted` is true only for a checkpoint's inner ops:
103    /// they were validated when first written and replay under the
104    /// checkpoint's own ontology, and build_checkpoint_ops emits AddNode with
105    /// empty properties by design (EXP-02) — re-validation would fail any
106    /// required property and quarantine the entity (Bug 14b). Quarantining
107    /// them is meaningless anyway: synthetic entry hashes don't exist in the
108    /// oplog. Compaction must reproduce the materialized graph exactly.
109    fn apply_entry(&mut self, entry: &Entry, trusted: bool) {
110        match &entry.payload {
111            GraphOp::Checkpoint { ops, op_clocks, .. } => {
112                // R-08: Replay synthetic ops to restore graph state.
113                // Bug 6 fix: use per-op clocks (preserves LWW metadata).
114                for (i, op) in ops.iter().enumerate() {
115                    // Bug 14 fix: compaction folds every ExtendOntology into the
116                    // checkpoint's inner DefineOntology. Apply it — otherwise
117                    // extension-typed entities fail validation and quarantine on
118                    // replay, and a reopened store materializes without them.
119                    // The oplog is authoritative; a declared ontology only seeds
120                    // new stores.
121                    if let GraphOp::DefineOntology { ontology } = op {
122                        self.ontology = ontology.clone();
123                        continue;
124                    }
125                    let clock = if i < op_clocks.len() {
126                        LamportClock::with_values(&entry.author, op_clocks[i].0, op_clocks[i].1)
127                    } else {
128                        entry.clock.clone() // fallback for old checkpoints without op_clocks
129                    };
130                    let synthetic = Entry::new(op.clone(), vec![], vec![], clock, &entry.author);
131                    self.apply_entry(&synthetic, true);
132                }
133            }
134            GraphOp::DefineOntology { .. } => {
135                // Genesis — nothing to materialize.
136            }
137            GraphOp::ExtendOntology { extension } => {
138                if let Err(_e) = self.ontology.merge_extension(extension) {
139                    if !trusted {
140                        self.quarantined.insert(entry.hash);
141                    }
142                }
143            }
144            GraphOp::AddNode {
145                node_id,
146                node_type,
147                subtype,
148                label,
149                properties,
150            } => {
151                // R-02: validate against ontology, quarantine if invalid
152                if !trusted {
153                    if let Err(_e) =
154                        self.ontology
155                            .validate_node(node_type, subtype.as_deref(), properties)
156                    {
157                        self.quarantined.insert(entry.hash);
158                        return;
159                    }
160                }
161                self.apply_add_node(
162                    node_id,
163                    node_type,
164                    subtype.as_deref(),
165                    label,
166                    properties,
167                    &entry.clock,
168                );
169            }
170            GraphOp::AddEdge {
171                edge_id,
172                edge_type,
173                source_id,
174                target_id,
175                properties,
176            } => {
177                if !trusted {
178                    // R-02: validate edge type exists.
179                    if !self.ontology.edge_types.contains_key(edge_type.as_str()) {
180                        self.quarantined.insert(entry.hash);
181                        return;
182                    }
183                    // Bug 13 fix: validate source/target type constraints when both nodes
184                    // are materialized. If one is missing (out-of-order sync), skip —
185                    // validation happens on rebuild.
186                    if let (Some(src), Some(tgt)) = (
187                        self.nodes.get(source_id.as_str()),
188                        self.nodes.get(target_id.as_str()),
189                    ) {
190                        if self
191                            .ontology
192                            .validate_edge(edge_type, &src.node_type, &tgt.node_type, properties)
193                            .is_err()
194                        {
195                            self.quarantined.insert(entry.hash);
196                            return;
197                        }
198                    }
199                }
200                self.apply_add_edge(
201                    edge_id,
202                    edge_type,
203                    source_id,
204                    target_id,
205                    properties,
206                    &entry.clock,
207                );
208            }
209            GraphOp::UpdateProperty {
210                entity_id,
211                key,
212                value,
213            } => {
214                self.apply_update_property(entity_id, key, value, &entry.clock);
215            }
216            GraphOp::RemoveNode { node_id } => {
217                self.apply_remove_node(node_id, &entry.clock);
218            }
219            GraphOp::RemoveEdge { edge_id } => {
220                self.apply_remove_edge(edge_id, &entry.clock);
221            }
222            GraphOp::DefineLens { .. } => {
223                // Reserved. No materialization — lenses are metadata, not graph state.
224            }
225        }
226    }
227
228    /// Apply a sequence of entries (full rematerialization from op log).
229    pub fn apply_all(&mut self, entries: &[&Entry]) {
230        for entry in entries {
231            self.apply(entry);
232        }
233    }
234
235    /// Rebuild from scratch: clear everything and replay all entries.
236    pub fn rebuild(&mut self, entries: &[&Entry]) {
237        self.nodes.clear();
238        self.edges.clear();
239        self.outgoing.clear();
240        self.incoming.clear();
241        self.by_type.clear();
242        self.quarantined.clear();
243        self.apply_all(entries);
244    }
245
246    // -- Queries --
247
248    /// Get a node by ID (returns None if not found or tombstoned).
249    pub fn get_node(&self, node_id: &str) -> Option<&Node> {
250        self.nodes.get(node_id).filter(|n| !n.tombstoned)
251    }
252
253    /// Get an edge by ID (returns None if not found or tombstoned).
254    pub fn get_edge(&self, edge_id: &str) -> Option<&Edge> {
255        self.edges.get(edge_id).filter(|e| !e.tombstoned)
256    }
257
258    /// Query all live nodes of a given type, including descendants (RDFS rdfs9).
259    /// If "entity" has children "server" and "project", querying "entity" returns all three.
260    pub fn nodes_by_type(&self, node_type: &str) -> Vec<&Node> {
261        let mut types = vec![node_type.to_string()];
262        types.extend(
263            self.ontology
264                .descendants(node_type)
265                .into_iter()
266                .map(|s| s.to_string()),
267        );
268        types
269            .iter()
270            .flat_map(|t| self.by_type.get(t.as_str()))
271            .flatten()
272            .filter_map(|id| self.get_node(id))
273            .collect()
274    }
275
276    /// Query all live nodes of a given subtype.
277    pub fn nodes_by_subtype(&self, subtype: &str) -> Vec<&Node> {
278        self.nodes
279            .values()
280            .filter(|n| !n.tombstoned && n.subtype.as_deref() == Some(subtype))
281            .collect()
282    }
283
284    /// Query nodes by a property value.
285    pub fn nodes_by_property(&self, key: &str, value: &Value) -> Vec<&Node> {
286        self.nodes
287            .values()
288            .filter(|n| !n.tombstoned && n.properties.get(key) == Some(value))
289            .collect()
290    }
291
292    /// Get outgoing edges for a node (only live edges with live endpoints).
293    pub fn outgoing_edges(&self, node_id: &str) -> Vec<&Edge> {
294        match self.outgoing.get(node_id) {
295            Some(edge_ids) => edge_ids
296                .iter()
297                .filter_map(|eid| self.get_edge(eid))
298                .filter(|e| self.is_node_live(&e.target_id))
299                .collect(),
300            None => vec![],
301        }
302    }
303
304    /// Get incoming edges for a node (only live edges with live endpoints).
305    pub fn incoming_edges(&self, node_id: &str) -> Vec<&Edge> {
306        match self.incoming.get(node_id) {
307            Some(edge_ids) => edge_ids
308                .iter()
309                .filter_map(|eid| self.get_edge(eid))
310                .filter(|e| self.is_node_live(&e.source_id))
311                .collect(),
312            None => vec![],
313        }
314    }
315
316    /// Approximate heap memory used by the materialized graph (bytes).
317    /// Uses fixed overhead estimates per node/edge. Does not account for heap
318    /// allocations behind String/Vec in property values or allocator fragmentation.
319    /// Actual memory may be 2-3x higher for string-heavy graphs.
320    pub fn estimated_memory_bytes(&self) -> usize {
321        let mut total = 0;
322        // Nodes: id string + type string + label + properties + clocks + overhead
323        for node in self.nodes.values() {
324            total += node.node_id.len() + node.node_type.len() + node.label.len();
325            total += node.subtype.as_ref().map_or(0, |s| s.len());
326            // Properties: key + estimated value size + clock per property
327            for (k, v) in &node.properties {
328                total += k.len() + std::mem::size_of_val(v) + 48; // key + value + clock overhead
329            }
330            total += 128; // fixed struct overhead (clocks, bools, HashMap shells)
331        }
332        // Edges: similar structure
333        for edge in self.edges.values() {
334            total += edge.edge_id.len() + edge.edge_type.len();
335            total += edge.source_id.len() + edge.target_id.len();
336            for (k, v) in &edge.properties {
337                total += k.len() + std::mem::size_of_val(v) + 48;
338            }
339            total += 128;
340        }
341        // Adjacency indexes: outgoing + incoming (id strings + HashSet overhead)
342        for (k, set) in &self.outgoing {
343            total += k.len() + set.len() * 32;
344        }
345        for (k, set) in &self.incoming {
346            total += k.len() + set.len() * 32;
347        }
348        // Type index
349        for (k, set) in &self.by_type {
350            total += k.len() + set.len() * 32;
351        }
352        // Quarantine set
353        total += self.quarantined.len() * 48;
354        total
355    }
356
357    /// All live nodes.
358    pub fn all_nodes(&self) -> Vec<&Node> {
359        self.nodes.values().filter(|n| !n.tombstoned).collect()
360    }
361
362    /// All live edges (with live endpoints).
363    pub fn all_edges(&self) -> Vec<&Edge> {
364        self.edges
365            .values()
366            .filter(|e| {
367                !e.tombstoned && self.is_node_live(&e.source_id) && self.is_node_live(&e.target_id)
368            })
369            .collect()
370    }
371
372    /// Neighbors of a node (connected via outgoing edges).
373    pub fn neighbors(&self, node_id: &str) -> Vec<&str> {
374        self.outgoing_edges(node_id)
375            .iter()
376            .map(|e| e.target_id.as_str())
377            .collect()
378    }
379
380    /// Reverse neighbors (connected via incoming edges).
381    pub fn reverse_neighbors(&self, node_id: &str) -> Vec<&str> {
382        self.incoming_edges(node_id)
383            .iter()
384            .map(|e| e.source_id.as_str())
385            .collect()
386    }
387
388    // -- CRDT application helpers --
389
390    fn apply_add_node(
391        &mut self,
392        node_id: &str,
393        node_type: &str,
394        subtype: Option<&str>,
395        label: &str,
396        properties: &BTreeMap<String, Value>,
397        clock: &LamportClock,
398    ) {
399        if let Some(existing) = self.nodes.get_mut(node_id) {
400            // Add-wins: always resurrect from tombstone.
401            existing.tombstoned = false;
402            // Track the latest add clock for add-wins semantics.
403            if clock_wins(clock, &existing.last_add_clock) {
404                existing.last_add_clock = clock.clone();
405            }
406            // LWW merge for label, subtype, and properties.
407            if clock_wins(clock, &existing.last_clock) {
408                existing.label = label.to_string();
409                existing.subtype = subtype.map(|s| s.to_string());
410                existing.last_clock = clock.clone();
411            }
412            merge_properties_lww(
413                &mut existing.properties,
414                &mut existing.property_clocks,
415                properties,
416                clock,
417            );
418        } else {
419            let property_clocks: HashMap<String, LamportClock> = properties
420                .keys()
421                .map(|k| (k.clone(), clock.clone()))
422                .collect();
423            let node = Node {
424                node_id: node_id.to_string(),
425                node_type: node_type.to_string(),
426                subtype: subtype.map(|s| s.to_string()),
427                label: label.to_string(),
428                properties: properties.clone(),
429                property_clocks,
430                last_clock: clock.clone(),
431                last_add_clock: clock.clone(),
432                tombstoned: false,
433            };
434            self.by_type
435                .entry(node_type.to_string())
436                .or_default()
437                .insert(node_id.to_string());
438            self.nodes.insert(node_id.to_string(), node);
439        }
440    }
441
442    fn apply_add_edge(
443        &mut self,
444        edge_id: &str,
445        edge_type: &str,
446        source_id: &str,
447        target_id: &str,
448        properties: &BTreeMap<String, Value>,
449        clock: &LamportClock,
450    ) {
451        if let Some(existing) = self.edges.get_mut(edge_id) {
452            // Add-wins: always resurrect if tombstoned.
453            existing.tombstoned = false;
454            if clock_wins(clock, &existing.last_add_clock) {
455                existing.last_add_clock = clock.clone();
456            }
457            if clock_wins(clock, &existing.last_clock) {
458                existing.last_clock = clock.clone();
459            }
460            merge_properties_lww(
461                &mut existing.properties,
462                &mut existing.property_clocks,
463                properties,
464                clock,
465            );
466        } else {
467            let property_clocks: HashMap<String, LamportClock> = properties
468                .keys()
469                .map(|k| (k.clone(), clock.clone()))
470                .collect();
471            let edge = Edge {
472                edge_id: edge_id.to_string(),
473                edge_type: edge_type.to_string(),
474                source_id: source_id.to_string(),
475                target_id: target_id.to_string(),
476                properties: properties.clone(),
477                property_clocks,
478                last_clock: clock.clone(),
479                last_add_clock: clock.clone(),
480                tombstoned: false,
481            };
482            self.outgoing
483                .entry(source_id.to_string())
484                .or_default()
485                .insert(edge_id.to_string());
486            self.incoming
487                .entry(target_id.to_string())
488                .or_default()
489                .insert(edge_id.to_string());
490            self.edges.insert(edge_id.to_string(), edge);
491        }
492    }
493
494    fn apply_update_property(
495        &mut self,
496        entity_id: &str,
497        key: &str,
498        value: &Value,
499        clock: &LamportClock,
500    ) {
501        // Try node first, then edge. Per-property LWW: each key competes
502        // only with other writes to the same key, not the entire entity.
503        if let Some(node) = self.nodes.get_mut(entity_id) {
504            let dominated = node
505                .property_clocks
506                .get(key)
507                .map(|c| clock_wins(clock, c))
508                .unwrap_or(true);
509            if dominated {
510                node.properties.insert(key.to_string(), value.clone());
511                node.property_clocks.insert(key.to_string(), clock.clone());
512            }
513            // Update entity-level clock for add-wins tracking.
514            if clock_wins(clock, &node.last_clock) {
515                node.last_clock = clock.clone();
516            }
517        } else if let Some(edge) = self.edges.get_mut(entity_id) {
518            let dominated = edge
519                .property_clocks
520                .get(key)
521                .map(|c| clock_wins(clock, c))
522                .unwrap_or(true);
523            if dominated {
524                edge.properties.insert(key.to_string(), value.clone());
525                edge.property_clocks.insert(key.to_string(), clock.clone());
526            }
527            if clock_wins(clock, &edge.last_clock) {
528                edge.last_clock = clock.clone();
529            }
530        }
531        // If entity not found, silently ignore (may arrive out of order in sync).
532    }
533
534    fn apply_remove_node(&mut self, node_id: &str, clock: &LamportClock) {
535        if let Some(node) = self.nodes.get_mut(node_id) {
536            // Add-wins: only tombstone if the remove clock is strictly greater
537            // than the last add clock. If a concurrent (or later) add exists,
538            // the node stays alive.
539            if clock_wins(clock, &node.last_add_clock) {
540                node.tombstoned = true;
541                node.last_clock = clock.clone();
542            }
543        }
544        // Tombstoning a node doesn't physically remove edges — they just become
545        // invisible via is_node_live() checks in queries.
546    }
547
548    fn apply_remove_edge(&mut self, edge_id: &str, clock: &LamportClock) {
549        if let Some(edge) = self.edges.get_mut(edge_id) {
550            // Add-wins: only tombstone if remove clock > last add clock.
551            if clock_wins(clock, &edge.last_add_clock) {
552                edge.tombstoned = true;
553                edge.last_clock = clock.clone();
554            }
555        }
556    }
557
558    fn is_node_live(&self, node_id: &str) -> bool {
559        self.nodes
560            .get(node_id)
561            .map(|n| !n.tombstoned)
562            .unwrap_or(false)
563    }
564}
565
566/// Per-property LWW merge: each property from `new_props` competes with
567/// existing properties. Higher clock wins per key.
568fn merge_properties_lww(
569    existing_props: &mut BTreeMap<String, Value>,
570    existing_clocks: &mut HashMap<String, LamportClock>,
571    new_props: &BTreeMap<String, Value>,
572    clock: &LamportClock,
573) {
574    for (k, v) in new_props {
575        let dominated = existing_clocks
576            .get(k)
577            .map(|c| clock_wins(clock, c))
578            .unwrap_or(true);
579        if dominated {
580            existing_props.insert(k.clone(), v.clone());
581            existing_clocks.insert(k.clone(), clock.clone());
582        }
583    }
584}
585
586/// LWW comparison: returns true if `new_clock` wins over `existing_clock`.
587/// Uses HybridClock total ordering: (physical_ms, logical, id).
588fn clock_wins(new_clock: &LamportClock, existing_clock: &LamportClock) -> bool {
589    new_clock.cmp_order(existing_clock) == std::cmp::Ordering::Greater
590}
591
592#[cfg(test)]
593mod tests {
594    use super::*;
595    use crate::entry::Entry;
596    use crate::ontology::{EdgeTypeDef, NodeTypeDef};
597
598    fn test_ontology() -> Ontology {
599        Ontology {
600            node_types: BTreeMap::from([
601                (
602                    "entity".into(),
603                    NodeTypeDef {
604                        description: None,
605                        properties: BTreeMap::new(),
606                        subtypes: None,
607                        parent_type: None,
608                    },
609                ),
610                (
611                    "signal".into(),
612                    NodeTypeDef {
613                        description: None,
614                        properties: BTreeMap::new(),
615                        subtypes: None,
616                        parent_type: None,
617                    },
618                ),
619            ]),
620            edge_types: BTreeMap::from([
621                (
622                    "RUNS_ON".into(),
623                    EdgeTypeDef {
624                        description: None,
625                        source_types: vec!["entity".into()],
626                        target_types: vec!["entity".into()],
627                        properties: BTreeMap::new(),
628                    },
629                ),
630                (
631                    "OBSERVES".into(),
632                    EdgeTypeDef {
633                        description: None,
634                        source_types: vec!["signal".into()],
635                        target_types: vec!["entity".into()],
636                        properties: BTreeMap::new(),
637                    },
638                ),
639            ]),
640        }
641    }
642
643    fn make_entry(op: GraphOp, clock_time: u64, author: &str) -> Entry {
644        Entry::new(
645            op,
646            vec![],
647            vec![],
648            LamportClock::with_values(author, clock_time, 0),
649            author,
650        )
651    }
652
653    // -- test_graph.rs spec from docs/silk.md --
654
655    #[test]
656    fn add_node_appears_in_query() {
657        let mut g = MaterializedGraph::new(test_ontology());
658        let entry = make_entry(
659            GraphOp::AddNode {
660                node_id: "server-1".into(),
661                node_type: "entity".into(),
662                label: "Server 1".into(),
663                properties: BTreeMap::from([("ip".into(), Value::String("10.0.0.1".into()))]),
664                subtype: None,
665            },
666            1,
667            "inst-a",
668        );
669        g.apply(&entry);
670
671        let node = g.get_node("server-1").unwrap();
672        assert_eq!(node.node_type, "entity");
673        assert_eq!(node.label, "Server 1");
674        assert_eq!(
675            node.properties.get("ip"),
676            Some(&Value::String("10.0.0.1".into()))
677        );
678    }
679
680    #[test]
681    fn add_edge_creates_adjacency() {
682        let mut g = MaterializedGraph::new(test_ontology());
683        g.apply(&make_entry(
684            GraphOp::AddNode {
685                node_id: "svc".into(),
686                node_type: "entity".into(),
687                label: "svc".into(),
688                properties: BTreeMap::new(),
689                subtype: None,
690            },
691            1,
692            "inst-a",
693        ));
694        g.apply(&make_entry(
695            GraphOp::AddNode {
696                node_id: "srv".into(),
697                node_type: "entity".into(),
698                label: "srv".into(),
699                properties: BTreeMap::new(),
700                subtype: None,
701            },
702            2,
703            "inst-a",
704        ));
705        g.apply(&make_entry(
706            GraphOp::AddEdge {
707                edge_id: "e1".into(),
708                edge_type: "RUNS_ON".into(),
709                source_id: "svc".into(),
710                target_id: "srv".into(),
711                properties: BTreeMap::new(),
712            },
713            3,
714            "inst-a",
715        ));
716
717        // Both endpoints know about the edge.
718        let out = g.outgoing_edges("svc");
719        assert_eq!(out.len(), 1);
720        assert_eq!(out[0].target_id, "srv");
721
722        let inc = g.incoming_edges("srv");
723        assert_eq!(inc.len(), 1);
724        assert_eq!(inc[0].source_id, "svc");
725
726        assert_eq!(g.neighbors("svc"), vec!["srv"]);
727    }
728
729    #[test]
730    fn update_property_reflected() {
731        let mut g = MaterializedGraph::new(test_ontology());
732        g.apply(&make_entry(
733            GraphOp::AddNode {
734                node_id: "s1".into(),
735                node_type: "entity".into(),
736                label: "s1".into(),
737                properties: BTreeMap::new(),
738                subtype: None,
739            },
740            1,
741            "inst-a",
742        ));
743        g.apply(&make_entry(
744            GraphOp::UpdateProperty {
745                entity_id: "s1".into(),
746                key: "cpu".into(),
747                value: Value::Float(85.5),
748            },
749            2,
750            "inst-a",
751        ));
752
753        let node = g.get_node("s1").unwrap();
754        assert_eq!(node.properties.get("cpu"), Some(&Value::Float(85.5)));
755    }
756
757    #[test]
758    fn remove_node_cascades_edges() {
759        let mut g = MaterializedGraph::new(test_ontology());
760        g.apply(&make_entry(
761            GraphOp::AddNode {
762                node_id: "a".into(),
763                node_type: "entity".into(),
764                label: "a".into(),
765                properties: BTreeMap::new(),
766                subtype: None,
767            },
768            1,
769            "inst-a",
770        ));
771        g.apply(&make_entry(
772            GraphOp::AddNode {
773                node_id: "b".into(),
774                node_type: "entity".into(),
775                label: "b".into(),
776                properties: BTreeMap::new(),
777                subtype: None,
778            },
779            2,
780            "inst-a",
781        ));
782        g.apply(&make_entry(
783            GraphOp::AddEdge {
784                edge_id: "e1".into(),
785                edge_type: "RUNS_ON".into(),
786                source_id: "a".into(),
787                target_id: "b".into(),
788                properties: BTreeMap::new(),
789            },
790            3,
791            "inst-a",
792        ));
793        assert_eq!(g.all_edges().len(), 1);
794
795        // Remove node 'b' — edge becomes invisible (dangling target).
796        g.apply(&make_entry(
797            GraphOp::RemoveNode {
798                node_id: "b".into(),
799            },
800            4,
801            "inst-a",
802        ));
803        assert!(g.get_node("b").is_none());
804        // Edge still exists but not returned by all_edges (target tombstoned).
805        assert_eq!(g.all_edges().len(), 0);
806        // Outgoing from 'a' also filters out dangling edges.
807        assert_eq!(g.outgoing_edges("a").len(), 0);
808    }
809
810    #[test]
811    fn remove_edge_preserves_nodes() {
812        let mut g = MaterializedGraph::new(test_ontology());
813        g.apply(&make_entry(
814            GraphOp::AddNode {
815                node_id: "a".into(),
816                node_type: "entity".into(),
817                label: "a".into(),
818                properties: BTreeMap::new(),
819                subtype: None,
820            },
821            1,
822            "inst-a",
823        ));
824        g.apply(&make_entry(
825            GraphOp::AddNode {
826                node_id: "b".into(),
827                node_type: "entity".into(),
828                label: "b".into(),
829                properties: BTreeMap::new(),
830                subtype: None,
831            },
832            2,
833            "inst-a",
834        ));
835        g.apply(&make_entry(
836            GraphOp::AddEdge {
837                edge_id: "e1".into(),
838                edge_type: "RUNS_ON".into(),
839                source_id: "a".into(),
840                target_id: "b".into(),
841                properties: BTreeMap::new(),
842            },
843            3,
844            "inst-a",
845        ));
846        g.apply(&make_entry(
847            GraphOp::RemoveEdge {
848                edge_id: "e1".into(),
849            },
850            4,
851            "inst-a",
852        ));
853
854        // Nodes still exist.
855        assert!(g.get_node("a").is_some());
856        assert!(g.get_node("b").is_some());
857        // Edge is gone.
858        assert!(g.get_edge("e1").is_none());
859        assert_eq!(g.all_edges().len(), 0);
860    }
861
862    #[test]
863    fn query_by_type_filters() {
864        let mut g = MaterializedGraph::new(test_ontology());
865        g.apply(&make_entry(
866            GraphOp::AddNode {
867                node_id: "s1".into(),
868                node_type: "entity".into(),
869                label: "s1".into(),
870                properties: BTreeMap::new(),
871                subtype: None,
872            },
873            1,
874            "inst-a",
875        ));
876        g.apply(&make_entry(
877            GraphOp::AddNode {
878                node_id: "s2".into(),
879                node_type: "entity".into(),
880                label: "s2".into(),
881                properties: BTreeMap::new(),
882                subtype: None,
883            },
884            2,
885            "inst-a",
886        ));
887        g.apply(&make_entry(
888            GraphOp::AddNode {
889                node_id: "alert".into(),
890                node_type: "signal".into(),
891                label: "alert".into(),
892                properties: BTreeMap::new(),
893                subtype: None,
894            },
895            3,
896            "inst-a",
897        ));
898
899        let entities = g.nodes_by_type("entity");
900        assert_eq!(entities.len(), 2);
901        let signals = g.nodes_by_type("signal");
902        assert_eq!(signals.len(), 1);
903        assert_eq!(signals[0].node_id, "alert");
904    }
905
906    #[test]
907    fn query_by_property_filters() {
908        let mut g = MaterializedGraph::new(test_ontology());
909        g.apply(&make_entry(
910            GraphOp::AddNode {
911                node_id: "s1".into(),
912                node_type: "entity".into(),
913                label: "s1".into(),
914                properties: BTreeMap::from([("status".into(), Value::String("alive".into()))]),
915                subtype: None,
916            },
917            1,
918            "inst-a",
919        ));
920        g.apply(&make_entry(
921            GraphOp::AddNode {
922                node_id: "s2".into(),
923                node_type: "entity".into(),
924                label: "s2".into(),
925                properties: BTreeMap::from([("status".into(), Value::String("dead".into()))]),
926                subtype: None,
927            },
928            2,
929            "inst-a",
930        ));
931
932        let alive = g.nodes_by_property("status", &Value::String("alive".into()));
933        assert_eq!(alive.len(), 1);
934        assert_eq!(alive[0].node_id, "s1");
935    }
936
937    #[test]
938    fn materialization_from_empty() {
939        // Build graph incrementally.
940        let mut g1 = MaterializedGraph::new(test_ontology());
941        let entries = vec![
942            make_entry(
943                GraphOp::DefineOntology {
944                    ontology: test_ontology(),
945                },
946                0,
947                "inst-a",
948            ),
949            make_entry(
950                GraphOp::AddNode {
951                    node_id: "a".into(),
952                    node_type: "entity".into(),
953                    label: "a".into(),
954                    properties: BTreeMap::new(),
955                    subtype: None,
956                },
957                1,
958                "inst-a",
959            ),
960            make_entry(
961                GraphOp::AddNode {
962                    node_id: "b".into(),
963                    node_type: "entity".into(),
964                    label: "b".into(),
965                    properties: BTreeMap::new(),
966                    subtype: None,
967                },
968                2,
969                "inst-a",
970            ),
971            make_entry(
972                GraphOp::AddEdge {
973                    edge_id: "e1".into(),
974                    edge_type: "RUNS_ON".into(),
975                    source_id: "a".into(),
976                    target_id: "b".into(),
977                    properties: BTreeMap::new(),
978                },
979                3,
980                "inst-a",
981            ),
982        ];
983        for e in &entries {
984            g1.apply(e);
985        }
986
987        // Rebuild from scratch.
988        let mut g2 = MaterializedGraph::new(test_ontology());
989        let refs: Vec<&Entry> = entries.iter().collect();
990        g2.rebuild(&refs);
991
992        // Same result.
993        assert_eq!(g1.all_nodes().len(), g2.all_nodes().len());
994        assert_eq!(g1.all_edges().len(), g2.all_edges().len());
995        for node in g1.all_nodes() {
996            let n2 = g2.get_node(&node.node_id).unwrap();
997            assert_eq!(node.node_type, n2.node_type);
998            assert_eq!(node.properties, n2.properties);
999        }
1000    }
1001
1002    #[test]
1003    fn incremental_equals_full() {
1004        let entries = vec![
1005            make_entry(
1006                GraphOp::DefineOntology {
1007                    ontology: test_ontology(),
1008                },
1009                0,
1010                "inst-a",
1011            ),
1012            make_entry(
1013                GraphOp::AddNode {
1014                    node_id: "a".into(),
1015                    node_type: "entity".into(),
1016                    label: "a".into(),
1017                    properties: BTreeMap::from([("x".into(), Value::Int(1))]),
1018                    subtype: None,
1019                },
1020                1,
1021                "inst-a",
1022            ),
1023            make_entry(
1024                GraphOp::UpdateProperty {
1025                    entity_id: "a".into(),
1026                    key: "x".into(),
1027                    value: Value::Int(2),
1028                },
1029                2,
1030                "inst-a",
1031            ),
1032            make_entry(
1033                GraphOp::AddNode {
1034                    node_id: "b".into(),
1035                    node_type: "entity".into(),
1036                    label: "b".into(),
1037                    properties: BTreeMap::new(),
1038                    subtype: None,
1039                },
1040                3,
1041                "inst-a",
1042            ),
1043            make_entry(
1044                GraphOp::AddEdge {
1045                    edge_id: "e1".into(),
1046                    edge_type: "RUNS_ON".into(),
1047                    source_id: "a".into(),
1048                    target_id: "b".into(),
1049                    properties: BTreeMap::new(),
1050                },
1051                4,
1052                "inst-a",
1053            ),
1054            make_entry(
1055                GraphOp::RemoveEdge {
1056                    edge_id: "e1".into(),
1057                },
1058                5,
1059                "inst-a",
1060            ),
1061        ];
1062
1063        // Incremental.
1064        let mut g_inc = MaterializedGraph::new(test_ontology());
1065        for e in &entries {
1066            g_inc.apply(e);
1067        }
1068
1069        // Full replay.
1070        let mut g_full = MaterializedGraph::new(test_ontology());
1071        let refs: Vec<&Entry> = entries.iter().collect();
1072        g_full.rebuild(&refs);
1073
1074        // Property should be 2 (updated).
1075        assert_eq!(
1076            g_inc.get_node("a").unwrap().properties.get("x"),
1077            Some(&Value::Int(2))
1078        );
1079        assert_eq!(
1080            g_full.get_node("a").unwrap().properties.get("x"),
1081            Some(&Value::Int(2))
1082        );
1083        // Edge should be removed.
1084        assert_eq!(g_inc.all_edges().len(), 0);
1085        assert_eq!(g_full.all_edges().len(), 0);
1086    }
1087
1088    #[test]
1089    fn lww_concurrent_property_update() {
1090        // Two instances update the same property — higher clock wins.
1091        let mut g = MaterializedGraph::new(test_ontology());
1092        g.apply(&make_entry(
1093            GraphOp::AddNode {
1094                node_id: "s1".into(),
1095                node_type: "entity".into(),
1096                label: "s1".into(),
1097                properties: BTreeMap::new(),
1098                subtype: None,
1099            },
1100            1,
1101            "inst-a",
1102        ));
1103        // inst-a sets status=alive at time 2
1104        g.apply(&make_entry(
1105            GraphOp::UpdateProperty {
1106                entity_id: "s1".into(),
1107                key: "status".into(),
1108                value: Value::String("alive".into()),
1109            },
1110            2,
1111            "inst-a",
1112        ));
1113        // inst-b sets status=dead at time 3 — wins (higher clock)
1114        g.apply(&make_entry(
1115            GraphOp::UpdateProperty {
1116                entity_id: "s1".into(),
1117                key: "status".into(),
1118                value: Value::String("dead".into()),
1119            },
1120            3,
1121            "inst-b",
1122        ));
1123        assert_eq!(
1124            g.get_node("s1").unwrap().properties.get("status"),
1125            Some(&Value::String("dead".into()))
1126        );
1127    }
1128
1129    #[test]
1130    fn lww_tiebreak_by_instance_id() {
1131        // Same clock time — higher instance ID wins.
1132        let mut g = MaterializedGraph::new(test_ontology());
1133        g.apply(&make_entry(
1134            GraphOp::AddNode {
1135                node_id: "s1".into(),
1136                node_type: "entity".into(),
1137                label: "s1".into(),
1138                properties: BTreeMap::new(),
1139                subtype: None,
1140            },
1141            1,
1142            "inst-a",
1143        ));
1144        // Both at physical_ms=5, logical=0. Lower id wins → "inst-a" wins.
1145        g.apply(&make_entry(
1146            GraphOp::UpdateProperty {
1147                entity_id: "s1".into(),
1148                key: "x".into(),
1149                value: Value::Int(1),
1150            },
1151            5,
1152            "inst-a",
1153        ));
1154        g.apply(&make_entry(
1155            GraphOp::UpdateProperty {
1156                entity_id: "s1".into(),
1157                key: "x".into(),
1158                value: Value::Int(2),
1159            },
1160            5,
1161            "inst-b",
1162        ));
1163        // inst-a has lower id → wins the tiebreak → value stays Int(1).
1164        assert_eq!(
1165            g.get_node("s1").unwrap().properties.get("x"),
1166            Some(&Value::Int(1))
1167        );
1168    }
1169
1170    #[test]
1171    fn lww_per_property_concurrent_different_keys() {
1172        // Two instances concurrently update DIFFERENT properties at the same
1173        // clock time. Both updates must be accepted — they don't conflict.
1174        // This requires per-property LWW, not node-level LWW.
1175        let mut g = MaterializedGraph::new(test_ontology());
1176        g.apply(&make_entry(
1177            GraphOp::AddNode {
1178                node_id: "s1".into(),
1179                node_type: "entity".into(),
1180                label: "s1".into(),
1181                properties: BTreeMap::from([
1182                    ("x".into(), Value::Int(0)),
1183                    ("y".into(), Value::Int(0)),
1184                ]),
1185                subtype: None,
1186            },
1187            1,
1188            "inst-a",
1189        ));
1190        // inst-a updates "x" at time 3
1191        g.apply(&make_entry(
1192            GraphOp::UpdateProperty {
1193                entity_id: "s1".into(),
1194                key: "x".into(),
1195                value: Value::Int(42),
1196            },
1197            3,
1198            "inst-a",
1199        ));
1200        // inst-b updates "y" at time 3 (concurrent, different property)
1201        g.apply(&make_entry(
1202            GraphOp::UpdateProperty {
1203                entity_id: "s1".into(),
1204                key: "y".into(),
1205                value: Value::Int(99),
1206            },
1207            3,
1208            "inst-b",
1209        ));
1210
1211        let node = g.get_node("s1").unwrap();
1212        // Both updates must be applied — no conflict.
1213        assert_eq!(
1214            node.properties.get("x"),
1215            Some(&Value::Int(42)),
1216            "update to 'x' must not be rejected by concurrent update to 'y'"
1217        );
1218        assert_eq!(
1219            node.properties.get("y"),
1220            Some(&Value::Int(99)),
1221            "update to 'y' must not be rejected by concurrent update to 'x'"
1222        );
1223    }
1224
1225    #[test]
1226    fn lww_per_property_order_independent() {
1227        // Same scenario but applied in reverse order — result must be identical.
1228        let mut g = MaterializedGraph::new(test_ontology());
1229        g.apply(&make_entry(
1230            GraphOp::AddNode {
1231                node_id: "s1".into(),
1232                node_type: "entity".into(),
1233                label: "s1".into(),
1234                properties: BTreeMap::from([
1235                    ("x".into(), Value::Int(0)),
1236                    ("y".into(), Value::Int(0)),
1237                ]),
1238                subtype: None,
1239            },
1240            1,
1241            "inst-a",
1242        ));
1243        // Apply inst-b first this time
1244        g.apply(&make_entry(
1245            GraphOp::UpdateProperty {
1246                entity_id: "s1".into(),
1247                key: "y".into(),
1248                value: Value::Int(99),
1249            },
1250            3,
1251            "inst-b",
1252        ));
1253        g.apply(&make_entry(
1254            GraphOp::UpdateProperty {
1255                entity_id: "s1".into(),
1256                key: "x".into(),
1257                value: Value::Int(42),
1258            },
1259            3,
1260            "inst-a",
1261        ));
1262
1263        let node = g.get_node("s1").unwrap();
1264        assert_eq!(node.properties.get("x"), Some(&Value::Int(42)));
1265        assert_eq!(node.properties.get("y"), Some(&Value::Int(99)));
1266    }
1267
1268    #[test]
1269    fn add_wins_over_remove() {
1270        // Concurrent add + remove → node should exist (add-wins).
1271        let mut g = MaterializedGraph::new(test_ontology());
1272        g.apply(&make_entry(
1273            GraphOp::AddNode {
1274                node_id: "s1".into(),
1275                node_type: "entity".into(),
1276                label: "s1".into(),
1277                properties: BTreeMap::new(),
1278                subtype: None,
1279            },
1280            1,
1281            "inst-a",
1282        ));
1283        // Remove at time 2.
1284        g.apply(&make_entry(
1285            GraphOp::RemoveNode {
1286                node_id: "s1".into(),
1287            },
1288            2,
1289            "inst-a",
1290        ));
1291        assert!(g.get_node("s1").is_none());
1292
1293        // Re-add at time 3 (add-wins — resurrects).
1294        g.apply(&make_entry(
1295            GraphOp::AddNode {
1296                node_id: "s1".into(),
1297                node_type: "entity".into(),
1298                label: "s1 v2".into(),
1299                properties: BTreeMap::new(),
1300                subtype: None,
1301            },
1302            3,
1303            "inst-b",
1304        ));
1305        let node = g.get_node("s1").unwrap();
1306        assert_eq!(node.label, "s1 v2");
1307        assert!(!node.tombstoned);
1308    }
1309
1310    /// Bug 14: a checkpoint's inner DefineOntology carries the merged ontology
1311    /// (compaction folds ExtendOntology entries into it). Replay must apply it,
1312    /// or entities typed by an extension quarantine and vanish from the graph.
1313    #[test]
1314    fn checkpoint_replay_applies_inner_define_ontology() {
1315        // Base ontology: "entity" only — no "signal".
1316        let mut base = test_ontology();
1317        base.node_types.remove("signal");
1318        base.edge_types.remove("OBSERVES");
1319        let mut g = MaterializedGraph::new(base);
1320
1321        let checkpoint = make_entry(
1322            GraphOp::Checkpoint {
1323                ops: vec![
1324                    GraphOp::DefineOntology {
1325                        ontology: test_ontology(), // merged: has "signal"
1326                    },
1327                    GraphOp::AddNode {
1328                        node_id: "n1".into(),
1329                        node_type: "entity".into(),
1330                        label: "base-typed".into(),
1331                        properties: BTreeMap::new(),
1332                        subtype: None,
1333                    },
1334                    GraphOp::AddNode {
1335                        node_id: "s1".into(),
1336                        node_type: "signal".into(),
1337                        label: "extension-typed".into(),
1338                        properties: BTreeMap::new(),
1339                        subtype: None,
1340                    },
1341                ],
1342                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1343                compacted_at_physical_ms: 1,
1344                compacted_at_logical: 2,
1345            },
1346            1,
1347            "inst-a",
1348        );
1349        g.apply(&checkpoint);
1350
1351        assert!(g.get_node("n1").is_some());
1352        assert!(g.get_node("s1").is_some(), "extension-typed node lost");
1353        assert!(g.ontology.node_types.contains_key("signal"));
1354        assert!(g.quarantined.is_empty());
1355    }
1356
1357    /// Bug 14b: build_checkpoint_ops emits AddNode with empty properties
1358    /// (EXP-02: per-property clocks ride in separate UpdateProperty ops).
1359    /// Replay must trust checkpoint inner ops — validating them fails any
1360    /// required property against the empty map and quarantines the node.
1361    #[test]
1362    fn checkpoint_replay_trusts_inner_ops() {
1363        use crate::ontology::{PropertyDef, ValueType};
1364
1365        let mut ont = test_ontology();
1366        ont.node_types.get_mut("entity").unwrap().properties.insert(
1367            "name".into(),
1368            PropertyDef {
1369                value_type: ValueType::String,
1370                required: true,
1371                description: None,
1372                constraints: None,
1373            },
1374        );
1375        let mut g = MaterializedGraph::new(ont.clone());
1376
1377        let checkpoint = make_entry(
1378            GraphOp::Checkpoint {
1379                ops: vec![
1380                    GraphOp::DefineOntology { ontology: ont },
1381                    GraphOp::AddNode {
1382                        node_id: "n1".into(),
1383                        node_type: "entity".into(),
1384                        label: "req".into(),
1385                        properties: BTreeMap::new(), // empty by design
1386                        subtype: None,
1387                    },
1388                    GraphOp::UpdateProperty {
1389                        entity_id: "n1".into(),
1390                        key: "name".into(),
1391                        value: Value::String("x".into()),
1392                    },
1393                ],
1394                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1395                compacted_at_physical_ms: 1,
1396                compacted_at_logical: 2,
1397            },
1398            1,
1399            "inst-a",
1400        );
1401        g.apply(&checkpoint);
1402
1403        let node = g.get_node("n1").expect("required-property node lost");
1404        assert_eq!(
1405            node.properties.get("name"),
1406            Some(&Value::String("x".into()))
1407        );
1408        assert!(g.quarantined.is_empty());
1409    }
1410}