1use std::io::{BufRead, BufReader, Read, Write};
47use std::path::Path;
48
49use serde::{Deserialize, Serialize};
50use serde_json::{json, Map, Value};
51
52use crate::error::{MnemoError, Result};
53use crate::memory::{Memory, MemoryType, Metric, Scope};
54use crate::result_cache::{CachePutOpts, SemanticCachePutOpts, DEFAULT_SEMANTIC_THRESHOLD};
55use crate::store::{Mnemo, RecallRequest};
56
57const PROTOCOL_VERSION: &str = "2025-06-18";
59
60const PASSPHRASE_ENV: &str = "MNEMO_PASSPHRASE";
62
63#[doc(hidden)]
68pub fn __fuzz_parse_request_line(line: &str) -> std::result::Result<(), String> {
69 serde_json::from_str::<Request>(line)
70 .map(|_| ())
71 .map_err(|e| e.to_string())
72}
73
74#[derive(Debug, Deserialize)]
79struct Request {
80 #[serde(default)]
81 jsonrpc: String,
82 #[serde(default)]
83 id: Option<Value>,
84 method: String,
85 #[serde(default)]
86 params: Value,
87}
88
89#[derive(Debug, Serialize)]
91struct Response {
92 jsonrpc: &'static str,
93 id: Value,
94 #[serde(skip_serializing_if = "Option::is_none")]
95 result: Option<Value>,
96 #[serde(skip_serializing_if = "Option::is_none")]
97 error: Option<RpcError>,
98}
99
100#[derive(Debug, Serialize)]
102struct RpcError {
103 code: i32,
104 message: String,
105 #[serde(skip_serializing_if = "Option::is_none")]
106 data: Option<Value>,
107}
108
109impl RpcError {
110 fn new(code: i32, message: impl Into<String>) -> Self {
111 Self {
112 code,
113 message: message.into(),
114 data: None,
115 }
116 }
117}
118
119const PARSE_ERROR: i32 = -32700;
121const INVALID_REQUEST: i32 = -32600;
122const METHOD_NOT_FOUND: i32 = -32601;
123const INVALID_PARAMS: i32 = -32602;
124const INTERNAL_ERROR: i32 = -32603;
125
126pub fn serve_stdio(path: &Path) -> Result<()> {
139 serve_stdio_with_embedder(path, None)
140}
141
142pub fn serve_stdio_with_embedder(
152 path: &Path,
153 embed_spec: Option<(String, String, bool, std::time::Duration)>,
154) -> Result<()> {
155 let stdin = std::io::stdin();
156 let stdout = std::io::stdout();
157 serve_with_streams(path, stdin.lock(), stdout.lock(), embed_spec)
158}
159
160pub fn serve_with_streams<R: Read, W: Write>(
164 path: &Path,
165 reader: R,
166 mut writer: W,
167 embed_spec: Option<(String, String, bool, std::time::Duration)>,
168) -> Result<()> {
169 let passphrase = std::env::var(PASSPHRASE_ENV).map_err(|_| {
170 MnemoError::Invalid(format!(
171 "{PASSPHRASE_ENV} is required to serve — set it in the shell before starting"
172 ))
173 })?;
174
175 eprintln!(
176 "mnemo mcp: opening {} (protocol {PROTOCOL_VERSION})",
177 path.display()
178 );
179 let mut db = Mnemo::open(path, &passphrase)?;
180 if let Some((cmd, name, normalize, timeout)) = embed_spec {
181 let identity = crate::EmbedderIdentity {
182 name,
183 dimensions: db.dimensions(),
184 normalize,
185 };
186 let embedder = crate::EmbedCmd::new(&cmd, identity)?.with_timeout(timeout);
187 db.set_embedder(embedder)?;
188 eprintln!(
189 "mnemo mcp: embedder ready (text params accepted, timeout {}s)",
190 timeout.as_secs()
191 );
192 }
193 eprintln!("mnemo mcp: ready ({} live memories)", db.len());
194
195 let reader = BufReader::new(reader);
196 for line in reader.lines() {
197 let line = line.map_err(MnemoError::Io)?;
198 let trimmed = line.trim();
199 if trimmed.is_empty() {
200 continue;
201 }
202
203 let (id, out) = match serde_json::from_str::<Request>(trimmed) {
204 Ok(req) => {
205 let id = req.id.clone();
206 let is_notification = id.is_none();
207 let result = handle(&mut db, &req);
208 if is_notification {
209 continue;
211 }
212 (id.unwrap_or(Value::Null), result)
213 }
214 Err(e) => (
215 Value::Null,
216 Err(RpcError::new(PARSE_ERROR, format!("parse error: {e}"))),
217 ),
218 };
219
220 let resp = match out {
221 Ok(value) => Response {
222 jsonrpc: "2.0",
223 id,
224 result: Some(value),
225 error: None,
226 },
227 Err(err) => Response {
228 jsonrpc: "2.0",
229 id,
230 result: None,
231 error: Some(err),
232 },
233 };
234
235 let line =
236 serde_json::to_string(&resp).map_err(|e| MnemoError::Serialize(e.to_string()))?;
237 writeln!(writer, "{line}").map_err(MnemoError::Io)?;
238 writer.flush().map_err(MnemoError::Io)?;
239 }
240 if let Err(e) = db.close() {
254 eprintln!("mnemo mcp: flush on shutdown failed: {e}");
255 }
256 eprintln!("mnemo mcp: stdin closed, shutting down");
257 Ok(())
258}
259
260fn handle(db: &mut Mnemo, req: &Request) -> std::result::Result<Value, RpcError> {
265 if req.jsonrpc != "2.0" {
266 return Err(RpcError::new(
267 INVALID_REQUEST,
268 format!("expected jsonrpc \"2.0\", got {:?}", req.jsonrpc),
269 ));
270 }
271 match req.method.as_str() {
272 "initialize" => Ok(initialize_result(&req.params)),
273 "notifications/initialized" | "notifications/cancelled" => Ok(Value::Null),
274 "ping" => Ok(json!({})),
275 "tools/list" => Ok(tools_list()),
276 "tools/call" => tools_call(db, &req.params),
277 other => Err(RpcError::new(
278 METHOD_NOT_FOUND,
279 format!("method '{other}' not implemented"),
280 )),
281 }
282}
283
284fn initialize_result(params: &Value) -> Value {
287 let client_version = params
288 .get("protocolVersion")
289 .and_then(|v| v.as_str())
290 .unwrap_or(PROTOCOL_VERSION)
291 .to_string();
292 json!({
293 "protocolVersion": client_version,
294 "capabilities": {
295 "tools": { "listChanged": false },
296 },
297 "serverInfo": {
298 "name": "mnemo",
299 "version": env!("CARGO_PKG_VERSION"),
300 },
301 })
302}
303
304fn tools_list() -> Value {
310 json!({
311 "tools": [
312 {
313 "name": "about",
314 "description": "Read-only briefing for this .mnemo file: manifest scaffold plus any memories tagged as onboarding, then the project essence (conventions, decisions, open items) as a final `{kind: \"essence\", render}` element when the file has one. Run this first to learn how the file expects to be used.",
315 "inputSchema": {
316 "type": "object",
317 "properties": {},
318 "additionalProperties": false
319 }
320 },
321 {
322 "name": "remember",
323 "description": "Insert, overwrite, or UPSERT a memory. Pass EITHER `vector` (explicit embedding) OR `text` (embedded by the server if it was started with --embed-cmd / --embed-name). Phase 12.1: when `key` is set, this becomes a keyed upsert — if a live memory already holds this `(agent + key)` slot the prior record is marked `_mnemo.superseded_by` and a new one is inserted, and an identical re-write is a no-op. **PREFER `key` for any fact that replaces an earlier one** (versions, current status, user preferences, plan outcomes). Persists to disk before returning. Returns `{id, outcome}` where `outcome` is `added`, `superseded`, `unchanged`, or (no key) `inserted`.",
324 "inputSchema": {
325 "type": "object",
326 "required": ["content", "memory_type"],
327 "properties": {
328 "content": { "type": "string" },
329 "memory_type": { "type": "string", "enum": ["episodic", "semantic", "procedural", "working"] },
330 "vector": { "type": "array", "items": { "type": "number" }, "description": "Explicit embedding; mutually exclusive with `text`." },
331 "text": { "type": "string", "description": "Content to embed via the server's registered embedder; mutually exclusive with `vector`. Requires the server to have been started with --embed-cmd." },
332 "importance": { "type": "number", "minimum": 0, "maximum": 1 },
333 "agent_id": { "type": "string" },
334 "session_id": { "type": "string" },
335 "ttl_secs": { "type": "integer" },
336 "shared": { "type": "boolean" },
337 "metadata": { "type": "object" },
338 "key": { "type": "string", "description": "Phase 12.1 upsert slot name. A later `remember` with the same `key` + agent scope supersedes the prior record; an identical re-write is a no-op. Use this for any fact that replaces an earlier one." }
339 },
340 "additionalProperties": false
341 }
342 },
343 {
344 "name": "update",
345 "description": "Edit a memory in place (Phase 12.1). Preserves the ULID, `created_at` and access stats. Any of `content` (needs `vector` to avoid embedding drift), `text` (re-embeds via the server), `importance`, `metadata`, `ttl_secs` can change. An identical patch returns `{id, outcome: 'unchanged'}` without writing. Prefer `remember` with `key` for facts that replace earlier ones; use `update` for in-place corrections to one specific record.",
346 "inputSchema": {
347 "type": "object",
348 "required": ["id"],
349 "properties": {
350 "id": { "type": "string", "description": "ULID of the memory to update." },
351 "content": { "type": "string", "description": "New content. Requires a matching `vector` (use `text` to re-embed instead)." },
352 "vector": { "type": "array", "items": { "type": "number" } },
353 "text": { "type": "string", "description": "New content; the server embeds it through its registered embedder. Mutually exclusive with `content` + `vector`." },
354 "importance": { "type": "number", "minimum": 0, "maximum": 1 },
355 "metadata": { "type": "object", "description": "Replace the user-visible metadata. Reserved `_mnemo.*` keys on the record are preserved." },
356 "ttl_secs": { "type": "integer", "description": "Set the TTL to this many seconds. Use `clear_ttl: true` to drop any TTL." },
357 "clear_ttl": { "type": "boolean", "description": "Clear any TTL on the record." }
358 },
359 "additionalProperties": false
360 }
361 },
362 {
363 "name": "essence",
364 "description": "Phase 12.7b: edit or render the essence layer of a thread (or the project layer). ONE tool for every op — set `op` to `add` / `append` / `update` / `resolve` / `remove` / `render`. **Call it when the goal, a decision, an open problem, the next step or the user's way of working changes, and append a timeline entry at each milestone.** Section items (`intent`, `conventions`, `decisions`, `state`, `open`, `next`, `lessons`) are ≤ 280 bytes and have an `open → resolved` lifecycle; `append` adds an append-only `timeline` entry (≤ 600 bytes). Thread-scoped (`thread`) or project-scoped (`project: true`). `render` returns the deterministic markdown view, budgeted at `budget` bytes (default 6144).",
365 "inputSchema": {
366 "type": "object",
367 "required": ["op"],
368 "properties": {
369 "op": { "type": "string", "enum": ["add", "append", "update", "resolve", "remove", "render"] },
370 "thread": { "type": "string", "description": "Thread id. Omit with `project: true` for the project layer." },
371 "project": { "type": "boolean", "default": false, "description": "Project-scoped item (shows in every thread's render). Mutually exclusive with `thread`." },
372 "section": { "type": "string", "enum": ["intent", "conventions", "decisions", "state", "open", "next", "lessons"], "description": "Required for `op: add`. Timeline uses `op: append`, not add." },
373 "id": { "type": "string", "description": "ULID of the section item. Required for update / resolve / remove." },
374 "text": { "type": "string", "description": "Required for add / append / update." },
375 "refs": { "type": "array", "items": { "type": "string" }, "description": "Optional free-form reference strings (turn ids, memory ids, commits, paths)." },
376 "budget": { "type": "integer", "default": 6144, "description": "Byte budget for `op: render`." }
377 },
378 "additionalProperties": false
379 }
380 },
381 {
382 "name": "thread_start",
383 "description": "Phase 12.7c: start a new thread that continues from a previous one. `essence {op: render, thread: <new>}` then pulls the prior thread's open items, `next` and last-N timeline entries ahead of the new thread's own essence — the 'new thread, same experience' path. Returns `{id}` (the link-record ULID).",
384 "inputSchema": {
385 "type": "object",
386 "required": ["new", "from"],
387 "properties": {
388 "new": { "type": "string", "description": "New thread id." },
389 "from": { "type": "string", "description": "Previous thread id to continue from." }
390 },
391 "additionalProperties": false
392 }
393 },
394 {
395 "name": "commit_deltas",
396 "description": "Atomic batch write: facts, essence ops and thread archive turns in one call. Every item is checked before anything is written, so an invalid item writes nothing; one flush on success. Facts may omit `vector`: the server embeds every fact `content`, essence text and turn in one batched call with its embedder (`--embed-cmd` or builtin). Use at a session end to save a burst of updates in one round trip. `turn_policy` defaults to `archive`; `expire_after` takes a TTL in seconds; `drop` discards turns. Returns a counts summary.",
397 "inputSchema": {
398 "type": "object",
399 "properties": {
400 "facts": {
401 "type": "array",
402 "items": {
403 "type": "object",
404 "required": ["content", "memory_type"],
405 "properties": {
406 "content": { "type": "string" },
407 "memory_type": { "type": "string", "enum": ["episodic", "semantic", "procedural", "working"] },
408 "vector": { "type": "array", "items": { "type": "number" }, "description": "Optional. Omit it to have the server embed `content`." },
409 "importance": { "type": "number", "minimum": 0, "maximum": 1 },
410 "agent_id": { "type": "string" },
411 "session_id": { "type": "string" },
412 "shared": { "type": "boolean" },
413 "metadata": { "type": "object" },
414 "key": { "type": "string" }
415 }
416 }
417 },
418 "essence_ops": {
419 "type": "array",
420 "items": {
421 "type": "object",
422 "required": ["kind"],
423 "properties": {
424 "kind": { "type": "string", "enum": ["add", "append_timeline", "update", "resolve", "remove"] },
425 "thread": { "type": "string" },
426 "project": { "type": "boolean" },
427 "section": { "type": "string" },
428 "id": { "type": "string" },
429 "text": { "type": "string" },
430 "refs": { "type": "array", "items": { "type": "string" } }
431 }
432 }
433 },
434 "turns": {
435 "type": "array",
436 "items": {
437 "type": "object",
438 "required": ["turn_id", "role", "content"],
439 "properties": {
440 "turn_id": { "type": "string" },
441 "role": { "type": "string" },
442 "content": { "type": "string" },
443 "timestamp": { "type": "integer" }
444 }
445 }
446 },
447 "thread_id": { "type": "string", "description": "Required when `turns` is non-empty." },
448 "turn_policy": { "type": "string", "enum": ["archive", "expire_after", "drop"], "default": "archive" },
449 "turn_policy_ttl": { "type": "integer", "description": "TTL in seconds; required when `turn_policy = 'expire_after'`." }
450 },
451 "additionalProperties": false
452 }
453 },
454 {
455 "name": "thread_ingest",
456 "description": "Phase 12.7a: ingest a transcript file into a thread's archive layer. Idempotent per `turn_id` — re-running on a longer transcript appends only the new turns. Requires an embedder on the server (`--embed-cmd` at serve start) because every turn is embedded. Returns `{added_records, skipped_existing_turns, chunked_turns, truncated_turns}`.",
457 "inputSchema": {
458 "type": "object",
459 "required": ["thread", "transcript_path"],
460 "properties": {
461 "thread": { "type": "string", "description": "Thread id to append under." },
462 "transcript_path": { "type": "string", "description": "Local path to the transcript file." },
463 "format": { "type": "string", "enum": ["jsonl", "claude-code-jsonl"], "default": "jsonl" }
464 },
465 "additionalProperties": false
466 }
467 },
468 {
469 "name": "thread_export",
470 "description": "Return a thread's archive in the ingest shape: `[{turn_id, role, content, timestamp}]` in order, one entry per turn (long turns' chunks joined), plus `sha256` and `original_length` on a truncated tool result. Feeding it back to ingest for the same thread adds nothing.",
471 "inputSchema": {
472 "type": "object",
473 "required": ["thread"],
474 "properties": {
475 "thread": { "type": "string", "description": "Thread id to export." }
476 },
477 "additionalProperties": false
478 }
479 },
480 {
481 "name": "history",
482 "description": "Walk the full chain for a keyed slot (Phase 12.1), newest first. Entry 0 is the live record (or empty if the slot is empty); subsequent entries are the superseded history. Slot resolution matches keyed `remember`: with no `agent` and `shared=false`, walks the default agent's private slot (v0.8.1 bugfix — pre-v0.8.1 this walked the shared slot by accident and returned empty for the common case). Pass `shared: true` for the shared slot; `agent` + `shared=true` is rejected.",
483 "inputSchema": {
484 "type": "object",
485 "required": ["key"],
486 "properties": {
487 "key": { "type": "string" },
488 "agent": { "type": "string", "description": "Query the private slot for this agent. Mutually exclusive with `shared`." },
489 "shared": { "type": "boolean", "default": false, "description": "Query the shared slot (one per file). Mutually exclusive with `agent`." }
490 },
491 "additionalProperties": false
492 }
493 },
494 {
495 "name": "recall",
496 "description": "Score every live memory (or the ANN-narrowed candidate set) against a query and return the top-k. Pass EITHER `query` (explicit vector) OR `text` (embedded by the server; requires --embed-cmd at serve start). Phase 12.6: add a `filter` object to restrict candidates before scoring (`session_id` plus per-metadata-key conditions: equality, `$in`, `$exists`, `$gte`/`$lte`). Phase 12.1: superseded records never appear. Phase 12.8: set `max_bytes: N` to cap the total UTF-8 bytes of returned `content` fields; the response is then shaped as `{hits: [...], truncated: bool}` so the caller can tell whether the budget cut hits. Without `max_bytes`, the response is a bare JSON array of hits (unchanged).",
497 "inputSchema": {
498 "type": "object",
499 "properties": {
500 "query": { "type": "array", "items": { "type": "number" }, "description": "Explicit query vector; mutually exclusive with `text`." },
501 "text": { "type": "string", "description": "Query text; embedded by the server's registered embedder. Mutually exclusive with `query`." },
502 "top_k": { "type": "integer", "default": 10 },
503 "types": { "type": "array", "items": { "type": "string" } },
504 "agent": { "type": "string" },
505 "metric": { "type": "string", "enum": ["cosine", "l2", "dot"], "default": "cosine" },
506 "max_bytes": { "type": "integer", "minimum": 1, "description": "Phase 12.8 budget: stop adding hits once the UTF-8 bytes of their `content` fields would exceed N. Always returns at least the top hit. When set, the response is `{hits, truncated}` instead of a bare array." },
507 "filter": {
508 "type": "object",
509 "description": "Phase 12.6 pre-scoring filter. All conditions AND together. Shape: `{\"session_id\": \"...\", \"metadata\": {key: value | {\"$in\": [...]} | {\"$exists\": true} | {\"$gte\": n, \"$lte\": n}}}`.",
510 "properties": {
511 "session_id": { "type": "string" },
512 "metadata": { "type": "object" }
513 },
514 "additionalProperties": false
515 }
516 },
517 "additionalProperties": false
518 }
519 },
520 {
521 "name": "forget",
522 "description": "Soft-delete a memory by ULID. Space is reclaimed on the next `mnemo compact`. Persists to disk before returning.",
523 "inputSchema": {
524 "type": "object",
525 "required": ["id"],
526 "properties": {
527 "id": { "type": "string", "description": "ULID of the memory to delete" }
528 },
529 "additionalProperties": false
530 }
531 },
532 {
533 "name": "list",
534 "description": "Return every live memory as JSON. Sorted by creation time, oldest first. Vectors are omitted by default to keep payloads small.",
535 "inputSchema": {
536 "type": "object",
537 "properties": {
538 "include_vectors": { "type": "boolean", "default": false }
539 },
540 "additionalProperties": false
541 }
542 },
543 {
544 "name": "snapshot_list",
545 "description": "List every restorable snapshot (one per committed flush). Each entry gives the transaction ID, creation timestamp, and memory count.",
546 "inputSchema": {
547 "type": "object",
548 "properties": {},
549 "additionalProperties": false
550 }
551 },
552 {
553 "name": "stats",
554 "description": "Summary statistics: memory count, deleted count, agents present, snapshot count, page cache occupancy.",
555 "inputSchema": {
556 "type": "object",
557 "properties": {},
558 "additionalProperties": false
559 }
560 },
561 {
562 "name": "cache_put",
563 "description": "Insert or overwrite an exact-key cache entry. Value is a UTF-8 string; use `content_type` to label the payload. Persists to disk before returning. Namespace and key are caller-chosen; the key is SHA-256 hashed by the engine.",
564 "inputSchema": {
565 "type": "object",
566 "required": ["namespace", "key", "value"],
567 "properties": {
568 "namespace": { "type": "string" },
569 "key": { "type": "string" },
570 "value": { "type": "string" },
571 "content_type": { "type": "string", "default": "text" },
572 "ttl_secs": { "type": "integer", "minimum": 0 }
573 },
574 "additionalProperties": false
575 }
576 },
577 {
578 "name": "cache_get",
579 "description": "Look up an exact-key cache entry. Returns `{value, content_type, hit}` on a hit, `{hit: false}` on miss (missing or TTL-expired). Bumps access stats catalog-only.",
580 "inputSchema": {
581 "type": "object",
582 "required": ["namespace", "key"],
583 "properties": {
584 "namespace": { "type": "string" },
585 "key": { "type": "string" }
586 },
587 "additionalProperties": false
588 }
589 },
590 {
591 "name": "cache_stats",
592 "description": "Cache statistics for a namespace (omit for the whole cache). Returns entries, bytes, hits, misses, hit_rate, evictions.",
593 "inputSchema": {
594 "type": "object",
595 "properties": {
596 "namespace": { "type": "string" }
597 },
598 "additionalProperties": false
599 }
600 },
601 {
602 "name": "cache_put_semantic",
603 "description": "Insert an entry with an embedding vector for semantic recall (Phase 10.2). Pass EITHER `vector` + `model` (explicit) OR `text` (embedded by the server; `model` defaults to the server embedder's identity name). Later cache_get_semantic calls under a different `model` won't see this entry (a different-model hit would be a semantics bug).",
604 "inputSchema": {
605 "type": "object",
606 "required": ["namespace", "key", "value"],
607 "properties": {
608 "namespace": { "type": "string" },
609 "key": { "type": "string" },
610 "vector": { "type": "array", "items": { "type": "number" }, "description": "Explicit embedding; mutually exclusive with `text`. Requires `model`." },
611 "text": { "type": "string", "description": "Content to embed via the server's registered embedder; mutually exclusive with `vector`." },
612 "value": { "type": "string" },
613 "model": { "type": "string", "description": "Required with `vector`; optional with `text` (defaults to the embedder's identity name)." },
614 "content_type": { "type": "string", "default": "text" },
615 "ttl_secs": { "type": "integer", "minimum": 0 }
616 },
617 "additionalProperties": false
618 }
619 },
620 {
621 "name": "cache_get_semantic",
622 "description": "Semantic recall over the cache: top-1 cosine over live vectored entries in the namespace whose `model` matches. Pass EITHER `query` + `model` OR `text` (which embeds via the server and defaults the model to the embedder's identity). Returns `{hit: true, value, similarity, ...}` iff similarity >= threshold, else `{hit: false}`. Threshold defaults to 0.97.",
623 "inputSchema": {
624 "type": "object",
625 "required": ["namespace"],
626 "properties": {
627 "namespace": { "type": "string" },
628 "query": { "type": "array", "items": { "type": "number" }, "description": "Explicit query vector; mutually exclusive with `text`. Requires `model`." },
629 "text": { "type": "string", "description": "Query text; embedded by the server. Mutually exclusive with `query`." },
630 "model": { "type": "string", "description": "Required with `query`; optional with `text`." },
631 "threshold": { "type": "number", "minimum": -1.0, "maximum": 1.0 }
632 },
633 "additionalProperties": false
634 }
635 }
636 ]
637 })
638}
639
640fn tools_call(db: &mut Mnemo, params: &Value) -> std::result::Result<Value, RpcError> {
645 let name = params
646 .get("name")
647 .and_then(|v| v.as_str())
648 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "missing tool name"))?;
649 let empty = json!({});
650 let args = params.get("arguments").unwrap_or(&empty);
651
652 let text = match name {
653 "about" => tool_about(db)?,
654 "remember" => tool_remember(db, args)?,
655 "update" => tool_update(db, args)?,
656 "history" => tool_history(db, args)?,
657 "thread_ingest" => tool_thread_ingest(db, args)?,
658 "thread_export" => tool_thread_export(db, args)?,
659 "thread_start" => tool_thread_start(db, args)?,
660 "commit_deltas" => tool_commit_deltas(db, args)?,
661 "essence" => tool_essence(db, args)?,
662 "recall" => tool_recall(db, args)?,
663 "forget" => tool_forget(db, args)?,
664 "list" => tool_list(db, args)?,
665 "snapshot_list" => tool_snapshot_list(db)?,
666 "stats" => tool_stats(db)?,
667 "cache_put" => tool_cache_put(db, args)?,
668 "cache_get" => tool_cache_get(db, args)?,
669 "cache_stats" => tool_cache_stats(db, args)?,
670 "cache_put_semantic" => tool_cache_put_semantic(db, args)?,
671 "cache_get_semantic" => tool_cache_get_semantic(db, args)?,
672 other => {
673 return Err(RpcError::new(
674 METHOD_NOT_FOUND,
675 format!("unknown tool '{other}'"),
676 ))
677 }
678 };
679
680 Ok(json!({
681 "content": [ { "type": "text", "text": text } ],
682 "isError": false,
683 }))
684}
685
686fn embed_text_via_server(
697 db: &mut Mnemo,
698 tool: &str,
699 text: &str,
700) -> std::result::Result<Vec<f32>, RpcError> {
701 let embedder = db.embedder().ok_or_else(|| {
702 RpcError::new(
703 INVALID_PARAMS,
704 format!(
705 "{tool}: 'text' requires an embedder on the server. \
706 Start with: mnemo serve --mcp <file> --embed-cmd '<cmd>' --embed-name <name>"
707 ),
708 )
709 })?;
710 let mut vs = embedder.embed(&[text]).map_err(mnemo_err)?;
711 vs.pop().ok_or_else(|| {
712 RpcError::new(
713 INTERNAL_ERROR,
714 format!("{tool}: embedder returned no vector for the input text"),
715 )
716 })
717}
718
719fn mnemo_err(e: MnemoError) -> RpcError {
720 RpcError::new(INTERNAL_ERROR, e.to_string())
721}
722
723fn tool_about(db: &mut Mnemo) -> std::result::Result<String, RpcError> {
724 let mems = db.about().map_err(mnemo_err)?;
725 let mut arr: Vec<Value> = mems.iter().map(memory_to_json).collect();
726 if let Some(render) = db.about_essence().map_err(mnemo_err)? {
729 arr.push(json!({ "kind": "essence", "scope": "project", "render": render }));
730 }
731 Ok(serde_json::to_string_pretty(&arr).unwrap_or_else(|_| "[]".into()))
732}
733
734fn tool_thread_export(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
735 let thread = args
736 .get("thread")
737 .and_then(|v| v.as_str())
738 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_export: missing 'thread'"))?;
739 let turns = db.thread_export_turns(thread).map_err(mnemo_err)?;
740 let arr: Vec<Value> = turns.iter().map(|t| t.to_json()).collect();
741 Ok(Value::Array(arr).to_string())
742}
743
744fn tool_remember(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
745 let content = args
746 .get("content")
747 .and_then(|v| v.as_str())
748 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "remember: missing 'content'"))?;
749 let mtype_str = args
750 .get("memory_type")
751 .and_then(|v| v.as_str())
752 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "remember: missing 'memory_type'"))?;
753 let mtype = MemoryType::parse(mtype_str).ok_or_else(|| {
754 RpcError::new(
755 INVALID_PARAMS,
756 format!("remember: unknown memory_type '{mtype_str}'"),
757 )
758 })?;
759 let vector = match (args.get("vector"), args.get("text")) {
762 (Some(_), Some(_)) => {
763 return Err(RpcError::new(
764 INVALID_PARAMS,
765 "remember: pass EITHER 'vector' or 'text', not both",
766 ));
767 }
768 (Some(v), None) => v
769 .as_array()
770 .ok_or_else(|| {
771 RpcError::new(INVALID_PARAMS, "remember: 'vector' must be a JSON array")
772 })?
773 .iter()
774 .map(|v| v.as_f64().map(|f| f as f32))
775 .collect::<Option<Vec<f32>>>()
776 .ok_or_else(|| {
777 RpcError::new(
778 INVALID_PARAMS,
779 "remember: 'vector' must be an array of numbers",
780 )
781 })?,
782 (None, Some(t)) => {
783 let text = t.as_str().ok_or_else(|| {
784 RpcError::new(INVALID_PARAMS, "remember: 'text' must be a string")
785 })?;
786 embed_text_via_server(db, "remember", text)?
787 }
788 (None, None) => {
789 return Err(RpcError::new(
790 INVALID_PARAMS,
791 "remember: missing 'vector' or 'text'",
792 ))
793 }
794 };
795
796 let mut mem = Memory::new(content, mtype, vector);
797 if let Some(agent) = args.get("agent_id").and_then(|v| v.as_str()) {
798 mem = mem.with_agent(agent);
799 }
800 if let Some(sess) = args.get("session_id").and_then(|v| v.as_str()) {
801 mem = mem.with_session(sess);
802 }
803 if let Some(imp) = args.get("importance").and_then(|v| v.as_f64()) {
804 mem = mem.with_importance(imp as f32);
805 }
806 if let Some(ttl) = args.get("ttl_secs").and_then(|v| v.as_i64()) {
807 mem = mem.with_ttl(ttl);
808 }
809 if let Some(shared) = args.get("shared").and_then(|v| v.as_bool()) {
810 if shared {
811 mem = mem.with_scope(Scope::Shared);
812 }
813 }
814 if let Some(meta) = args.get("metadata").and_then(|v| v.as_object()) {
815 for (k, v) in meta {
816 mem.metadata.insert(k.clone(), v.clone());
817 }
818 }
819
820 match args.get("key").and_then(|v| v.as_str()) {
825 None => {
826 let id = db.remember(mem).map_err(mnemo_err)?;
827 db.flush().map_err(mnemo_err)?;
828 Ok(json!({ "id": id.to_string(), "outcome": "inserted" }).to_string())
829 }
830 Some(k) => {
831 let (id, outcome) = db.remember_keyed(mem, k).map_err(mnemo_err)?;
832 db.flush().map_err(mnemo_err)?;
833 let outcome_str = match outcome {
834 crate::WriteOutcome::Added => "added",
835 crate::WriteOutcome::Superseded => "superseded",
836 crate::WriteOutcome::Unchanged => "unchanged",
837 crate::WriteOutcome::Updated => "superseded",
838 };
839 Ok(json!({ "id": id.to_string(), "outcome": outcome_str }).to_string())
840 }
841 }
842}
843
844fn tool_update(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
845 use crate::Ulid;
846 use crate::{MemoryPatch, WriteOutcome};
847 let id_str = args
848 .get("id")
849 .and_then(|v| v.as_str())
850 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: missing 'id'"))?;
851 let id = Ulid::from_string(id_str)
852 .map_err(|_| RpcError::new(INVALID_PARAMS, format!("update: invalid ULID '{id_str}'")))?;
853 let has_text = args.get("text").is_some();
857 let has_content = args.get("content").is_some();
858 let has_vector = args.get("vector").is_some();
859 if has_text && (has_content || has_vector) {
860 return Err(RpcError::new(
861 INVALID_PARAMS,
862 "update: pass EITHER 'text' or 'content'+'vector', not both",
863 ));
864 }
865 let mut patch = MemoryPatch::default();
866 if has_content {
867 let c = args
868 .get("content")
869 .and_then(|v| v.as_str())
870 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: 'content' must be a string"))?;
871 patch.content = Some(c.to_string());
872 }
873 if has_vector {
874 let v = args
875 .get("vector")
876 .and_then(|v| v.as_array())
877 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: 'vector' must be a JSON array"))?
878 .iter()
879 .map(|v| v.as_f64().map(|f| f as f32))
880 .collect::<Option<Vec<f32>>>()
881 .ok_or_else(|| {
882 RpcError::new(
883 INVALID_PARAMS,
884 "update: 'vector' must be an array of numbers",
885 )
886 })?;
887 patch.vector = Some(v);
888 }
889 if let Some(imp) = args.get("importance").and_then(|v| v.as_f64()) {
890 patch.importance = Some(imp as f32);
891 }
892 if let Some(meta) = args.get("metadata").and_then(|v| v.as_object()) {
893 patch.metadata = Some(meta.clone());
894 }
895 let clear_ttl = args
896 .get("clear_ttl")
897 .and_then(|v| v.as_bool())
898 .unwrap_or(false);
899 if clear_ttl {
900 patch.ttl_secs = Some(None);
901 } else if let Some(ttl) = args.get("ttl_secs").and_then(|v| v.as_i64()) {
902 patch.ttl_secs = Some(Some(ttl));
903 }
904
905 let outcome = if has_text {
906 let t = args
907 .get("text")
908 .and_then(|v| v.as_str())
909 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: 'text' must be a string"))?;
910 db.update_text(&id, t, patch).map_err(mnemo_err)?
911 } else {
912 db.update(&id, patch).map_err(mnemo_err)?
913 };
914 db.flush().map_err(mnemo_err)?;
915 let outcome_str = match outcome {
916 WriteOutcome::Updated => "updated",
917 WriteOutcome::Unchanged => "unchanged",
918 WriteOutcome::Added => "updated",
919 WriteOutcome::Superseded => "updated",
920 };
921 Ok(json!({ "id": id.to_string(), "outcome": outcome_str }).to_string())
922}
923
924fn tool_thread_start(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
925 let new = args
926 .get("new")
927 .and_then(|v| v.as_str())
928 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_start: missing 'new'"))?;
929 let from = args
930 .get("from")
931 .and_then(|v| v.as_str())
932 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_start: missing 'from'"))?;
933 let id = db.thread_start(new, from).map_err(mnemo_err)?;
934 db.flush().map_err(mnemo_err)?;
935 Ok(json!({ "id": id.to_string() }).to_string())
936}
937
938fn tool_commit_deltas(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
939 use crate::{
940 CommitBatch, EssenceOp, EssenceOpKind, EssenceScope, EssenceSection, KeyedMemory,
941 ThreadTurn, TurnPolicy, Ulid,
942 };
943 let mut batch = CommitBatch::default();
944
945 if let Some(arr) = args.get("facts").and_then(|v| v.as_array()) {
947 for (i, f) in arr.iter().enumerate() {
948 let obj = f.as_object().ok_or_else(|| {
949 RpcError::new(
950 INVALID_PARAMS,
951 format!("commit_deltas: facts[{i}] must be an object"),
952 )
953 })?;
954 let content = obj.get("content").and_then(|v| v.as_str()).ok_or_else(|| {
955 RpcError::new(
956 INVALID_PARAMS,
957 format!("commit_deltas: facts[{i}] missing 'content'"),
958 )
959 })?;
960 let mtype_str = obj
961 .get("memory_type")
962 .and_then(|v| v.as_str())
963 .ok_or_else(|| {
964 RpcError::new(
965 INVALID_PARAMS,
966 format!("commit_deltas: facts[{i}] missing 'memory_type'"),
967 )
968 })?;
969 let mtype = MemoryType::parse(mtype_str).ok_or_else(|| {
970 RpcError::new(
971 INVALID_PARAMS,
972 format!("commit_deltas: facts[{i}] unknown memory_type '{mtype_str}'"),
973 )
974 })?;
975 let vector = match obj.get("vector") {
979 None | Some(Value::Null) => Vec::new(),
980 Some(v) => v
981 .as_array()
982 .and_then(|a| {
983 a.iter()
984 .map(|x| x.as_f64().map(|f| f as f32))
985 .collect::<Option<Vec<f32>>>()
986 })
987 .ok_or_else(|| {
988 RpcError::new(
989 INVALID_PARAMS,
990 format!("commit_deltas: facts[{i}] vector must be an array of numbers"),
991 )
992 })?,
993 };
994 let mut mem = Memory::new(content, mtype, vector);
995 if let Some(a) = obj.get("agent_id").and_then(|v| v.as_str()) {
996 mem = mem.with_agent(a);
997 }
998 if let Some(s) = obj.get("session_id").and_then(|v| v.as_str()) {
999 mem = mem.with_session(s);
1000 }
1001 if let Some(imp) = obj.get("importance").and_then(|v| v.as_f64()) {
1002 mem = mem.with_importance(imp as f32);
1003 }
1004 if let Some(shared) = obj.get("shared").and_then(|v| v.as_bool()) {
1005 if shared {
1006 mem = mem.with_scope(Scope::Shared);
1007 }
1008 }
1009 if let Some(meta) = obj.get("metadata").and_then(|v| v.as_object()) {
1010 for (k, v) in meta {
1011 mem.metadata.insert(k.clone(), v.clone());
1012 }
1013 }
1014 let key = obj.get("key").and_then(|v| v.as_str()).map(String::from);
1015 batch.facts.push(KeyedMemory { memory: mem, key });
1016 }
1017 }
1018
1019 if let Some(arr) = args.get("essence_ops").and_then(|v| v.as_array()) {
1021 for (i, o) in arr.iter().enumerate() {
1022 let obj = o.as_object().ok_or_else(|| {
1023 RpcError::new(
1024 INVALID_PARAMS,
1025 format!("commit_deltas: essence_ops[{i}] must be an object"),
1026 )
1027 })?;
1028 let kind_str = obj.get("kind").and_then(|v| v.as_str()).ok_or_else(|| {
1029 RpcError::new(
1030 INVALID_PARAMS,
1031 format!("commit_deltas: essence_ops[{i}] missing 'kind'"),
1032 )
1033 })?;
1034 let kind = match kind_str {
1035 "add" => EssenceOpKind::Add,
1036 "append_timeline" => EssenceOpKind::AppendTimeline,
1037 "update" => EssenceOpKind::Update,
1038 "resolve" => EssenceOpKind::Resolve,
1039 "remove" => EssenceOpKind::Remove,
1040 other => {
1041 return Err(RpcError::new(
1042 INVALID_PARAMS,
1043 format!("commit_deltas: essence_ops[{i}] unknown kind '{other}'"),
1044 ));
1045 }
1046 };
1047 let thread = obj.get("thread").and_then(|v| v.as_str()).map(String::from);
1048 let project = obj
1049 .get("project")
1050 .and_then(|v| v.as_bool())
1051 .unwrap_or(false);
1052 let scope = match (thread, project) {
1053 (Some(t), false) => Some(EssenceScope::Thread(t)),
1054 (None, true) => Some(EssenceScope::Project),
1055 (None, false) => None,
1056 (Some(_), true) => {
1057 return Err(RpcError::new(
1058 INVALID_PARAMS,
1059 format!(
1060 "commit_deltas: essence_ops[{i}] can't set both 'thread' and 'project'"
1061 ),
1062 ));
1063 }
1064 };
1065 let section = obj
1066 .get("section")
1067 .and_then(|v| v.as_str())
1068 .and_then(EssenceSection::parse);
1069 let id = obj
1070 .get("id")
1071 .and_then(|v| v.as_str())
1072 .map(Ulid::from_string)
1073 .transpose()
1074 .map_err(|_| {
1075 RpcError::new(
1076 INVALID_PARAMS,
1077 format!("commit_deltas: essence_ops[{i}] 'id' is not a valid ULID"),
1078 )
1079 })?;
1080 let text = obj.get("text").and_then(|v| v.as_str()).map(String::from);
1081 let refs = obj.get("refs").and_then(|v| v.as_array()).map(|a| {
1082 a.iter()
1083 .filter_map(|x| x.as_str().map(String::from))
1084 .collect()
1085 });
1086 batch.essence_ops.push(EssenceOp {
1087 kind,
1088 scope,
1089 section,
1090 id,
1091 text,
1092 refs,
1093 });
1094 }
1095 }
1096
1097 if let Some(arr) = args.get("turns").and_then(|v| v.as_array()) {
1099 for (i, t) in arr.iter().enumerate() {
1100 let obj = t.as_object().ok_or_else(|| {
1101 RpcError::new(
1102 INVALID_PARAMS,
1103 format!("commit_deltas: turns[{i}] must be an object"),
1104 )
1105 })?;
1106 let turn_id = obj
1107 .get("turn_id")
1108 .and_then(|v| v.as_str())
1109 .ok_or_else(|| {
1110 RpcError::new(
1111 INVALID_PARAMS,
1112 format!("commit_deltas: turns[{i}] missing 'turn_id'"),
1113 )
1114 })?
1115 .to_string();
1116 let role = obj
1117 .get("role")
1118 .and_then(|v| v.as_str())
1119 .ok_or_else(|| {
1120 RpcError::new(
1121 INVALID_PARAMS,
1122 format!("commit_deltas: turns[{i}] missing 'role'"),
1123 )
1124 })?
1125 .to_string();
1126 let content = obj
1127 .get("content")
1128 .and_then(|v| v.as_str())
1129 .ok_or_else(|| {
1130 RpcError::new(
1131 INVALID_PARAMS,
1132 format!("commit_deltas: turns[{i}] missing 'content'"),
1133 )
1134 })?
1135 .to_string();
1136 let timestamp = obj.get("timestamp").and_then(|v| v.as_i64());
1137 batch.turns.push(ThreadTurn {
1138 turn_id,
1139 role,
1140 content,
1141 timestamp,
1142 });
1143 }
1144 }
1145 batch.thread_id = args
1146 .get("thread_id")
1147 .and_then(|v| v.as_str())
1148 .map(String::from);
1149 let policy_str = args
1150 .get("turn_policy")
1151 .and_then(|v| v.as_str())
1152 .unwrap_or("archive");
1153 batch.turn_policy = match policy_str {
1154 "archive" => TurnPolicy::Archive,
1155 "drop" => TurnPolicy::Drop,
1156 "expire_after" => {
1157 let ttl = args
1158 .get("turn_policy_ttl")
1159 .and_then(|v| v.as_i64())
1160 .ok_or_else(|| {
1161 RpcError::new(
1162 INVALID_PARAMS,
1163 "commit_deltas: turn_policy='expire_after' requires 'turn_policy_ttl' (seconds)",
1164 )
1165 })?;
1166 TurnPolicy::ExpireAfter(ttl)
1167 }
1168 other => {
1169 return Err(RpcError::new(
1170 INVALID_PARAMS,
1171 format!("commit_deltas: unknown turn_policy '{other}'"),
1172 ));
1173 }
1174 };
1175
1176 let report = db.commit_deltas(batch).map_err(mnemo_err)?;
1177 Ok(json!({
1178 "facts_added": report.facts_added,
1179 "facts_superseded": report.facts_superseded,
1180 "facts_unchanged": report.facts_unchanged,
1181 "essence_ops_applied": report.essence_ops_applied,
1182 "essence_ops_unchanged": report.essence_ops_unchanged,
1183 "turns_added": report.turns_added,
1184 "turns_skipped": report.turns_skipped,
1185 "turns_truncated": report.turns_truncated,
1186 "turns_chunked": report.turns_chunked,
1187 })
1188 .to_string())
1189}
1190
1191fn tool_essence(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1192 use crate::{EssenceScope, EssenceSection, Ulid, WriteOutcome};
1193 let op = args
1194 .get("op")
1195 .and_then(|v| v.as_str())
1196 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence: missing 'op'"))?;
1197 let thread = args
1198 .get("thread")
1199 .and_then(|v| v.as_str())
1200 .map(String::from);
1201 let project = args
1202 .get("project")
1203 .and_then(|v| v.as_bool())
1204 .unwrap_or(false);
1205 let scope_from = |t: Option<String>, p: bool| -> std::result::Result<EssenceScope, RpcError> {
1206 match (t, p) {
1207 (Some(s), false) => Ok(EssenceScope::Thread(s)),
1208 (None, true) => Ok(EssenceScope::Project),
1209 (None, false) => Err(RpcError::new(
1210 INVALID_PARAMS,
1211 "essence: pass either `thread` or `project: true`",
1212 )),
1213 (Some(_), true) => Err(RpcError::new(
1214 INVALID_PARAMS,
1215 "essence: `thread` and `project` are mutually exclusive",
1216 )),
1217 }
1218 };
1219 let refs: Option<Vec<String>> = args.get("refs").and_then(|v| v.as_array()).map(|arr| {
1220 arr.iter()
1221 .filter_map(|x| x.as_str().map(String::from))
1222 .collect()
1223 });
1224 let text = args.get("text").and_then(|v| v.as_str()).map(String::from);
1225 let id_str = args.get("id").and_then(|v| v.as_str());
1226 let outcome_to_str = |o: WriteOutcome| -> &'static str {
1227 match o {
1228 WriteOutcome::Updated => "updated",
1229 WriteOutcome::Unchanged => "unchanged",
1230 WriteOutcome::Added => "updated",
1231 WriteOutcome::Superseded => "updated",
1232 }
1233 };
1234 match op {
1235 "add" => {
1236 let section_str = args
1237 .get("section")
1238 .and_then(|v| v.as_str())
1239 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence add: missing 'section'"))?;
1240 let section = EssenceSection::parse(section_str).ok_or_else(|| {
1241 RpcError::new(
1242 INVALID_PARAMS,
1243 format!("essence add: unknown section '{section_str}'"),
1244 )
1245 })?;
1246 if section == EssenceSection::Timeline {
1247 return Err(RpcError::new(
1248 INVALID_PARAMS,
1249 "essence add: timeline uses op: append, not add",
1250 ));
1251 }
1252 let text =
1253 text.ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence add: missing 'text'"))?;
1254 let scope = scope_from(thread, project)?;
1255 let id = db
1256 .essence_add(scope, section, &text, refs)
1257 .map_err(mnemo_err)?;
1258 db.flush().map_err(mnemo_err)?;
1259 Ok(json!({ "id": id.to_string(), "outcome": "added" }).to_string())
1260 }
1261 "append" => {
1262 let text = text
1263 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence append: missing 'text'"))?;
1264 let scope = scope_from(thread, project)?;
1265 let id = db
1266 .essence_append_timeline(scope, &text, refs)
1267 .map_err(mnemo_err)?;
1268 db.flush().map_err(mnemo_err)?;
1269 Ok(json!({ "id": id.to_string(), "outcome": "appended" }).to_string())
1270 }
1271 "update" => {
1272 let id = id_str
1273 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence update: missing 'id'"))?;
1274 let ulid = Ulid::from_string(id).map_err(|_| {
1275 RpcError::new(
1276 INVALID_PARAMS,
1277 format!("essence update: invalid ULID '{id}'"),
1278 )
1279 })?;
1280 let text = text
1281 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence update: missing 'text'"))?;
1282 let outcome = db.essence_update(&ulid, &text, refs).map_err(mnemo_err)?;
1283 db.flush().map_err(mnemo_err)?;
1284 Ok(json!({ "id": id, "outcome": outcome_to_str(outcome) }).to_string())
1285 }
1286 "resolve" => {
1287 let id = id_str
1288 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence resolve: missing 'id'"))?;
1289 let ulid = Ulid::from_string(id).map_err(|_| {
1290 RpcError::new(
1291 INVALID_PARAMS,
1292 format!("essence resolve: invalid ULID '{id}'"),
1293 )
1294 })?;
1295 let outcome = db.essence_resolve(&ulid).map_err(mnemo_err)?;
1296 db.flush().map_err(mnemo_err)?;
1297 let s = match outcome {
1298 WriteOutcome::Updated => "resolved",
1299 WriteOutcome::Unchanged => "already-resolved",
1300 _ => "resolved",
1301 };
1302 Ok(json!({ "id": id, "outcome": s }).to_string())
1303 }
1304 "remove" => {
1305 let id = id_str
1306 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence remove: missing 'id'"))?;
1307 let ulid = Ulid::from_string(id).map_err(|_| {
1308 RpcError::new(
1309 INVALID_PARAMS,
1310 format!("essence remove: invalid ULID '{id}'"),
1311 )
1312 })?;
1313 db.essence_remove(&ulid).map_err(mnemo_err)?;
1314 db.flush().map_err(mnemo_err)?;
1315 Ok(json!({ "id": id, "outcome": "removed" }).to_string())
1316 }
1317 "render" => {
1318 let budget = args.get("budget").and_then(|v| v.as_u64()).unwrap_or(6144) as usize;
1319 let scope = scope_from(thread, project)?;
1320 let rendered = db.essence(scope, budget).map_err(mnemo_err)?;
1321 Ok(rendered)
1322 }
1323 other => Err(RpcError::new(
1324 INVALID_PARAMS,
1325 format!("essence: unknown op '{other}'"),
1326 )),
1327 }
1328}
1329
1330fn tool_thread_ingest(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1331 let thread = args
1332 .get("thread")
1333 .and_then(|v| v.as_str())
1334 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_ingest: missing 'thread'"))?;
1335 let path = args
1336 .get("transcript_path")
1337 .and_then(|v| v.as_str())
1338 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_ingest: missing 'transcript_path'"))?;
1339 let format_str = args
1340 .get("format")
1341 .and_then(|v| v.as_str())
1342 .unwrap_or("jsonl");
1343 let format = crate::TranscriptFormat::parse(format_str).ok_or_else(|| {
1344 RpcError::new(
1345 INVALID_PARAMS,
1346 format!(
1347 "thread_ingest: unknown format '{format_str}' (want 'jsonl' or 'claude-code-jsonl')"
1348 ),
1349 )
1350 })?;
1351 let file = std::fs::File::open(path).map_err(|e| {
1352 RpcError::new(
1353 INVALID_PARAMS,
1354 format!("thread_ingest: cannot open {path}: {e}"),
1355 )
1356 })?;
1357 let turns = crate::read_transcript(file, format).map_err(mnemo_err)?;
1358 let report = db.ingest_thread(thread, &turns).map_err(mnemo_err)?;
1359 db.flush().map_err(mnemo_err)?;
1360 Ok(json!({
1361 "added_records": report.added_records,
1362 "skipped_existing_turns": report.skipped_existing_turns,
1363 "chunked_turns": report.chunked_turns,
1364 "truncated_turns": report.truncated_turns,
1365 })
1366 .to_string())
1367}
1368
1369fn tool_history(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1370 use crate::KeyScope;
1371 let key = args
1372 .get("key")
1373 .and_then(|v| v.as_str())
1374 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "history: missing 'key'"))?;
1375 let agent = args.get("agent").and_then(|v| v.as_str());
1376 let shared = args
1377 .get("shared")
1378 .and_then(|v| v.as_bool())
1379 .unwrap_or(false);
1380 if shared && agent.is_some() {
1381 return Err(RpcError::new(
1382 INVALID_PARAMS,
1383 "history: 'agent' and 'shared' are mutually exclusive",
1384 ));
1385 }
1386 let scope = if shared {
1388 KeyScope::Shared
1389 } else {
1390 match agent {
1391 Some(a) => KeyScope::Private(a.to_string()),
1392 None => KeyScope::private_default(),
1393 }
1394 };
1395 let chain = db.history_in(key, scope).map_err(mnemo_err)?;
1396 let out: Vec<Value> = chain
1397 .iter()
1398 .enumerate()
1399 .map(|(i, m)| {
1400 let mut v = memory_to_json(m);
1401 if let Some(obj) = v.as_object_mut() {
1402 obj.insert(
1403 "state".into(),
1404 Value::String(if i == 0 {
1405 "live".into()
1406 } else {
1407 "superseded".into()
1408 }),
1409 );
1410 }
1411 v
1412 })
1413 .collect();
1414 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1415}
1416
1417fn tool_recall(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1418 let query = match (args.get("query"), args.get("text")) {
1420 (Some(_), Some(_)) => {
1421 return Err(RpcError::new(
1422 INVALID_PARAMS,
1423 "recall: pass EITHER 'query' or 'text', not both",
1424 ))
1425 }
1426 (Some(v), None) => v
1427 .as_array()
1428 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "recall: 'query' must be a JSON array"))?
1429 .iter()
1430 .map(|v| v.as_f64().map(|f| f as f32))
1431 .collect::<Option<Vec<f32>>>()
1432 .ok_or_else(|| {
1433 RpcError::new(
1434 INVALID_PARAMS,
1435 "recall: 'query' must be an array of numbers",
1436 )
1437 })?,
1438 (None, Some(t)) => {
1439 let text = t
1440 .as_str()
1441 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "recall: 'text' must be a string"))?;
1442 embed_text_via_server(db, "recall", text)?
1443 }
1444 (None, None) => {
1445 return Err(RpcError::new(
1446 INVALID_PARAMS,
1447 "recall: missing 'query' or 'text'",
1448 ))
1449 }
1450 };
1451 let top_k = args.get("top_k").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
1452 let metric = match args
1453 .get("metric")
1454 .and_then(|v| v.as_str())
1455 .unwrap_or("cosine")
1456 {
1457 "cosine" => Metric::Cosine,
1458 "l2" => Metric::L2,
1459 "dot" => Metric::Dot,
1460 other => {
1461 return Err(RpcError::new(
1462 INVALID_PARAMS,
1463 format!("recall: unknown metric '{other}'"),
1464 ))
1465 }
1466 };
1467 let mut req = RecallRequest::new(query)
1480 .top_k(top_k)
1481 .metric(metric)
1482 .track_access(true);
1483 if let Some(types) = args.get("types").and_then(|v| v.as_array()) {
1484 let parsed: Vec<MemoryType> = types
1485 .iter()
1486 .filter_map(|t| t.as_str().and_then(MemoryType::parse))
1487 .collect();
1488 if !parsed.is_empty() {
1489 req = req.types(parsed);
1490 }
1491 }
1492 if let Some(agent) = args.get("agent").and_then(|v| v.as_str()) {
1493 req = req.agent(agent.to_string());
1494 }
1495 if let Some(filter_obj) = args.get("filter").and_then(|v| v.as_object()) {
1502 req = req.filter(parse_mcp_filter(filter_obj)?);
1503 }
1504 let has_max_bytes = args.get("max_bytes").is_some();
1506 if let Some(n) = args.get("max_bytes").and_then(|v| v.as_u64()) {
1507 req = req.max_bytes(n as usize);
1508 }
1509
1510 let resp = db.recall_budgeted(&req).map_err(mnemo_err)?;
1511 let out: Vec<Value> = resp
1512 .hits
1513 .into_iter()
1514 .map(|h| {
1515 let mut obj = memory_to_json(&h.memory);
1516 if let Some(map) = obj.as_object_mut() {
1517 map.insert("score".into(), json!(h.score));
1518 map.insert("similarity".into(), json!(h.similarity));
1519 }
1520 obj
1521 })
1522 .collect();
1523 if has_max_bytes {
1524 let wrapped = json!({
1527 "hits": out,
1528 "truncated": resp.truncated,
1529 });
1530 Ok(serde_json::to_string_pretty(&wrapped).unwrap_or_else(|_| "{}".into()))
1531 } else {
1532 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1533 }
1534}
1535
1536fn parse_mcp_filter(
1546 obj: &serde_json::Map<String, Value>,
1547) -> std::result::Result<crate::Filter, RpcError> {
1548 let mut f = crate::Filter::default();
1549 if let Some(sid) = obj.get("session_id").and_then(|v| v.as_str()) {
1550 f = f.with_session(sid.to_string());
1551 }
1552 if let Some(meta) = obj.get("metadata").and_then(|v| v.as_object()) {
1553 for (k, v) in meta {
1554 match v {
1555 Value::Object(ops) => {
1557 for (op_key, op_val) in ops {
1558 match op_key.as_str() {
1559 "$exists" => match op_val.as_bool() {
1560 Some(true) => {
1561 f = f.metadata_exists(k.clone());
1562 }
1563 Some(false) => {
1564 return Err(RpcError::new(
1565 INVALID_PARAMS,
1566 format!(
1567 "filter.metadata.{k}.$exists=false isn't supported; \
1568 drop the filter term instead"
1569 ),
1570 ));
1571 }
1572 None => {
1573 return Err(RpcError::new(
1574 INVALID_PARAMS,
1575 format!("filter.metadata.{k}.$exists must be a boolean"),
1576 ));
1577 }
1578 },
1579 "$in" => {
1580 let arr = op_val.as_array().ok_or_else(|| {
1581 RpcError::new(
1582 INVALID_PARAMS,
1583 format!("filter.metadata.{k}.$in must be an array"),
1584 )
1585 })?;
1586 f = f.metadata_in(k.clone(), arr.clone());
1587 }
1588 "$gte" | "$lte" => {
1589 let mut min = None;
1590 let mut max = None;
1591 for (bound, bound_val) in ops {
1592 let n = bound_val.as_f64().ok_or_else(|| {
1593 RpcError::new(
1594 INVALID_PARAMS,
1595 format!("filter.metadata.{k}.{bound} must be a number"),
1596 )
1597 })?;
1598 match bound.as_str() {
1599 "$gte" => min = Some(n),
1600 "$lte" => max = Some(n),
1601 other => {
1602 return Err(RpcError::new(
1603 INVALID_PARAMS,
1604 format!(
1605 "filter.metadata.{k}: unsupported range \
1606 operator '{other}' (want $gte, $lte)"
1607 ),
1608 ));
1609 }
1610 }
1611 }
1612 f = f.metadata_range(k.clone(), min, max);
1613 break; }
1615 other => {
1616 return Err(RpcError::new(
1617 INVALID_PARAMS,
1618 format!(
1619 "filter.metadata.{k}: unknown operator '{other}' \
1620 (want $exists, $in, $gte, $lte)"
1621 ),
1622 ));
1623 }
1624 }
1625 }
1626 }
1627 other => {
1629 f = f.metadata_eq(k.clone(), other.clone());
1630 }
1631 }
1632 }
1633 }
1634 Ok(f)
1635}
1636
1637fn tool_forget(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1638 let id_str = args
1639 .get("id")
1640 .and_then(|v| v.as_str())
1641 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "forget: missing 'id'"))?;
1642 let ulid = crate::Ulid::from_string(id_str)
1643 .map_err(|_| RpcError::new(INVALID_PARAMS, format!("forget: invalid ULID '{id_str}'")))?;
1644 db.delete(&ulid).map_err(mnemo_err)?;
1645 db.flush().map_err(mnemo_err)?;
1646 Ok(json!({ "deleted": id_str }).to_string())
1647}
1648
1649fn tool_list(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1650 let include_vectors = args
1651 .get("include_vectors")
1652 .and_then(|v| v.as_bool())
1653 .unwrap_or(false);
1654 let mems = db.memories().map_err(mnemo_err)?;
1655 let out: Vec<Value> = mems
1656 .iter()
1657 .map(|m| {
1658 let mut v = memory_to_json(m);
1659 if !include_vectors {
1660 if let Some(o) = v.as_object_mut() {
1661 o.remove("vector");
1662 }
1663 }
1664 v
1665 })
1666 .collect();
1667 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1668}
1669
1670fn tool_snapshot_list(db: &mut Mnemo) -> std::result::Result<String, RpcError> {
1671 let snaps = db.snapshots();
1672 let out: Vec<Value> = snaps
1673 .into_iter()
1674 .map(|s| {
1675 json!({
1676 "txn_id": s.txn_id,
1677 "created_at": s.created_at,
1678 "memory_count": s.memory_count,
1679 })
1680 })
1681 .collect();
1682 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1683}
1684
1685fn tool_stats(db: &mut Mnemo) -> std::result::Result<String, RpcError> {
1686 let s = db.stats().map_err(mnemo_err)?;
1687 let (cache_pages, cache_capacity) = db.page_cache_stats();
1688 let snapshot_count = db.snapshots().len();
1689 Ok(json!({
1692 "memories": s.memories,
1693 "deleted": s.deleted,
1694 "dimensions": s.dimensions,
1695 "file_bytes": s.file_bytes,
1696 "agents": s.agents,
1697 "encrypted": s.encrypted,
1698 "created_at": s.created_at,
1699 "wal_pages": s.wal_pages,
1700 "index": s.index,
1701 "snapshot_count": snapshot_count,
1702 "cache_pages": cache_pages,
1703 "cache_capacity": cache_capacity,
1704 })
1705 .to_string())
1706}
1707
1708fn tool_cache_put(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1711 let ns = args
1712 .get("namespace")
1713 .and_then(|v| v.as_str())
1714 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put: missing 'namespace'"))?;
1715 let key = args
1716 .get("key")
1717 .and_then(|v| v.as_str())
1718 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put: missing 'key'"))?;
1719 let value = args
1720 .get("value")
1721 .and_then(|v| v.as_str())
1722 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put: missing 'value'"))?;
1723 let content_type = args
1724 .get("content_type")
1725 .and_then(|v| v.as_str())
1726 .unwrap_or("text")
1727 .to_string();
1728 let ttl_secs = args.get("ttl_secs").and_then(|v| v.as_u64());
1729 let opts = CachePutOpts {
1730 content_type,
1731 ttl_secs,
1732 cost_hint_ms: None,
1733 };
1734 db.cache_put(ns, key, value.as_bytes(), opts)
1735 .map_err(mnemo_err)?;
1736 db.flush().map_err(mnemo_err)?;
1737 Ok(json!({ "cached": true, "namespace": ns, "key": key, "bytes": value.len() }).to_string())
1738}
1739
1740fn tool_cache_get(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1741 let ns = args
1742 .get("namespace")
1743 .and_then(|v| v.as_str())
1744 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_get: missing 'namespace'"))?;
1745 let key = args
1746 .get("key")
1747 .and_then(|v| v.as_str())
1748 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_get: missing 'key'"))?;
1749 match db.cache_get(ns, key).map_err(mnemo_err)? {
1750 Some(hit) => Ok(json!({
1751 "hit": true,
1752 "value": String::from_utf8_lossy(&hit.value),
1753 "content_type": hit.content_type,
1754 "created_at": hit.created_at,
1755 "accessed_at": hit.accessed_at,
1756 "access_count": hit.access_count,
1757 })
1758 .to_string()),
1759 None => Ok(json!({ "hit": false }).to_string()),
1760 }
1761}
1762
1763fn tool_cache_stats(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1764 let ns = args.get("namespace").and_then(|v| v.as_str());
1765 let s = db.cache_stats(ns);
1766 Ok(json!({
1767 "namespace": ns,
1768 "entries": s.entries,
1769 "bytes": s.bytes,
1770 "hits": s.hits,
1771 "misses": s.misses,
1772 "hit_rate": s.hit_rate,
1773 "evictions": s.evictions,
1774 })
1775 .to_string())
1776}
1777
1778fn tool_cache_put_semantic(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1779 let ns = args
1780 .get("namespace")
1781 .and_then(|v| v.as_str())
1782 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'namespace'"))?;
1783 let key = args
1784 .get("key")
1785 .and_then(|v| v.as_str())
1786 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'key'"))?;
1787 let value = args
1788 .get("value")
1789 .and_then(|v| v.as_str())
1790 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'value'"))?;
1791 let (vector, model) = match (args.get("vector"), args.get("text")) {
1794 (Some(_), Some(_)) => {
1795 return Err(RpcError::new(
1796 INVALID_PARAMS,
1797 "cache_put_semantic: pass EITHER 'vector' or 'text', not both",
1798 ));
1799 }
1800 (Some(v), None) => {
1801 let vec = v
1802 .as_array()
1803 .ok_or_else(|| {
1804 RpcError::new(
1805 INVALID_PARAMS,
1806 "cache_put_semantic: 'vector' must be a JSON array",
1807 )
1808 })?
1809 .iter()
1810 .map(|v| v.as_f64().map(|f| f as f32))
1811 .collect::<Option<Vec<f32>>>()
1812 .ok_or_else(|| {
1813 RpcError::new(
1814 INVALID_PARAMS,
1815 "cache_put_semantic: 'vector' must be an array of numbers",
1816 )
1817 })?;
1818 let model = args
1819 .get("model")
1820 .and_then(|v| v.as_str())
1821 .ok_or_else(|| {
1822 RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'model'")
1823 })?
1824 .to_string();
1825 (vec, model)
1826 }
1827 (None, Some(t)) => {
1828 let text = t.as_str().ok_or_else(|| {
1829 RpcError::new(
1830 INVALID_PARAMS,
1831 "cache_put_semantic: 'text' must be a string",
1832 )
1833 })?;
1834 let vec = embed_text_via_server(db, "cache_put_semantic", text)?;
1835 let model = match args.get("model").and_then(|v| v.as_str()) {
1837 Some(m) => m.to_string(),
1838 None => {
1839 db.embedder()
1840 .expect("embed_text_via_server would have errored otherwise")
1841 .identity()
1842 .name
1843 }
1844 };
1845 (vec, model)
1846 }
1847 (None, None) => {
1848 return Err(RpcError::new(
1849 INVALID_PARAMS,
1850 "cache_put_semantic: missing 'vector' or 'text'",
1851 ))
1852 }
1853 };
1854 let content_type = args
1855 .get("content_type")
1856 .and_then(|v| v.as_str())
1857 .unwrap_or("text")
1858 .to_string();
1859 let ttl_secs = args.get("ttl_secs").and_then(|v| v.as_u64());
1860 let opts = SemanticCachePutOpts {
1861 content_type,
1862 ttl_secs,
1863 cost_hint_ms: None,
1864 model,
1865 };
1866 db.cache_put_semantic(ns, key, vector, value.as_bytes(), opts)
1867 .map_err(mnemo_err)?;
1868 db.flush().map_err(mnemo_err)?;
1869 Ok(json!({ "cached": true, "namespace": ns, "key": key, "bytes": value.len() }).to_string())
1870}
1871
1872fn tool_cache_get_semantic(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1873 let ns = args
1874 .get("namespace")
1875 .and_then(|v| v.as_str())
1876 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_get_semantic: missing 'namespace'"))?;
1877 let (query, model_owned) = match (args.get("query"), args.get("text")) {
1880 (Some(_), Some(_)) => {
1881 return Err(RpcError::new(
1882 INVALID_PARAMS,
1883 "cache_get_semantic: pass EITHER 'query' or 'text', not both",
1884 ));
1885 }
1886 (Some(q), None) => {
1887 let vec = q
1888 .as_array()
1889 .ok_or_else(|| {
1890 RpcError::new(
1891 INVALID_PARAMS,
1892 "cache_get_semantic: 'query' must be a JSON array",
1893 )
1894 })?
1895 .iter()
1896 .map(|v| v.as_f64().map(|f| f as f32))
1897 .collect::<Option<Vec<f32>>>()
1898 .ok_or_else(|| {
1899 RpcError::new(
1900 INVALID_PARAMS,
1901 "cache_get_semantic: 'query' must be an array of numbers",
1902 )
1903 })?;
1904 let model = args
1905 .get("model")
1906 .and_then(|v| v.as_str())
1907 .ok_or_else(|| {
1908 RpcError::new(INVALID_PARAMS, "cache_get_semantic: missing 'model'")
1909 })?
1910 .to_string();
1911 (vec, model)
1912 }
1913 (None, Some(t)) => {
1914 let text = t.as_str().ok_or_else(|| {
1915 RpcError::new(
1916 INVALID_PARAMS,
1917 "cache_get_semantic: 'text' must be a string",
1918 )
1919 })?;
1920 let vec = embed_text_via_server(db, "cache_get_semantic", text)?;
1921 let model = match args.get("model").and_then(|v| v.as_str()) {
1922 Some(m) => m.to_string(),
1923 None => {
1924 db.embedder()
1925 .expect("embed_text_via_server would have errored otherwise")
1926 .identity()
1927 .name
1928 }
1929 };
1930 (vec, model)
1931 }
1932 (None, None) => {
1933 return Err(RpcError::new(
1934 INVALID_PARAMS,
1935 "cache_get_semantic: missing 'query' or 'text'",
1936 ))
1937 }
1938 };
1939 let model = model_owned.as_str();
1940 let threshold = args
1941 .get("threshold")
1942 .and_then(|v| v.as_f64())
1943 .map(|f| f as f32)
1944 .unwrap_or(DEFAULT_SEMANTIC_THRESHOLD);
1945 match db
1946 .cache_get_semantic(ns, &query, threshold, model)
1947 .map_err(mnemo_err)?
1948 {
1949 Some((hit, similarity)) => Ok(json!({
1950 "hit": true,
1951 "similarity": similarity,
1952 "value": String::from_utf8_lossy(&hit.value),
1953 "content_type": hit.content_type,
1954 "created_at": hit.created_at,
1955 "accessed_at": hit.accessed_at,
1956 "access_count": hit.access_count,
1957 })
1958 .to_string()),
1959 None => Ok(json!({ "hit": false }).to_string()),
1960 }
1961}
1962
1963fn memory_to_json(m: &Memory) -> Value {
1968 serde_json::to_value(m).unwrap_or_else(|_| {
1969 let mut map = Map::new();
1970 map.insert("id".into(), json!(m.id.to_string()));
1971 Value::Object(map)
1972 })
1973}