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

restore_from_file() dropped HLC timestamps from snapshots, assigning fresh
(wall, 0, uuid) values instead. After disaster recovery, any causal
successor write with the pre-crash HLC was rejected with TimestampRegression,
causing silent data loss on replay/gossip.

- Add _parse_hlc() for wire/snapshot normalisation with legacy fallback
- restore_from_file: restore hlc, update global _hlc, hold lock during restore
- bulk_write: pass through hlc from row payloads
- gossip.receive: apply hlc, reject stale overwrites, skip missing-hlc updates

Regression tests cover snapshot round-trip and gossip HLC ordering.

Co-authored-by: Daniel <DJLougen@users.noreply.github.com>

Author: cursoragent

hive/gossip.py

@@ -122,11 +122,28 @@ 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,
                 )
                 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
@@ -505,25 +521,30 @@ def restore_from_file(self, path: str) -> int:
                 raise ValueError(
                     "snapshot checksum mismatch: file is corrupt or tampered"
                 )
-        self._nodes.clear()
-        self._order.clear()
-        self._order_index.clear()
-        for node_dict in nodes:
-            node = MemoryNode(
-                key=node_dict["key"],
-                value=node_dict["value"],
-                ts_ns=node_dict["ts_ns"],
-                trust=node_dict.get("trust", 1.0),
-                tags=set(node_dict.get("tags", [])),
-                node_id=node_dict.get("id", uuid.uuid4().hex[:12]),
-            )
-            for kind, neighbours in node_dict.get("edges", {}).items():
-                for n in neighbours:
-                    node.attach(kind, n)
-            storage_key = self._prefix(node.key)
-            self._nodes[storage_key] = node
-            self._order_index[storage_key] = len(self._order)
-            self._order.append(storage_key)
+        with self._lock:
+            self._nodes.clear()
+            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=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]),
+                )
+                for kind, neighbours in node_dict.get("edges", {}).items():
+                    for n in neighbours:
+                        node.attach(kind, n)
+                storage_key = self._prefix(node.key)
+                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

@@ -149,6 +149,34 @@ 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."""
+    from hive.rust_brain import TimestampRegression
+
+    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

@@ -32,3 +32,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"