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, OntologyExtension, ValidationMode};
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/// Why an entry was quarantined (S1).
47///
48/// The validators produce a precise diagnosis — type name, property,
49/// expected type, allowed set, constraint name — and every quarantine site
50/// used to drop it one line before the operator needed it, leaving a bare
51/// hash and no way to tell an unknown type from a constraint violation.
52#[derive(Debug, Clone, PartialEq)]
53pub struct QuarantineRecord {
54    /// The operation kind that was rejected, e.g. "add_node".
55    pub op_kind: String,
56    /// The validator's own message.
57    pub reason: String,
58    /// Hex content hash of the ontology this decision was made against, so a
59    /// stale diagnosis is recognizable after the schema moves.
60    pub ontology_hash: String,
61}
62
63/// Materialized graph — derived from the op log.
64///
65/// Provides fast queries without replaying the full log.
66/// Updated incrementally as new entries arrive, or rebuilt
67/// from scratch by replaying the entire op log.
68///
69/// CRDT semantics:
70/// - **Add-wins** for topology (concurrent add + remove → node/edge exists)
71/// - **LWW** (Last-Writer-Wins) per property key (highest Lamport clock wins)
72/// - **Tombstones** for deletes (mark as deleted, don't physically remove)
73#[derive(Clone)]
74pub struct MaterializedGraph {
75    /// node_id → Node
76    pub nodes: HashMap<String, Node>,
77    /// edge_id → Edge
78    pub edges: HashMap<String, Edge>,
79    /// node_id → set of outgoing edge_ids
80    pub outgoing: HashMap<String, HashSet<String>>,
81    /// node_id → set of incoming edge_ids
82    pub incoming: HashMap<String, HashSet<String>>,
83    /// node_type → set of node_ids (type index)
84    pub by_type: HashMap<String, HashSet<String>>,
85    /// The ontology (for validation during materialization)
86    pub ontology: Ontology,
87    /// The ontology this graph started from, before any `ExtendOntology` was
88    /// replayed. `rebuild` resets to it so that replaying the log is
89    /// idempotent: without this, an extension already folded into `ontology`
90    /// is re-merged on rebuild, fails as a duplicate, and quarantines the
91    /// store's own schema entry. It also makes the effective ontology a
92    /// function of (base, oplog) rather than of how many rebuilds have run,
93    /// which is what I-06's proof assumes when it says both peers replay
94    /// "against the same evolved ontology".
95    base_ontology: Ontology,
96    /// R-02: entries that failed ontology validation during apply().
97    /// These entries exist in the oplog (for CRDT convergence) but are
98    /// invisible in the materialized graph. Grow-only within a single
99    /// materialization pass. Cleared and rebuilt on `rebuild()` — this
100    /// allows previously-quarantined entries to be re-evaluated when the
101    /// ontology evolves (e.g., after ExtendOntology arrives via sync).
102    pub quarantined: HashMap<Hash, QuarantineRecord>,
103    /// S9: edges that arrived before their endpoints. Held (and reported
104    /// quarantined) until both endpoints materialize, then re-evaluated with
105    /// full endpoint-type validation. Previously such edges were admitted
106    /// unvalidated on the promise of a rebuild that only fires when the merge
107    /// batch happens to carry a schema change.
108    pending_edges: HashMap<Hash, Entry>,
109}
110
111impl MaterializedGraph {
112    /// Create an empty materialized graph with the given ontology.
113    pub fn new(ontology: Ontology) -> Self {
114        Self {
115            nodes: HashMap::new(),
116            edges: HashMap::new(),
117            outgoing: HashMap::new(),
118            incoming: HashMap::new(),
119            by_type: HashMap::new(),
120            base_ontology: ontology.clone(),
121            ontology,
122            quarantined: HashMap::new(),
123            pending_edges: HashMap::new(),
124        }
125    }
126
127    /// Apply a single entry to the graph (incremental materialization).
128    ///
129    /// R-02: Validates AddNode/AddEdge payloads against the ontology.
130    /// Invalid entries are quarantined (added to `self.quarantined`) and
131    /// skipped for materialization. They remain in the oplog for CRDT
132    /// convergence — quarantine is a graph-layer concern, not an oplog concern.
133    pub fn apply(&mut self, entry: &Entry) {
134        self.apply_entry(entry, ValidationMode::Full)
135    }
136
137    fn apply_entry(&mut self, entry: &Entry, mode: ValidationMode) {
138        self.apply_inner(entry, mode, true)
139    }
140
141    /// Core of `apply`. `mode` is `SkipRequired` only for a checkpoint's inner
142    /// ops, because `build_checkpoint_ops` emits `AddNode` with an empty
143    /// property map by design (EXP-02) with the values following as separate
144    /// `UpdateProperty` ops — enforcing required-presence there would
145    /// quarantine every such entity (Bug 14b). S2: the bypass is limited to
146    /// that one rule. Declared types, constraints and edge endpoints are
147    /// enforced on checkpoint inner ops like anything else.
148    ///
149    /// The checkpoint's own trustworthiness is established at the merge
150    /// boundary, not here: a foreign checkpoint may only enter a store whose
151    /// oplog is still genesis-only (H1). Replay has no valid baseline of its
152    /// own to judge against, since the ontology it carries is precisely what
153    /// the rest of the replay is relative to.
154    /// `adopt_checkpoint_ontology` is false only during `rebuild`'s second
155    /// pass, where the effective ontology has already been folded and a
156    /// checkpoint's inner `DefineOntology` must not clobber the extensions
157    /// that causally follow it.
158    fn apply_inner(
159        &mut self,
160        entry: &Entry,
161        mode: ValidationMode,
162        adopt_checkpoint_ontology: bool,
163    ) {
164        macro_rules! quarantine {
165            ($kind:expr, $reason:expr) => {{
166                let record = QuarantineRecord {
167                    op_kind: $kind.to_string(),
168                    reason: $reason.to_string(),
169                    ontology_hash: hex::encode(self.ontology.content_hash()),
170                };
171                self.quarantined.insert(entry.hash, record);
172            }};
173        }
174        match &entry.payload {
175            GraphOp::Checkpoint { ops, op_clocks, .. } => {
176                // R-08: Replay synthetic ops to restore graph state.
177                // Bug 6 fix: use per-op clocks (preserves LWW metadata).
178                for (i, op) in ops.iter().enumerate() {
179                    // Bug 14 fix: compaction folds every ExtendOntology into the
180                    // checkpoint's inner DefineOntology. Apply it — otherwise
181                    // extension-typed entities fail validation and quarantine on
182                    // replay, and a reopened store materializes without them.
183                    // The oplog is authoritative; a declared ontology only seeds
184                    // new stores.
185                    if let GraphOp::DefineOntology { ontology } = op {
186                        if adopt_checkpoint_ontology {
187                            self.ontology = ontology.clone();
188                        }
189                        continue;
190                    }
191                    let clock = if i < op_clocks.len() {
192                        LamportClock::with_values(&entry.author, op_clocks[i].0, op_clocks[i].1)
193                    } else {
194                        entry.clock.clone() // fallback for old checkpoints without op_clocks
195                    };
196                    let synthetic = Entry::new(op.clone(), vec![], vec![], clock, &entry.author);
197                    self.apply_entry(&synthetic, ValidationMode::SkipRequired);
198                }
199                // H6: the checkpoint itself is now materialized.
200                self.quarantined.remove(&entry.hash);
201            }
202            GraphOp::DefineOntology { .. } => {
203                // Genesis — nothing to materialize.
204            }
205            GraphOp::ExtendOntology { extension } => {
206                if let Err(e) = self.ontology.merge_extension(extension) {
207                    quarantine!("extend_ontology", e);
208                } else {
209                    self.quarantined.remove(&entry.hash);
210                }
211            }
212            GraphOp::AddNode {
213                node_id,
214                node_type,
215                subtype,
216                label,
217                properties,
218            } => {
219                // R-02: validate against ontology, quarantine if invalid
220                if let Err(e) = self.ontology.validate_node_mode(
221                    node_type,
222                    subtype.as_deref(),
223                    properties,
224                    mode,
225                ) {
226                    quarantine!("add_node", e);
227                    return;
228                }
229                // H6: quarantine must be a function of the oplog, not of sync
230                // history. An entry that validates now is not quarantined,
231                // even if a previous pass rejected it.
232                self.quarantined.remove(&entry.hash);
233                self.apply_add_node(
234                    node_id,
235                    node_type,
236                    subtype.as_deref(),
237                    label,
238                    properties,
239                    &entry.clock,
240                );
241                // S9: an endpoint just arrived — edges held for it can proceed.
242                self.retry_pending_edges();
243            }
244            GraphOp::AddEdge {
245                edge_id,
246                edge_type,
247                source_id,
248                target_id,
249                properties,
250            } => {
251                // R-02: validate edge type exists.
252                if !self.ontology.edge_types.contains_key(edge_type.as_str()) {
253                    quarantine!(
254                        "add_edge",
255                        crate::ontology::ValidationError::UnknownEdgeType(edge_type.clone())
256                    );
257                    return;
258                }
259                // Bug 13 fix: validate source/target type constraints when both nodes
260                // are materialized. S9: when an endpoint is missing the edge is
261                // held pending instead of admitted unvalidated, because the
262                // rebuild the old comment promised only fires when the batch
263                // happens to carry a schema change.
264                match (
265                    self.nodes.get(source_id.as_str()),
266                    self.nodes.get(target_id.as_str()),
267                ) {
268                    (Some(src), Some(tgt)) => {
269                        if let Err(e) = self.ontology.validate_edge_mode(
270                            edge_type,
271                            &src.node_type,
272                            &tgt.node_type,
273                            properties,
274                            mode,
275                        ) {
276                            quarantine!("add_edge", e);
277                            return;
278                        }
279                    }
280                    _ => {
281                        self.pending_edges.insert(entry.hash, entry.clone());
282                        quarantine!(
283                            "add_edge",
284                            format!(
285                                "endpoint not yet materialized (source '{source_id}', \
286                                 target '{target_id}'); held pending until both arrive"
287                            )
288                        );
289                        return;
290                    }
291                }
292                self.quarantined.remove(&entry.hash);
293                self.apply_add_edge(
294                    edge_id,
295                    edge_type,
296                    source_id,
297                    target_id,
298                    properties,
299                    &entry.clock,
300                );
301            }
302            GraphOp::UpdateProperty {
303                entity_id,
304                key,
305                value,
306            } => {
307                // H2: this arm applied the value with no validation at all, so
308                // a peer's UpdateProperty entered the graph regardless of the
309                // local ontology — the v0.1.6 bug reopened on the sync side.
310                // The entity's declared type is the one the graph already
311                // holds; if it is not materialized yet there is nothing to
312                // validate against and the value is accepted, exactly as the
313                // local path does.
314                let verdict = if let Some(node) = self.nodes.get(entity_id.as_str()) {
315                    self.ontology.validate_property_update(
316                        &node.node_type,
317                        node.subtype.as_deref(),
318                        key,
319                        value,
320                    )
321                } else if let Some(edge) = self.edges.get(entity_id.as_str()) {
322                    self.ontology
323                        .validate_edge_property_update(&edge.edge_type, key, value)
324                } else {
325                    Ok(())
326                };
327                if let Err(e) = verdict {
328                    quarantine!("update_property", e);
329                    return;
330                }
331                self.quarantined.remove(&entry.hash);
332                self.apply_update_property(entity_id, key, value, &entry.clock);
333            }
334            GraphOp::RemoveNode { node_id } => {
335                self.quarantined.remove(&entry.hash);
336                self.apply_remove_node(node_id, &entry.clock);
337            }
338            GraphOp::RemoveEdge { edge_id } => {
339                self.quarantined.remove(&entry.hash);
340                self.apply_remove_edge(edge_id, &entry.clock);
341            }
342            GraphOp::DefineLens { .. } => {
343                // Reserved. No materialization — lenses are metadata, not graph state.
344            }
345        }
346    }
347
348    /// Apply a sequence of entries (full rematerialization from op log).
349    pub fn apply_all(&mut self, entries: &[&Entry]) {
350        for entry in entries {
351            self.apply(entry);
352        }
353    }
354
355    /// Rebuild from scratch: clear everything and replay all entries.
356    pub fn rebuild(&mut self, entries: &[&Entry]) {
357        self.rebuild_inner(entries, None)
358    }
359
360    /// Rebuild as if `extension` had already been appended to the log.
361    ///
362    /// Backs `preview_extension`: the only honest way to answer "what would
363    /// this extension cost me" is to run the materialization that would
364    /// follow it. An extension merges into the folded ontology and nothing
365    /// else, so appending it to pass 1 is exactly equivalent to writing the
366    /// entry and replaying, without writing anything.
367    pub fn rebuild_with_pending_extension(
368        &mut self,
369        entries: &[&Entry],
370        extension: &OntologyExtension,
371    ) {
372        self.rebuild_inner(entries, Some(extension))
373    }
374
375    fn rebuild_inner(&mut self, entries: &[&Entry], pending: Option<&OntologyExtension>) {
376        self.nodes.clear();
377        self.edges.clear();
378        self.outgoing.clear();
379        self.incoming.clear();
380        self.by_type.clear();
381        self.quarantined.clear();
382        self.pending_edges.clear();
383
384        // Pass 1: fold the effective ontology from the log, starting at the
385        // base so each extension merges exactly once.
386        //
387        // Validity is then judged against the FINAL ontology, not against the
388        // ontology as of each entry's position. That is what makes the
389        // documented un-quarantine promise true in the ordinary case: an
390        // operator receives data whose type is unknown, extends the schema,
391        // and the data becomes visible — even though the extension is
392        // causally later than the data it rescues. Judging each entry against
393        // the schema as of its own position would re-quarantine it on every
394        // replay, forever.
395        self.ontology = self.base_ontology.clone();
396        for entry in entries {
397            match &entry.payload {
398                GraphOp::DefineOntology { .. } => {}
399                GraphOp::ExtendOntology { extension } => {
400                    if let Err(e) = self.ontology.merge_extension(extension) {
401                        self.quarantined.insert(
402                            entry.hash,
403                            QuarantineRecord {
404                                op_kind: "extend_ontology".to_string(),
405                                reason: e.to_string(),
406                                ontology_hash: hex::encode(self.ontology.content_hash()),
407                            },
408                        );
409                    }
410                }
411                GraphOp::Checkpoint { ops, .. } => {
412                    for op in ops {
413                        if let GraphOp::DefineOntology { ontology } = op {
414                            self.ontology = ontology.clone();
415                        }
416                    }
417                }
418                _ => {}
419            }
420        }
421
422        // The previewed extension folds last, where a freshly appended entry
423        // would land.
424        if let Some(extension) = pending {
425            self.ontology
426                .merge_extension(extension)
427                .expect("caller validates the extension before previewing it");
428        }
429
430        // Pass 2: materialize everything else against that ontology.
431        // ExtendOntology entries are already folded and already judged.
432        for entry in entries {
433            if matches!(entry.payload, GraphOp::ExtendOntology { .. }) {
434                continue;
435            }
436            self.apply_inner(entry, ValidationMode::Full, false);
437        }
438    }
439
440    /// S9: re-evaluate edges that were waiting on an endpoint. Called whenever
441    /// a node materializes. Edges whose endpoints are still missing stay
442    /// pending; the rest go through full validation and are admitted or
443    /// quarantined on their merits.
444    fn retry_pending_edges(&mut self) {
445        if self.pending_edges.is_empty() {
446            return;
447        }
448        let ready: Vec<Entry> = self
449            .pending_edges
450            .values()
451            .filter(|entry| match &entry.payload {
452                GraphOp::AddEdge {
453                    source_id,
454                    target_id,
455                    ..
456                } => {
457                    self.nodes.contains_key(source_id.as_str())
458                        && self.nodes.contains_key(target_id.as_str())
459                }
460                _ => false,
461            })
462            .cloned()
463            .collect();
464        for entry in ready {
465            self.pending_edges.remove(&entry.hash);
466            self.apply_entry(&entry, ValidationMode::Full);
467        }
468    }
469
470    // -- Queries --
471
472    /// Get a node by ID (returns None if not found or tombstoned).
473    pub fn get_node(&self, node_id: &str) -> Option<&Node> {
474        self.nodes.get(node_id).filter(|n| !n.tombstoned)
475    }
476
477    /// Get an edge by ID (returns None if not found or tombstoned).
478    pub fn get_edge(&self, edge_id: &str) -> Option<&Edge> {
479        self.edges.get(edge_id).filter(|e| !e.tombstoned)
480    }
481
482    /// Query all live nodes of a given type, including descendants (RDFS rdfs9).
483    /// If "entity" has children "server" and "project", querying "entity" returns all three.
484    pub fn nodes_by_type(&self, node_type: &str) -> Vec<&Node> {
485        let mut types = vec![node_type.to_string()];
486        types.extend(
487            self.ontology
488                .descendants(node_type)
489                .into_iter()
490                .map(|s| s.to_string()),
491        );
492        types
493            .iter()
494            .flat_map(|t| self.by_type.get(t.as_str()))
495            .flatten()
496            .filter_map(|id| self.get_node(id))
497            .collect()
498    }
499
500    /// Query all live nodes of a given subtype.
501    pub fn nodes_by_subtype(&self, subtype: &str) -> Vec<&Node> {
502        self.nodes
503            .values()
504            .filter(|n| !n.tombstoned && n.subtype.as_deref() == Some(subtype))
505            .collect()
506    }
507
508    /// Query nodes by a property value.
509    pub fn nodes_by_property(&self, key: &str, value: &Value) -> Vec<&Node> {
510        self.nodes
511            .values()
512            .filter(|n| !n.tombstoned && n.properties.get(key) == Some(value))
513            .collect()
514    }
515
516    /// Get outgoing edges for a node (only live edges with live endpoints).
517    pub fn outgoing_edges(&self, node_id: &str) -> Vec<&Edge> {
518        match self.outgoing.get(node_id) {
519            Some(edge_ids) => edge_ids
520                .iter()
521                .filter_map(|eid| self.get_edge(eid))
522                .filter(|e| self.is_node_live(&e.target_id))
523                .collect(),
524            None => vec![],
525        }
526    }
527
528    /// Get incoming edges for a node (only live edges with live endpoints).
529    pub fn incoming_edges(&self, node_id: &str) -> Vec<&Edge> {
530        match self.incoming.get(node_id) {
531            Some(edge_ids) => edge_ids
532                .iter()
533                .filter_map(|eid| self.get_edge(eid))
534                .filter(|e| self.is_node_live(&e.source_id))
535                .collect(),
536            None => vec![],
537        }
538    }
539
540    /// Approximate heap memory used by the materialized graph (bytes).
541    /// Uses fixed overhead estimates per node/edge. Does not account for heap
542    /// allocations behind String/Vec in property values or allocator fragmentation.
543    /// Actual memory may be 2-3x higher for string-heavy graphs.
544    pub fn estimated_memory_bytes(&self) -> usize {
545        let mut total = 0;
546        // Nodes: id string + type string + label + properties + clocks + overhead
547        for node in self.nodes.values() {
548            total += node.node_id.len() + node.node_type.len() + node.label.len();
549            total += node.subtype.as_ref().map_or(0, |s| s.len());
550            // Properties: key + estimated value size + clock per property
551            for (k, v) in &node.properties {
552                total += k.len() + std::mem::size_of_val(v) + 48; // key + value + clock overhead
553            }
554            total += 128; // fixed struct overhead (clocks, bools, HashMap shells)
555        }
556        // Edges: similar structure
557        for edge in self.edges.values() {
558            total += edge.edge_id.len() + edge.edge_type.len();
559            total += edge.source_id.len() + edge.target_id.len();
560            for (k, v) in &edge.properties {
561                total += k.len() + std::mem::size_of_val(v) + 48;
562            }
563            total += 128;
564        }
565        // Adjacency indexes: outgoing + incoming (id strings + HashSet overhead)
566        for (k, set) in &self.outgoing {
567            total += k.len() + set.len() * 32;
568        }
569        for (k, set) in &self.incoming {
570            total += k.len() + set.len() * 32;
571        }
572        // Type index
573        for (k, set) in &self.by_type {
574            total += k.len() + set.len() * 32;
575        }
576        // Quarantine set
577        total += self.quarantined.len() * 48;
578        total
579    }
580
581    /// All live nodes.
582    pub fn all_nodes(&self) -> Vec<&Node> {
583        self.nodes.values().filter(|n| !n.tombstoned).collect()
584    }
585
586    /// All live edges (with live endpoints).
587    pub fn all_edges(&self) -> Vec<&Edge> {
588        self.edges
589            .values()
590            .filter(|e| {
591                !e.tombstoned && self.is_node_live(&e.source_id) && self.is_node_live(&e.target_id)
592            })
593            .collect()
594    }
595
596    /// Neighbors of a node (connected via outgoing edges).
597    pub fn neighbors(&self, node_id: &str) -> Vec<&str> {
598        self.outgoing_edges(node_id)
599            .iter()
600            .map(|e| e.target_id.as_str())
601            .collect()
602    }
603
604    /// Reverse neighbors (connected via incoming edges).
605    pub fn reverse_neighbors(&self, node_id: &str) -> Vec<&str> {
606        self.incoming_edges(node_id)
607            .iter()
608            .map(|e| e.source_id.as_str())
609            .collect()
610    }
611
612    // -- CRDT application helpers --
613
614    fn apply_add_node(
615        &mut self,
616        node_id: &str,
617        node_type: &str,
618        subtype: Option<&str>,
619        label: &str,
620        properties: &BTreeMap<String, Value>,
621        clock: &LamportClock,
622    ) {
623        if let Some(existing) = self.nodes.get_mut(node_id) {
624            // Add-wins: always resurrect from tombstone.
625            existing.tombstoned = false;
626            // Track the latest add clock for add-wins semantics.
627            if clock_wins(clock, &existing.last_add_clock) {
628                existing.last_add_clock = clock.clone();
629            }
630            // LWW merge for label, subtype, and properties.
631            if clock_wins(clock, &existing.last_clock) {
632                existing.label = label.to_string();
633                existing.subtype = subtype.map(|s| s.to_string());
634                existing.last_clock = clock.clone();
635            }
636            merge_properties_lww(
637                &mut existing.properties,
638                &mut existing.property_clocks,
639                properties,
640                clock,
641            );
642        } else {
643            let property_clocks: HashMap<String, LamportClock> = properties
644                .keys()
645                .map(|k| (k.clone(), clock.clone()))
646                .collect();
647            let node = Node {
648                node_id: node_id.to_string(),
649                node_type: node_type.to_string(),
650                subtype: subtype.map(|s| s.to_string()),
651                label: label.to_string(),
652                properties: properties.clone(),
653                property_clocks,
654                last_clock: clock.clone(),
655                last_add_clock: clock.clone(),
656                tombstoned: false,
657            };
658            self.by_type
659                .entry(node_type.to_string())
660                .or_default()
661                .insert(node_id.to_string());
662            self.nodes.insert(node_id.to_string(), node);
663        }
664    }
665
666    fn apply_add_edge(
667        &mut self,
668        edge_id: &str,
669        edge_type: &str,
670        source_id: &str,
671        target_id: &str,
672        properties: &BTreeMap<String, Value>,
673        clock: &LamportClock,
674    ) {
675        if let Some(existing) = self.edges.get_mut(edge_id) {
676            // Add-wins: always resurrect if tombstoned.
677            existing.tombstoned = false;
678            if clock_wins(clock, &existing.last_add_clock) {
679                existing.last_add_clock = clock.clone();
680            }
681            if clock_wins(clock, &existing.last_clock) {
682                existing.last_clock = clock.clone();
683            }
684            merge_properties_lww(
685                &mut existing.properties,
686                &mut existing.property_clocks,
687                properties,
688                clock,
689            );
690        } else {
691            let property_clocks: HashMap<String, LamportClock> = properties
692                .keys()
693                .map(|k| (k.clone(), clock.clone()))
694                .collect();
695            let edge = Edge {
696                edge_id: edge_id.to_string(),
697                edge_type: edge_type.to_string(),
698                source_id: source_id.to_string(),
699                target_id: target_id.to_string(),
700                properties: properties.clone(),
701                property_clocks,
702                last_clock: clock.clone(),
703                last_add_clock: clock.clone(),
704                tombstoned: false,
705            };
706            self.outgoing
707                .entry(source_id.to_string())
708                .or_default()
709                .insert(edge_id.to_string());
710            self.incoming
711                .entry(target_id.to_string())
712                .or_default()
713                .insert(edge_id.to_string());
714            self.edges.insert(edge_id.to_string(), edge);
715        }
716    }
717
718    fn apply_update_property(
719        &mut self,
720        entity_id: &str,
721        key: &str,
722        value: &Value,
723        clock: &LamportClock,
724    ) {
725        // Try node first, then edge. Per-property LWW: each key competes
726        // only with other writes to the same key, not the entire entity.
727        if let Some(node) = self.nodes.get_mut(entity_id) {
728            let dominated = node
729                .property_clocks
730                .get(key)
731                .map(|c| clock_wins(clock, c))
732                .unwrap_or(true);
733            if dominated {
734                node.properties.insert(key.to_string(), value.clone());
735                node.property_clocks.insert(key.to_string(), clock.clone());
736            }
737            // Update entity-level clock for add-wins tracking.
738            if clock_wins(clock, &node.last_clock) {
739                node.last_clock = clock.clone();
740            }
741        } else if let Some(edge) = self.edges.get_mut(entity_id) {
742            let dominated = edge
743                .property_clocks
744                .get(key)
745                .map(|c| clock_wins(clock, c))
746                .unwrap_or(true);
747            if dominated {
748                edge.properties.insert(key.to_string(), value.clone());
749                edge.property_clocks.insert(key.to_string(), clock.clone());
750            }
751            if clock_wins(clock, &edge.last_clock) {
752                edge.last_clock = clock.clone();
753            }
754        }
755        // If entity not found, silently ignore (may arrive out of order in sync).
756    }
757
758    fn apply_remove_node(&mut self, node_id: &str, clock: &LamportClock) {
759        if let Some(node) = self.nodes.get_mut(node_id) {
760            // Add-wins: only tombstone if the remove clock is strictly greater
761            // than the last add clock. If a concurrent (or later) add exists,
762            // the node stays alive.
763            if clock_wins(clock, &node.last_add_clock) {
764                node.tombstoned = true;
765                node.last_clock = clock.clone();
766            }
767        }
768        // Tombstoning a node doesn't physically remove edges — they just become
769        // invisible via is_node_live() checks in queries.
770    }
771
772    fn apply_remove_edge(&mut self, edge_id: &str, clock: &LamportClock) {
773        if let Some(edge) = self.edges.get_mut(edge_id) {
774            // Add-wins: only tombstone if remove clock > last add clock.
775            if clock_wins(clock, &edge.last_add_clock) {
776                edge.tombstoned = true;
777                edge.last_clock = clock.clone();
778            }
779        }
780    }
781
782    fn is_node_live(&self, node_id: &str) -> bool {
783        self.nodes
784            .get(node_id)
785            .map(|n| !n.tombstoned)
786            .unwrap_or(false)
787    }
788}
789
790/// Per-property LWW merge: each property from `new_props` competes with
791/// existing properties. Higher clock wins per key.
792fn merge_properties_lww(
793    existing_props: &mut BTreeMap<String, Value>,
794    existing_clocks: &mut HashMap<String, LamportClock>,
795    new_props: &BTreeMap<String, Value>,
796    clock: &LamportClock,
797) {
798    for (k, v) in new_props {
799        let dominated = existing_clocks
800            .get(k)
801            .map(|c| clock_wins(clock, c))
802            .unwrap_or(true);
803        if dominated {
804            existing_props.insert(k.clone(), v.clone());
805            existing_clocks.insert(k.clone(), clock.clone());
806        }
807    }
808}
809
810/// LWW comparison: returns true if `new_clock` wins over `existing_clock`.
811/// Uses HybridClock total ordering: (physical_ms, logical, id).
812fn clock_wins(new_clock: &LamportClock, existing_clock: &LamportClock) -> bool {
813    new_clock.cmp_order(existing_clock) == std::cmp::Ordering::Greater
814}
815
816#[cfg(test)]
817mod tests {
818    use super::*;
819    use crate::entry::Entry;
820    use crate::ontology::{EdgeTypeDef, NodeTypeDef};
821
822    fn test_ontology() -> Ontology {
823        Ontology {
824            node_types: BTreeMap::from([
825                (
826                    "entity".into(),
827                    NodeTypeDef {
828                        description: None,
829                        properties: BTreeMap::new(),
830                        subtypes: None,
831                        parent_type: None,
832                    },
833                ),
834                (
835                    "signal".into(),
836                    NodeTypeDef {
837                        description: None,
838                        properties: BTreeMap::new(),
839                        subtypes: None,
840                        parent_type: None,
841                    },
842                ),
843            ]),
844            edge_types: BTreeMap::from([
845                (
846                    "RUNS_ON".into(),
847                    EdgeTypeDef {
848                        description: None,
849                        source_types: vec!["entity".into()],
850                        target_types: vec!["entity".into()],
851                        properties: BTreeMap::new(),
852                    },
853                ),
854                (
855                    "OBSERVES".into(),
856                    EdgeTypeDef {
857                        description: None,
858                        source_types: vec!["signal".into()],
859                        target_types: vec!["entity".into()],
860                        properties: BTreeMap::new(),
861                    },
862                ),
863            ]),
864        }
865    }
866
867    fn make_entry(op: GraphOp, clock_time: u64, author: &str) -> Entry {
868        Entry::new(
869            op,
870            vec![],
871            vec![],
872            LamportClock::with_values(author, clock_time, 0),
873            author,
874        )
875    }
876
877    // -- test_graph.rs spec from docs/silk.md --
878
879    #[test]
880    fn add_node_appears_in_query() {
881        let mut g = MaterializedGraph::new(test_ontology());
882        let entry = make_entry(
883            GraphOp::AddNode {
884                node_id: "server-1".into(),
885                node_type: "entity".into(),
886                label: "Server 1".into(),
887                properties: BTreeMap::from([("ip".into(), Value::String("10.0.0.1".into()))]),
888                subtype: None,
889            },
890            1,
891            "inst-a",
892        );
893        g.apply(&entry);
894
895        let node = g.get_node("server-1").unwrap();
896        assert_eq!(node.node_type, "entity");
897        assert_eq!(node.label, "Server 1");
898        assert_eq!(
899            node.properties.get("ip"),
900            Some(&Value::String("10.0.0.1".into()))
901        );
902    }
903
904    #[test]
905    fn add_edge_creates_adjacency() {
906        let mut g = MaterializedGraph::new(test_ontology());
907        g.apply(&make_entry(
908            GraphOp::AddNode {
909                node_id: "svc".into(),
910                node_type: "entity".into(),
911                label: "svc".into(),
912                properties: BTreeMap::new(),
913                subtype: None,
914            },
915            1,
916            "inst-a",
917        ));
918        g.apply(&make_entry(
919            GraphOp::AddNode {
920                node_id: "srv".into(),
921                node_type: "entity".into(),
922                label: "srv".into(),
923                properties: BTreeMap::new(),
924                subtype: None,
925            },
926            2,
927            "inst-a",
928        ));
929        g.apply(&make_entry(
930            GraphOp::AddEdge {
931                edge_id: "e1".into(),
932                edge_type: "RUNS_ON".into(),
933                source_id: "svc".into(),
934                target_id: "srv".into(),
935                properties: BTreeMap::new(),
936            },
937            3,
938            "inst-a",
939        ));
940
941        // Both endpoints know about the edge.
942        let out = g.outgoing_edges("svc");
943        assert_eq!(out.len(), 1);
944        assert_eq!(out[0].target_id, "srv");
945
946        let inc = g.incoming_edges("srv");
947        assert_eq!(inc.len(), 1);
948        assert_eq!(inc[0].source_id, "svc");
949
950        assert_eq!(g.neighbors("svc"), vec!["srv"]);
951    }
952
953    #[test]
954    fn update_property_reflected() {
955        let mut g = MaterializedGraph::new(test_ontology());
956        g.apply(&make_entry(
957            GraphOp::AddNode {
958                node_id: "s1".into(),
959                node_type: "entity".into(),
960                label: "s1".into(),
961                properties: BTreeMap::new(),
962                subtype: None,
963            },
964            1,
965            "inst-a",
966        ));
967        g.apply(&make_entry(
968            GraphOp::UpdateProperty {
969                entity_id: "s1".into(),
970                key: "cpu".into(),
971                value: Value::Float(85.5),
972            },
973            2,
974            "inst-a",
975        ));
976
977        let node = g.get_node("s1").unwrap();
978        assert_eq!(node.properties.get("cpu"), Some(&Value::Float(85.5)));
979    }
980
981    #[test]
982    fn remove_node_cascades_edges() {
983        let mut g = MaterializedGraph::new(test_ontology());
984        g.apply(&make_entry(
985            GraphOp::AddNode {
986                node_id: "a".into(),
987                node_type: "entity".into(),
988                label: "a".into(),
989                properties: BTreeMap::new(),
990                subtype: None,
991            },
992            1,
993            "inst-a",
994        ));
995        g.apply(&make_entry(
996            GraphOp::AddNode {
997                node_id: "b".into(),
998                node_type: "entity".into(),
999                label: "b".into(),
1000                properties: BTreeMap::new(),
1001                subtype: None,
1002            },
1003            2,
1004            "inst-a",
1005        ));
1006        g.apply(&make_entry(
1007            GraphOp::AddEdge {
1008                edge_id: "e1".into(),
1009                edge_type: "RUNS_ON".into(),
1010                source_id: "a".into(),
1011                target_id: "b".into(),
1012                properties: BTreeMap::new(),
1013            },
1014            3,
1015            "inst-a",
1016        ));
1017        assert_eq!(g.all_edges().len(), 1);
1018
1019        // Remove node 'b' — edge becomes invisible (dangling target).
1020        g.apply(&make_entry(
1021            GraphOp::RemoveNode {
1022                node_id: "b".into(),
1023            },
1024            4,
1025            "inst-a",
1026        ));
1027        assert!(g.get_node("b").is_none());
1028        // Edge still exists but not returned by all_edges (target tombstoned).
1029        assert_eq!(g.all_edges().len(), 0);
1030        // Outgoing from 'a' also filters out dangling edges.
1031        assert_eq!(g.outgoing_edges("a").len(), 0);
1032    }
1033
1034    #[test]
1035    fn remove_edge_preserves_nodes() {
1036        let mut g = MaterializedGraph::new(test_ontology());
1037        g.apply(&make_entry(
1038            GraphOp::AddNode {
1039                node_id: "a".into(),
1040                node_type: "entity".into(),
1041                label: "a".into(),
1042                properties: BTreeMap::new(),
1043                subtype: None,
1044            },
1045            1,
1046            "inst-a",
1047        ));
1048        g.apply(&make_entry(
1049            GraphOp::AddNode {
1050                node_id: "b".into(),
1051                node_type: "entity".into(),
1052                label: "b".into(),
1053                properties: BTreeMap::new(),
1054                subtype: None,
1055            },
1056            2,
1057            "inst-a",
1058        ));
1059        g.apply(&make_entry(
1060            GraphOp::AddEdge {
1061                edge_id: "e1".into(),
1062                edge_type: "RUNS_ON".into(),
1063                source_id: "a".into(),
1064                target_id: "b".into(),
1065                properties: BTreeMap::new(),
1066            },
1067            3,
1068            "inst-a",
1069        ));
1070        g.apply(&make_entry(
1071            GraphOp::RemoveEdge {
1072                edge_id: "e1".into(),
1073            },
1074            4,
1075            "inst-a",
1076        ));
1077
1078        // Nodes still exist.
1079        assert!(g.get_node("a").is_some());
1080        assert!(g.get_node("b").is_some());
1081        // Edge is gone.
1082        assert!(g.get_edge("e1").is_none());
1083        assert_eq!(g.all_edges().len(), 0);
1084    }
1085
1086    #[test]
1087    fn query_by_type_filters() {
1088        let mut g = MaterializedGraph::new(test_ontology());
1089        g.apply(&make_entry(
1090            GraphOp::AddNode {
1091                node_id: "s1".into(),
1092                node_type: "entity".into(),
1093                label: "s1".into(),
1094                properties: BTreeMap::new(),
1095                subtype: None,
1096            },
1097            1,
1098            "inst-a",
1099        ));
1100        g.apply(&make_entry(
1101            GraphOp::AddNode {
1102                node_id: "s2".into(),
1103                node_type: "entity".into(),
1104                label: "s2".into(),
1105                properties: BTreeMap::new(),
1106                subtype: None,
1107            },
1108            2,
1109            "inst-a",
1110        ));
1111        g.apply(&make_entry(
1112            GraphOp::AddNode {
1113                node_id: "alert".into(),
1114                node_type: "signal".into(),
1115                label: "alert".into(),
1116                properties: BTreeMap::new(),
1117                subtype: None,
1118            },
1119            3,
1120            "inst-a",
1121        ));
1122
1123        let entities = g.nodes_by_type("entity");
1124        assert_eq!(entities.len(), 2);
1125        let signals = g.nodes_by_type("signal");
1126        assert_eq!(signals.len(), 1);
1127        assert_eq!(signals[0].node_id, "alert");
1128    }
1129
1130    #[test]
1131    fn query_by_property_filters() {
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::from([("status".into(), Value::String("alive".into()))]),
1139                subtype: None,
1140            },
1141            1,
1142            "inst-a",
1143        ));
1144        g.apply(&make_entry(
1145            GraphOp::AddNode {
1146                node_id: "s2".into(),
1147                node_type: "entity".into(),
1148                label: "s2".into(),
1149                properties: BTreeMap::from([("status".into(), Value::String("dead".into()))]),
1150                subtype: None,
1151            },
1152            2,
1153            "inst-a",
1154        ));
1155
1156        let alive = g.nodes_by_property("status", &Value::String("alive".into()));
1157        assert_eq!(alive.len(), 1);
1158        assert_eq!(alive[0].node_id, "s1");
1159    }
1160
1161    #[test]
1162    fn materialization_from_empty() {
1163        // Build graph incrementally.
1164        let mut g1 = MaterializedGraph::new(test_ontology());
1165        let entries = vec![
1166            make_entry(
1167                GraphOp::DefineOntology {
1168                    ontology: test_ontology(),
1169                },
1170                0,
1171                "inst-a",
1172            ),
1173            make_entry(
1174                GraphOp::AddNode {
1175                    node_id: "a".into(),
1176                    node_type: "entity".into(),
1177                    label: "a".into(),
1178                    properties: BTreeMap::new(),
1179                    subtype: None,
1180                },
1181                1,
1182                "inst-a",
1183            ),
1184            make_entry(
1185                GraphOp::AddNode {
1186                    node_id: "b".into(),
1187                    node_type: "entity".into(),
1188                    label: "b".into(),
1189                    properties: BTreeMap::new(),
1190                    subtype: None,
1191                },
1192                2,
1193                "inst-a",
1194            ),
1195            make_entry(
1196                GraphOp::AddEdge {
1197                    edge_id: "e1".into(),
1198                    edge_type: "RUNS_ON".into(),
1199                    source_id: "a".into(),
1200                    target_id: "b".into(),
1201                    properties: BTreeMap::new(),
1202                },
1203                3,
1204                "inst-a",
1205            ),
1206        ];
1207        for e in &entries {
1208            g1.apply(e);
1209        }
1210
1211        // Rebuild from scratch.
1212        let mut g2 = MaterializedGraph::new(test_ontology());
1213        let refs: Vec<&Entry> = entries.iter().collect();
1214        g2.rebuild(&refs);
1215
1216        // Same result.
1217        assert_eq!(g1.all_nodes().len(), g2.all_nodes().len());
1218        assert_eq!(g1.all_edges().len(), g2.all_edges().len());
1219        for node in g1.all_nodes() {
1220            let n2 = g2.get_node(&node.node_id).unwrap();
1221            assert_eq!(node.node_type, n2.node_type);
1222            assert_eq!(node.properties, n2.properties);
1223        }
1224    }
1225
1226    #[test]
1227    fn incremental_equals_full() {
1228        let entries = vec![
1229            make_entry(
1230                GraphOp::DefineOntology {
1231                    ontology: test_ontology(),
1232                },
1233                0,
1234                "inst-a",
1235            ),
1236            make_entry(
1237                GraphOp::AddNode {
1238                    node_id: "a".into(),
1239                    node_type: "entity".into(),
1240                    label: "a".into(),
1241                    properties: BTreeMap::from([("x".into(), Value::Int(1))]),
1242                    subtype: None,
1243                },
1244                1,
1245                "inst-a",
1246            ),
1247            make_entry(
1248                GraphOp::UpdateProperty {
1249                    entity_id: "a".into(),
1250                    key: "x".into(),
1251                    value: Value::Int(2),
1252                },
1253                2,
1254                "inst-a",
1255            ),
1256            make_entry(
1257                GraphOp::AddNode {
1258                    node_id: "b".into(),
1259                    node_type: "entity".into(),
1260                    label: "b".into(),
1261                    properties: BTreeMap::new(),
1262                    subtype: None,
1263                },
1264                3,
1265                "inst-a",
1266            ),
1267            make_entry(
1268                GraphOp::AddEdge {
1269                    edge_id: "e1".into(),
1270                    edge_type: "RUNS_ON".into(),
1271                    source_id: "a".into(),
1272                    target_id: "b".into(),
1273                    properties: BTreeMap::new(),
1274                },
1275                4,
1276                "inst-a",
1277            ),
1278            make_entry(
1279                GraphOp::RemoveEdge {
1280                    edge_id: "e1".into(),
1281                },
1282                5,
1283                "inst-a",
1284            ),
1285        ];
1286
1287        // Incremental.
1288        let mut g_inc = MaterializedGraph::new(test_ontology());
1289        for e in &entries {
1290            g_inc.apply(e);
1291        }
1292
1293        // Full replay.
1294        let mut g_full = MaterializedGraph::new(test_ontology());
1295        let refs: Vec<&Entry> = entries.iter().collect();
1296        g_full.rebuild(&refs);
1297
1298        // Property should be 2 (updated).
1299        assert_eq!(
1300            g_inc.get_node("a").unwrap().properties.get("x"),
1301            Some(&Value::Int(2))
1302        );
1303        assert_eq!(
1304            g_full.get_node("a").unwrap().properties.get("x"),
1305            Some(&Value::Int(2))
1306        );
1307        // Edge should be removed.
1308        assert_eq!(g_inc.all_edges().len(), 0);
1309        assert_eq!(g_full.all_edges().len(), 0);
1310    }
1311
1312    #[test]
1313    fn lww_concurrent_property_update() {
1314        // Two instances update the same property — higher clock wins.
1315        let mut g = MaterializedGraph::new(test_ontology());
1316        g.apply(&make_entry(
1317            GraphOp::AddNode {
1318                node_id: "s1".into(),
1319                node_type: "entity".into(),
1320                label: "s1".into(),
1321                properties: BTreeMap::new(),
1322                subtype: None,
1323            },
1324            1,
1325            "inst-a",
1326        ));
1327        // inst-a sets status=alive at time 2
1328        g.apply(&make_entry(
1329            GraphOp::UpdateProperty {
1330                entity_id: "s1".into(),
1331                key: "status".into(),
1332                value: Value::String("alive".into()),
1333            },
1334            2,
1335            "inst-a",
1336        ));
1337        // inst-b sets status=dead at time 3 — wins (higher clock)
1338        g.apply(&make_entry(
1339            GraphOp::UpdateProperty {
1340                entity_id: "s1".into(),
1341                key: "status".into(),
1342                value: Value::String("dead".into()),
1343            },
1344            3,
1345            "inst-b",
1346        ));
1347        assert_eq!(
1348            g.get_node("s1").unwrap().properties.get("status"),
1349            Some(&Value::String("dead".into()))
1350        );
1351    }
1352
1353    #[test]
1354    fn lww_tiebreak_by_instance_id() {
1355        // Same clock time — higher instance ID wins.
1356        let mut g = MaterializedGraph::new(test_ontology());
1357        g.apply(&make_entry(
1358            GraphOp::AddNode {
1359                node_id: "s1".into(),
1360                node_type: "entity".into(),
1361                label: "s1".into(),
1362                properties: BTreeMap::new(),
1363                subtype: None,
1364            },
1365            1,
1366            "inst-a",
1367        ));
1368        // Both at physical_ms=5, logical=0. Lower id wins → "inst-a" wins.
1369        g.apply(&make_entry(
1370            GraphOp::UpdateProperty {
1371                entity_id: "s1".into(),
1372                key: "x".into(),
1373                value: Value::Int(1),
1374            },
1375            5,
1376            "inst-a",
1377        ));
1378        g.apply(&make_entry(
1379            GraphOp::UpdateProperty {
1380                entity_id: "s1".into(),
1381                key: "x".into(),
1382                value: Value::Int(2),
1383            },
1384            5,
1385            "inst-b",
1386        ));
1387        // inst-a has lower id → wins the tiebreak → value stays Int(1).
1388        assert_eq!(
1389            g.get_node("s1").unwrap().properties.get("x"),
1390            Some(&Value::Int(1))
1391        );
1392    }
1393
1394    #[test]
1395    fn lww_per_property_concurrent_different_keys() {
1396        // Two instances concurrently update DIFFERENT properties at the same
1397        // clock time. Both updates must be accepted — they don't conflict.
1398        // This requires per-property LWW, not node-level LWW.
1399        let mut g = MaterializedGraph::new(test_ontology());
1400        g.apply(&make_entry(
1401            GraphOp::AddNode {
1402                node_id: "s1".into(),
1403                node_type: "entity".into(),
1404                label: "s1".into(),
1405                properties: BTreeMap::from([
1406                    ("x".into(), Value::Int(0)),
1407                    ("y".into(), Value::Int(0)),
1408                ]),
1409                subtype: None,
1410            },
1411            1,
1412            "inst-a",
1413        ));
1414        // inst-a updates "x" at time 3
1415        g.apply(&make_entry(
1416            GraphOp::UpdateProperty {
1417                entity_id: "s1".into(),
1418                key: "x".into(),
1419                value: Value::Int(42),
1420            },
1421            3,
1422            "inst-a",
1423        ));
1424        // inst-b updates "y" at time 3 (concurrent, different property)
1425        g.apply(&make_entry(
1426            GraphOp::UpdateProperty {
1427                entity_id: "s1".into(),
1428                key: "y".into(),
1429                value: Value::Int(99),
1430            },
1431            3,
1432            "inst-b",
1433        ));
1434
1435        let node = g.get_node("s1").unwrap();
1436        // Both updates must be applied — no conflict.
1437        assert_eq!(
1438            node.properties.get("x"),
1439            Some(&Value::Int(42)),
1440            "update to 'x' must not be rejected by concurrent update to 'y'"
1441        );
1442        assert_eq!(
1443            node.properties.get("y"),
1444            Some(&Value::Int(99)),
1445            "update to 'y' must not be rejected by concurrent update to 'x'"
1446        );
1447    }
1448
1449    #[test]
1450    fn lww_per_property_order_independent() {
1451        // Same scenario but applied in reverse order — result must be identical.
1452        let mut g = MaterializedGraph::new(test_ontology());
1453        g.apply(&make_entry(
1454            GraphOp::AddNode {
1455                node_id: "s1".into(),
1456                node_type: "entity".into(),
1457                label: "s1".into(),
1458                properties: BTreeMap::from([
1459                    ("x".into(), Value::Int(0)),
1460                    ("y".into(), Value::Int(0)),
1461                ]),
1462                subtype: None,
1463            },
1464            1,
1465            "inst-a",
1466        ));
1467        // Apply inst-b first this time
1468        g.apply(&make_entry(
1469            GraphOp::UpdateProperty {
1470                entity_id: "s1".into(),
1471                key: "y".into(),
1472                value: Value::Int(99),
1473            },
1474            3,
1475            "inst-b",
1476        ));
1477        g.apply(&make_entry(
1478            GraphOp::UpdateProperty {
1479                entity_id: "s1".into(),
1480                key: "x".into(),
1481                value: Value::Int(42),
1482            },
1483            3,
1484            "inst-a",
1485        ));
1486
1487        let node = g.get_node("s1").unwrap();
1488        assert_eq!(node.properties.get("x"), Some(&Value::Int(42)));
1489        assert_eq!(node.properties.get("y"), Some(&Value::Int(99)));
1490    }
1491
1492    #[test]
1493    fn add_wins_over_remove() {
1494        // Concurrent add + remove → node should exist (add-wins).
1495        let mut g = MaterializedGraph::new(test_ontology());
1496        g.apply(&make_entry(
1497            GraphOp::AddNode {
1498                node_id: "s1".into(),
1499                node_type: "entity".into(),
1500                label: "s1".into(),
1501                properties: BTreeMap::new(),
1502                subtype: None,
1503            },
1504            1,
1505            "inst-a",
1506        ));
1507        // Remove at time 2.
1508        g.apply(&make_entry(
1509            GraphOp::RemoveNode {
1510                node_id: "s1".into(),
1511            },
1512            2,
1513            "inst-a",
1514        ));
1515        assert!(g.get_node("s1").is_none());
1516
1517        // Re-add at time 3 (add-wins — resurrects).
1518        g.apply(&make_entry(
1519            GraphOp::AddNode {
1520                node_id: "s1".into(),
1521                node_type: "entity".into(),
1522                label: "s1 v2".into(),
1523                properties: BTreeMap::new(),
1524                subtype: None,
1525            },
1526            3,
1527            "inst-b",
1528        ));
1529        let node = g.get_node("s1").unwrap();
1530        assert_eq!(node.label, "s1 v2");
1531        assert!(!node.tombstoned);
1532    }
1533
1534    /// Bug 14: a checkpoint's inner DefineOntology carries the merged ontology
1535    /// (compaction folds ExtendOntology entries into it). Replay must apply it,
1536    /// or entities typed by an extension quarantine and vanish from the graph.
1537    #[test]
1538    fn checkpoint_replay_applies_inner_define_ontology() {
1539        // Base ontology: "entity" only — no "signal".
1540        let mut base = test_ontology();
1541        base.node_types.remove("signal");
1542        base.edge_types.remove("OBSERVES");
1543        let mut g = MaterializedGraph::new(base);
1544
1545        let checkpoint = make_entry(
1546            GraphOp::Checkpoint {
1547                ops: vec![
1548                    GraphOp::DefineOntology {
1549                        ontology: test_ontology(), // merged: has "signal"
1550                    },
1551                    GraphOp::AddNode {
1552                        node_id: "n1".into(),
1553                        node_type: "entity".into(),
1554                        label: "base-typed".into(),
1555                        properties: BTreeMap::new(),
1556                        subtype: None,
1557                    },
1558                    GraphOp::AddNode {
1559                        node_id: "s1".into(),
1560                        node_type: "signal".into(),
1561                        label: "extension-typed".into(),
1562                        properties: BTreeMap::new(),
1563                        subtype: None,
1564                    },
1565                ],
1566                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1567                compacted_at_physical_ms: 1,
1568                compacted_at_logical: 2,
1569            },
1570            1,
1571            "inst-a",
1572        );
1573        g.apply(&checkpoint);
1574
1575        assert!(g.get_node("n1").is_some());
1576        assert!(g.get_node("s1").is_some(), "extension-typed node lost");
1577        assert!(g.ontology.node_types.contains_key("signal"));
1578        assert!(g.quarantined.is_empty());
1579    }
1580
1581    /// Bug 14b: build_checkpoint_ops emits AddNode with empty properties
1582    /// (EXP-02: per-property clocks ride in separate UpdateProperty ops).
1583    /// Replay must trust checkpoint inner ops — validating them fails any
1584    /// required property against the empty map and quarantines the node.
1585    #[test]
1586    fn checkpoint_replay_trusts_inner_ops() {
1587        use crate::ontology::{PropertyDef, ValueType};
1588
1589        let mut ont = test_ontology();
1590        ont.node_types.get_mut("entity").unwrap().properties.insert(
1591            "name".into(),
1592            PropertyDef {
1593                value_type: ValueType::String,
1594                required: true,
1595                description: None,
1596                constraints: None,
1597            },
1598        );
1599        let mut g = MaterializedGraph::new(ont.clone());
1600
1601        let checkpoint = make_entry(
1602            GraphOp::Checkpoint {
1603                ops: vec![
1604                    GraphOp::DefineOntology { ontology: ont },
1605                    GraphOp::AddNode {
1606                        node_id: "n1".into(),
1607                        node_type: "entity".into(),
1608                        label: "req".into(),
1609                        properties: BTreeMap::new(), // empty by design
1610                        subtype: None,
1611                    },
1612                    GraphOp::UpdateProperty {
1613                        entity_id: "n1".into(),
1614                        key: "name".into(),
1615                        value: Value::String("x".into()),
1616                    },
1617                ],
1618                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
1619                compacted_at_physical_ms: 1,
1620                compacted_at_logical: 2,
1621            },
1622            1,
1623            "inst-a",
1624        );
1625        g.apply(&checkpoint);
1626
1627        let node = g.get_node("n1").expect("required-property node lost");
1628        assert_eq!(
1629            node.properties.get("name"),
1630            Some(&Value::String("x".into()))
1631        );
1632        assert!(g.quarantined.is_empty());
1633    }
1634}