fix(rust_brain): preserve HLC on restore and gossip receive

hive/gossip.py

@@ -122,11 +122,29 @@ def receive(self, events: list[dict[str, Any]]) -> int:
         applied = 0
         for ev in events:
             try:
+                key = ev["key"]
+                raw_hlc = ev.get("hlc")
+                if raw_hlc is None:
+                    # Without an HLC we cannot establish causal order for updates.
+                    if self._brain.get(key) is not None:
+                        _log.debug(
+                            "Skipping gossip update for %r: missing hlc on existing key",
+                            key,
+                        )
+                        continue
+                    hlc = None
+                else:
+                    hlc = tuple(raw_hlc)
+                    self._brain.update_hlc(hlc)
+
                 self._brain.remember(
-                    ev["key"],
+                    key,
                     ev["value"],
                     trust=ev.get("trust", 1.0),
                     tags=set(ev.get("tags", [])),
+                    ts_ns=ev.get("ts_ns"),
+                    hlc=hlc,
+                    edges=ev.get("edges"),
                 )
                 applied += 1
             except Exception as exc:

hive/rust_brain/__init__.py

@@ -125,6 +125,19 @@ def _now_ns() -> int:
     return time.time_ns() - HIVE_EPOCH_NS
 
 
+def _parse_hlc(raw: Any, *, ts_ns: int | None = None) -> tuple[int, int, str]:
+    """Normalise an HLC from wire/snapshot form (list or tuple).
+
+    Falls back to a synthetic HLC from ``ts_ns`` for pre-v0.6.0 snapshots.
+    """
+    if raw is not None:
+        wall, logical, node_id = raw
+        return (int(wall), int(logical), str(node_id))
+    if ts_ns is not None:
+        return (ts_ns + HIVE_EPOCH_NS, 0, "legacy")
+    return _hlc.now()
+
+
 # ---------------------------------------------------------------------------
 # Edge / node model
 # ---------------------------------------------------------------------------
@@ -389,6 +402,9 @@ def bulk_write(self, rows: Iterable[Mapping[str, Any]]) -> int:
                 tags=row.get("tags", ()),
                 edges=row.get("edges"),
                 ts_ns=row.get("ts_ns"),
+                hlc=_parse_hlc(row.get("hlc"), ts_ns=row.get("ts_ns"))
+                if row.get("hlc") is not None or row.get("ts_ns") is not None
+                else None,
             )
             n += 1
         return n
@@ -509,10 +525,13 @@ def restore_from_file(self, path: str) -> int:
         self._order.clear()
         self._order_index.clear()
         for node_dict in nodes:
+            ts_ns = node_dict["ts_ns"]
+            node_hlc = _parse_hlc(node_dict.get("hlc"), ts_ns=ts_ns)
             node = MemoryNode(
                 key=node_dict["key"],
                 value=node_dict["value"],
-                ts_ns=node_dict["ts_ns"],
+                ts_ns=ts_ns,
+                hlc=node_hlc,
                 trust=node_dict.get("trust", 1.0),
                 tags=set(node_dict.get("tags", [])),
                 node_id=node_dict.get("id", uuid.uuid4().hex[:12]),
@@ -524,6 +543,7 @@ def restore_from_file(self, path: str) -> int:
             self._nodes[storage_key] = node
             self._order_index[storage_key] = len(self._order)
             self._order.append(storage_key)
+            _hlc.update(node_hlc)
         return len(nodes)
 
     def __repr__(self) -> str:

tests/test_enterprise_backup.py

@@ -9,7 +9,7 @@
 
 import pytest
 
-from hive.rust_brain import RustBrain
+from hive.rust_brain import RustBrain, TimestampRegression
 
 
 def test_snapshot_to_file_and_restore():
@@ -149,6 +149,32 @@ def test_version_mismatch():
         os.unlink(path)
 
 
+def test_restore_preserves_hlc_for_replication():
+    """HLC must survive snapshot round-trip so post-restore writes stay ordered."""
+    brain = RustBrain()
+    brain.remember("k", "v1", hlc=(5000, 10, "nodeA"))
+    orig_hlc = brain.get("k").hlc
+
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".gz") as f:
+        path = f.name
+
+    try:
+        brain.snapshot_to_file(path)
+        brain2 = RustBrain()
+        brain2.restore_from_file(path)
+        assert brain2.get("k").hlc == orig_hlc
+
+        # Causal successor from the original writer should apply after restore.
+        brain2.remember("k", "v2", hlc=(5000, 11, "nodeA"))
+        assert brain2.recall("k") == "v2"
+
+        # Stale replay must still be rejected.
+        with pytest.raises(TimestampRegression):
+            brain2.remember("k", "stale", hlc=(5000, 5, "nodeA"))
+    finally:
+        os.unlink(path)
+
+
 def test_snapshot_with_tenant_isolation():
     brain = RustBrain(tenant_id="org_a", tenant_isolation=True)
     brain.remember("secret", "data")

tests/test_gossip.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import pytest
+
 from hive.gossip import GossipProtocol
 from hive.rust_brain import RustBrain
 
@@ -32,3 +34,45 @@ def test_gossip_start_stop():
     gossip.start()
     gossip.stop()
     # No crash
+
+
+def test_gossip_rejects_stale_hlc_overwrite():
+    brain = RustBrain()
+    gossip = GossipProtocol(brain, peers=[])
+    brain.remember("shared", "FRESH", hlc=(2000, 0, "local"))
+
+    applied = gossip.receive(
+        [{"key": "shared", "value": "STALE", "hlc": [1000, 0, "remote"]}]
+    )
+    assert applied == 0
+    assert brain.recall("shared") == "FRESH"
+
+
+def test_gossip_applies_newer_hlc():
+    brain = RustBrain()
+    gossip = GossipProtocol(brain, peers=[])
+    brain.remember("shared", "OLD", hlc=(1000, 0, "local"))
+
+    applied = gossip.receive(
+        [{"key": "shared", "value": "NEW", "hlc": [2000, 0, "remote"]}]
+    )
+    assert applied == 1
+    assert brain.recall("shared") == "NEW"
+
+
+def test_gossip_skips_missing_hlc_on_existing_key():
+    brain = RustBrain()
+    gossip = GossipProtocol(brain, peers=[])
+    brain.remember("k", "local")
+
+    applied = gossip.receive([{"key": "k", "value": "remote"}])
+    assert applied == 0
+    assert brain.recall("k") == "local"
+
+
+def test_gossip_allows_missing_hlc_for_new_key():
+    brain = RustBrain()
+    gossip = GossipProtocol(brain, peers=[])
+    applied = gossip.receive([{"key": "new_key", "value": "remote"}])
+    assert applied == 1
+    assert brain.recall("new_key") == "remote"