car_server_core/assistant/

memory.rs

//! Durable memory for the assistant: `remember` and `recall`.
//!
//! Backed by CAR's own graph memory ([`car_memgine::MemgineEngine`]) rather than
//! a bespoke store — this is the capability the assistant exists to showcase.
//! Remembered facts are persisted to `~/.car/memory/assistant.json` and
//! re-ingested on startup, so the assistant remembers across sessions and
//! process restarts.
//!
//! Recall runs the memgine's **full** retrieval path (`build_context`), not the
//! Fast one. Fast mode skips exactly the parts that make this a graph rather
//! than a list: skill lookup, PPR-based fact scoring, inline repair of stale
//! dependents, and known-unknowns extraction. The assistant was on Fast, so the
//! flagship agent advertised graph memory while doing keyword recall over flat
//! notes — the differentiator was bypassed on the one path users actually hit.
//!
//! Full is affordable here specifically because this engine is built with
//! `MemgineEngine::new(None)` and never has inference attached: the one
//! genuinely expensive step Fast avoids is the embedding flush, and that is
//! guarded by `self.inference.is_some()`, so it no-ops either way. What Full
//! adds is in-memory graph work — no I/O, no model call. If inference is ever
//! wired to this engine, re-measure before assuming that still holds.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use car_engine::ToolExecutor;
use car_eventlog::Event;
use car_memgine::{
    MemgineEngine, ProactiveMaintenanceRequest, ProactiveMemoryDecision, ProactiveMemoryRequest,
};
use serde_json::{json, Value};

/// The on-disk note format and the memory taxonomy now live in
/// [`car_memgine::note_store`], so the MCP server can read and write the SAME
/// store this assistant does. Re-exported here because `NoteKind` is part of
/// this module's published surface (`car_server_core::assistant::NoteKind`)
/// and callers should not have to care that it moved.
pub use car_memgine::note_store::{ingest, Note, NoteKind};

/// One fact as it crosses the sync wire.
///
/// `kind` rides along because it is not decoration — a `Preference` is saved as
/// a memgine *constraint* and therefore lands in the always-included Active
/// Constraints layer, while a `Fact` surfaces only when a recall query matches
/// it. Reconstructing every peer-synced note as a plain fact (which is what
/// happened before car#665) silently downgraded a standing rule to something
/// the assistant would only sometimes see, on the second device and not the
/// first, with nothing anywhere reporting an error.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SyncedFact {
    pub subject: String,
    pub body: String,
    pub kind: NoteKind,
}

/// A sink that mirrors remembered facts into CAR's synced oplog
/// (`Surface::Knowledge`) so the assistant's memory can converge across a user's
/// devices. Supplied ONLY in the daemon-attached `car do --serve` process, which
/// reaches the daemon's single oplog owner over WS; `None` for one-shot `car do`,
/// so the standalone CLI never becomes a second writer to the lock-guarded oplog.
/// The write is best-effort: a sync failure must never fail the local remember.
#[async_trait]
pub trait MemorySync: Send + Sync {
    /// Append a remembered fact to the knowledge oplog. The fact is
    /// content-addressed by its `{subject, body, kind}` payload (no explicit id),
    /// so on the grow-only Knowledge tier each distinct value is its own
    /// immutable entity: an edit is a NEW op (both survive the fold), and an
    /// identical re-remember dedups. The read side ([`Self::pull_knowledge`])
    /// converges a subject to its newest-by-HLC body. `subject` is normalized
    /// (trimmed) to match the local store's entity boundary.
    ///
    /// **`kind` is part of the content address, and that is a deliberate
    /// migration decision (car#665).** Adding it changes the address of a fact
    /// that was already synced under the old `{subject, body}` payload, so such
    /// a fact re-remembered after the upgrade folds as a *second* entity rather
    /// than deduping against the first. That is harmless here because
    /// [`Self::pull_knowledge`] reduces newest-per-subject before anything sees
    /// the facts: both entities share a subject, the kinded one carries the
    /// later HLC, and the reduce keeps it. So the duplicate never reaches a
    /// reader — it only costs one extra op on a tier that is grow-only by
    /// design (an edit already appends). The alternative — inferring a kind on
    /// the receiving side from a locally-known preference with the same subject
    /// — would leave the wire lossy and only work when the same rule had been
    /// set on both devices, which is exactly the case that needs no fixing.
    async fn append_knowledge(
        &self,
        subject: &str,
        body: &str,
        kind: NoteKind,
    ) -> Result<(), String>;

