Merge pull request #874 from ruvnet/feat/pir-wp18-atomic-observation

feat: AtomicObservation + causal episodic graph (PIR WP18, ADR-320)

Cargo.lock

@@ -8793,6 +8793,7 @@ version = "0.1.0"
 dependencies = [
  "rand 0.8.6",
  "ruvector-proof-gate",
+ "rvf-types",
  "serde",
  "serde_json",
 ]

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 }

crates/ruvector-agent-memory/src/fusion.rs

@@ -0,0 +1,436 @@
+//! 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<NodeRef>,
+    /// 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<ObservationId, AtomicObservation>,
+    clusters: BTreeMap<ClusterId, FusedCluster>,
+    /// Ledger entry ids for observations admitted through [`Self::ingest_governed`].
+    ledger_ids: BTreeMap<ObservationId, u64>,
+    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<ObservationId, FusionError> {
+        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<S: WitnessSink, G: ProofGate>(
+        &mut self,
+        obs: AtomicObservation,
+        ledger: &mut TransactionalLedger<S, G>,
+        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<String>,
+    ) -> Result<ClusterId, FusionError> {
+        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.
+    ///
+    /// 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<BTreeSet<ObservationId>, FusionError> {
+        let mut atomic = BTreeSet::new();
+        let mut visited_clusters = BTreeSet::new();
+        let mut worklist: Vec<NodeRef> = 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)
+    }
+
+    /// 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<u64> {
+        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<ObservationId, FusionError> {
+        // 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<f32, FusionError> {
+        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)),
+        }
+    }
+
+}
+
+#[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<ObservationId>,
+        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:?}"),
+        }
+    }
+}

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,

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<String>) -> 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<ObservationId>,
+    /// Original evidence body this observation records; provenance resolves
+    /// back to these bytes.
+    pub payload: Vec<u8>,
+    /// 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<ObservationId>,
+        payload: Vec<u8>,
+        keypair: &Ed25519Keypair,
+    ) -> Result<Self, ObservationError> {
+        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<u8> {
+        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<u8>, 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<ObservationId>| {
+            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);
+        }
+    }
+}

crates/ruvector-agent-memory/tests/atomic_observation_fusion.rs

@@ -0,0 +1,302 @@
+//! 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<ObservationId>,
+    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<ObservationId> =
+        [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());
+}
+
+/// 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"));
+    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:?}"),
+    }
+}