clawft_kernel/
spatial_bvh.rs

1//! BVH-backed [`SpatialBackend`] (WEFT-718 / ADR-056 Phase C).
2//!
3//! Thin adapter: `Arc<Mutex<BvhStore>>` + optional [`ChainSink`] wired to
4//! the kernel [`ChainManager`]. Epoch versioning follows `vector_hnsw`.
5
6use std::sync::{Arc, Mutex};
7
8use clawft_bvh::{
9    Aabb, BranchId, BranchMeta, BvhChainKind, BvhError, BvhStore, BvhStoreConfig, ChainSink,
10    DiffEntry, Frustum, Leaf, LeafId, Ray, RayHit, Vec3,
11};
12
13use crate::spatial_backend::{SpatialBackend, SpatialError, SpatialResult};
14
15#[cfg(feature = "exochain")]
16use crate::chain::ChainManager;
17
18// ── Chain sink adapter ───────────────────────────────────────────────────
19
20/// Kernel `ChainSink` implementation: maps BVH mutations to ExoChain events.
21///
22/// Payloads are JSON (chain path already CBOR-encodes the envelope per
23/// ADR-030 when the chain is dual-signed). Dual-sign is applied by
24/// `ChainManager` when keys are configured (ADR-028).
25pub struct KernelBvhChainSink {
26    #[cfg(feature = "exochain")]
27    chain: Arc<ChainManager>,
28}
29
30impl KernelBvhChainSink {
31    /// Construct from a live chain manager.
32    #[cfg(feature = "exochain")]
33    pub fn new(chain: Arc<ChainManager>) -> Self {
34        Self { chain }
35    }
36
37    #[cfg(not(feature = "exochain"))]
38    pub fn new_noop() -> Self {
39        Self {}
40    }
41}
42
43impl ChainSink for KernelBvhChainSink {
44    fn on_event(&self, event: &BvhChainKind) {
45        #[cfg(feature = "exochain")]
46        {
47            let (kind, payload) = match event {
48                BvhChainKind::Insert {
49                    leaf_id,
50                    leaf,
51                    branch,
52                } => (
53                    crate::chain::EVENT_KIND_BVH_INSERT,
54                    serde_json::json!({
55                        "leaf_id": leaf_id.0,
56                        "branch": branch.0,
57                        "tag": leaf.tag,
58                        "identity": match leaf.identity_kind {
59                            clawft_bvh::IdentityKind::Object => "object",
60                            clawft_bvh::IdentityKind::Event => "event",
61                        },
62                        "bound": {
63                            "min": [leaf.bound.min.x, leaf.bound.min.y, leaf.bound.min.z],
64                            "max": [leaf.bound.max.x, leaf.bound.max.y, leaf.bound.max.z],
65                        },
66                        // Payload as hex for replay without embedding raw binary in logs.
67                        "payload_hex": hex_encode(&leaf.payload),
68                    }),
69                ),
70                BvhChainKind::Remove { leaf_id, branch } => (
71                    crate::chain::EVENT_KIND_BVH_REMOVE,
72                    serde_json::json!({
73                        "leaf_id": leaf_id.0,
74                        "branch": branch.0,
75                    }),
76                ),
77                BvhChainKind::Derive {
78                    parent,
79                    child,
80                    meta,
81                } => (
82                    crate::chain::EVENT_KIND_BVH_DERIVE,
83                    serde_json::json!({
84                        "parent": parent.0,
85                        "child": child.0,
86                        "name": meta.name,
87                        "priority_tier": meta.priority_tier,
88                    }),
89                ),
90                BvhChainKind::RebalanceSeal {
91                    branch,
92                    leaf_count,
93                    epoch,
94                } => (
95                    crate::chain::EVENT_KIND_BVH_REBALANCE_SEAL,
96                    serde_json::json!({
97                        "branch": branch.0,
98                        "leaf_count": leaf_count,
99                        "epoch": epoch,
100                    }),
101                ),
102            };
103            self.chain.append("spatial_bvh", kind, Some(payload));
104        }
105        #[cfg(not(feature = "exochain"))]
106        {
107            let _ = event;
108        }
109    }
110}
111
112fn hex_encode(bytes: &[u8]) -> String {
113    const HEX: &[u8; 16] = b"0123456789abcdef";
114    let mut out = String::with_capacity(bytes.len() * 2);
115    for &b in bytes {
116        out.push(HEX[(b >> 4) as usize] as char);
117        out.push(HEX[(b & 0xf) as usize] as char);
118    }
119    out
120}
121
122/// Decode hex produced by [`hex_encode`] (replay helper).
123pub fn hex_decode(s: &str) -> Option<Vec<u8>> {
124    if !s.len().is_multiple_of(2) {
125        return None;
126    }
127    let mut out = Vec::with_capacity(s.len() / 2);
128    let bytes = s.as_bytes();
129    let mut i = 0;
130    while i < bytes.len() {
131        let hi = hex_nibble(bytes[i])?;
132        let lo = hex_nibble(bytes[i + 1])?;
133        out.push((hi << 4) | lo);
134        i += 2;
135    }
136    Some(out)
137}
138
139fn hex_nibble(c: u8) -> Option<u8> {
140    match c {
141        b'0'..=b'9' => Some(c - b'0'),
142        b'a'..=b'f' => Some(c - b'a' + 10),
143        b'A'..=b'F' => Some(c - b'A' + 10),
144        _ => None,
145    }
146}
147
148// ── Backend ──────────────────────────────────────────────────────────────
149
150/// BVH spatial backend wrapping [`BvhStore`].
151pub struct BvhBackend {
152    store: Mutex<BvhStore>,
153    max_leaves: Option<usize>,
154    /// In-process record of every mutation (for `ecc.spatial.events` / replay checks), fed
155    /// alongside any ExoChain sink.
156    log: Arc<EventLog>,
157}
158
159/// Most mutation events kept in memory; older ones are dropped and replay is reported as
160/// incomplete rather than wrong.
161pub const EVENT_LOG_CAP: usize = 100_000;
162
163/// Bounded in-memory mutation log.
164#[derive(Default)]
165pub struct EventLog {
166    events: Mutex<Vec<clawft_bvh::BvhChainKind>>,
167    truncated: std::sync::atomic::AtomicBool,
168}
169
170impl ChainSink for EventLog {
171    fn on_event(&self, event: &clawft_bvh::BvhChainKind) {
172        let mut ev = self.events.lock().expect("event log poisoned");
173        if ev.len() >= EVENT_LOG_CAP {
174            ev.remove(0);
175            self.truncated.store(true, std::sync::atomic::Ordering::Relaxed);
176        }
177        ev.push(event.clone());
178    }
179}
180
181/// Fans one event out to the log and the optional external (ExoChain) sink.
182struct TeeSink {
183    log: Arc<EventLog>,
184    inner: Option<Arc<dyn ChainSink>>,
185}
186
187impl ChainSink for TeeSink {
188    fn on_event(&self, event: &clawft_bvh::BvhChainKind) {
189        self.log.on_event(event);
190        if let Some(ref s) = self.inner {
191            s.on_event(event);
192        }
193    }
194}
195
196/// Result of rebuilding the store from its mutation log.
197#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct ReplayReport {
199    /// Events replayed.
200    pub events: usize,
201    /// The replica has the same branches and the same leaf ids per branch as the live store.
202    pub ok: bool,
203    /// The log dropped old events (replay cannot be complete).
204    pub truncated: bool,
205    /// Branches compared.
206    pub branches: usize,
207}
208
209impl BvhBackend {
210    /// Create from store config (no chain sink yet).
211    pub fn new(config: BvhStoreConfig) -> Self {
212        let max = if config.max_leaves == 0 {
213            None
214        } else {
215            Some(config.max_leaves)
216        };
217        let log = Arc::new(EventLog::default());
218        let tee: Arc<dyn ChainSink> = Arc::new(TeeSink { log: log.clone(), inner: None });
219        Self {
220            store: Mutex::new(BvhStore::with_sink(config, tee)),
221            max_leaves: max,
222            log,
223        }
224    }
225
226    /// Defaults (1M leaves).
227    pub fn with_defaults() -> Self {
228        Self::new(BvhStoreConfig::default())
229    }
230
231    /// Create with a chain sink attached before first mutation.
232    pub fn with_sink(config: BvhStoreConfig, sink: Arc<dyn ChainSink>) -> Self {
233        let max = if config.max_leaves == 0 {
234            None
235        } else {
236            Some(config.max_leaves)
237        };
238        let log = Arc::new(EventLog::default());
239        let tee: Arc<dyn ChainSink> = Arc::new(TeeSink { log: log.clone(), inner: Some(sink) });
240        Self {
241            store: Mutex::new(BvhStore::with_sink(config, tee)),
242            max_leaves: max,
243            log,
244        }
245    }
246
247    /// Attach / replace the chain sink after construction (boot order).
248    pub fn set_chain_sink(&self, sink: Arc<dyn ChainSink>) {
249        let mut store = self.store.lock().expect("BvhStore lock poisoned");
250        store.set_sink(Arc::new(TeeSink { log: self.log.clone(), inner: Some(sink) }));
251    }
252
253    /// Run `f` with the store locked (branch-scoped insert / remove / query for the daemon's
254    /// `ecc.spatial.*` RPCs). Mutations still go through the store's chain sink.
255    pub fn with_store<R>(&self, f: impl FnOnce(&mut BvhStore) -> R) -> R {
256        let mut store = self.store.lock().expect("BvhStore lock poisoned");
257        f(&mut store)
258    }
259
260    /// Mutation events recorded so far, and whether old ones were dropped.
261    pub fn events(&self) -> (Vec<clawft_bvh::BvhChainKind>, bool) {
262        let ev = self.log.events.lock().expect("event log poisoned").clone();
263        (ev, self.log.truncated.load(std::sync::atomic::Ordering::Relaxed))
264    }
265
266    /// Rebuild a replica from the mutation log and compare it with the live store (branches
267    /// and leaf ids per branch). The live store is not replaced.
268    pub fn verify_replay(&self) -> ReplayReport {
269        let (events, truncated) = self.events();
270        let store = self.store.lock().expect("BvhStore lock poisoned");
271        let mut replica = BvhStore::new(store.config().clone());
272        let applied = events.iter().all(|e| replica.apply_chain_event(e).is_ok());
273        let branches = store.branch_count();
274        let ids = |s: &BvhStore, b: u64| -> Option<Vec<u64>> {
275            let mut v: Vec<u64> = s.snapshot_leaves(BranchId(b)).ok()?.into_iter().map(|(id, _)| id.0).collect();
276            v.sort_unstable();
277            Some(v)
278        };
279        let same = replica.branch_count() == branches && (0..branches as u64).all(|b| ids(&store, b) == ids(&replica, b));
280        ReplayReport { events: events.len(), ok: applied && same && !truncated, truncated, branches }
281    }
282
283    /// Snapshot leaves on the main branch (restart equality tests).
284    pub fn snapshot_main(&self) -> Vec<(LeafId, Leaf)> {
285        let store = self.store.lock().expect("BvhStore lock poisoned");
286        store
287            .snapshot_leaves(BranchId::MAIN)
288            .unwrap_or_default()
289    }
290
291    /// Restore leaves on the main branch (restart helper).
292    pub fn restore_main(&self, leaves: &[(LeafId, Leaf)]) -> SpatialResult<()> {
293        let mut store = self.store.lock().expect("BvhStore lock poisoned");
294        store
295            .restore_leaves(BranchId::MAIN, leaves)
296            .map_err(map_bvh_err)
297    }
298
299    /// Number of branches.
300    pub fn branch_count(&self) -> usize {
301        self.store
302            .lock()
303            .expect("BvhStore lock poisoned")
304            .branch_count()
305    }
306
307    /// Seal / rebalance witness for the active branch.
308    pub fn rebalance_seal(&self) -> SpatialResult<()> {
309        let mut store = self.store.lock().expect("BvhStore lock poisoned");
310        let branch = store.active_branch();
311        store.rebalance_seal(branch).map_err(map_bvh_err)
312    }
313}
314
315fn map_bvh_err(e: BvhError) -> SpatialError {
316    match e {
317        BvhError::CapacityExceeded { max, current } => {
318            SpatialError::CapacityExceeded { max, current }
319        }
320        BvhError::UnknownBranch(id) => SpatialError::UnknownBranch(id.0),
321        BvhError::Other(msg) => SpatialError::Other(msg.to_string()),
322    }
323}
324
325impl SpatialBackend for BvhBackend {
326    fn insert(&self, leaf: Leaf) -> SpatialResult<LeafId> {
327        let mut store = self.store.lock().expect("BvhStore lock poisoned");
328        store.insert(leaf).map_err(map_bvh_err)
329    }
330
331    fn remove(&self, id: LeafId) -> bool {
332        let mut store = self.store.lock().expect("BvhStore lock poisoned");
333        store.remove(id)
334    }
335
336    fn get(&self, id: LeafId) -> Option<Leaf> {
337        let store = self.store.lock().expect("BvhStore lock poisoned");
338        store.get(id).cloned()
339    }
340
341    fn query_point(&self, p: Vec3) -> Vec<LeafId> {
342        let store = self.store.lock().expect("BvhStore lock poisoned");
343        store.query_point(p)
344    }
345
346    fn query_aabb(&self, bb: Aabb) -> Vec<LeafId> {
347        let store = self.store.lock().expect("BvhStore lock poisoned");
348        store.query_aabb(bb)
349    }
350
351    fn query_sphere(&self, center: Vec3, radius: f32) -> Vec<LeafId> {
352        let store = self.store.lock().expect("BvhStore lock poisoned");
353        store.query_sphere(center, radius)
354    }
355
356    fn query_ray(&self, ray: Ray, max_t: f32) -> Vec<RayHit> {
357        let store = self.store.lock().expect("BvhStore lock poisoned");
358        store.query_ray(ray, max_t)
359    }
360
361    fn query_frustum(&self, frustum: &Frustum) -> Vec<LeafId> {
362        let store = self.store.lock().expect("BvhStore lock poisoned");
363        store.query_frustum(frustum)
364    }
365
366    fn query_knn(&self, p: Vec3, k: usize) -> Vec<LeafId> {
367        let store = self.store.lock().expect("BvhStore lock poisoned");
368        store.query_knn(p, k)
369    }
370
371    fn derive_branch(&self, meta: BranchMeta) -> SpatialResult<BranchId> {
372        let mut store = self.store.lock().expect("BvhStore lock poisoned");
373        store.derive(meta).map_err(map_bvh_err)
374    }
375
376    fn branch_diff(&self, a: BranchId, b: BranchId, region: Aabb) -> Vec<DiffEntry> {
377        let store = self.store.lock().expect("BvhStore lock poisoned");
378        store.branch_diff(a, b, region).unwrap_or_default()
379    }
380
381    fn epoch(&self) -> u64 {
382        self.store.lock().expect("BvhStore lock poisoned").epoch()
383    }
384
385    fn len(&self) -> usize {
386        self.store.lock().expect("BvhStore lock poisoned").len()
387    }
388
389    fn backend_name(&self) -> &str {
390        "bvh"
391    }
392
393    fn max_leaves(&self) -> Option<usize> {
394        self.max_leaves
395    }
396}
397
398#[cfg(test)]
399mod tests {
400    use super::*;
401    use clawft_bvh::{IdentityKind, RecordingChainSink};
402
403    fn leaf_at(x: f32, tag: u32) -> Leaf {
404        Leaf::empty_payload(
405            Aabb::from_min_max(
406                Vec3::new(x, 0.0, 0.0),
407                Vec3::new(x + 0.5, 0.5, 0.5),
408            ),
409            IdentityKind::Object,
410            tag,
411        )
412    }
413
414    #[test]
415    fn insert_query_remove() {
416        let b = BvhBackend::with_defaults();
417        let id = b.insert(leaf_at(0.0, 1)).unwrap();
418        assert_eq!(b.len(), 1);
419        assert!(b.query_point(Vec3::new(0.25, 0.25, 0.25)).contains(&id));
420        assert!(b.remove(id));
421        assert!(b.is_empty());
422    }
423
424    #[test]
425    fn chain_sink_receives_insert() {
426        let sink = Arc::new(RecordingChainSink::new());
427        let b = BvhBackend::with_sink(BvhStoreConfig::default(), sink.clone());
428        let _ = b.insert(leaf_at(1.0, 9)).unwrap();
429        assert_eq!(sink.len(), 1);
430    }
431
432    #[test]
433    fn snapshot_restore_byte_identical() {
434        let b = BvhBackend::with_defaults();
435        for i in 0..20 {
436            b.insert(leaf_at(i as f32, i)).unwrap();
437        }
438        let snap = b.snapshot_main();
439        let b2 = BvhBackend::with_defaults();
440        b2.restore_main(&snap).unwrap();
441        assert_eq!(b.snapshot_main(), b2.snapshot_main());
442        assert_eq!(b.len(), b2.len());
443    }
444
445    #[test]
446    fn hex_roundtrip() {
447        let raw = vec![0u8, 1, 0xab, 0xff];
448        let h = hex_encode(&raw);
449        assert_eq!(hex_decode(&h).unwrap(), raw);
450    }
451}
clawft_kernel/spatial_bvh.rs

clawft_kernel/
spatial_bvh.rs
