car-server-core 0.36.0

//! 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 uses the memgine's keyword/graph retrieval
//! (`build_context_fast`, no inference required).

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,
    ProactiveMemorySave,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

#[derive(Clone, Serialize, Deserialize)]
struct Note {
    subject: String,
    body: String,
}

/// 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}` 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.
    async fn append_knowledge(&self, subject: &str, body: &str) -> Result<(), String>;

    /// Pull peer-synced knowledge facts — the read side — as `(subject, body)`
    /// pairs 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<(String, String)>, 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 notes: Vec<Note> = std::fs::read_to_string(&path)
            .ok()
            .and_then(|s| serde_json::from_str(&s).ok())
            .unwrap_or_default();
        let mut engine = MemgineEngine::new(None);
        for (i, n) in notes.iter().enumerate() {
            ingest(&mut engine, i, n);
        }
        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 (e.g. a preference, a project name, a decision). \
                                Use for things worth persisting, not transient details.",
                "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." }
                    },
                    "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"]
                }
            }),
        ]
    }

    fn remember(&self, params: &Value) -> Result<Value, 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 mut g = self.inner.lock().map_err(|_| "memory lock poisoned")?;
        let note = Note {
            subject: subject.to_string(),
            body: body.to_string(),
        };
        // Supersede an existing fact with the same subject (case-insensitive) so
        // an updated fact replaces the stale one instead of accumulating.
        match g
            .notes
            .iter_mut()
            .find(|n| n.subject.eq_ignore_ascii_case(subject))
        {
            Some(existing) => existing.body = body.to_string(),
            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).
        if let Some(parent) = self.path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let serialized =
            serde_json::to_string_pretty(&g.notes).map_err(|e| format!("serialize: {e}"))?;
        std::fs::write(&self.path, serialized).map_err(|e| format!("persist memory: {e}"))?;
        Ok(json!({ "remembered": subject, "total_facts": g.notes.len() }))
    }

    /// 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) {
        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} on the sync side (no explicit id).
        if let Err(e) = sync
            .append_knowledge(normalize_subject(subject), body)
            .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<(String, String)> {
        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<&(String, String)> = peers
            .iter()
            .filter(|(subject, _)| !local_keys.contains(&subject_key(subject)))
            .collect();
        if gaps.is_empty() {
            // Nothing new from peers — the maintained local engine has the answer.
            let context = g.engine.build_context_fast(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 (subject, body) in gaps {
            ingest(
                &mut engine,
                idx,
                &Note {
                    subject: subject.clone(),
                    body: body.clone(),
                },
            );
            idx += 1;
        }
        let context = engine.build_context_fast(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<&(String, String)> = peers
            .iter()
            .filter(|(subject, _)| !local_keys.contains(&subject_key(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 (subject, body) in gaps {
            ingest(
                &mut engine,
                idx,
                &Note {
                    subject: subject.clone(),
                    body: body.clone(),
                },
            );
            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()
}

/// Ingest one note as a memgine fact. Deterministic id per index so a reload
/// re-creates the same graph.
fn ingest(engine: &mut MemgineEngine, idx: usize, n: &Note) {
    engine.save_proactive_knowledge(ProactiveMemorySave {
        id: Some(format!("assistant-note-{idx}")),
        subject: n.subject.clone(),
        body: n.body.clone(),
        tags: vec!["assistant_memory".to_string()],
        confidence: Some("high".to_string()),
        tenant_id: None,
        is_constraint: false,
    });
}

#[async_trait]
impl ToolExecutor for MemoryTools {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        match tool {
            "remember" => {
                let out = self.remember(params)?;
                // Mirror into the synced oplog after the local write succeeds.
                self.mirror_to_sync(params).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");
    }

    #[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"], "");
    }

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

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

    #[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![("Editor".into(), "vim".into())]);
    }

    #[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) -> Result<(), String> {
            Err("oplog unreachable".into())
        }
        async fn pull_knowledge(&self) -> Result<Vec<(String, String)>, 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![(
            "project".into(),
            "the project is Zephyr".into(),
        )]));
        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"]
        );
    }

    #[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![("editor".into(), "emacs".into())]));
        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![("pet".into(), "a peer dog".into())]));
        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);
    }
}