    /// Pull peer-synced knowledge facts — the read side — already reduced to
    /// newest-per-subject. Best-effort: an `Err` means "no peer facts available
    /// right now", never a failure; `recall` degrades to local memory. The impl
    /// pumps the oplog then reads the folded Knowledge facts, reducing
    /// newest-per-subject by LAST-in-ascending-hlc order.
    async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String>;
}

/// `remember` / `recall`, backed by a persistent memgine graph.
pub struct MemoryTools {
    inner: Mutex<Inner>,
    path: PathBuf,
    /// Optional mirror into the synced knowledge oplog (daemon serve only).
    sync: Option<Arc<dyn MemorySync>>,
}

struct Inner {
    engine: MemgineEngine,
    notes: Vec<Note>,
}

impl MemoryTools {
    /// Open (or create) the assistant's memory at `path`, re-ingesting any
    /// previously remembered facts so recall works immediately.
    pub fn open(path: PathBuf) -> Self {
        let (engine, notes) = car_memgine::note_store::engine_from(&path);
        Self {
            inner: Mutex::new(Inner { engine, notes }),
            path,
            sync: None,
        }
    }

    /// Attach a synced-oplog mirror so remembered facts converge across the
    /// user's devices. Set only on the daemon-attached serve path; leaving it
    /// unset keeps `remember` a purely local write (one-shot `car do`).
    pub fn with_sync(mut self, sync: Option<Arc<dyn MemorySync>>) -> Self {
        self.sync = sync;
        self
    }

