From 23cf4e1da1d09b86a207d9100c082540702f9f38 Mon Sep 17 00:00:00 2001 From: ruv Date: Thu, 20 Aug 2026 10:01:28 -0400 Subject: [PATCH 1/3] feat: AtomicObservation + causal episodic graph (PIR WP18, ADR-320) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the cross-source causal fusion layer to ruvector-agent-memory, extending the WP4 TARL ledger (ADR-307) into the multi-agent setting. Pattern informed by MemFuse (arXiv:2608.18704, Darwin-Agent/Mi-Memory), and explicitly distinct from the unrelated pre-existing memfuse/memfuse OSS memory layer — per ADR-320's binding naming discipline, no artifact here is named "memfuse". - observation.rs: AtomicObservation — the event-layer atomic unit per ADR-320 §1 (source, time, confidence, tenant, signature, causal_parents + evidence payload). Content-addressed via the repo's SHA-256 (rvf-types); per-observation Ed25519 signature via rvf-types' RFC 8032 primitives. No new hash or signature scheme (ADR-320 gate). - fusion.rs: CausalEpisodicGraph — cluster-layer fusion with the load- bearing provenance guarantee (every derived node resolves back to its atomic sources, transitively). Enforces ADR-320 security gates: per-observation signature, causal-parents integrity (resolvable + acyclic), tenant isolation. ingest_governed routes admission through the WP4 TransactionalLedger (causal_parents -> ledger depends_on), reusing the ledger discipline rather than rebuilding it. - Tests: signature round-trip, multi-source fusion, provenance resolution proof, confidence/tenant carry-through, tampered-observation rejection, cross-tenant/unresolved-parent rejection, and governed ledger composition. 24 tests green; clippy clean; #861 guard OK. Refs #865 #837 Co-Authored-By: claude-flow Claude-Session: https://claude.ai/code/session_012Jib2gQyJpqCoo2xYAbb4X --- Cargo.lock | 1 + crates/ruvector-agent-memory/Cargo.toml | 6 + crates/ruvector-agent-memory/src/fusion.rs | 438 ++++++++++++++++++ crates/ruvector-agent-memory/src/lib.rs | 15 + .../ruvector-agent-memory/src/observation.rs | 328 +++++++++++++ .../tests/atomic_observation_fusion.rs | 274 +++++++++++ 6 files changed, 1062 insertions(+) create mode 100644 crates/ruvector-agent-memory/src/fusion.rs create mode 100644 crates/ruvector-agent-memory/src/observation.rs create mode 100644 crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs diff --git a/Cargo.lock b/Cargo.lock index cd1c8d921..4a2b07501 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8793,6 +8793,7 @@ version = "0.1.0" dependencies = [ "rand 0.8.6", "ruvector-proof-gate", + "rvf-types", "serde", "serde_json", ] diff --git a/crates/ruvector-agent-memory/Cargo.toml b/crates/ruvector-agent-memory/Cargo.toml index 1182bbbf1..1fe70bb3f 100644 --- a/crates/ruvector-agent-memory/Cargo.toml +++ b/crates/ruvector-agent-memory/Cargo.toml @@ -16,6 +16,12 @@ path = "src/main.rs" [dependencies] rand = "0.8" serde = { workspace = true } +# ADR-320 (PIR WP18): reuse the repo's SHA-256 (content addressing) and +# RFC 8032 Ed25519 (per-observation signatures) for AtomicObservation — no new +# hash or signature scheme (ADR-320 Security Gates). `ed25519` feature pulls in +# ed25519-dalek + rand_core; the always-compiled `sha256` module needs no +# feature. +rvf-types = { version = "0.2", path = "../rvf/rvf-types", features = ["ed25519"] } # Existing ADR-194/047 proof-gated write machinery; adapts to the ledger's # ProofGate trait behind the `proof-gate` feature (see src/ledger.rs). ruvector-proof-gate = { path = "../ruvector-proof-gate", optional = true } diff --git a/crates/ruvector-agent-memory/src/fusion.rs b/crates/ruvector-agent-memory/src/fusion.rs new file mode 100644 index 000000000..169b1622a --- /dev/null +++ b/crates/ruvector-agent-memory/src/fusion.rs @@ -0,0 +1,438 @@ +//! The causal episodic graph — the cluster-layer fusion structure of ADR-320 +//! (PIR WP18), built over [`AtomicObservation`]s. +//! +//! Informed by **MemFuse (arXiv:2608.18704, `Darwin-Agent/Mi-Memory`)**'s +//! event-layer → cluster-layer architecture, and **explicitly distinct from the +//! unrelated pre-existing `memfuse/memfuse` open-source project**. Atomic, +//! source-tagged observations (the event layer) fuse into +//! [`FusedCluster`]s (the cluster layer), and **every fused/derived node +//! resolves back to the atomic source observations it was built from** — the +//! load-bearing provenance guarantee of this layer (see +//! [`CausalEpisodicGraph::resolve_provenance`]). +//! +//! ## Security / validation gates (ADR-320) +//! +//! - **Per-observation signature** verified before an observation enters the +//! graph ([`CausalEpisodicGraph::ingest`]). +//! - **Causal-parents integrity**: every `causal_parents` reference must +//! resolve to an already-present observation; an unresolvable parent, or a +//! self-referential (cyclic) parent, is a hard rejection at ingest, not a +//! warning. (Because parents must pre-exist, the parent relation is +//! necessarily a DAG.) +//! - **Tenant isolation**: a [`CausalEpisodicGraph`] is bound to one +//! [`Tenant`]; an observation from any other tenant is rejected and never +//! fuses in. +//! +//! ## Transactional integrity (ADR-320 §3, reusing WP4) +//! +//! [`CausalEpisodicGraph::ingest_governed`] routes an observation's admission +//! through the WP4 [`TransactionalLedger`] (ADR-307 TARL): the observation +//! enters memory as a governed `add` transition that emits an ADR-134 witness +//! record ("no witness, no mutation"), with its `causal_parents` mapped to the +//! ledger's `depends_on` edges so the ledger's poisoning-containment cascade +//! extends to observation lineage. This crate does **not** rebuild the ledger; +//! it interfaces to it. + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::ledger::{ProofGate, TransactionalLedger}; +use crate::observation::{AtomicObservation, ObservationId, Tenant}; +use crate::ops::{LedgerError, WitnessSink}; + +/// Identifier for a cluster-layer [`FusedCluster`] (graph-local, monotonic). +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ClusterId(pub u64); + +/// A reference to a node in the causal episodic graph: either an event-layer +/// atomic observation or a cluster-layer fused cluster. A cluster may fuse +/// other clusters, so provenance resolution is transitive down to atomic +/// observations. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum NodeRef { + Observation(ObservationId), + Cluster(ClusterId), +} + +/// A cluster-layer node: related observations (and/or sub-clusters) fused into +/// one unit, carrying aggregated confidence and the (single) tenant its members +/// share. +#[derive(Debug, Clone, PartialEq)] +pub struct FusedCluster { + pub id: ClusterId, + /// Event-layer observations and/or sub-clusters fused into this node. + pub members: Vec, + /// The tenant all members share (equal to the owning graph's tenant). + pub tenant: Tenant, + /// Aggregated confidence: the **weakest-link minimum** over member + /// confidences. A fused cluster is only as trustworthy as its least + /// confident evidence. + pub confidence: f32, + /// Human-readable fusion key / topic label. + pub label: String, +} + +/// Errors from fusion-graph operations. +#[derive(Debug)] +pub enum FusionError { + /// The observation's signature did not verify (per-observation gate). + SignatureInvalid(ObservationId), + /// The observation's tenant does not match the graph's tenant boundary. + TenantMismatch { expected: Tenant, got: Tenant }, + /// A `causal_parents` reference does not resolve to a known observation. + UnresolvedParent { + observation: ObservationId, + missing: ObservationId, + }, + /// The observation lists its own id as a causal parent (a trivial cycle). + CausalCycle(ObservationId), + /// An observation with this id is already present. + DuplicateObservation(ObservationId), + /// A referenced observation is not in the graph. + UnknownObservation(ObservationId), + /// A referenced cluster is not in the graph. + UnknownCluster(ClusterId), + /// A fusion was requested with no members. + EmptyCluster, + /// A governed ingest referenced a parent that has no ledger entry (it was + /// not itself admitted through the ledger). + ParentNotGoverned(ObservationId), + /// The underlying WP4 ledger refused the governed transition. + Ledger(LedgerError), +} + +impl std::fmt::Display for FusionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + FusionError::SignatureInvalid(id) => { + write!(f, "observation {} failed signature verification", id.to_hex()) + } + FusionError::TenantMismatch { expected, got } => write!( + f, + "tenant isolation violation: graph tenant {:?}, observation tenant {:?}", + expected.as_str(), + got.as_str() + ), + FusionError::UnresolvedParent { + observation, + missing, + } => write!( + f, + "observation {} references unresolved causal parent {}", + observation.to_hex(), + missing.to_hex() + ), + FusionError::CausalCycle(id) => { + write!(f, "observation {} is its own causal parent", id.to_hex()) + } + FusionError::DuplicateObservation(id) => { + write!(f, "observation {} already present", id.to_hex()) + } + FusionError::UnknownObservation(id) => { + write!(f, "unknown observation {}", id.to_hex()) + } + FusionError::UnknownCluster(id) => write!(f, "unknown cluster {}", id.0), + FusionError::EmptyCluster => write!(f, "cannot fuse an empty member set"), + FusionError::ParentNotGoverned(id) => write!( + f, + "governed ingest parent {} has no ledger entry", + id.to_hex() + ), + FusionError::Ledger(e) => write!(f, "ledger refused governed ingest: {e}"), + } + } +} + +impl std::error::Error for FusionError {} + +/// A tenant-scoped causal episodic graph: an event layer of atomic +/// observations and a cluster layer of fused clusters over them. +pub struct CausalEpisodicGraph { + tenant: Tenant, + observations: BTreeMap, + clusters: BTreeMap, + /// Ledger entry ids for observations admitted through [`Self::ingest_governed`]. + ledger_ids: BTreeMap, + next_cluster: u64, +} + +impl CausalEpisodicGraph { + /// A new, empty graph bound to `tenant`. + pub fn new(tenant: Tenant) -> Self { + Self { + tenant, + observations: BTreeMap::new(), + clusters: BTreeMap::new(), + ledger_ids: BTreeMap::new(), + next_cluster: 0, + } + } + + /// The tenant boundary this graph enforces. + pub fn tenant(&self) -> &Tenant { + &self.tenant + } + + /// Admit an observation into the event layer after running every + /// ADR-320 ingest gate (signature, tenant, causal-parents integrity, + /// duplicate). Returns the observation's content address on success. + /// + /// This is the pure-graph path; [`Self::ingest_governed`] additionally + /// routes admission through the WP4 ledger. + pub fn ingest(&mut self, obs: AtomicObservation) -> Result { + let id = self.validate_for_ingest(&obs)?; + self.observations.insert(id, obs); + Ok(id) + } + + /// Admit an observation as a **governed transition** through the WP4 TARL + /// ledger (ADR-320 §3). Runs the same ingest gates, then records the + /// admission in `ledger` as an `add` (emitting an ADR-134 witness record), + /// wiring `causal_parents` to the ledger's `depends_on` edges. On ledger + /// refusal nothing is inserted into the graph. + /// + /// Returns the observation id and its ledger entry id. + pub fn ingest_governed( + &mut self, + obs: AtomicObservation, + ledger: &mut TransactionalLedger, + actor: &str, + ) -> Result<(ObservationId, u64), FusionError> { + let id = self.validate_for_ingest(&obs)?; + + // Map causal parents to ledger dependency edges. Parents must have been + // admitted through the ledger too, else there is no governed dependency + // to record. + let mut depends_on = Vec::with_capacity(obs.causal_parents.len()); + for parent in &obs.causal_parents { + let ledger_id = self + .ledger_ids + .get(parent) + .ok_or(FusionError::ParentNotGoverned(*parent))?; + depends_on.push(*ledger_id); + } + + // Witness-first governed transition: on refusal nothing mutates here. + let entry_id = ledger + .add( + id.to_hex(), + &depends_on, + actor, + "ADR-320 atomic observation admitted to continuous-latent-state tier", + ) + .map_err(FusionError::Ledger)?; + + self.ledger_ids.insert(id, entry_id); + self.observations.insert(id, obs); + Ok((id, entry_id)) + } + + /// Fuse `members` (event-layer observations and/or sub-clusters, all of + /// this graph's tenant) into a new cluster-layer node. Confidence is the + /// weakest-link minimum over members. Rejects an empty set or any member + /// not present in the graph. + pub fn fuse( + &mut self, + members: &[NodeRef], + label: impl Into, + ) -> Result { + if members.is_empty() { + return Err(FusionError::EmptyCluster); + } + let mut confidence = f32::INFINITY; + for member in members { + confidence = confidence.min(self.node_confidence(*member)?); + } + let id = ClusterId(self.next_cluster); + self.next_cluster += 1; + self.clusters.insert( + id, + FusedCluster { + id, + members: members.to_vec(), + tenant: self.tenant.clone(), + confidence, + label: label.into(), + }, + ); + Ok(id) + } + + /// Resolve `node` to the set of **atomic source observations** it derives + /// from — the load-bearing provenance guarantee. For an observation this is + /// the observation itself; for a cluster it is the transitive union of its + /// members' atomic sources. The traversal is cycle-safe (the cluster layer + /// is acyclic by construction, but a `visited` guard makes resolution total + /// regardless). + pub fn resolve_provenance( + &self, + node: NodeRef, + ) -> Result, FusionError> { + let mut atomic = BTreeSet::new(); + let mut visited_clusters = BTreeSet::new(); + self.collect_provenance(node, &mut atomic, &mut visited_clusters)?; + Ok(atomic) + } + + /// Look up an event-layer observation. + pub fn observation(&self, id: ObservationId) -> Option<&AtomicObservation> { + self.observations.get(&id) + } + + /// Look up a cluster-layer node. + pub fn cluster(&self, id: ClusterId) -> Option<&FusedCluster> { + self.clusters.get(&id) + } + + /// The ledger entry id for an observation admitted via + /// [`Self::ingest_governed`], if any. + pub fn ledger_id(&self, id: ObservationId) -> Option { + self.ledger_ids.get(&id).copied() + } + + /// Number of event-layer observations. + pub fn observation_count(&self) -> usize { + self.observations.len() + } + + /// Number of cluster-layer nodes. + pub fn cluster_count(&self) -> usize { + self.clusters.len() + } + + // ── Internals ──────────────────────────────────────────────────────────── + + /// Run every ingest gate and return the validated content address without + /// mutating the graph. + fn validate_for_ingest(&self, obs: &AtomicObservation) -> Result { + // Tenant isolation (hard boundary). + if obs.tenant != self.tenant { + return Err(FusionError::TenantMismatch { + expected: self.tenant.clone(), + got: obs.tenant.clone(), + }); + } + // Per-observation signature. + if !obs.verify() { + return Err(FusionError::SignatureInvalid(obs.id())); + } + let id = obs.id(); + if self.observations.contains_key(&id) { + return Err(FusionError::DuplicateObservation(id)); + } + // Causal-parents integrity: acyclic (no self-parent) and resolvable. + for parent in &obs.causal_parents { + if *parent == id { + return Err(FusionError::CausalCycle(id)); + } + if !self.observations.contains_key(parent) { + return Err(FusionError::UnresolvedParent { + observation: id, + missing: *parent, + }); + } + } + Ok(id) + } + + /// Confidence of a node, for weakest-link aggregation. + fn node_confidence(&self, node: NodeRef) -> Result { + match node { + NodeRef::Observation(id) => self + .observations + .get(&id) + .map(|o| o.confidence) + .ok_or(FusionError::UnknownObservation(id)), + NodeRef::Cluster(id) => self + .clusters + .get(&id) + .map(|c| c.confidence) + .ok_or(FusionError::UnknownCluster(id)), + } + } + + fn collect_provenance( + &self, + node: NodeRef, + atomic: &mut BTreeSet, + visited_clusters: &mut BTreeSet, + ) -> Result<(), FusionError> { + match node { + NodeRef::Observation(id) => { + if !self.observations.contains_key(&id) { + return Err(FusionError::UnknownObservation(id)); + } + atomic.insert(id); + Ok(()) + } + NodeRef::Cluster(id) => { + if !visited_clusters.insert(id) { + return Ok(()); // already expanded; cycle-safe + } + let cluster = self + .clusters + .get(&id) + .ok_or(FusionError::UnknownCluster(id))?; + for member in &cluster.members { + self.collect_provenance(*member, atomic, visited_clusters)?; + } + Ok(()) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::observation::{ObservationSource, SourceKind}; + use rand::rngs::OsRng; + use rvf_types::Ed25519Keypair; + + fn signed( + keypair: &Ed25519Keypair, + kind: SourceKind, + src_id: &str, + tenant: &str, + confidence: f32, + parents: Vec, + payload: &[u8], + ) -> AtomicObservation { + AtomicObservation::new_signed( + ObservationSource { + kind, + id: src_id.to_string(), + public_key: [0u8; 32], + }, + 0, + confidence, + Tenant::new(tenant), + parents, + payload.to_vec(), + keypair, + ) + .unwrap() + } + + #[test] + fn cross_tenant_observation_rejected() { + let kp = Ed25519Keypair::generate(&mut OsRng); + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let foreign = signed(&kp, SourceKind::AgentObservation, "a", "globex", 0.9, vec![], b"x"); + match graph.ingest(foreign) { + Err(FusionError::TenantMismatch { .. }) => {} + other => panic!("expected TenantMismatch, got {other:?}"), + } + } + + #[test] + fn unresolved_parent_rejected() { + let kp = Ed25519Keypair::generate(&mut OsRng); + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let ghost = ObservationId([9u8; 32]); + let child = signed(&kp, SourceKind::AgentObservation, "a", "acme", 0.8, vec![ghost], b"c"); + match graph.ingest(child) { + Err(FusionError::UnresolvedParent { .. }) => {} + other => panic!("expected UnresolvedParent, got {other:?}"), + } + } +} diff --git a/crates/ruvector-agent-memory/src/lib.rs b/crates/ruvector-agent-memory/src/lib.rs index 507f30e5e..fdf950096 100644 --- a/crates/ruvector-agent-memory/src/lib.rs +++ b/crates/ruvector-agent-memory/src/lib.rs @@ -24,6 +24,15 @@ //! five executable memory operations over accepted/pending/rejected states, //! with witness-record emission (ADR-134 schema) and proof-gated acceptance. //! +//! The `observation` and `fusion` modules add the cross-source causal fusion +//! layer (ADR-320, PIR WP18): source-tagged [`AtomicObservation`]s fuse into a +//! [`CausalEpisodicGraph`] with provenance preserved back to each atomic +//! source. Informed by MemFuse (arXiv:2608.18704, `Darwin-Agent/Mi-Memory`) and +//! explicitly distinct from the unrelated `memfuse/memfuse` OSS project. It +//! reuses this crate's WP4 ledger for governed admission and `rvf-types`' +//! SHA-256/Ed25519 for content addressing and per-observation signatures — no +//! new hash or signature scheme is introduced. +//! //! - Park et al. 2023, "Generative Agents" (arXiv:2304.03442) //! - Zhong et al. 2023, "MemoryBank" (arXiv:2305.10250) //! - Xu 2026, "Self-Aware Vector Embeddings for RAG" (arXiv:2604.20598) @@ -31,15 +40,21 @@ //! - Survey 2026, "From Storage to Experience" (arXiv:2605.06716) pub mod compaction; +pub mod fusion; pub mod ledger; pub mod memory; +pub mod observation; pub mod ops; pub mod scoring; pub use compaction::{CoherencePolicy, CoherenceWeights, CompactionPolicy, LfuPolicy, LruPolicy}; +pub use fusion::{CausalEpisodicGraph, ClusterId, FusedCluster, FusionError, NodeRef}; #[cfg(feature = "proof-gate")] pub use ledger::WriteGateAdapter; pub use ledger::{replay_history, AlwaysAdmitGate, LedgerEntry, ProofGate, TransactionalLedger}; +pub use observation::{ + AtomicObservation, ObservationError, ObservationId, ObservationSource, SourceKind, Tenant, +}; pub use memory::{MemoryEntry, MemoryStore, SearchResult}; pub use ops::{ AcceptanceReceipt, EvidenceGrade, LedgerError, LedgerState, LedgerWitnessRecord, MemoryOp, diff --git a/crates/ruvector-agent-memory/src/observation.rs b/crates/ruvector-agent-memory/src/observation.rs new file mode 100644 index 000000000..649594469 --- /dev/null +++ b/crates/ruvector-agent-memory/src/observation.rs @@ -0,0 +1,328 @@ +//! `AtomicObservation` — the atomic memory unit of the cross-source causal +//! fusion layer (ADR-320, PIR WP18). +//! +//! Informed by **MemFuse (arXiv:2608.18704, `Darwin-Agent/Mi-Memory`)**, which +//! preserves source-level evidence in an event-layer atomic memory and organizes +//! related atomic events into a causal fusion graph. This is deliberately a +//! first-party implementation of that *pattern* and is **explicitly distinct +//! from the unrelated, pre-existing `memfuse/memfuse` open-source memory layer** +//! — per ADR-320's binding naming discipline, no artifact here is ever named +//! `memfuse`. +//! +//! An [`AtomicObservation`] is the atomic unit written into ADR-307's +//! continuous-latent-state tier by any agent or sensor. Per ADR-320 §1 it +//! carries, at minimum: `source`, `time`, `confidence`, `tenant`, `signature`, +//! and `causal_parents`. It is **content-addressed**: its [`ObservationId`] is +//! the SHA-256 of its canonical byte encoding (everything except the +//! signature), so provenance — the load-bearing property of this layer — is a +//! cryptographic fact, not a bookkeeping convention: the id *is* a commitment +//! to the exact source, time, confidence, tenant, parents, and evidence. +//! +//! ## Reused primitives (no new hash or signature scheme — ADR-320 gate) +//! +//! - Hashing: [`rvf_types::sha256::sha256`] (the repo's FIPS 180-4 SHA-256). +//! - Signing: [`rvf_types::ed25519_sign`] / [`rvf_types::ed25519_verify`] +//! (the repo's RFC 8032 Ed25519, the same `SignatureVerified`-grade evidence +//! the ADR-322C witness contract anticipates). + +use rvf_types::sha256::sha256; +use rvf_types::{ed25519_sign, ed25519_verify, Ed25519Keypair}; + +/// Domain-separation tag bound into every observation's signed preimage, so a +/// signature over an `AtomicObservation` can never be replayed as a signature +/// over some other RVF/ledger structure. +const OBSERVATION_DOMAIN: &[u8] = b"ADR-320-atomic-observation-v1"; + +/// Multi-tenant isolation boundary carried by every observation. Enforced as a +/// hard boundary at fusion time (ADR-320 Security Gates: an observation from +/// one tenant never fuses into another tenant's causal graph). +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct Tenant(pub String); + +impl Tenant { + pub fn new(id: impl Into) -> Self { + Tenant(id.into()) + } + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// The heterogeneous source kinds that converge into one causal graph. ADR-320 +/// models RuView RF events, network telemetry, agent observations, and user +/// activity flowing into the same fusion; `Other` keeps the schema extensible +/// without a layout break (reserve codes >= 16 for out-of-tree kinds). +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum SourceKind { + /// RuView radio-frequency event stream. + RuViewRf, + /// Network / infrastructure telemetry. + NetworkTelemetry, + /// An agent's own first-party observation. + AgentObservation, + /// End-user activity signal. + UserActivity, + /// Extension point for out-of-tree source kinds (code >= 16). + Other(u16), +} + +impl SourceKind { + /// Stable numeric code bound into the canonical (hashed) encoding. + pub fn code(self) -> u16 { + match self { + SourceKind::RuViewRf => 1, + SourceKind::NetworkTelemetry => 2, + SourceKind::AgentObservation => 3, + SourceKind::UserActivity => 4, + SourceKind::Other(x) => x, + } + } +} + +/// The originating identity of an observation: its kind, a stable string id +/// (e.g. `"sensor-7"`, `"agent-alpha"`), and the Ed25519 public key the +/// observation is signed under. Carrying the key inline makes verification +/// self-contained for this slice; a production deployment would resolve the key +/// from a tenant-scoped key registry (see Deferred scope). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ObservationSource { + pub kind: SourceKind, + pub id: String, + pub public_key: [u8; 32], +} + +/// Content address of an [`AtomicObservation`]: SHA-256 over the domain tag and +/// the observation's canonical encoding, excluding the signature. Two +/// observations share an id **iff** they commit to the identical source, time, +/// confidence, tenant, causal parents, and evidence payload. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ObservationId(pub [u8; 32]); + +impl ObservationId { + /// Lowercase hex rendering — used as the provenance content string when an + /// observation is recorded as a governed transition in the WP4 ledger. + pub fn to_hex(self) -> String { + let mut s = String::with_capacity(64); + for b in self.0 { + s.push(char::from_digit((b >> 4) as u32, 16).unwrap()); + s.push(char::from_digit((b & 0xf) as u32, 16).unwrap()); + } + s + } +} + +/// Errors constructing or validating an [`AtomicObservation`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ObservationError { + /// `confidence` was not a finite value in `[0.0, 1.0]`. + ConfidenceOutOfRange, +} + +/// An atomic, source-tagged observation — the event-layer memory unit of +/// ADR-320. Fields follow ADR-320 §1 exactly; `payload` is the original +/// evidence body that provenance resolves back to. +#[derive(Debug, Clone, PartialEq)] +pub struct AtomicObservation { + /// Originating agent/sensor identity (ADR-320 `source`). + pub source: ObservationSource, + /// Observation timestamp in nanoseconds (ADR-320 `time`). + pub time_ns: u64, + /// The source's own confidence in the observation, in `[0.0, 1.0]` + /// (ADR-320 `confidence`). + pub confidence: f32, + /// Multi-tenant isolation boundary (ADR-320 `tenant`). + pub tenant: Tenant, + /// The `AtomicObservation`(s) that causally preceded and informed this one + /// (ADR-320 `causal_parents`). Canonicalized (sorted, de-duplicated). + pub causal_parents: Vec, + /// Original evidence body this observation records; provenance resolves + /// back to these bytes. + pub payload: Vec, + /// Ed25519 signature over this observation's content address (ADR-320 + /// `signature`; RFC 8032, reused from `rvf-types`). + pub signature: [u8; 64], +} + +impl AtomicObservation { + /// Construct and **sign** an observation. The `source.public_key` is + /// overwritten with `keypair`'s public key so the stored source identity + /// and the signing key can never disagree. + /// + /// `causal_parents` is canonicalized (sorted + de-duplicated) before the + /// content address is computed, so parent ordering never affects identity. + /// + /// Returns [`ObservationError::ConfidenceOutOfRange`] if `confidence` is + /// not finite in `[0.0, 1.0]`. + pub fn new_signed( + mut source: ObservationSource, + time_ns: u64, + confidence: f32, + tenant: Tenant, + mut causal_parents: Vec, + payload: Vec, + keypair: &Ed25519Keypair, + ) -> Result { + if !confidence.is_finite() || !(0.0..=1.0).contains(&confidence) { + return Err(ObservationError::ConfidenceOutOfRange); + } + source.public_key = keypair.public_key(); + causal_parents.sort_unstable(); + causal_parents.dedup(); + + let mut obs = AtomicObservation { + source, + time_ns, + confidence, + tenant, + causal_parents, + payload, + signature: [0u8; 64], + }; + let id = obs.compute_id(); + obs.signature = ed25519_sign(&keypair.secret_key(), &id.0); + Ok(obs) + } + + /// The observation's content address (recomputed from current field + /// values; never trusts a stored value). + pub fn id(&self) -> ObservationId { + self.compute_id() + } + + /// Verify the signature binds this exact content under `source.public_key`. + /// + /// Because the signed message is the content address, any post-signing + /// mutation to any field (source, time, confidence, tenant, parents, or + /// payload) changes the id and makes verification fail — this is the + /// per-observation authenticity gate of ADR-320. + pub fn verify(&self) -> bool { + let id = self.compute_id(); + ed25519_verify(&self.source.public_key, &id.0, &self.signature) + } + + /// Canonical, deterministic byte encoding of everything the id commits to + /// (the signature is excluded — it is a function of the id, not an input). + fn canonical_bytes(&self) -> Vec { + let mut b = Vec::new(); + b.extend_from_slice(OBSERVATION_DOMAIN); + b.extend_from_slice(&self.source.kind.code().to_le_bytes()); + write_len_prefixed(&mut b, self.source.id.as_bytes()); + b.extend_from_slice(&self.source.public_key); + b.extend_from_slice(&self.time_ns.to_le_bytes()); + // f32 committed by its bit pattern for exact, portable determinism. + b.extend_from_slice(&self.confidence.to_bits().to_le_bytes()); + write_len_prefixed(&mut b, self.tenant.as_str().as_bytes()); + b.extend_from_slice(&(self.causal_parents.len() as u32).to_le_bytes()); + for parent in &self.causal_parents { + b.extend_from_slice(&parent.0); + } + write_len_prefixed(&mut b, &self.payload); + b + } + + fn compute_id(&self) -> ObservationId { + ObservationId(sha256(&self.canonical_bytes())) + } +} + +/// Append a `u32` length prefix (LE) followed by the bytes. +fn write_len_prefixed(buf: &mut Vec, data: &[u8]) { + buf.extend_from_slice(&(data.len() as u32).to_le_bytes()); + buf.extend_from_slice(data); +} + +#[cfg(test)] +mod tests { + use super::*; + use rand::rngs::OsRng; + + fn kp() -> Ed25519Keypair { + Ed25519Keypair::generate(&mut OsRng) + } + + fn source(kind: SourceKind, id: &str) -> ObservationSource { + ObservationSource { + kind, + id: id.to_string(), + public_key: [0u8; 32], + } + } + + #[test] + fn construction_and_signature_round_trip() { + let keypair = kp(); + let obs = AtomicObservation::new_signed( + source(SourceKind::AgentObservation, "agent-alpha"), + 1_000, + 0.9, + Tenant::new("acme"), + vec![], + b"observed evidence".to_vec(), + &keypair, + ) + .unwrap(); + + assert!(obs.verify(), "freshly signed observation must verify"); + // public key was bound from the signing keypair. + assert_eq!(obs.source.public_key, keypair.public_key()); + } + + #[test] + fn id_is_stable_and_parent_order_independent() { + let keypair = kp(); + let p1 = ObservationId([1u8; 32]); + let p2 = ObservationId([2u8; 32]); + let mk = |parents: Vec| { + AtomicObservation::new_signed( + source(SourceKind::RuViewRf, "rf-1"), + 42, + 0.5, + Tenant::new("t"), + parents, + b"e".to_vec(), + &keypair, + ) + .unwrap() + }; + // Parent list order must not change identity (canonicalized). + assert_eq!(mk(vec![p1, p2]).id(), mk(vec![p2, p1]).id()); + // Duplicate parents collapse. + assert_eq!(mk(vec![p1, p1, p2]).id(), mk(vec![p1, p2]).id()); + } + + #[test] + fn tampered_payload_fails_verification() { + let keypair = kp(); + let mut obs = AtomicObservation::new_signed( + source(SourceKind::NetworkTelemetry, "netprobe"), + 7, + 0.75, + Tenant::new("acme"), + vec![], + b"authentic".to_vec(), + &keypair, + ) + .unwrap(); + assert!(obs.verify()); + + obs.payload = b"forged".to_vec(); + assert!(!obs.verify(), "post-signing mutation must break the signature"); + } + + #[test] + fn confidence_out_of_range_rejected() { + let keypair = kp(); + for bad in [1.5f32, -0.1, f32::NAN] { + let r = AtomicObservation::new_signed( + source(SourceKind::UserActivity, "u"), + 0, + bad, + Tenant::new("t"), + vec![], + b"x".to_vec(), + &keypair, + ); + assert_eq!(r.unwrap_err(), ObservationError::ConfidenceOutOfRange); + } + } +} diff --git a/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs b/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs new file mode 100644 index 000000000..95fa4600e --- /dev/null +++ b/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs @@ -0,0 +1,274 @@ +//! Integration tests for the ADR-320 (PIR WP18) cross-source causal fusion +//! layer: AtomicObservation construction + signature round-trip, multi-source +//! fusion into the causal episodic graph, provenance resolution back to atomic +//! sources, confidence/tenant carry-through, tampered-observation rejection, +//! and composition with the WP4 (ADR-307) transactional ledger. +//! +//! Pattern informed by MemFuse (arXiv:2608.18704, `Darwin-Agent/Mi-Memory`); +//! explicitly distinct from the unrelated `memfuse/memfuse` OSS project. + +use rand::rngs::OsRng; +use ruvector_agent_memory::{ + AlwaysAdmitGate, AtomicObservation, CausalEpisodicGraph, FusionError, NodeRef, ObservationId, + ObservationSource, SourceKind, Tenant, TransactionalLedger, +}; +use ruvector_agent_memory::ops::{LedgerState, MemoryWitnessLog}; +use rvf_types::Ed25519Keypair; + +/// Sign an observation from a fresh per-source keypair (each source authenticates +/// under its own key, as heterogeneous real sources would). +fn observe( + kind: SourceKind, + src_id: &str, + tenant: &str, + confidence: f32, + parents: Vec, + payload: &[u8], +) -> AtomicObservation { + let keypair = Ed25519Keypair::generate(&mut OsRng); + AtomicObservation::new_signed( + ObservationSource { + kind, + id: src_id.to_string(), + public_key: [0u8; 32], // overwritten from the keypair on signing + }, + 1_700_000_000_000, + confidence, + Tenant::new(tenant), + parents, + payload.to_vec(), + &keypair, + ) + .expect("valid observation") +} + +#[test] +fn atomic_observation_signature_round_trip() { + let obs = observe( + SourceKind::AgentObservation, + "agent-alpha", + "acme", + 0.92, + vec![], + b"door opened at 14:03", + ); + assert!(obs.verify(), "a freshly signed observation must verify"); + + // Round-tripping through the content address is stable. + let id_a = obs.id(); + let id_b = obs.id(); + assert_eq!(id_a, id_b); +} + +#[test] +fn multi_source_fusion_into_graph() { + // Three heterogeneous source kinds converging into one tenant's graph. + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let rf = observe(SourceKind::RuViewRf, "rf-antenna-2", "acme", 0.80, vec![], b"rf-burst"); + let net = observe( + SourceKind::NetworkTelemetry, + "netprobe-7", + "acme", + 0.95, + vec![], + b"tcp-syn-spike", + ); + let user = observe( + SourceKind::UserActivity, + "session-42", + "acme", + 0.70, + vec![], + b"login-attempt", + ); + + let rf_id = graph.ingest(rf).unwrap(); + let net_id = graph.ingest(net).unwrap(); + let user_id = graph.ingest(user).unwrap(); + assert_eq!(graph.observation_count(), 3); + + let cluster = graph + .fuse( + &[ + NodeRef::Observation(rf_id), + NodeRef::Observation(net_id), + NodeRef::Observation(user_id), + ], + "correlated-access-event", + ) + .unwrap(); + assert_eq!(graph.cluster_count(), 1); + assert_eq!(graph.cluster(cluster).unwrap().members.len(), 3); +} + +/// THE provenance-resolution proof (ADR-320's load-bearing guarantee): a +/// derived, fused node resolves back to exactly the atomic source observations +/// it was built from — including transitively through a cluster-of-clusters — +/// and each resolved id looks up the original source evidence. +#[test] +fn provenance_resolves_derived_node_to_atomic_sources() { + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + + let rf = observe(SourceKind::RuViewRf, "rf-1", "acme", 0.6, vec![], b"rf-evidence"); + let net = observe(SourceKind::NetworkTelemetry, "net-1", "acme", 0.9, vec![], b"net-evidence"); + let user = observe(SourceKind::UserActivity, "user-1", "acme", 0.8, vec![], b"user-evidence"); + + let rf_id = graph.ingest(rf).unwrap(); + let net_id = graph.ingest(net).unwrap(); + let user_id = graph.ingest(user).unwrap(); + + // Cluster A fuses two event-layer sources; cluster B fuses cluster A with a + // third source — so B is a *derived* node two hops from some atomic sources. + let cluster_a = graph + .fuse( + &[NodeRef::Observation(rf_id), NodeRef::Observation(net_id)], + "rf+net", + ) + .unwrap(); + let cluster_b = graph + .fuse( + &[NodeRef::Cluster(cluster_a), NodeRef::Observation(user_id)], + "everything", + ) + .unwrap(); + + // Provenance of the derived node B is exactly the three atomic sources. + let provenance = graph.resolve_provenance(NodeRef::Cluster(cluster_b)).unwrap(); + let expected: std::collections::BTreeSet = + [rf_id, net_id, user_id].into_iter().collect(); + assert_eq!( + provenance, expected, + "derived node must resolve to exactly its atomic sources" + ); + + // And every resolved id traces to the original source evidence. + for id in &provenance { + let atomic = graph.observation(*id).expect("resolved id must be present"); + assert!(atomic.verify(), "resolved atomic source stays authentic"); + } + // Spot-check one payload round-trips to the original evidence bytes. + assert_eq!(graph.observation(rf_id).unwrap().payload, b"rf-evidence"); + + // Provenance of an intermediate cluster is its two atomic sources only. + let prov_a = graph.resolve_provenance(NodeRef::Cluster(cluster_a)).unwrap(); + assert_eq!(prov_a, [rf_id, net_id].into_iter().collect()); +} + +#[test] +fn confidence_and_tenant_carried_through_fusion() { + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let a = observe(SourceKind::RuViewRf, "rf", "acme", 0.9, vec![], b"a"); + let b = observe(SourceKind::NetworkTelemetry, "net", "acme", 0.4, vec![], b"b"); + let c = observe(SourceKind::UserActivity, "user", "acme", 0.7, vec![], b"c"); + + let a_id = graph.ingest(a).unwrap(); + let b_id = graph.ingest(b).unwrap(); + let c_id = graph.ingest(c).unwrap(); + + let cluster = graph + .fuse( + &[ + NodeRef::Observation(a_id), + NodeRef::Observation(b_id), + NodeRef::Observation(c_id), + ], + "mixed", + ) + .unwrap(); + let fused = graph.cluster(cluster).unwrap(); + + // Weakest-link confidence carries through fusion. + assert!((fused.confidence - 0.4).abs() < 1e-6); + // Tenant is preserved as the graph's isolation boundary. + assert_eq!(fused.tenant, Tenant::new("acme")); +} + +#[test] +fn tampered_observation_is_rejected_at_ingest() { + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let mut obs = observe(SourceKind::AgentObservation, "a", "acme", 0.9, vec![], b"authentic"); + assert!(obs.verify()); + + // A compromised or malfunctioning source flips the evidence after signing. + obs.payload = b"forged".to_vec(); + + match graph.ingest(obs) { + Err(FusionError::SignatureInvalid(_)) => {} + other => panic!("tampered observation must be rejected, got {other:?}"), + } + assert_eq!(graph.observation_count(), 0, "nothing tampered enters the graph"); +} + +/// ADR-320 §3 transactional integrity: an observation entering memory is a +/// governed transition through the WP4 (ADR-307 TARL) ledger, emitting a +/// witness record, with causal parents mapped to ledger dependency edges. +#[test] +fn governed_ingest_composes_with_wp4_ledger() { + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let mut ledger = TransactionalLedger::new(MemoryWitnessLog::default(), AlwaysAdmitGate::default()); + + let parent = observe(SourceKind::RuViewRf, "rf", "acme", 0.9, vec![], b"parent-evidence"); + let (parent_id, parent_entry) = graph + .ingest_governed(parent, &mut ledger, "fusion-layer") + .unwrap(); + + // A child observation citing the parent as causal lineage. + let child = observe( + SourceKind::AgentObservation, + "agent", + "acme", + 0.85, + vec![parent_id], + b"child-evidence", + ); + let (child_id, child_entry) = graph + .ingest_governed(child, &mut ledger, "fusion-layer") + .unwrap(); + + // Both admissions are governed ledger entries in Pending (each was + // witnessed before it mutated memory — "no witness, no mutation"). + assert_eq!(ledger.state_of(parent_entry), Some(LedgerState::Pending)); + assert_eq!(ledger.state_of(child_entry), Some(LedgerState::Pending)); + assert_eq!(graph.ledger_id(child_id), Some(child_entry)); + + // The causal parent became a ledger dependency edge. + assert_eq!(ledger.entry(child_entry).unwrap().depends_on, vec![parent_entry]); + + // Witness chain over the governed transitions is intact and non-empty. + assert!(ledger.witness_sink().verify_chain()); + assert!(!ledger.witness_sink().records.is_empty()); + + // Provenance still resolves through the governed graph. + let cluster = graph + .fuse( + &[NodeRef::Observation(parent_id), NodeRef::Observation(child_id)], + "lineage", + ) + .unwrap(); + let prov = graph.resolve_provenance(NodeRef::Cluster(cluster)).unwrap(); + assert_eq!(prov, [parent_id, child_id].into_iter().collect()); +} + +#[test] +fn governed_ingest_rejects_ungoverned_parent() { + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let mut ledger = TransactionalLedger::new(MemoryWitnessLog::default(), AlwaysAdmitGate::default()); + + // Parent admitted on the pure path (no ledger entry)... + let parent = observe(SourceKind::RuViewRf, "rf", "acme", 0.9, vec![], b"p"); + let parent_id = graph.ingest(parent).unwrap(); + + // ...so a governed child citing it has no ledger dependency to record. + let child = observe( + SourceKind::AgentObservation, + "agent", + "acme", + 0.8, + vec![parent_id], + b"c", + ); + match graph.ingest_governed(child, &mut ledger, "fusion-layer") { + Err(FusionError::ParentNotGoverned(id)) => assert_eq!(id, parent_id), + other => panic!("expected ParentNotGoverned, got {other:?}"), + } +} From d43c6d97e3fdbb3fc84c814fce9195f26ce729ae Mon Sep 17 00:00:00 2001 From: ruv Date: Thu, 20 Aug 2026 10:08:19 -0400 Subject: [PATCH 2/3] fix: make provenance resolution depth-safe (iterative traversal) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Security audit on #874 (MEDIUM): resolve_provenance/collect_provenance was recursive — cycle-safe via the visited_clusters guard but NOT depth-safe. A deep linear fuse chain (C1=fuse([obs]); C2=fuse([C1]); … CN) recursed ~N deep and overflowed the stack (reproduced N=200k → SIGABRT) on this load-bearing query. Convert to an iterative worklist (heap-allocated Vec) + the existing visited_clusters guard, so traversal depth is bounded by heap, not the call stack. Cycle-safety and error semantics are unchanged. Add a 200k-level deep-chain test asserting resolve_provenance returns the correct atomic source set without aborting (runs in <0.5s iteratively). Refs #865 #837 Co-Authored-By: claude-flow Claude-Session: https://claude.ai/code/session_012Jib2gQyJpqCoo2xYAbb4X --- crates/ruvector-agent-memory/src/fusion.rs | 35 ++++++++++++++++--- .../tests/atomic_observation_fusion.rs | 28 +++++++++++++++ 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/crates/ruvector-agent-memory/src/fusion.rs b/crates/ruvector-agent-memory/src/fusion.rs index 169b1622a..80b773a06 100644 --- a/crates/ruvector-agent-memory/src/fusion.rs +++ b/crates/ruvector-agent-memory/src/fusion.rs @@ -260,16 +260,43 @@ impl CausalEpisodicGraph { /// Resolve `node` to the set of **atomic source observations** it derives /// from — the load-bearing provenance guarantee. For an observation this is /// the observation itself; for a cluster it is the transitive union of its - /// members' atomic sources. The traversal is cycle-safe (the cluster layer - /// is acyclic by construction, but a `visited` guard makes resolution total - /// regardless). + /// members' atomic sources. + /// + /// Traversal is **iterative** (an explicit heap-allocated worklist, not the + /// call stack), so a deeply nested fuse chain (`C1=fuse([obs])`, + /// `C2=fuse([C1])`, …) is bounded by heap, not stack depth, and cannot + /// overflow the stack / abort the process on this load-bearing query. It is + /// also **cycle-safe**: the `visited_clusters` guard makes resolution total + /// even though the cluster layer is acyclic by construction. pub fn resolve_provenance( &self, node: NodeRef, ) -> Result, FusionError> { let mut atomic = BTreeSet::new(); let mut visited_clusters = BTreeSet::new(); - self.collect_provenance(node, &mut atomic, &mut visited_clusters)?; + let mut worklist: Vec = vec![node]; + while let Some(current) = worklist.pop() { + match current { + NodeRef::Observation(id) => { + if !self.observations.contains_key(&id) { + return Err(FusionError::UnknownObservation(id)); + } + atomic.insert(id); + } + NodeRef::Cluster(id) => { + if !visited_clusters.insert(id) { + continue; // already expanded; cycle-safe + } + let cluster = self + .clusters + .get(&id) + .ok_or(FusionError::UnknownCluster(id))?; + for member in &cluster.members { + worklist.push(*member); + } + } + } + } Ok(atomic) } diff --git a/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs b/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs index 95fa4600e..bcfa796dd 100644 --- a/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs +++ b/crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs @@ -249,6 +249,34 @@ fn governed_ingest_composes_with_wp4_ledger() { assert_eq!(prov, [parent_id, child_id].into_iter().collect()); } +/// Depth-safety (security audit #874, MEDIUM): a deeply nested linear fuse +/// chain (`C1=fuse([obs])`, `C2=fuse([C1])`, … `CN`) must resolve without +/// overflowing the stack. The old recursive `collect_provenance` aborted the +/// process (SIGABRT) at this depth; the iterative worklist bounds depth by heap. +#[test] +fn deep_fuse_chain_provenance_is_depth_safe() { + let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); + let obs = observe(SourceKind::RuViewRf, "rf", "acme", 0.5, vec![], b"root-evidence"); + let obs_id = graph.ingest(obs).unwrap(); + + // 200_000 levels deep — far past what the recursive version could survive, + // but linear and fast iteratively. + const DEPTH: usize = 200_000; + let mut current = graph.fuse(&[NodeRef::Observation(obs_id)], "level-0").unwrap(); + for level in 1..DEPTH { + current = graph + .fuse(&[NodeRef::Cluster(current)], format!("level-{level}")) + .unwrap(); + } + + // The load-bearing provenance query resolves to exactly the one atomic + // source, without aborting. + let provenance = graph.resolve_provenance(NodeRef::Cluster(current)).unwrap(); + assert_eq!(provenance, [obs_id].into_iter().collect()); + // Weakest-link confidence propagated unchanged through the whole chain. + assert!((graph.cluster(current).unwrap().confidence - 0.5).abs() < 1e-6); +} + #[test] fn governed_ingest_rejects_ungoverned_parent() { let mut graph = CausalEpisodicGraph::new(Tenant::new("acme")); From d652936d1256f0537b212c088af8beef2e3029b7 Mon Sep 17 00:00:00 2001 From: ruv Date: Thu, 20 Aug 2026 10:08:51 -0400 Subject: [PATCH 3/3] refactor: remove now-dead recursive collect_provenance helper The iterative resolve_provenance fully replaces it; drop the unused method (cleared the dead_code warning). Co-Authored-By: claude-flow Claude-Session: https://claude.ai/code/session_012Jib2gQyJpqCoo2xYAbb4X --- crates/ruvector-agent-memory/src/fusion.rs | 29 ---------------------- 1 file changed, 29 deletions(-) diff --git a/crates/ruvector-agent-memory/src/fusion.rs b/crates/ruvector-agent-memory/src/fusion.rs index 80b773a06..7e4a9b7b6 100644 --- a/crates/ruvector-agent-memory/src/fusion.rs +++ b/crates/ruvector-agent-memory/src/fusion.rs @@ -377,35 +377,6 @@ impl CausalEpisodicGraph { } } - fn collect_provenance( - &self, - node: NodeRef, - atomic: &mut BTreeSet, - visited_clusters: &mut BTreeSet, - ) -> Result<(), FusionError> { - match node { - NodeRef::Observation(id) => { - if !self.observations.contains_key(&id) { - return Err(FusionError::UnknownObservation(id)); - } - atomic.insert(id); - Ok(()) - } - NodeRef::Cluster(id) => { - if !visited_clusters.insert(id) { - return Ok(()); // already expanded; cycle-safe - } - let cluster = self - .clusters - .get(&id) - .ok_or(FusionError::UnknownCluster(id))?; - for member in &cluster.members { - self.collect_provenance(*member, atomic, visited_clusters)?; - } - Ok(()) - } - } - } } #[cfg(test)]