klieo-memory-graph-rag 2.1.0

//! `ProvenanceKnowledgeGraph` — wraps any
//! [`klieo_memory_graph::KnowledgeGraph`] with a
//! [`klieo_provenance::ProvenanceRepository`] so every successful
//! `index()` mints a signed
//! [`klieo_provenance::ChainEntry`] and every `recall_paths()`
//! attaches the matching entry per [`klieo_memory_graph::PathHop`].
//!
//! See ADR-038 (`docs/adr/adr-038-provenance-graph-bridge.md`) for
//! the design decision and ADR-037
//! (`adr-037-chain-canonicalisation-v2.md`) for the v2 chain entry
//! shape this crate emits.
//!
//! ### Wire
//!
//! ```text
//! GraphAwareLongTerm
//!   └── graph: Arc<dyn KnowledgeGraph>
//!       └── ProvenanceKnowledgeGraph
//!           ├── inner: Arc<dyn KnowledgeGraph>  (Neo4j or InMemory)
//!           └── repository: Arc<dyn ProvenanceRepository>
//! ```
//!
//! `EpisodeProjector` from M3 also drops in here unchanged — wrapping
//! its `graph` handle means every projected episode auto-mints a
//! chain entry with no projector code change.
//!
//! ### Failure modes
//!
//! - Inner `index()` fails → return the error; **no** chain entry
//!   minted. (Vector remains authoritative per the M2 contract.)
//! - Inner `index()` succeeds, `repository.append()` fails →
//!   `tracing::warn!` with `scope` + `fact_id`; return success. The
//!   index landed; the audit chain has a documented gap.
//! - `forget()` emits a `Custom("MemoryForget")` event BEFORE
//!   delegating to `inner.forget()` so the audit chain records the
//!   intent even if the deletion fails downstream.

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use klieo_core::error::MemoryError;
use klieo_core::ids::FactId;
use klieo_core::memory::Scope;
use klieo_memory_graph::{ChainEntry, EntityRef, IndexEntry, KnowledgeGraph, RetrievalPath};
use klieo_provenance::{ProvenanceEvent, ProvenanceEventKind, ProvenanceRepository};
use lru::LruCache;
use parking_lot::Mutex;
use sha2::{Digest, Sha256};
use std::num::NonZeroUsize;
use std::sync::Arc;

/// Event kind label used in the `Custom` variant when `forget` is
/// called. Stable string so downstream verifiers can grep for it.
pub const FORGET_EVENT_LABEL: &str = "MemoryForget";

/// Default LRU capacity applied by [`ProvenanceKnowledgeGraph::new`].
/// Cache eviction surfaces `chain_entry = None` in `recall_paths`;
/// the [`ProvenanceRepository`] remains the source of truth. Size
/// up via [`ProvenanceKnowledgeGraph::with_fact_index_capacity`].
pub const DEFAULT_FACT_INDEX_CAPACITY: usize = 10_000;

/// `KnowledgeGraph` wrapper that emits a
/// [`klieo_provenance::ChainEntry`] on every successful `index()`
/// and attaches the matching entry to each
/// [`klieo_memory_graph::PathHop`] returned by `recall_paths()`.
#[non_exhaustive]
pub struct ProvenanceKnowledgeGraph {
    inner: Arc<dyn KnowledgeGraph>,
    repository: Arc<dyn ProvenanceRepository>,
    actor: String,
    fact_index: Mutex<LruCache<(String, FactId), ChainEntry>>,
}