    /// The two model-facing tool schemas.
    pub fn tool_defs() -> Vec<Value> {
        vec![
            json!({
                "name": "remember",
                "description": "Save a durable fact about the user or task so you can recall it in \
                                future sessions. Use for things worth persisting, not transient \
                                details. `kind` decides how the fact is used later, so pick it \
                                deliberately.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "subject": { "type": "string", "description": "Short label for the fact (e.g. 'project name')." },
                        "body": { "type": "string", "description": "The fact to remember." },
                        "kind": {
                            "type": "string",
                            "enum": ["fact", "preference", "procedure"],
                            "description": "'fact' (default): something true about the user, \
                                            project, or environment; retrieved when relevant to \
                                            the query. 'preference': a standing instruction about \
                                            how the user wants you to work — surfaced in EVERY \
                                            future session, so reserve it for rules you should \
                                            never violate. 'procedure': how a task was done and \
                                            whether it worked, for reuse next time."
                        }
                    },
                    "required": ["subject", "body"]
                },
                "mutating": true,
                "tier": "full_access"
            }),
            json!({
                "name": "recall",
                "description": "Retrieve previously remembered facts relevant to a query. Use at the \
                                start of a task to recall what you know about the user or project.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": { "type": "string", "description": "What to recall." }
                    },
                    "required": ["query"]
                }
            }),
        ]
    }

    /// Returns the tool result plus the validated [`NoteKind`], so the caller's
    /// sync mirror sends the kind this call actually stored rather than
    /// re-parsing the raw params and risking a second, divergent answer.
    fn remember(&self, params: &Value) -> Result<(Value, NoteKind), String> {
        // Trim the subject up front so the local entity boundary matches the
        // synced one (the mirror sends the same normalized subject).
        let subject = normalize_subject(
            params
                .get("subject")
                .and_then(Value::as_str)
                .ok_or("remember requires a 'subject' string")?,
        );
        let body = params
            .get("body")
            .and_then(Value::as_str)
            .ok_or("remember requires a 'body' string")?;
        let kind = NoteKind::parse(params.get("kind").and_then(Value::as_str))
            .map_err(|e| format!("remember: {e}"))?;
        let mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;
        let note = Note {
            subject: subject.to_string(),
            body: body.to_string(),
            kind,
        };
        // Supersede an existing fact with the same subject (case-insensitive) so
        // an updated fact replaces the stale one instead of accumulating. A
        // re-remember may also RECLASSIFY (a fact promoted to a preference), so
        // the kind is superseded with the body.
        match g
            .notes
            .iter_mut()
            .find(|n| n.subject.eq_ignore_ascii_case(subject))
        {
            Some(existing) => {
                existing.body = body.to_string();
                existing.kind = kind;
            }
            None => g.notes.push(note),
        }
        // Re-ingest the whole set so the memgine graph mirrors the current notes
        // (cheap for a personal store; correctness over cleverness).
        let mut engine = MemgineEngine::new(None);
        for (i, n) in g.notes.iter().enumerate() {
            ingest(&mut engine, i, n);
        }
        g.engine = engine;
        // Persist the whole set (small; correctness over cleverness).
        car_memgine::note_store::save(&self.path, &g.notes)?;
        Ok((
            json!({ "remembered": subject, "total_facts": g.notes.len() }),
            kind,
        ))
    }

    /// Best-effort mirror of a just-remembered fact into the synced knowledge
    /// oplog. Only active when a [`MemorySync`] is attached (daemon serve). A
    /// sync failure is logged and swallowed — the fact is already saved locally,
    /// and convergence is best-effort, so it must never fail the tool call.
    async fn mirror_to_sync(&self, params: &Value, kind: NoteKind) {
        let Some(sync) = &self.sync else {
            return;
        };
        let (Some(subject), Some(body)) = (
            params.get("subject").and_then(Value::as_str),
            params.get("body").and_then(Value::as_str),
        ) else {
            return;
        };
        // Same normalized subject as the local store; the fact is content-addressed
        // by {subject, body, kind} on the sync side (no explicit id). The kind is
        // the one `remember` already validated, not a re-parse of the raw params.
        if let Err(e) = sync
            .append_knowledge(normalize_subject(subject), body, kind)
            .await
        {
            tracing::debug!(
                target: "assistant.memory",
                "knowledge sync append failed (non-fatal): {e}"
            );
        }
    }

    /// Best-effort pull of peer-synced facts. Returns empty on no sink or any
    /// error — recall must never fail because sync is unavailable.
    async fn pull_peer_knowledge(&self) -> Vec<SyncedFact> {
        let Some(sync) = &self.sync else {
            return Vec::new();
        };
        match sync.pull_knowledge().await {
            Ok(facts) => facts,
            Err(e) => {
                tracing::debug!(
                    target: "assistant.memory",
                    "peer knowledge pull failed (non-fatal): {e}"
                );
                Vec::new()
            }
        }
    }

    async fn recall(&self, params: &Value) -> Result<Value, String> {
        let query = params
            .get("query")
            .and_then(Value::as_str)
            .ok_or("recall requires a 'query' string")?;
        // Best-effort peer pull FIRST — no lock held across the await. Peers are
        // already reduced to newest-per-subject by the sink.
        let peers = self.pull_peer_knowledge().await;
        let mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;
        if g.notes.is_empty() && peers.is_empty() {
            return Ok(json!({ "query": query, "context": "", "note": "no facts remembered yet" }));
        }
        // Option B: local notes are authoritative for the subjects THIS device
        // holds; peer facts fill only the gaps. A peer's newer edit to a locally
        // held subject is intentionally NOT applied — local has no HLC to compare
        // against, so this is conservative local authority, not cross-device
        // newest-wins. Same case-insensitive trimmed key as the local supersede.
        let local_keys: std::collections::HashSet<String> =
            g.notes.iter().map(|n| subject_key(&n.subject)).collect();
        let gaps: Vec<&SyncedFact> = peers
            .iter()
            .filter(|fact| !local_keys.contains(&subject_key(&fact.subject)))
            .collect();
        if gaps.is_empty() {
            // Nothing new from peers — the maintained local engine has the answer.
            let context = g.engine.build_context(query);
            return Ok(json!({ "query": query, "context": context }));
        }
        // Build a fresh view = local notes + peer-gap facts for this recall (the
        // peer facts live in the oplog and are re-pulled each recall, never
        // written into the local notes file).
        let mut engine = MemgineEngine::new(None);
        let mut idx = 0;
        for n in &g.notes {
            ingest(&mut engine, idx, n);
            idx += 1;
        }
        for fact in gaps {
            ingest(
                &mut engine,
                idx,
                &Note {
                    subject: fact.subject.clone(),
                    body: fact.body.clone(),
                    // The peer's own kind, carried over the wire (car#665). A
                    // preference must stay a preference here: reconstructing it
                    // as a plain fact demoted the constraint to something
                    // recall-only, so a standing rule quietly stopped applying
                    // on this device and kept applying on the one that set it.
                    kind: fact.kind,
                },
            );
            idx += 1;
        }
        let context = engine.build_context(query);
        Ok(json!({ "query": query, "context": context }))
    }

    /// Run the paper-style proactive memory pass for the assistant loop.
    ///
    /// This is deliberately host-side and deterministic: update compact memory
    /// from the runtime trajectory, then decide whether one remembered fact is
    /// strong enough to interrupt the next model turn. The model still has the
    /// explicit `recall` tool, but it no longer has to remember to call it before
    /// CAR can surface high-value memory.
    pub async fn proactive_intervention(
        &self,
        query: &str,
        recent: Vec<String>,
        events: &[Event],
    ) -> Result<
        (
            car_memgine::ProactiveMaintenanceReport,
            ProactiveMemoryDecision,
        ),
        String,
    > {
        let peers = self.pull_peer_knowledge().await;
        let mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;

        let local_keys: std::collections::HashSet<String> =
            g.notes.iter().map(|n| subject_key(&n.subject)).collect();
        let gaps: Vec<&SyncedFact> = peers
            .iter()
            .filter(|fact| !local_keys.contains(&subject_key(&fact.subject)))
            .collect();

        let maintenance_request = ProactiveMaintenanceRequest {
            max_recent: 32,
            tenant_id: None,
        };
        let mut request = ProactiveMemoryRequest {
            query: query.to_string(),
            recent,
            ..Default::default()
        };

        if gaps.is_empty() {
            let maintenance = g
                .engine
                .maintain_proactive_memory_from_events(events, &maintenance_request);
            request.trigger.merge(maintenance.trigger.clone());
            let decision = g.engine.proactive_intervention(&request);
            return Ok((maintenance, decision));
        }

        let mut engine = MemgineEngine::new(None);
        let mut idx = 0;
        for n in &g.notes {
            ingest(&mut engine, idx, n);
            idx += 1;
        }
        for fact in gaps {
            ingest(
                &mut engine,
                idx,
                &Note {
                    subject: fact.subject.clone(),
                    body: fact.body.clone(),
                    // The peer's own kind, carried over the wire (car#665). A
                    // preference must stay a preference here: reconstructing it
                    // as a plain fact demoted the constraint to something
                    // recall-only, so a standing rule quietly stopped applying
                    // on this device and kept applying on the one that set it.
                    kind: fact.kind,
                },
            );
            idx += 1;
        }
        let maintenance =
            engine.maintain_proactive_memory_from_events(events, &maintenance_request);
        request.trigger.merge(maintenance.trigger.clone());
        let decision = engine.proactive_intervention(&request);
        Ok((maintenance, decision))
    }
}

