diff --git a/docs/cli.mdx b/docs/cli.mdx index 44e1ff9..91a33b9 100644 --- a/docs/cli.mdx +++ b/docs/cli.mdx @@ -36,6 +36,24 @@ Memory types: `fact`, `preference`, `decision`, `task`, `project_overview`, `agent_instruction`, `file_annotation`, `code_note`, `reference`. When omitted, the type is inferred heuristically. +## History + +An audit timeline of memory changes — what was created, updated, deleted, or +crawled, when, and by which source (`cli`, `mcp`, or `crawler`). Metadata only: +no old content is retained. + +```sh +# Recent changes (newest first) +memd history + +# Filter by action, type, scope, or time +memd history --action delete --type fact --since 7d --limit 50 +``` + +Each crawl pass logs a single rolled-up event (e.g. `12 indexed, 3 deleted`) +rather than one event per file. The log starts empty on first run and accrues +going forward — past mutations are not backfilled. + ## Crawler | Command | Description | diff --git a/docs/mcp.mdx b/docs/mcp.mdx index f6af1e3..76f3a5b 100644 --- a/docs/mcp.mdx +++ b/docs/mcp.mdx @@ -109,6 +109,7 @@ highlighting, and grouped counts happen server-side. | `forget_memory` | `id` | Delete a memory. | | `list_memories` | `type?`, `scope?`, `limit?`, `offset?`, `include_content?`, `crop_length?`, `facets?` | List recent memories — metadata-only by default. | | `stats` | `group_by?` | Document count + server-side facet counts (default: type/scope/source). | +| `history` | `action?`, `type?`, `scope?`, `since?`, `limit?` | Audit timeline of memory changes (create/update/delete/crawl), newest first. Metadata only. | ## Protocol diff --git a/docs/superpowers/specs/2026-06-15-memory-history-design.md b/docs/superpowers/specs/2026-06-15-memory-history-design.md new file mode 100644 index 0000000..0506539 --- /dev/null +++ b/docs/superpowers/specs/2026-06-15-memory-history-design.md @@ -0,0 +1,115 @@ +# memd memory history — design + +**Date:** 2026-06-15 +**Status:** Approved +**Author:** Quentin de Quelen (with Claude) + +## Goal + +Give users an auditable timeline of what changed in their memory store: which +memories were created, updated, or deleted, when, and by which source. Today, +once a memory is updated or forgotten its prior state is gone and there is no +record the event happened. + +Scope is an **audit timeline** — metadata only. We do not retain old content, +diffs, or an undo/restore capability (explicitly out of scope). + +## Storage — a `memory_events` index + +A second Meilisearch index alongside `memories`, created idempotently at daemon +startup. It is a plain log: **no embedder**. One document per event. + +| field | type | notes | +|---|---|---| +| `id` | string | UUIDv7 primary key (time-ordered) | +| `ts` | i64 | unix seconds — sortable | +| `action` | string | `create` \| `update` \| `delete` \| `crawl` — filterable | +| `memory_id` | string | affected memory id (empty for `crawl`) — filterable | +| `title` | string | snapshot of the memory title (readable after delete) — searchable | +| `type` | string | memory type snapshot — filterable | +| `scope` | string | scope snapshot — filterable | +| `source` | string | `cli` \| `mcp` \| `crawler` — filterable | +| `source_client` | string? | optional client id | +| `detail` | string? | for `crawl`: `"12 indexed, 3 updated, 2 deleted, 0 errors"` — searchable | + +Index settings: + +- `filterableAttributes = [action, type, scope, source, memory_id, ts]` +- `sortableAttributes = [ts]` +- `searchableAttributes = [title, detail, memory_id]` + +No backfill: the log starts empty and accrues going forward — past mutations +cannot be reconstructed. The events index is included automatically in dumps and +engine migrations because those operate instance-wide. + +## Recording — single chokepoint in `MemoryService` + +`MeiliClient` becomes index-aware: add an `index: String` field (defaulting to +`"memories"`, preserving every existing call site) and a `for_index(uid)` helper +returning a clone pointed at another index. Events reuse the same upsert/search +plumbing. + +`MemoryService` gains an `EventLog` (a client bound to `memory_events`): + +- `save` logs a `create` **only on a real write** — the content-hash dedup + early-return logs nothing. +- `update` and `forget` take a new `Source` argument so the event records who + performed it. `forget` reads the doc's metadata **before** deleting, to + snapshot title/type/scope into the event. +- Recording is **best-effort and off the critical path**: fire-and-forget, no + Meilisearch task wait. A logging failure never blocks or fails a memory + mutation — mirroring the existing `bump_accessed` pattern. (Consequence: the + log is eventually consistent; an event may not be queryable for a few hundred + ms after the mutation.) +- The **crawler** stays event-free per file. `crawler::scan` emits a single + `crawl` summary event at the end, built from its existing `CrawlSummary`. + +Access bumps (`bump_accessed`, the `read` last-accessed write) go straight to +the client, not through `save/update`, so they correctly produce no events. + +Rejected alternative: logging at each call site (CLI/MCP/crawler). Three places +to keep in sync and easy to miss a path; the service is the single funnel every +mutation already passes through. + +## Surface — CLI + MCP + +### CLI + +``` +memd history [--action ] [--type ] \ + [--scope ] [--since ] [--limit ] +``` + +Newest first (`ts:desc`). Reuses `parse_since` (`30d`, `12h`, `45m`, unix secs). +One line per event: + +``` +2026-06-15T10:30:00Z update [fact] My title (mcp · scope: global) +2026-06-15T10:28:11Z crawl 12 indexed, 3 updated, 2 deleted, 0 errors (crawler) +``` + +### MCP + +A `history` tool with the same filters (`action`, `type`, `scope`, `since`, +`limit`), returning lightweight JSON rows so an agent can ask "what changed +recently." + +### Status (optional) + +A `history: N events` line in `memd status`. + +## Testing + +Unit tests for the pure pieces (no live Meilisearch), matching the existing +unit-test style: + +- event construction from each mutation kind (create/update/delete/crawl) +- the history filter-expression builder +- the CLI line renderer +- `Source` round-trips into the event record + +## Out of scope + +- Before/after content diffs +- Undo / restore / revert +- Per-file crawler events (summary only) diff --git a/src/cli.rs b/src/cli.rs index 73ac372..9563406 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -95,6 +95,9 @@ pub async fn status() -> Result<()> { .unwrap_or(0); println!(" memories: {count}"); } + if let Some(events) = svc.events().count().await { + println!(" events: {events}"); + } } let mcp_up = daemon_healthy(&cfg).await; @@ -290,7 +293,7 @@ pub async fn search( pub async fn forget(id: String) -> Result<()> { let cfg = Config::load_or_init()?; let svc = require_daemon(&cfg).await?; - if svc.forget(&id).await? { + if svc.forget(&id, Source::Cli).await? { println!("Forgot memory {id}"); } else { println!("No memory with id {id}"); @@ -298,6 +301,46 @@ pub async fn forget(id: String) -> Result<()> { Ok(()) } +/// Show the history of memory changes (most recent first). +pub async fn history( + action: Option, + ty: Option, + scope: Option, + since: Option, + limit: usize, +) -> Result<()> { + let cfg = Config::load_or_init()?; + let svc = require_daemon(&cfg).await?; + + let parsed_action = match &action { + Some(s) => Some(crate::history::EventAction::parse(s).ok_or_else(|| { + anyhow::anyhow!("unknown action: {s} (use create/update/delete/crawl)") + })?), + None => None, + }; + let since_ts = match since { + Some(s) => Some(parse_since(&s)?), + None => None, + }; + + let query = crate::history::EventQuery { + action: parsed_action, + r#type: ty, + scope, + since: since_ts, + limit, + }; + let events = svc.history(&query).await?; + if events.is_empty() { + println!("No history yet."); + return Ok(()); + } + for ev in &events { + println!("{}", crate::history::render_event(ev)); + } + Ok(()) +} + /// Run a one-off crawl. pub async fn crawl_run() -> Result<()> { let cfg = Config::load_or_init()?; diff --git a/src/crawler/mod.rs b/src/crawler/mod.rs index df473a6..75619b8 100644 --- a/src/crawler/mod.rs +++ b/src/crawler/mod.rs @@ -109,6 +109,19 @@ pub async fn scan(cfg: &Config, svc: &MemoryService) -> Result { summary.deleted = reconcile_deletions(svc, &seen).await.unwrap_or(0); summary.finished_at = crate::memory::model::now_secs(); save_summary(&summary)?; + + // Record one rolled-up history event per crawl — but only when something + // actually changed, so periodic reconciles that find nothing don't flood + // the audit log. + if summary.indexed + summary.deleted + summary.errors > 0 { + let detail = format!( + "{} indexed, {} skipped, {} deleted, {} errors", + summary.indexed, summary.skipped, summary.deleted, summary.errors + ); + svc.events() + .record(crate::history::MemoryEvent::crawl(detail)) + .await; + } Ok(summary) } @@ -139,7 +152,7 @@ pub async fn index_one(cfg: &Config, svc: &MemoryService, path: &Path) -> Result /// Remove the document for a deleted/removed file path. pub async fn remove_one(svc: &MemoryService, path: &Path) -> Result { let id = path_id(&path.to_string_lossy()); - svc.forget(&id).await + svc.forget(&id, crate::memory::Source::Crawler).await } /// Watch configured roots and keep the index in sync. Runs until cancelled. diff --git a/src/daemon.rs b/src/daemon.rs index 14748a8..958b081 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -41,7 +41,11 @@ pub async fn serve() -> Result<()> { .ensure_index(&cfg.embedder.source, &cfg.embedder.model) .await .context("ensuring memories index")?; - tracing::info!("Meilisearch ready; index configured"); + svc.events() + .ensure() + .await + .context("ensuring memory_events index")?; + tracing::info!("Meilisearch ready; indexes configured"); // 3. Crawler + watcher in the background. let crawl_cfg = cfg.clone(); diff --git a/src/history.rs b/src/history.rs new file mode 100644 index 0000000..2d291b1 --- /dev/null +++ b/src/history.rs @@ -0,0 +1,332 @@ +//! Memory mutation history: an audit timeline of create/update/delete events +//! (plus one rolled-up event per crawl), stored in a dedicated `memory_events` +//! Meilisearch index. +//! +//! The log is metadata-only — it records *that* a memory changed, by which +//! source and when, but never retains old content or diffs. Writes are +//! best-effort and off the critical path: a logging failure must never block +//! or fail a memory mutation. + +use crate::meili::MeiliClient; +use crate::memory::Source; +use crate::memory::model::now_secs; +use anyhow::Result; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +/// The dedicated audit-log index. +pub const EVENTS_INDEX: &str = "memory_events"; + +/// What happened to a memory. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EventAction { + Create, + Update, + Delete, + /// A crawl pass — one summary event, not one per file. + Crawl, +} + +impl EventAction { + pub fn as_str(&self) -> &'static str { + match self { + EventAction::Create => "create", + EventAction::Update => "update", + EventAction::Delete => "delete", + EventAction::Crawl => "crawl", + } + } + + /// Parse a user-supplied action string (lenient). + pub fn parse(s: &str) -> Option { + match s.trim().to_lowercase().as_str() { + "create" | "created" | "add" => Some(EventAction::Create), + "update" | "updated" | "edit" => Some(EventAction::Update), + "delete" | "deleted" | "forget" | "remove" => Some(EventAction::Delete), + "crawl" | "crawled" => Some(EventAction::Crawl), + _ => None, + } + } +} + +/// One audit-log document, mirroring the `memory_events` index schema. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryEvent { + pub id: String, + pub ts: i64, + pub action: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub memory_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub r#type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub scope: Option, + pub source: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source_client: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +impl MemoryEvent { + /// A create/update/delete event affecting a single memory. + #[allow(clippy::too_many_arguments)] + pub fn mutation( + action: EventAction, + memory_id: &str, + title: Option, + r#type: Option, + scope: Option, + source: Source, + source_client: Option, + ) -> Self { + Self { + id: uuid::Uuid::now_v7().to_string(), + ts: now_secs(), + action: action.as_str().to_string(), + memory_id: Some(memory_id.to_string()), + title, + r#type, + scope, + source: source.as_str().to_string(), + source_client, + detail: None, + } + } + + /// A rolled-up summary event for one crawl pass. + pub fn crawl(detail: String) -> Self { + Self { + id: uuid::Uuid::now_v7().to_string(), + ts: now_secs(), + action: EventAction::Crawl.as_str().to_string(), + memory_id: None, + title: None, + r#type: None, + scope: None, + source: Source::Crawler.as_str().to_string(), + source_client: None, + detail: Some(detail), + } + } +} + +/// Filters for querying the history (most recent first). +#[derive(Debug, Clone)] +pub struct EventQuery { + pub action: Option, + pub r#type: Option, + pub scope: Option, + pub since: Option, + pub limit: usize, +} + +impl Default for EventQuery { + fn default() -> Self { + Self { + action: None, + r#type: None, + scope: None, + since: None, + limit: 20, + } + } +} + +/// The audit log: a thin wrapper over a [`MeiliClient`] bound to the +/// `memory_events` index. +#[derive(Clone)] +pub struct EventLog { + client: MeiliClient, +} + +impl EventLog { + /// Build a log from any client (rebinds it to the events index). + pub fn from_client(client: &MeiliClient) -> Self { + Self { + client: client.for_index(EVENTS_INDEX), + } + } + + /// Ensure the events index exists with its settings. + pub async fn ensure(&self) -> Result<()> { + self.client.ensure_log_index().await + } + + /// Append one event. Best-effort: logs and swallows any error so a logging + /// failure never affects the caller's mutation. + pub async fn record(&self, event: MemoryEvent) { + if let Err(e) = self.client.insert_no_wait(&[event]).await { + tracing::debug!("recording memory event failed: {e}"); + } + } + + /// Query the timeline, newest first. Returns raw event rows. + pub async fn query(&self, q: &EventQuery) -> Result> { + let mut body = json!({ + "q": "", + "limit": q.limit.max(1), + "sort": ["ts:desc"], + }); + let filters = build_event_filters(q); + if !filters.is_empty() { + body["filter"] = Value::String(filters.join(" AND ")); + } + let resp = self.client.search(&body).await?; + Ok(resp + .get("hits") + .and_then(|h| h.as_array()) + .cloned() + .unwrap_or_default()) + } + + /// Total number of recorded events, if the index is reachable. + pub async fn count(&self) -> Option { + self.client + .stats() + .await + .ok()? + .get("numberOfDocuments") + .and_then(|n| n.as_u64()) + } +} + +/// Build a Meilisearch filter expression list from an event query. +fn build_event_filters(q: &EventQuery) -> Vec { + let mut f = Vec::new(); + if let Some(action) = q.action { + f.push(format!("action = '{}'", action.as_str())); + } + if let Some(ty) = &q.r#type { + f.push(format!("type = '{}'", escape(ty))); + } + if let Some(scope) = &q.scope { + f.push(format!("scope = '{}'", escape(scope))); + } + if let Some(since) = q.since { + f.push(format!("ts >= {since}")); + } + f +} + +/// Escape single quotes in a filter literal. +fn escape(s: &str) -> String { + s.replace('\'', "\\'") +} + +/// Render one event row as a single human-readable line (for `memd history`). +pub fn render_event(ev: &Value) -> String { + let s = |k: &str| ev.get(k).and_then(|v| v.as_str()).unwrap_or(""); + let when = format_ts(ev.get("ts").and_then(|v| v.as_i64()).unwrap_or(0)); + let action = s("action"); + + if action == "crawl" { + let detail = s("detail"); + return format!("{when} crawl {detail} — crawler"); + } + + let ty = s("type"); + let ty_tag = if ty.is_empty() { + String::new() + } else { + format!("[{ty}] ") + }; + let title = { + let t = s("title"); + if t.is_empty() { "(untitled)" } else { t } + }; + let scope = { + let sc = s("scope"); + if sc.is_empty() { "—" } else { sc } + }; + format!( + "{when} {action:<6} {ty_tag}{title} — {source} · {scope} · id {id}", + source = s("source"), + id = s("memory_id"), + ) +} + +/// Format unix seconds as RFC3339 (falls back to the raw number). +fn format_ts(secs: i64) -> String { + time::OffsetDateTime::from_unix_timestamp(secs) + .ok() + .and_then(|t| { + t.format(&time::format_description::well_known::Rfc3339) + .ok() + }) + .unwrap_or_else(|| secs.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn action_parse_is_lenient() { + assert_eq!(EventAction::parse("created"), Some(EventAction::Create)); + assert_eq!(EventAction::parse("FORGET"), Some(EventAction::Delete)); + assert_eq!(EventAction::parse("crawl"), Some(EventAction::Crawl)); + assert_eq!(EventAction::parse("bogus"), None); + } + + #[test] + fn mutation_event_carries_metadata() { + let ev = MemoryEvent::mutation( + EventAction::Update, + "abc", + Some("My title".into()), + Some("fact".into()), + Some("global".into()), + Source::Mcp, + None, + ); + assert_eq!(ev.action, "update"); + assert_eq!(ev.memory_id.as_deref(), Some("abc")); + assert_eq!(ev.source, "mcp"); + assert!(ev.detail.is_none()); + } + + #[test] + fn crawl_event_has_detail_and_no_memory() { + let ev = MemoryEvent::crawl("3 indexed, 1 deleted".into()); + assert_eq!(ev.action, "crawl"); + assert_eq!(ev.source, "crawler"); + assert!(ev.memory_id.is_none()); + assert_eq!(ev.detail.as_deref(), Some("3 indexed, 1 deleted")); + } + + #[test] + fn builds_event_filters() { + let q = EventQuery { + action: Some(EventAction::Delete), + scope: Some("global".into()), + since: Some(100), + ..Default::default() + }; + let f = build_event_filters(&q); + assert!(f.contains(&"action = 'delete'".to_string())); + assert!(f.contains(&"scope = 'global'".to_string())); + assert!(f.contains(&"ts >= 100".to_string())); + } + + #[test] + fn renders_mutation_and_crawl_lines() { + let mutation = json!({ + "ts": 0, "action": "update", "type": "fact", + "title": "My title", "source": "mcp", "scope": "global", "memory_id": "abc" + }); + let line = render_event(&mutation); + assert!(line.contains("update")); + assert!(line.contains("[fact] My title")); + assert!(line.contains("mcp · global · id abc")); + + let crawl = json!({ "ts": 0, "action": "crawl", "detail": "3 indexed" }); + let line = render_event(&crawl); + assert!(line.contains("crawl")); + assert!(line.contains("3 indexed")); + assert!(line.contains("crawler")); + } +} diff --git a/src/main.rs b/src/main.rs index 9ffb657..1fc66bf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,7 @@ mod cli; mod config; mod crawler; mod daemon; +mod history; mod launchd; mod mcp; mod meili; @@ -90,6 +91,24 @@ enum Command { /// The memory id. id: String, }, + /// Show the history of memory changes (created, updated, deleted, crawled). + History { + /// Filter by action: create, update, delete, or crawl. + #[arg(long)] + action: Option, + /// Filter by memory type. + #[arg(long, value_name = "TYPE")] + r#type: Option, + /// Filter by scope. + #[arg(long)] + scope: Option, + /// Only events since this unix timestamp or relative span (e.g. 30d, 12h). + #[arg(long)] + since: Option, + /// Maximum number of events. + #[arg(long, default_value_t = 20)] + limit: usize, + }, /// Manage passive ingestion (the crawler). Crawl { #[command(subcommand)] @@ -199,6 +218,13 @@ async fn main() -> anyhow::Result<()> { limit, } => cli::search(query, r#type, since, semantic_ratio, limit).await, Command::Forget { id } => cli::forget(id).await, + Command::History { + action, + r#type, + scope, + since, + limit, + } => cli::history(action, r#type, scope, since, limit).await, Command::Crawl { action } => match action { CrawlAction::Run => cli::crawl_run().await, CrawlAction::Status => cli::crawl_status().await, diff --git a/src/mcp/protocol.rs b/src/mcp/protocol.rs index a1fa100..d6a1728 100644 --- a/src/mcp/protocol.rs +++ b/src/mcp/protocol.rs @@ -5,6 +5,7 @@ //! and the `notifications/initialized` no-op. Tool calls are translated into //! [`MemoryService`] operations. +use crate::history::{EventAction, EventQuery}; use crate::memory::{ GetRequest, MemoryService, MemoryType, ProjectionOptions, QueryResult, SaveRequest, Source, }; @@ -161,6 +162,20 @@ fn tool_defs() -> Value { "group_by": { "type": "array", "items": { "type": "string" }, "description": "Fields to group counts by (default: type, scope, source)." } } } + }, + { + "name": "history", + "description": "Audit timeline of memory changes (most recent first): which memories were created, updated, deleted, or crawled, when, and by which source. Metadata only — no old content. Optionally filter by action/type/scope/time.", + "inputSchema": { + "type": "object", + "properties": { + "action": { "type": "string", "description": "create, update, delete, or crawl." }, + "type": { "type": "string", "description": "Filter by memory type." }, + "scope": { "type": "string", "description": "Filter by scope." }, + "since": { "type": "integer", "description": "Unix seconds lower bound on the event timestamp." }, + "limit": { "type": "integer", "description": "Max events (default 20)." } + } + } } ]) } @@ -177,6 +192,7 @@ async fn handle_tool_call(svc: &MemoryService, id: Value, params: Value) -> Valu "forget_memory" => forget_memory(svc, args).await, "list_memories" => list_memories(svc, args).await, "stats" => stats(svc, args).await, + "history" => history(svc, args).await, other => Err(anyhow::anyhow!("unknown tool: {other}")), }; @@ -268,6 +284,7 @@ async fn update_memory(svc: &MemoryService, args: Value) -> anyhow::Result anyhow::Result anyhow::Result { svc.stats(&group_by).await } +async fn history(svc: &MemoryService, args: Value) -> anyhow::Result { + let query = EventQuery { + action: args + .get("action") + .and_then(|a| a.as_str()) + .and_then(EventAction::parse), + r#type: str_field(&args, "type"), + scope: str_field(&args, "scope"), + since: args.get("since").and_then(|s| s.as_i64()), + limit: args + .get("limit") + .and_then(|l| l.as_u64()) + .map(|n| n as usize) + .unwrap_or(20), + }; + let events = svc.history(&query).await?; + Ok(json!({ "count": events.len(), "events": events })) +} + // --- helpers --------------------------------------------------------------- fn str_field(args: &Value, key: &str) -> Option { diff --git a/src/meili/client.rs b/src/meili/client.rs index c5f7e65..4051347 100644 --- a/src/meili/client.rs +++ b/src/meili/client.rs @@ -15,6 +15,9 @@ pub struct MeiliClient { http: reqwest::Client, base: String, key: String, + /// The index this client targets. Defaults to [`INDEX`]; use + /// [`MeiliClient::for_index`] to bind a clone to another index. + index: String, } impl MeiliClient { @@ -27,6 +30,16 @@ impl MeiliClient { http, base: base.into(), key: key.into(), + index: INDEX.to_string(), + } + } + + /// A clone of this client bound to a different index (shares the same HTTP + /// connection pool, base URL, and key). + pub fn for_index(&self, index: &str) -> Self { + Self { + index: index.to_string(), + ..self.clone() } } @@ -123,15 +136,7 @@ impl MeiliClient { .patch("/experimental-features", &json!({ "vectorStore": true })) .await; - // Create the index (ignore "already exists"). - let create = self - .post("/indexes", &json!({ "uid": INDEX, "primaryKey": "id" })) - .await; - if let Ok(v) = create - && let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) - { - let _ = self.wait_task(uid).await; // tolerate "index already exists" - } + self.create_index().await; let settings = json!({ "searchableAttributes": ["title", "content", "summary", "tags"], @@ -148,8 +153,40 @@ impl MeiliClient { } } }); + self.apply_settings(&settings).await + } + + /// Ensure the `memory_events` audit-log index exists with its settings. It + /// is a plain log — no embedder. Idempotent; safe on every daemon start. + pub async fn ensure_log_index(&self) -> Result<()> { + self.create_index().await; + let settings = json!({ + "searchableAttributes": ["title", "detail", "memory_id"], + "filterableAttributes": ["action", "type", "scope", "source", "memory_id", "ts"], + "sortableAttributes": ["ts"], + }); + self.apply_settings(&settings).await + } + + /// Create `self.index` (ignoring "already exists"). + async fn create_index(&self) { + let create = self + .post( + "/indexes", + &json!({ "uid": self.index, "primaryKey": "id" }), + ) + .await; + if let Ok(v) = create + && let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) + { + let _ = self.wait_task(uid).await; // tolerate "index already exists" + } + } + + /// PATCH settings onto `self.index` and wait for the task. + async fn apply_settings(&self, settings: &Value) -> Result<()> { let v = self - .patch(&format!("/indexes/{INDEX}/settings"), &settings) + .patch(&format!("/indexes/{}/settings", self.index), settings) .await .context("applying index settings")?; if let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) { @@ -158,11 +195,20 @@ impl MeiliClient { Ok(()) } + /// Insert documents without waiting for the indexing task to finish. Used + /// for the best-effort event log: eventual consistency is acceptable and + /// we must not add task-polling latency to every memory mutation. + pub async fn insert_no_wait(&self, docs: &[T]) -> Result<()> { + self.post(&format!("/indexes/{}/documents", self.index), &docs) + .await + .map(|_| ()) + } + /// Upsert one document and wait for the task to finish (so embeddings are /// computed before we return). pub async fn upsert(&self, doc: &T) -> Result<()> { let v = self - .post(&format!("/indexes/{INDEX}/documents"), &[doc]) + .post(&format!("/indexes/{}/documents", self.index), &[doc]) .await .context("adding document")?; if let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) { @@ -174,7 +220,7 @@ impl MeiliClient { /// Upsert many documents in a single task and wait for completion. pub async fn upsert_many(&self, docs: &[T]) -> Result<()> { let v = self - .post(&format!("/indexes/{INDEX}/documents"), &docs) + .post(&format!("/indexes/{}/documents", self.index), &docs) .await .context("adding documents")?; if let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) { @@ -186,7 +232,7 @@ impl MeiliClient { /// Delete one document by id and wait for completion. pub async fn delete_doc(&self, id: &str) -> Result { let v = self - .delete(&format!("/indexes/{INDEX}/documents/{id}")) + .delete(&format!("/indexes/{}/documents/{id}", self.index)) .await?; if let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) { let task = self.wait_task(uid).await?; @@ -207,7 +253,10 @@ impl MeiliClient { return Ok(0); } let v = self - .post(&format!("/indexes/{INDEX}/documents/delete-batch"), &ids) + .post( + &format!("/indexes/{}/documents/delete-batch", self.index), + &ids, + ) .await .context("batch deleting documents")?; if let Some(uid) = v.get("taskUid").and_then(|t| t.as_u64()) { @@ -226,7 +275,7 @@ impl MeiliClient { pub async fn get_doc(&self, id: &str) -> Result> { let resp = self .http - .get(self.url(&format!("/indexes/{INDEX}/documents/{id}"))) + .get(self.url(&format!("/indexes/{}/documents/{id}", self.index))) .bearer_auth(&self.key) .send() .await?; @@ -239,14 +288,14 @@ impl MeiliClient { /// Run a search. `hybrid` (embedder + semantic ratio) is included when /// `semantic_ratio` is `Some`. pub async fn search(&self, body: &Value) -> Result { - self.post(&format!("/indexes/{INDEX}/search"), body) + self.post(&format!("/indexes/{}/search", self.index), body) .await .context("searching index") } /// Index document/storage statistics. pub async fn stats(&self) -> Result { - self.get(&format!("/indexes/{INDEX}/stats")).await + self.get(&format!("/indexes/{}/stats", self.index)).await } /// Trigger a dump of the whole instance and wait for it to finish. diff --git a/src/memory/service.rs b/src/memory/service.rs index 1318443..f901735 100644 --- a/src/memory/service.rs +++ b/src/memory/service.rs @@ -6,6 +6,7 @@ use super::classify; use super::model::{MemoryItem, MemoryType, Source, now_secs}; use crate::config::Config; +use crate::history::{EventAction, EventLog, EventQuery, MemoryEvent}; use crate::meili::MeiliClient; use anyhow::{Context, Result}; use serde_json::{Value, json}; @@ -99,13 +100,16 @@ const META_FIELDS: &[&str] = &[ #[derive(Clone)] pub struct MemoryService { client: MeiliClient, + events: EventLog, default_semantic_ratio: f32, } impl MemoryService { pub fn new(client: MeiliClient, default_semantic_ratio: f32) -> Self { + let events = EventLog::from_client(&client); Self { client, + events, default_semantic_ratio, } } @@ -120,6 +124,45 @@ impl MemoryService { &self.client } + /// The audit log of memory mutations. + pub fn events(&self) -> &EventLog { + &self.events + } + + /// Query the mutation history (newest first). + pub async fn history(&self, q: &EventQuery) -> Result> { + self.events.query(q).await + } + + /// Record a create/update/delete event, best-effort. Crawler-sourced + /// mutations are skipped here — they are summarized once per crawl pass. + #[allow(clippy::too_many_arguments)] + async fn record_mutation( + &self, + action: EventAction, + memory_id: &str, + title: Option, + ty: Option, + scope: Option, + source: Source, + source_client: Option, + ) { + if source == Source::Crawler { + return; + } + self.events + .record(MemoryEvent::mutation( + action, + memory_id, + title, + ty, + scope, + source, + source_client, + )) + .await; + } + /// Persist a memory. Classifies the type if absent, dedups on content hash, /// stamps timestamps, and upserts (Meilisearch embeds it locally). /// Returns the document id. @@ -156,6 +199,16 @@ impl MemoryService { content_hash: hash, }; self.client.upsert(&item).await?; + self.record_mutation( + EventAction::Create, + &id, + item.title.clone(), + Some(item.r#type.clone()), + Some(item.scope.clone()), + source, + item.source_client.clone(), + ) + .await; Ok(id) } @@ -270,12 +323,36 @@ impl MemoryService { } } - /// Delete a memory by id. Returns whether something was deleted. - pub async fn forget(&self, id: &str) -> Result { - self.client.delete_doc(id).await + /// Delete a memory by id. Returns whether something was deleted. Records a + /// `delete` event (with a metadata snapshot taken before deletion), unless + /// the deletion came from the crawler. + pub async fn forget(&self, id: &str, source: Source) -> Result { + // Snapshot metadata before deletion so the audit row stays readable. + let meta = self.client.get_doc(id).await.ok().flatten(); + let deleted = self.client.delete_doc(id).await?; + if deleted { + let field = |k: &str| { + meta.as_ref() + .and_then(|d| d.get(k)) + .and_then(|v| v.as_str()) + .map(String::from) + }; + self.record_mutation( + EventAction::Delete, + id, + field("title"), + field("type"), + field("scope"), + source, + None, + ) + .await; + } + Ok(deleted) } /// Update fields on an existing memory (partial). Returns false if absent. + #[allow(clippy::too_many_arguments)] pub async fn update( &self, id: &str, @@ -284,6 +361,7 @@ impl MemoryService { tags: Option>, scope: Option, ty: Option, + source: Source, ) -> Result { let Some(mut doc) = self.client.get_doc(id).await? else { return Ok(false); @@ -306,6 +384,17 @@ impl MemoryService { } doc["updated_at"] = json!(now_secs()); self.client.upsert(&doc).await?; + let field = |k: &str| doc.get(k).and_then(|v| v.as_str()).map(String::from); + self.record_mutation( + EventAction::Update, + id, + field("title"), + field("type"), + field("scope"), + source, + None, + ) + .await; Ok(true) }