impl ProvenanceKnowledgeGraph {
    /// Wrap `inner` with provenance emission against `repository`.
    /// `actor` is recorded on every emitted [`ProvenanceEvent`]
    /// (e.g. `"graph-projector"` for episode-driven indexing or
    /// `"graph-aware-long-term"` for `GraphAwareLongTerm`).
    ///
    /// See [`Self::with_fact_index_capacity`] to size the LRU
    /// cache for higher per-process ingest volume.
    pub fn new(
        inner: Arc<dyn KnowledgeGraph>,
        repository: Arc<dyn ProvenanceRepository>,
        actor: impl Into<String>,
    ) -> Self {
        let cap = NonZeroUsize::new(DEFAULT_FACT_INDEX_CAPACITY)
            .expect("DEFAULT_FACT_INDEX_CAPACITY is non-zero");
        Self::with_fact_index_capacity(inner, repository, actor, cap)
    }

    /// `capacity` is `NonZeroUsize` so the zero-capacity invariant
    /// is enforced at the call site rather than via runtime panic.
    pub fn with_fact_index_capacity(
        inner: Arc<dyn KnowledgeGraph>,
        repository: Arc<dyn ProvenanceRepository>,
        actor: impl Into<String>,
        capacity: NonZeroUsize,
    ) -> Self {
        Self {
            inner,
            repository,
            actor: actor.into(),
            fact_index: Mutex::new(LruCache::new(capacity)),
        }
    }
}

#[async_trait]
impl KnowledgeGraph for ProvenanceKnowledgeGraph {
    async fn index(
        &self,
        scope: Scope,
        fact_id: &FactId,
        entities: &[EntityRef],
        text: &str,
        valid_from: Option<DateTime<Utc>>,
    ) -> Result<(), MemoryError> {
        // Single-fact path delegates to `index_many` so chain-entry
        // emission lives in one place. Same fail-fast semantics:
        // inner failure short-circuits before any chain entry is
        // minted (graph / vector is source of truth; chain is a
        // side-effect ledger).
        let entry = IndexEntry::new(fact_id.clone(), entities.to_vec(), text, valid_from);
        self.index_many(scope, std::slice::from_ref(&entry)).await
    }

    async fn index_many(&self, scope: Scope, batch: &[IndexEntry]) -> Result<(), MemoryError> {
        self.inner.index_many(scope.clone(), batch).await?;

        // Mint chain entries only for entries the inner backend
        // actually committed. `KnowledgeGraph::index_many`'s
        // skip-empty contract means empty-entity entries are no-ops
        // — minting an audit ledger for them would diverge from
        // graph state.
        let scope_key = scope_key(&scope);
        for entry in batch.iter().filter(|e| !e.entities.is_empty()) {
            let event = memory_write_event(
                &self.actor,
                &entry.fact_id,
                &entry.entities,
                entry.valid_from,
            );
            match self.repository.append(&scope_key, event).await {
                Ok(chain_entry) => {
                    self.fact_index
                        .lock()
                        .put((scope_key.clone(), entry.fact_id.clone()), chain_entry);
                }
                Err(error) => {
                    tracing::warn!(
                        operation = "index_many",
                        scope = %scope_key,
                        fact_id = %entry.fact_id,
                        %error,
                        "provenance chain append failed in batch; graph index succeeded — audit gap"
                    );
                }
            }
        }
        Ok(())
    }

    async fn neighbors(
        &self,
        scope: &Scope,
        entities: &[EntityRef],
    ) -> Result<Vec<FactId>, MemoryError> {
        self.inner.neighbors(scope, entities).await
    }

    async fn recall_paths(
        &self,
        scope: &Scope,
        entities: &[EntityRef],
    ) -> Result<Vec<RetrievalPath>, MemoryError> {
        let mut paths = self.inner.recall_paths(scope, entities).await?;
        let scope_key = scope_key(scope);
        let mut cache = self.fact_index.lock();
        for path in paths.iter_mut() {
            for hop in path.hops.iter_mut() {
                if let Some(entry) = cache.get(&(scope_key.clone(), hop.fact_id.clone())) {
                    hop.chain_entry = Some(entry.clone());
                }
            }
        }
        Ok(paths)
    }