/// The shared subject normalization for the local store and the synced mirror,
/// so both draw the fact's entity boundary at the same place (trimmed).
fn normalize_subject(subject: &str) -> &str {
    subject.trim()
}

/// The case-insensitive, trimmed key for a subject — the same entity boundary
/// the local supersede uses (`normalize_subject` + case-insensitive match). The
/// peer-merge gate and the sink's newest-per-subject reduce must use this exact
/// key, or `Pet`/`pet` split into two entities.
fn subject_key(subject: &str) -> String {
    normalize_subject(subject).to_ascii_lowercase()
}

#[async_trait]
impl ToolExecutor for MemoryTools {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        match tool {
            "remember" => {
                let (out, kind) = self.remember(params)?;
                // Mirror into the synced oplog after the local write succeeds.
                self.mirror_to_sync(params, kind).await;
                Ok(out)
            }
            "recall" => self.recall(params).await,
            other => Err(format!("unknown tool: '{other}'")),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn remember_declares_mutating_and_full_access() {
        let defs = MemoryTools::tool_defs();
        let remember = defs
            .iter()
            .find(|def| def["name"] == "remember")
            .expect("remember tool def");
        assert_eq!(remember["mutating"], true);
        assert_eq!(remember["tier"], "full_access");
    }

    #[test]
    fn remember_advertises_the_memory_taxonomy() {
        let defs = MemoryTools::tool_defs();
        let remember = defs.iter().find(|def| def["name"] == "remember").unwrap();
        let kinds = remember["parameters"]["properties"]["kind"]["enum"]
            .as_array()
            .expect("kind is an enum — the schema is what teaches the model");
        for expected in ["fact", "preference", "procedure"] {
            assert!(kinds.iter().any(|k| k == expected), "missing {expected}");
        }
        // Not required: an unspecified kind is a plain fact.
        let required = remember["parameters"]["required"].as_array().unwrap();
        assert!(!required.iter().any(|r| r == "kind"));
    }