    async fn forget(&self, scope: &Scope, fact_id: &FactId) -> Result<(), MemoryError> {
        // Emit the audit event BEFORE delegating so the chain records
        // intent even if the inner deletion fails downstream.
        let event = forget_event(&self.actor, fact_id);
        let scope_key = scope_key(scope);
        if let Err(error) = self.repository.append(&scope_key, event).await {
            tracing::warn!(
                scope = %scope_key,
                fact_id = %fact_id,
                %error,
                "provenance forget event append failed; proceeding with inner.forget"
            );
        }
        self.inner.forget(scope, fact_id).await?;
        self.fact_index.lock().pop(&(scope_key, fact_id.clone()));
        Ok(())
    }
}

fn scope_key(scope: &Scope) -> String {
    match scope {
        Scope::Workspace(s) => format!("workspace:{s}"),
        Scope::Agent(s) => format!("agent:{s}"),
        Scope::Global => "global".to_string(),
    }
}

fn entities_payload_hash(fact_id: &FactId, entities: &[EntityRef]) -> String {
    let canonical = serde_json::json!({
        "fact_id": fact_id.to_string(),
        "entities": entities,
    });
    let bytes = serde_json::to_vec(&canonical).expect("EntityRef + FactId serialise");
    let mut hasher = Sha256::new();
    hasher.update(&bytes);
    hex::encode(hasher.finalize())
}

fn memory_write_event(
    actor: &str,
    fact_id: &FactId,
    entities: &[EntityRef],
    valid_from: Option<DateTime<Utc>>,
) -> ProvenanceEvent {
    let metadata = serde_json::json!({
        "entities": entities,
        "valid_from": valid_from.map(|t| t.to_rfc3339()),
    });
    ProvenanceEvent {
        kind: ProvenanceEventKind::MemoryWrite,
        actor: actor.to_string(),
        resource_ref: fact_id.to_string(),
        payload_hash_hex: entities_payload_hash(fact_id, entities),
        metadata,
    }
}

fn forget_event(actor: &str, fact_id: &FactId) -> ProvenanceEvent {
    let payload_hash = {
        let bytes = fact_id.to_string();
        let mut hasher = Sha256::new();
        hasher.update(bytes.as_bytes());
        hex::encode(hasher.finalize())
    };
    ProvenanceEvent {
        kind: ProvenanceEventKind::Custom(FORGET_EVENT_LABEL.to_string()),
        actor: actor.to_string(),
        resource_ref: fact_id.to_string(),
        payload_hash_hex: payload_hash,
        metadata: serde_json::Value::Null,
    }
}

#[cfg(test)]
// Test bodies hand-construct `&[entity.clone()]` slices for readability —
// `std::slice::from_ref` is not clearer in spec-style assertions.
#[allow(clippy::cloned_ref_to_slice_refs)]
mod tests {
    use super::*;
    use klieo_memory_graph::{EntityType, InMemoryGraph};
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// In-memory repository — single linear chain per scope, suitable
    /// for tests. Production paths wire the Sqlite-backed
    /// [`ProvenanceRepository`] from `klieo-ops`.
    #[derive(Default)]
    struct InMemoryRepo {
        chains: Mutex<HashMap<String, klieo_provenance::ProvenanceChain>>,
        append_calls: AtomicUsize,
    }

    #[async_trait]
    impl ProvenanceRepository for InMemoryRepo {
        async fn append(
            &self,
            scope: &str,
            event: ProvenanceEvent,
        ) -> Result<ChainEntry, klieo_provenance::ProvenanceError> {
            self.append_calls.fetch_add(1, Ordering::SeqCst);
            let mut chains = self.chains.lock();
            let chain = chains
                .entry(scope.to_string())
                .or_insert_with(|| klieo_provenance::ProvenanceChain::new(scope.to_string()));
            Ok(chain.append(event))
        }

        async fn head(
            &self,
            scope: &str,
        ) -> Result<Option<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(self
                .chains
                .lock()
                .get(scope)
                .and_then(|c| c.entries.last().cloned()))
        }