    #[tokio::test]
    async fn preference_becomes_an_always_included_constraint() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("assistant.json"));
        mem.execute(
            "remember",
            &json!({
                "subject": "code review",
                "body": "Never merge without a green CI run.",
                "kind": "preference"
            }),
        )
        .await
        .unwrap();
        mem.execute(
            "remember",
            &json!({ "subject": "project name", "body": "The project is called Zephyr." }),
        )
        .await
        .unwrap();

        // A query unrelated to either fact: the preference still surfaces (it is
        // a constraint), which is the behavioral difference the enum encodes.
        let out = mem
            .execute("recall", &json!({ "query": "what time is the standup" }))
            .await
            .unwrap();
        let context = out["context"].as_str().unwrap();
        assert!(
            context.contains("Active Constraints") && context.contains("green CI run"),
            "preference must land in the always-included constraints layer: {context}"
        );
    }

    #[tokio::test]
    async fn unknown_kind_is_rejected_rather_than_silently_stored() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("assistant.json"));
        let err = mem
            .execute(
                "remember",
                &json!({ "subject": "s", "body": "b", "kind": "identity" }),
            )
            .await
            .unwrap_err();
        assert!(err.contains("unknown kind 'identity'"), "{err}");
    }

    #[tokio::test]
    async fn remembers_across_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("assistant.json");

        {
            let mem = MemoryTools::open(path.clone());
            mem.execute(
                "remember",
                &json!({ "subject": "project name", "body": "The project is called Zephyr." }),
            )
            .await
            .unwrap();
        }

        // Re-open (simulating a new process/session) and recall.
        let mem = MemoryTools::open(path.clone());
        let out = mem
            .execute("recall", &json!({ "query": "what is my project called?" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap();
        assert!(
            ctx.contains("Zephyr"),
            "recall should surface the fact: {ctx}"
        );
    }

    #[tokio::test]
    async fn remember_supersedes_same_subject() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("m.json");
        let mem = MemoryTools::open(path.clone());
        mem.execute("remember", &json!({ "subject": "editor", "body": "vim" }))
            .await
            .unwrap();
        let out = mem
            .execute("remember", &json!({ "subject": "Editor", "body": "emacs" }))
            .await
            .unwrap();
        // Same subject (case-insensitive) → replaced, not accumulated.
        assert_eq!(out["total_facts"], 1);
        let notes: Vec<Note> =
            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        assert_eq!(notes.len(), 1);
        assert_eq!(notes[0].body, "emacs");
    }

    #[tokio::test]
    async fn recall_on_empty_is_graceful() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("m.json"));
        let out = mem
            .execute("recall", &json!({ "query": "anything" }))
            .await
            .unwrap();
        assert_eq!(out["context"], "");
    }

    /// Recall must run the FULL memgine path, not Fast.
    ///
    /// The observable difference on the assistant's real data shape is **fact
    /// ordering**. The four-layer model is relevance-*ascending* — most relevant
    /// last, so it sits closest to the model's recency attention — and that
    /// ranking is PPR-scored, which Fast skips entirely. On Fast the assistant
    /// emitted facts in creation order: recall was not relevance-ranked at all,
    /// while the docs advertised spreading activation.
    ///
    /// So this asserts the matching fact lands LAST despite being created FIRST.
    /// Under Fast it comes back first and this fails, which is the regression
    /// worth catching: switching one call back to `build_context_fast` would
    /// silently downgrade the flagship agent to keyword recall over flat notes.
    ///
    /// (Known-unknowns extraction, the other Full-only step, reads *conversation*
    /// nodes; assistant notes ingest as facts, so it does not fire here.)
    #[tokio::test]
    async fn recall_ranks_by_relevance_not_creation_order() {
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("m.json"));

        // The match is remembered FIRST, so creation order and relevance order
        // disagree — otherwise the assertion would pass either way.
        for (subject, body) in [
            ("deployment target", "Unclear which region we deploy to."),
            ("db", "Postgres 16 in prod."),
            ("owner", "Matt owns the release process."),
        ] {
            mem.execute("remember", &json!({ "subject": subject, "body": body }))
                .await
                .unwrap();
        }

        let out = mem
            .execute("recall", &json!({ "query": "deployment target" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap_or_default();

        let facts: Vec<&str> = ctx
            .lines()
            .skip_while(|l| !l.starts_with("## Current Facts"))
            .filter(|l| l.starts_with("- "))
            .collect();
        assert!(facts.len() >= 3, "expected all facts back, got: {ctx}");
        assert!(
            facts.last().unwrap().contains("deployment target"),
            "the query-matching fact must rank last (relevance-ascending); \
             got creation order, which means Fast mode is running.\nfacts: {facts:#?}"
        );
    }

    struct RecordingSync(Mutex<Vec<SyncedFact>>);
    #[async_trait]
    impl MemorySync for RecordingSync {
        async fn append_knowledge(
            &self,
            subject: &str,
            body: &str,
            kind: NoteKind,
        ) -> Result<(), String> {
            self.0.lock().unwrap().push(SyncedFact {
                subject: subject.into(),
                body: body.into(),
                kind,
            });
            Ok(())
        }
        async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String> {
            Ok(Vec::new())
        }
    }

    /// A sink that returns a fixed set of peer facts on pull (append is a no-op).
    struct PeerSync(Vec<SyncedFact>);
    #[async_trait]
    impl MemorySync for PeerSync {
        async fn append_knowledge(&self, _: &str, _: &str, _: NoteKind) -> Result<(), String> {
            Ok(())
        }
        async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String> {
            Ok(self.0.clone())
        }
    }

    /// A peer fact, for the `PeerSync` fixtures.
    fn peer_fact(subject: &str, body: &str, kind: NoteKind) -> SyncedFact {
        SyncedFact {
            subject: subject.into(),
            body: body.into(),
            kind,
        }
    }

    #[tokio::test]
    async fn remember_mirrors_normalized_subject_and_body_to_sync() {
        let dir = tempfile::tempdir().unwrap();
        let rec = Arc::new(RecordingSync(Mutex::new(Vec::new())));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(rec.clone()));
        // Untrimmed subject → the mirror sees the same normalized subject the
        // local store uses, so both draw the entity boundary in the same place.
        mem.execute(
            "remember",
            &json!({ "subject": "  Editor  ", "body": "vim" }),
        )
        .await
        .unwrap();
        let got = rec.0.lock().unwrap().clone();
        assert_eq!(got, vec![peer_fact("Editor", "vim", NoteKind::Fact)]);
    }

    #[tokio::test]
    async fn remember_without_sync_stays_local_only() {
        // One-shot `car do`: no sink attached, remember still works.
        let dir = tempfile::tempdir().unwrap();
        let mem = MemoryTools::open(dir.path().join("m.json"));
        let out = mem
            .execute("remember", &json!({ "subject": "x", "body": "y" }))
            .await
            .unwrap();
        assert_eq!(out["remembered"], "x");
    }

    struct FailingSync;
    #[async_trait]
    impl MemorySync for FailingSync {
        async fn append_knowledge(&self, _: &str, _: &str, _: NoteKind) -> Result<(), String> {
            Err("oplog unreachable".into())
        }
        async fn pull_knowledge(&self) -> Result<Vec<SyncedFact>, String> {
            Err("oplog unreachable".into())
        }
    }

    #[tokio::test]
    async fn recall_surfaces_a_peer_fact_not_held_locally() {
        // No local notes; the fact exists only via a peer's sync.
        let dir = tempfile::tempdir().unwrap();
        let peer = Arc::new(PeerSync(vec![peer_fact(
            "project",
            "the project is Zephyr",
            NoteKind::Fact,
        )]));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
        let out = mem
            .execute("recall", &json!({ "query": "what is my project?" }))
            .await
            .unwrap();
        assert!(
            out["context"].as_str().unwrap().contains("Zephyr"),
            "{}",
            out["context"]
        );
    }

    /// car#665. A `preference` is saved as a memgine *constraint*, which context
    /// assembly includes in every session regardless of the query; a `fact`
    /// surfaces only when the query matches it. Reconstructing every peer-synced
    /// note as `NoteKind::default()` collapsed that distinction, so a standing
    /// rule set on device A quietly stopped being a standing rule on device B —
    /// with no error anywhere.
    ///
    /// The assertion is on the *layer* the note lands in, mirroring the local
    /// `preference_becomes_an_always_included_constraint`. Mere presence in the
    /// context does not discriminate: with a small store `build_context`
    /// lists an unmatched fact under Current Facts anyway. Active Constraints is
    /// what makes a rule unconditional, so that is what must be asserted.
    #[tokio::test]
    async fn a_peer_preference_stays_a_standing_rule_and_a_peer_fact_does_not() {
        let rule = "Never merge without a green CI run.";
        let unrelated = "what time is the standup";

        let recall_with = |kind| async move {
            let dir = tempfile::tempdir().unwrap();
            let peer = Arc::new(PeerSync(vec![peer_fact("code review", rule, kind)]));
            let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
            let out = mem
                .execute("recall", &json!({ "query": unrelated }))
                .await
                .unwrap();
            out["context"].as_str().unwrap().to_string()
        };

        let as_preference = recall_with(NoteKind::Preference).await;
        assert!(
            as_preference.contains("Active Constraints") && as_preference.contains(rule),
            "a synced preference must land in the always-included constraints \
             layer, exactly as it does on the device that set it: {as_preference}"
        );

        // The control: the same note synced as a plain fact — which is what the
        // receiving device used to reconstruct for *every* peer note — must not
        // reach that layer. Without this the assertion above would also pass if
        // the layer heading were always emitted, proving nothing.
        let as_fact = recall_with(NoteKind::Fact).await;
        assert!(
            !as_fact.contains("Active Constraints"),
            "control: a plain fact is not a standing rule — if it reaches the \
             constraints layer the test cannot tell the two kinds apart: {as_fact}"
        );
    }

    #[tokio::test]
    async fn remember_mirrors_the_kind_so_a_preference_crosses_the_wire() {
        let dir = tempfile::tempdir().unwrap();
        let rec = Arc::new(RecordingSync(Mutex::new(Vec::new())));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(rec.clone()));
        mem.execute(
            "remember",
            &json!({
                "subject": "code review",
                "body": "Never merge without a green CI run.",
                "kind": "preference"
            }),
        )
        .await
        .unwrap();
        assert_eq!(
            rec.0.lock().unwrap().clone(),
            vec![peer_fact(
                "code review",
                "Never merge without a green CI run.",
                NoteKind::Preference
            )],
            "the mirror must send the kind `remember` validated, not a default"
        );
    }

    #[tokio::test]
    async fn recall_keeps_local_over_a_peer_edit_of_the_same_subject() {
        // Local set editor=vim; a peer later set editor=emacs. Option B: local is
        // authoritative for its own subjects, so the peer edit is SUPPRESSED
        // (conservative local authority, not cross-device newest-wins). Pinned so
        // nobody "fixes" it by accident.
        let dir = tempfile::tempdir().unwrap();
        let peer = Arc::new(PeerSync(vec![peer_fact("editor", "emacs", NoteKind::Fact)]));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
        mem.execute("remember", &json!({ "subject": "editor", "body": "vim" }))
            .await
            .unwrap();
        let out = mem
            .execute("recall", &json!({ "query": "which editor?" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap().to_string();
        assert!(ctx.contains("vim"), "{ctx}");
        assert!(
            !ctx.contains("emacs"),
            "peer edit must not override local under Option B: {ctx}"
        );
    }

    #[tokio::test]
    async fn recall_peer_gate_is_case_insensitive() {
        // "Pet" (local) and "pet" (peer) are the same entity — local wins.
        let dir = tempfile::tempdir().unwrap();
        let peer = Arc::new(PeerSync(vec![peer_fact(
            "pet",
            "a peer dog",
            NoteKind::Fact,
        )]));
        let mem = MemoryTools::open(dir.path().join("m.json")).with_sync(Some(peer));
        mem.execute(
            "remember",
            &json!({ "subject": "Pet", "body": "my cat Mittens" }),
        )
        .await
        .unwrap();
        let out = mem
            .execute("recall", &json!({ "query": "what pet?" }))
            .await
            .unwrap();
        let ctx = out["context"].as_str().unwrap().to_string();
        assert!(ctx.contains("Mittens"), "{ctx}");
        assert!(
            !ctx.contains("peer dog"),
            "a case-variant peer subject must be gated: {ctx}"
        );
    }

    #[tokio::test]
    async fn recall_pull_failure_degrades_to_local() {
        // A pull error must never fail recall — it answers from local memory.
        let dir = tempfile::tempdir().unwrap();
        let mem =
            MemoryTools::open(dir.path().join("m.json")).with_sync(Some(Arc::new(FailingSync)));
        mem.execute("remember", &json!({ "subject": "city", "body": "Denver" }))
            .await
            .unwrap();
        let out = mem
            .execute("recall", &json!({ "query": "which city?" }))
            .await
            .unwrap();
        assert!(out["context"].as_str().unwrap().contains("Denver"));
    }

    #[tokio::test]
    async fn remember_survives_sync_failure() {
        // A sync hiccup must never fail the local remember — the fact is saved.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("m.json");
        let mem = MemoryTools::open(path.clone()).with_sync(Some(Arc::new(FailingSync)));
        let out = mem
            .execute("remember", &json!({ "subject": "a", "body": "b" }))
            .await
            .unwrap();
        assert_eq!(out["remembered"], "a");
        // And it persisted locally despite the sync error.
        let notes: Vec<Note> =
            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        assert_eq!(notes.len(), 1);
        assert_eq!(notes[0].body, "b");
    }

    // Fold two same-subject facts through the REAL car-sync fold to prove the
    // write format converges correctly on the grow-only Knowledge tier. The
    // earlier stable-per-subject key collapsed these to one op and kept the
    // EARLIEST (dropping the update); content-addressing keeps both, and the
    // read model (newest-by-HLC per subject) recovers the edit.
    #[test]
    fn synced_knowledge_survives_edits_and_converges_to_newest_body() {
        use car_sync::fold::fold;
        use car_sync::oplog::{DeviceLog, Scope, Surface};

        let mut log = DeviceLog::new("device-a");
        let vim = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "vim" }),
        );
        let emacs = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "emacs" }),
        );

        let state = fold(&[vim, emacs]);
        let entries = state.log_entries("knowledge");
        // Both edits survive as distinct immutable entities (no update dropped).
        assert_eq!(entries.len(), 2);
        // The read model converges a subject to its newest-by-HLC body.
        let newest = entries
            .iter()
            .filter(|r| r.payload["subject"] == "editor")
            .max_by(|a, b| a.hlc.cmp(&b.hlc))
            .unwrap();
        assert_eq!(newest.payload["body"], "emacs");
    }

    #[test]
    fn identical_reremember_dedups_in_the_synced_fold() {
        use car_sync::fold::fold;
        use car_sync::oplog::{DeviceLog, Scope, Surface};

        let mut log = DeviceLog::new("device-a");
        let a = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "vim" }),
        );
        let b = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({ "subject": "editor", "body": "vim" }),
        );
        // Same {subject, body} → same content-addressed key → one entity, no spew.
        let state = fold(&[a, b]);
        assert_eq!(state.log_entries("knowledge").len(), 1);
    }
}