        async fn list_range(
            &self,
            scope: &str,
            from_sequence: u64,
            to_sequence: u64,
        ) -> Result<Vec<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(self
                .chains
                .lock()
                .get(scope)
                .map(|c| {
                    c.entries
                        .iter()
                        .filter(|e| e.sequence >= from_sequence && e.sequence <= to_sequence)
                        .cloned()
                        .collect()
                })
                .unwrap_or_default())
        }

        async fn list_by_time(
            &self,
            _scope: &str,
            _from: DateTime<Utc>,
            _to: DateTime<Utc>,
        ) -> Result<Vec<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(Vec::new())
        }
    }

    /// Repository that always errors on `append`. Used to prove the
    /// graph-index path keeps succeeding when provenance is degraded.
    struct AlwaysFailRepo;
    #[async_trait]
    impl ProvenanceRepository for AlwaysFailRepo {
        async fn append(
            &self,
            _scope: &str,
            _event: ProvenanceEvent,
        ) -> Result<ChainEntry, klieo_provenance::ProvenanceError> {
            Err(klieo_provenance::ProvenanceError::Repository(
                "simulated repository failure".into(),
            ))
        }
        async fn head(
            &self,
            _scope: &str,
        ) -> Result<Option<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(None)
        }
        async fn list_range(
            &self,
            _scope: &str,
            _from: u64,
            _to: u64,
        ) -> Result<Vec<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(Vec::new())
        }
        async fn list_by_time(
            &self,
            _scope: &str,
            _from: DateTime<Utc>,
            _to: DateTime<Utc>,
        ) -> Result<Vec<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(Vec::new())
        }
    }

    /// Succeeds on the first `succeed_until` appends, fails after.
    struct FailAfterNAppendsRepo {
        chains: Mutex<HashMap<String, klieo_provenance::ProvenanceChain>>,
        call_count: AtomicUsize,
        succeed_until: usize,
    }

    impl FailAfterNAppendsRepo {
        fn new(succeed_until: usize) -> Self {
            Self {
                chains: Mutex::new(HashMap::new()),
                call_count: AtomicUsize::new(0),
                succeed_until,
            }
        }
    }

    #[async_trait]
    impl ProvenanceRepository for FailAfterNAppendsRepo {
        async fn append(
            &self,
            scope: &str,
            event: ProvenanceEvent,
        ) -> Result<ChainEntry, klieo_provenance::ProvenanceError> {
            let n = self.call_count.fetch_add(1, Ordering::SeqCst);
            if n >= self.succeed_until {
                return Err(klieo_provenance::ProvenanceError::Repository(
                    "simulated mid-batch failure".into(),
                ));
            }
            let mut chains = self.chains.lock();
            let chain = chains
                .entry(scope.to_string())
                .or_insert_with(|| klieo_provenance::ProvenanceChain::new(scope.to_string()));
            Ok(chain.append(event))
        }
        async fn head(
            &self,
            _scope: &str,
        ) -> Result<Option<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(None)
        }
        async fn list_range(
            &self,
            _scope: &str,
            _from: u64,
            _to: u64,
        ) -> Result<Vec<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(Vec::new())
        }
        async fn list_by_time(
            &self,
            _scope: &str,
            _from: DateTime<Utc>,
            _to: DateTime<Utc>,
        ) -> Result<Vec<ChainEntry>, klieo_provenance::ProvenanceError> {
            Ok(Vec::new())
        }
    }

    fn scope() -> Scope {
        Scope::Workspace("prov-test".into())
    }

    fn ticket(name: &str) -> EntityRef {
        EntityRef::new(EntityType::Ticket, name)
    }

    #[tokio::test]
    async fn index_mints_chain_entry_on_inner_success() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo.clone(),
            "test-actor",
        );

        let fid = FactId::new("f1");
        wrapper
            .index(scope(), &fid, &[ticket("T-1")], "", None)
            .await
            .unwrap();

        let entries = repo
            .list_range(&scope_key(&scope()), 0, u64::MAX)
            .await
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].event.resource_ref, "f1");
        assert!(matches!(
            entries[0].event.kind,
            ProvenanceEventKind::MemoryWrite
        ));
        assert_eq!(entries[0].chain_version, 2);
    }

    #[tokio::test]
    async fn recall_paths_attaches_chain_entry_per_hop() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo.clone(),
            "test-actor",
        );

        let ticket_entity = ticket("T-7");
        wrapper
            .index(
                scope(),
                &FactId::new("attached"),
                &[ticket_entity.clone()],
                "",
                None,
            )
            .await
            .unwrap();

        let paths = wrapper
            .recall_paths(&scope(), &[ticket_entity.clone()])
            .await
            .unwrap();
        assert_eq!(paths.len(), 1);
        let hop = &paths[0].hops[0];
        assert_eq!(hop.fact_id, FactId::new("attached"));
        let entry = hop
            .chain_entry
            .as_ref()
            .expect("wrapper must attach chain_entry on recall_paths");
        assert_eq!(entry.event.resource_ref, "attached");
        assert!(matches!(entry.event.kind, ProvenanceEventKind::MemoryWrite));
    }

    #[tokio::test]
    async fn recall_paths_emits_none_for_unrecorded_fact() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo,
            "test-actor",
        );

        // Index DIRECTLY on inner so the wrapper never sees the call and
        // never caches a chain entry for the fact_id.
        inner
            .index(scope(), &FactId::new("orphan"), &[ticket("T-9")], "", None)
            .await
            .unwrap();

        let paths = wrapper
            .recall_paths(&scope(), &[ticket("T-9")])
            .await
            .unwrap();
        let hop = &paths[0].hops[0];
        assert!(
            hop.chain_entry.is_none(),
            "facts indexed before the wrapper was attached must surface chain_entry=None"
        );
    }

    #[tokio::test]
    async fn repository_failure_does_not_fail_index() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(AlwaysFailRepo);
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo,
            "test-actor",
        );

        wrapper
            .index(
                scope(),
                &FactId::new("graceful"),
                &[ticket("T-1")],
                "",
                None,
            )
            .await
            .expect("inner success must not be cancelled by repository failure");

        let neighbors = inner.neighbors(&scope(), &[ticket("T-1")]).await.unwrap();
        assert!(neighbors.contains(&FactId::new("graceful")));
    }

    #[tokio::test]
    async fn inner_index_failure_skips_chain_emission() {
        struct AlwaysFailGraph;
        #[async_trait]
        impl KnowledgeGraph for AlwaysFailGraph {
            async fn index(
                &self,
                _scope: Scope,
                _fact_id: &FactId,
                _entities: &[EntityRef],
                _text: &str,
                _valid_from: Option<DateTime<Utc>>,
            ) -> Result<(), MemoryError> {
                Err(MemoryError::Store("simulated graph failure".into()))
            }
            async fn neighbors(
                &self,
                _scope: &Scope,
                _entities: &[EntityRef],
            ) -> Result<Vec<FactId>, MemoryError> {
                Ok(Vec::new())
            }
        }

        let inner: Arc<dyn KnowledgeGraph> = Arc::new(AlwaysFailGraph);
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(inner, repo.clone(), "test-actor");

        wrapper
            .index(scope(), &FactId::new("doomed"), &[ticket("T-1")], "", None)
            .await
            .expect_err("inner failure must propagate");
        assert_eq!(
            repo.append_calls.load(Ordering::SeqCst),
            0,
            "no chain entry must be emitted when inner.index fails"
        );
    }

    #[tokio::test]
    async fn forget_emits_audit_event_then_calls_inner() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo.clone(),
            "test-actor",
        );

        let fid = FactId::new("doomed");
        wrapper
            .index(scope(), &fid, &[ticket("T-1")], "", None)
            .await
            .unwrap();

        wrapper.forget(&scope(), &fid).await.unwrap();

        let entries = repo
            .list_range(&scope_key(&scope()), 0, u64::MAX)
            .await
            .unwrap();
        assert_eq!(entries.len(), 2, "MemoryWrite + MemoryForget");
        let forget_entry = &entries[1];
        assert!(
            matches!(&forget_entry.event.kind, ProvenanceEventKind::Custom(s) if s == FORGET_EVENT_LABEL)
        );

        let neighbors = inner.neighbors(&scope(), &[ticket("T-1")]).await.unwrap();
        assert!(
            !neighbors.contains(&fid),
            "inner forget must drop the fact's index entries"
        );

        // Cache-cleanup contract: after forget(), recall_paths through
        // the wrapper must not surface the dropped fact_id with a
        // dangling chain_entry. Inner is empty for this fact, so the
        // path set is empty regardless — the assertion pins that the
        // cache entry's stale ChainEntry is not leaked via some other
        // code path.
        let paths_after_forget = wrapper
            .recall_paths(&scope(), &[ticket("T-1")])
            .await
            .unwrap();
        assert!(
            paths_after_forget.is_empty(),
            "recall_paths must return no hops referencing a forgotten fact: {paths_after_forget:?}",
        );
    }

    #[tokio::test]
    async fn forget_propagates_inner_failure_after_emitting_audit_event() {
        struct ForgetFailGraph {
            forget_calls: AtomicUsize,
        }
        #[async_trait]
        impl KnowledgeGraph for ForgetFailGraph {
            async fn index(
                &self,
                _scope: Scope,
                _fact_id: &FactId,
                _entities: &[EntityRef],
                _text: &str,
                _valid_from: Option<DateTime<Utc>>,
            ) -> Result<(), MemoryError> {
                Ok(())
            }
            async fn neighbors(
                &self,
                _scope: &Scope,
                _entities: &[EntityRef],
            ) -> Result<Vec<FactId>, MemoryError> {
                Ok(Vec::new())
            }
            async fn forget(&self, _scope: &Scope, _fact_id: &FactId) -> Result<(), MemoryError> {
                self.forget_calls.fetch_add(1, Ordering::SeqCst);
                Err(MemoryError::Store("simulated inner forget failure".into()))
            }
        }

        let inner_stub = Arc::new(ForgetFailGraph {
            forget_calls: AtomicUsize::new(0),
        });
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner_stub.clone() as Arc<dyn KnowledgeGraph>,
            repo.clone(),
            "test-actor",
        );

        let err = wrapper
            .forget(&scope(), &FactId::new("doomed"))
            .await
            .expect_err("inner forget failure must propagate");
        assert!(
            matches!(err, MemoryError::Store(s) if s.contains("simulated inner forget failure"))
        );
        assert_eq!(inner_stub.forget_calls.load(Ordering::SeqCst), 1);

        // Audit event must still have been emitted BEFORE inner.forget
        // failed — the "intent before delegation" contract from ADR-038.
        let entries = repo
            .list_range(&scope_key(&scope()), 0, u64::MAX)
            .await
            .unwrap();
        assert_eq!(
            entries.len(),
            1,
            "MemoryForget audit event must be appended even when inner.forget fails"
        );
        assert!(matches!(
            &entries[0].event.kind,
            ProvenanceEventKind::Custom(s) if s == FORGET_EVENT_LABEL
        ));
    }

    #[test]
    fn scope_key_distinguishes_variants() {
        assert_eq!(scope_key(&Scope::Workspace("ws".into())), "workspace:ws");
        assert_eq!(scope_key(&Scope::Agent("a".into())), "agent:a");
        assert_eq!(scope_key(&Scope::Global), "global");
    }

    #[test]
    fn entities_payload_hash_is_deterministic() {
        let fid = FactId::new("f");
        let a = entities_payload_hash(&fid, &[ticket("T-1")]);
        let b = entities_payload_hash(&fid, &[ticket("T-1")]);
        assert_eq!(a, b);
        assert_eq!(a.len(), 64, "sha256 hex is 64 chars");
    }

    #[test]
    fn entities_payload_hash_distinguishes_inputs() {
        let fid = FactId::new("f");
        let a = entities_payload_hash(&fid, &[ticket("T-1")]);
        let b = entities_payload_hash(&fid, &[ticket("T-2")]);
        assert_ne!(a, b, "different entity sets must produce different hashes");

        let c = entities_payload_hash(&FactId::new("g"), &[ticket("T-1")]);
        assert_ne!(a, c, "different fact_ids must produce different hashes");
    }

    #[tokio::test]
    async fn index_many_mints_one_chain_entry_per_entry() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo.clone(),
            "test-actor",
        );

        let batch = vec![
            IndexEntry::new(FactId::new("f1"), vec![ticket("T-1")], "", None),
            IndexEntry::new(FactId::new("f2"), vec![ticket("T-2")], "", None),
            IndexEntry::new(FactId::new("f3"), vec![ticket("T-3")], "", None),
        ];

        wrapper.index_many(scope(), &batch).await.unwrap();

        let entries = repo
            .list_range(&scope_key(&scope()), 0, u64::MAX)
            .await
            .unwrap();
        assert_eq!(entries.len(), 3, "one ChainEntry per IndexEntry");
        for (i, entry) in entries.iter().enumerate() {
            assert_eq!(entry.event.resource_ref, format!("f{}", i + 1));
            assert!(matches!(entry.event.kind, ProvenanceEventKind::MemoryWrite));
        }
    }

    #[tokio::test]
    async fn index_many_inner_failure_skips_all_chain_emission() {
        // Single inner.index_many failure must short-circuit BEFORE
        // any chain entries are minted. Mirrors the single-fact
        // `inner_index_failure_skips_chain_emission` contract.
        struct AlwaysFailGraph;
        #[async_trait]
        impl KnowledgeGraph for AlwaysFailGraph {
            async fn index(
                &self,
                _scope: Scope,
                _fact_id: &FactId,
                _entities: &[EntityRef],
                _text: &str,
                _valid_from: Option<DateTime<Utc>>,
            ) -> Result<(), MemoryError> {
                Err(MemoryError::Store("simulated graph failure".into()))
            }
            async fn neighbors(
                &self,
                _scope: &Scope,
                _entities: &[EntityRef],
            ) -> Result<Vec<FactId>, MemoryError> {
                Ok(Vec::new())
            }
        }

        let inner: Arc<dyn KnowledgeGraph> = Arc::new(AlwaysFailGraph);
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(inner, repo.clone(), "test-actor");

        let batch = vec![
            IndexEntry::new(FactId::new("f1"), vec![ticket("T-1")], "", None),
            IndexEntry::new(FactId::new("f2"), vec![ticket("T-2")], "", None),
        ];
        wrapper
            .index_many(scope(), &batch)
            .await
            .expect_err("inner failure must propagate");
        assert_eq!(
            repo.append_calls.load(Ordering::SeqCst),
            0,
            "no chain entries must be emitted when inner.index_many fails"
        );
    }

    #[tokio::test]
    async fn index_many_mid_batch_repo_failure_logs_and_continues() {
        // Repository fails on the 2nd append. Inner succeeds for
        // every entry. Batch must return Ok (warn-only semantics);
        // entry 1 lands in fact_index, entry 2 is absent.
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(FailAfterNAppendsRepo::new(1));
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo,
            "test-actor",
        );

        let batch = vec![
            IndexEntry::new(FactId::new("f1"), vec![ticket("T-1")], "", None),
            IndexEntry::new(FactId::new("f2"), vec![ticket("T-2")], "", None),
        ];
        wrapper
            .index_many(scope(), &batch)
            .await
            .expect("per-entry append failure must not fail the batch");

        let scope_key = scope_key(&scope());
        let cache = wrapper.fact_index.lock();
        assert!(
            cache.contains(&(scope_key.clone(), FactId::new("f1"))),
            "entry 1 succeeded — must be in fact_index"
        );
        assert!(
            !cache.contains(&(scope_key, FactId::new("f2"))),
            "entry 2 append failed — must NOT be in fact_index"
        );
    }

    #[tokio::test]
    async fn index_many_skips_empty_entry_chain_emission() {
        // Empty-entity entries are no-ops on the inner backend per
        // the KnowledgeGraph::index_many skip-empty contract; the
        // provenance wrapper must NOT mint chain entries for them
        // (audit ledger ↔ graph state alignment).
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::new(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo.clone(),
            "test-actor",
        );

        let batch = vec![
            IndexEntry::new(FactId::new("f1"), vec![ticket("T-1")], "", None),
            IndexEntry::new(FactId::new("empty"), Vec::new(), "", None),
            IndexEntry::new(FactId::new("f3"), vec![ticket("T-3")], "", None),
        ];
        wrapper.index_many(scope(), &batch).await.unwrap();

        let entries = repo
            .list_range(&scope_key(&scope()), 0, u64::MAX)
            .await
            .unwrap();
        assert_eq!(entries.len(), 2, "empty-entity entry must skip chain mint");
        assert_eq!(entries[0].event.resource_ref, "f1");
        assert_eq!(entries[1].event.resource_ref, "f3");
    }

    #[tokio::test]
    async fn fact_index_evicts_least_recently_used_at_capacity() {
        let inner = Arc::new(InMemoryGraph::default());
        let repo = Arc::new(InMemoryRepo::default());
        let wrapper = ProvenanceKnowledgeGraph::with_fact_index_capacity(
            inner.clone() as Arc<dyn KnowledgeGraph>,
            repo,
            "test-actor",
            NonZeroUsize::new(2).unwrap(),
        );

        for n in 1..=3 {
            wrapper
                .index(
                    scope(),
                    &FactId::new(format!("f{n}")),
                    &[ticket(&format!("T-{n}"))],
                    "",
                    None,
                )
                .await
                .unwrap();
        }

        // Cache state: f1 evicted; f2, f3 retained.
        let scope_key = scope_key(&scope());
        {
            let cache = wrapper.fact_index.lock();
            assert_eq!(cache.len(), 2, "cache must respect capacity");
            assert!(
                !cache.contains(&(scope_key.clone(), FactId::new("f1"))),
                "oldest entry (f1) must be evicted"
            );
            assert!(cache.contains(&(scope_key.clone(), FactId::new("f2"))));
            assert!(cache.contains(&(scope_key, FactId::new("f3"))));
        }

        // Behavioural contract: `recall_paths` against the evicted
        // fact's entity surfaces `chain_entry = None`. Repository
        // stays the source of truth.
        let paths_evicted = wrapper
            .recall_paths(&scope(), &[ticket("T-1")])
            .await
            .unwrap();
        assert!(
            !paths_evicted.is_empty(),
            "InMemoryGraph must surface the f1 hop even after cache eviction"
        );
        for path in &paths_evicted {
            for hop in &path.hops {
                if hop.fact_id == FactId::new("f1") {
                    assert!(
                        hop.chain_entry.is_none(),
                        "evicted fact's hop must surface chain_entry = None; got {:?}",
                        hop.chain_entry
                    );
                }
            }
        }

        // Retained fact still attaches chain_entry.
        let paths_retained = wrapper
            .recall_paths(&scope(), &[ticket("T-3")])
            .await
            .unwrap();
        assert!(paths_retained.iter().any(|p| p
            .hops
            .iter()
            .any(|h| h.fact_id == FactId::new("f3") && h.chain_entry.is_some())));
    }
}