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.flush() {
252 eprintln!("mnemo mcp: flush on shutdown failed: {e}");
253 }
254 eprintln!("mnemo mcp: stdin closed, shutting down");
255 Ok(())
256}
257
258fn handle(db: &mut Mnemo, req: &Request) -> std::result::Result<Value, RpcError> {
263 if req.jsonrpc != "2.0" {
264 return Err(RpcError::new(
265 INVALID_REQUEST,
266 format!("expected jsonrpc \"2.0\", got {:?}", req.jsonrpc),
267 ));
268 }
269 match req.method.as_str() {
270 "initialize" => Ok(initialize_result(&req.params)),
271 "notifications/initialized" | "notifications/cancelled" => Ok(Value::Null),
272 "ping" => Ok(json!({})),
273 "tools/list" => Ok(tools_list()),
274 "tools/call" => tools_call(db, &req.params),
275 other => Err(RpcError::new(
276 METHOD_NOT_FOUND,
277 format!("method '{other}' not implemented"),
278 )),
279 }
280}
281
282fn initialize_result(params: &Value) -> Value {
285 let client_version = params
286 .get("protocolVersion")
287 .and_then(|v| v.as_str())
288 .unwrap_or(PROTOCOL_VERSION)
289 .to_string();
290 json!({
291 "protocolVersion": client_version,
292 "capabilities": {
293 "tools": { "listChanged": false },
294 },
295 "serverInfo": {
296 "name": "mnemo",
297 "version": env!("CARGO_PKG_VERSION"),
298 },
299 })
300}
301
302fn tools_list() -> Value {
308 json!({
309 "tools": [
310 {
311 "name": "about",
312 "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.",
313 "inputSchema": {
314 "type": "object",
315 "properties": {},
316 "additionalProperties": false
317 }
318 },
319 {
320 "name": "remember",
321 "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`.",
322 "inputSchema": {
323 "type": "object",
324 "required": ["content", "memory_type"],
325 "properties": {
326 "content": { "type": "string" },
327 "memory_type": { "type": "string", "enum": ["episodic", "semantic", "procedural", "working"] },
328 "vector": { "type": "array", "items": { "type": "number" }, "description": "Explicit embedding; mutually exclusive with `text`." },
329 "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." },
330 "importance": { "type": "number", "minimum": 0, "maximum": 1 },
331 "agent_id": { "type": "string" },
332 "session_id": { "type": "string" },
333 "ttl_secs": { "type": "integer" },
334 "shared": { "type": "boolean" },
335 "metadata": { "type": "object" },
336 "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." }
337 },
338 "additionalProperties": false
339 }
340 },
341 {
342 "name": "update",
343 "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.",
344 "inputSchema": {
345 "type": "object",
346 "required": ["id"],
347 "properties": {
348 "id": { "type": "string", "description": "ULID of the memory to update." },
349 "content": { "type": "string", "description": "New content. Requires a matching `vector` (use `text` to re-embed instead)." },
350 "vector": { "type": "array", "items": { "type": "number" } },
351 "text": { "type": "string", "description": "New content; the server embeds it through its registered embedder. Mutually exclusive with `content` + `vector`." },
352 "importance": { "type": "number", "minimum": 0, "maximum": 1 },
353 "metadata": { "type": "object", "description": "Replace the user-visible metadata. Reserved `_mnemo.*` keys on the record are preserved." },
354 "ttl_secs": { "type": "integer", "description": "Set the TTL to this many seconds. Use `clear_ttl: true` to drop any TTL." },
355 "clear_ttl": { "type": "boolean", "description": "Clear any TTL on the record." }
356 },
357 "additionalProperties": false
358 }
359 },
360 {
361 "name": "essence",
362 "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).",
363 "inputSchema": {
364 "type": "object",
365 "required": ["op"],
366 "properties": {
367 "op": { "type": "string", "enum": ["add", "append", "update", "resolve", "remove", "render"] },
368 "thread": { "type": "string", "description": "Thread id. Omit with `project: true` for the project layer." },
369 "project": { "type": "boolean", "default": false, "description": "Project-scoped item (shows in every thread's render). Mutually exclusive with `thread`." },
370 "section": { "type": "string", "enum": ["intent", "conventions", "decisions", "state", "open", "next", "lessons"], "description": "Required for `op: add`. Timeline uses `op: append`, not add." },
371 "id": { "type": "string", "description": "ULID of the section item. Required for update / resolve / remove." },
372 "text": { "type": "string", "description": "Required for add / append / update." },
373 "refs": { "type": "array", "items": { "type": "string" }, "description": "Optional free-form reference strings (turn ids, memory ids, commits, paths)." },
374 "budget": { "type": "integer", "default": 6144, "description": "Byte budget for `op: render`." }
375 },
376 "additionalProperties": false
377 }
378 },
379 {
380 "name": "thread_start",
381 "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).",
382 "inputSchema": {
383 "type": "object",
384 "required": ["new", "from"],
385 "properties": {
386 "new": { "type": "string", "description": "New thread id." },
387 "from": { "type": "string", "description": "Previous thread id to continue from." }
388 },
389 "additionalProperties": false
390 }
391 },
392 {
393 "name": "commit_deltas",
394 "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.",
395 "inputSchema": {
396 "type": "object",
397 "properties": {
398 "facts": {
399 "type": "array",
400 "items": {
401 "type": "object",
402 "required": ["content", "memory_type"],
403 "properties": {
404 "content": { "type": "string" },
405 "memory_type": { "type": "string", "enum": ["episodic", "semantic", "procedural", "working"] },
406 "vector": { "type": "array", "items": { "type": "number" }, "description": "Optional. Omit it to have the server embed `content`." },
407 "importance": { "type": "number", "minimum": 0, "maximum": 1 },
408 "agent_id": { "type": "string" },
409 "session_id": { "type": "string" },
410 "shared": { "type": "boolean" },
411 "metadata": { "type": "object" },
412 "key": { "type": "string" }
413 }
414 }
415 },
416 "essence_ops": {
417 "type": "array",
418 "items": {
419 "type": "object",
420 "required": ["kind"],
421 "properties": {
422 "kind": { "type": "string", "enum": ["add", "append_timeline", "update", "resolve", "remove"] },
423 "thread": { "type": "string" },
424 "project": { "type": "boolean" },
425 "section": { "type": "string" },
426 "id": { "type": "string" },
427 "text": { "type": "string" },
428 "refs": { "type": "array", "items": { "type": "string" } }
429 }
430 }
431 },
432 "turns": {
433 "type": "array",
434 "items": {
435 "type": "object",
436 "required": ["turn_id", "role", "content"],
437 "properties": {
438 "turn_id": { "type": "string" },
439 "role": { "type": "string" },
440 "content": { "type": "string" },
441 "timestamp": { "type": "integer" }
442 }
443 }
444 },
445 "thread_id": { "type": "string", "description": "Required when `turns` is non-empty." },
446 "turn_policy": { "type": "string", "enum": ["archive", "expire_after", "drop"], "default": "archive" },
447 "turn_policy_ttl": { "type": "integer", "description": "TTL in seconds; required when `turn_policy = 'expire_after'`." }
448 },
449 "additionalProperties": false
450 }
451 },
452 {
453 "name": "thread_ingest",
454 "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}`.",
455 "inputSchema": {
456 "type": "object",
457 "required": ["thread", "transcript_path"],
458 "properties": {
459 "thread": { "type": "string", "description": "Thread id to append under." },
460 "transcript_path": { "type": "string", "description": "Local path to the transcript file." },
461 "format": { "type": "string", "enum": ["jsonl", "claude-code-jsonl"], "default": "jsonl" }
462 },
463 "additionalProperties": false
464 }
465 },
466 {
467 "name": "thread_export",
468 "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.",
469 "inputSchema": {
470 "type": "object",
471 "required": ["thread"],
472 "properties": {
473 "thread": { "type": "string", "description": "Thread id to export." }
474 },
475 "additionalProperties": false
476 }
477 },
478 {
479 "name": "history",
480 "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.",
481 "inputSchema": {
482 "type": "object",
483 "required": ["key"],
484 "properties": {
485 "key": { "type": "string" },
486 "agent": { "type": "string", "description": "Query the private slot for this agent. Mutually exclusive with `shared`." },
487 "shared": { "type": "boolean", "default": false, "description": "Query the shared slot (one per file). Mutually exclusive with `agent`." }
488 },
489 "additionalProperties": false
490 }
491 },
492 {
493 "name": "recall",
494 "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).",
495 "inputSchema": {
496 "type": "object",
497 "properties": {
498 "query": { "type": "array", "items": { "type": "number" }, "description": "Explicit query vector; mutually exclusive with `text`." },
499 "text": { "type": "string", "description": "Query text; embedded by the server's registered embedder. Mutually exclusive with `query`." },
500 "top_k": { "type": "integer", "default": 10 },
501 "types": { "type": "array", "items": { "type": "string" } },
502 "agent": { "type": "string" },
503 "metric": { "type": "string", "enum": ["cosine", "l2", "dot"], "default": "cosine" },
504 "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." },
505 "filter": {
506 "type": "object",
507 "description": "Phase 12.6 pre-scoring filter. All conditions AND together. Shape: `{\"session_id\": \"...\", \"metadata\": {key: value | {\"$in\": [...]} | {\"$exists\": true} | {\"$gte\": n, \"$lte\": n}}}`.",
508 "properties": {
509 "session_id": { "type": "string" },
510 "metadata": { "type": "object" }
511 },
512 "additionalProperties": false
513 }
514 },
515 "additionalProperties": false
516 }
517 },
518 {
519 "name": "forget",
520 "description": "Soft-delete a memory by ULID. Space is reclaimed on the next `mnemo compact`. Persists to disk before returning.",
521 "inputSchema": {
522 "type": "object",
523 "required": ["id"],
524 "properties": {
525 "id": { "type": "string", "description": "ULID of the memory to delete" }
526 },
527 "additionalProperties": false
528 }
529 },
530 {
531 "name": "list",
532 "description": "Return every live memory as JSON. Sorted by creation time, oldest first. Vectors are omitted by default to keep payloads small.",
533 "inputSchema": {
534 "type": "object",
535 "properties": {
536 "include_vectors": { "type": "boolean", "default": false }
537 },
538 "additionalProperties": false
539 }
540 },
541 {
542 "name": "snapshot_list",
543 "description": "List every restorable snapshot (one per committed flush). Each entry gives the transaction ID, creation timestamp, and memory count.",
544 "inputSchema": {
545 "type": "object",
546 "properties": {},
547 "additionalProperties": false
548 }
549 },
550 {
551 "name": "stats",
552 "description": "Summary statistics: memory count, deleted count, agents present, snapshot count, page cache occupancy.",
553 "inputSchema": {
554 "type": "object",
555 "properties": {},
556 "additionalProperties": false
557 }
558 },
559 {
560 "name": "cache_put",
561 "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.",
562 "inputSchema": {
563 "type": "object",
564 "required": ["namespace", "key", "value"],
565 "properties": {
566 "namespace": { "type": "string" },
567 "key": { "type": "string" },
568 "value": { "type": "string" },
569 "content_type": { "type": "string", "default": "text" },
570 "ttl_secs": { "type": "integer", "minimum": 0 }
571 },
572 "additionalProperties": false
573 }
574 },
575 {
576 "name": "cache_get",
577 "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.",
578 "inputSchema": {
579 "type": "object",
580 "required": ["namespace", "key"],
581 "properties": {
582 "namespace": { "type": "string" },
583 "key": { "type": "string" }
584 },
585 "additionalProperties": false
586 }
587 },
588 {
589 "name": "cache_stats",
590 "description": "Cache statistics for a namespace (omit for the whole cache). Returns entries, bytes, hits, misses, hit_rate, evictions.",
591 "inputSchema": {
592 "type": "object",
593 "properties": {
594 "namespace": { "type": "string" }
595 },
596 "additionalProperties": false
597 }
598 },
599 {
600 "name": "cache_put_semantic",
601 "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).",
602 "inputSchema": {
603 "type": "object",
604 "required": ["namespace", "key", "value"],
605 "properties": {
606 "namespace": { "type": "string" },
607 "key": { "type": "string" },
608 "vector": { "type": "array", "items": { "type": "number" }, "description": "Explicit embedding; mutually exclusive with `text`. Requires `model`." },
609 "text": { "type": "string", "description": "Content to embed via the server's registered embedder; mutually exclusive with `vector`." },
610 "value": { "type": "string" },
611 "model": { "type": "string", "description": "Required with `vector`; optional with `text` (defaults to the embedder's identity name)." },
612 "content_type": { "type": "string", "default": "text" },
613 "ttl_secs": { "type": "integer", "minimum": 0 }
614 },
615 "additionalProperties": false
616 }
617 },
618 {
619 "name": "cache_get_semantic",
620 "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.",
621 "inputSchema": {
622 "type": "object",
623 "required": ["namespace"],
624 "properties": {
625 "namespace": { "type": "string" },
626 "query": { "type": "array", "items": { "type": "number" }, "description": "Explicit query vector; mutually exclusive with `text`. Requires `model`." },
627 "text": { "type": "string", "description": "Query text; embedded by the server. Mutually exclusive with `query`." },
628 "model": { "type": "string", "description": "Required with `query`; optional with `text`." },
629 "threshold": { "type": "number", "minimum": -1.0, "maximum": 1.0 }
630 },
631 "additionalProperties": false
632 }
633 }
634 ]
635 })
636}
637
638fn tools_call(db: &mut Mnemo, params: &Value) -> std::result::Result<Value, RpcError> {
643 let name = params
644 .get("name")
645 .and_then(|v| v.as_str())
646 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "missing tool name"))?;
647 let empty = json!({});
648 let args = params.get("arguments").unwrap_or(&empty);
649
650 let text = match name {
651 "about" => tool_about(db)?,
652 "remember" => tool_remember(db, args)?,
653 "update" => tool_update(db, args)?,
654 "history" => tool_history(db, args)?,
655 "thread_ingest" => tool_thread_ingest(db, args)?,
656 "thread_export" => tool_thread_export(db, args)?,
657 "thread_start" => tool_thread_start(db, args)?,
658 "commit_deltas" => tool_commit_deltas(db, args)?,
659 "essence" => tool_essence(db, args)?,
660 "recall" => tool_recall(db, args)?,
661 "forget" => tool_forget(db, args)?,
662 "list" => tool_list(db, args)?,
663 "snapshot_list" => tool_snapshot_list(db)?,
664 "stats" => tool_stats(db)?,
665 "cache_put" => tool_cache_put(db, args)?,
666 "cache_get" => tool_cache_get(db, args)?,
667 "cache_stats" => tool_cache_stats(db, args)?,
668 "cache_put_semantic" => tool_cache_put_semantic(db, args)?,
669 "cache_get_semantic" => tool_cache_get_semantic(db, args)?,
670 other => {
671 return Err(RpcError::new(
672 METHOD_NOT_FOUND,
673 format!("unknown tool '{other}'"),
674 ))
675 }
676 };
677
678 Ok(json!({
679 "content": [ { "type": "text", "text": text } ],
680 "isError": false,
681 }))
682}
683
684fn embed_text_via_server(
695 db: &mut Mnemo,
696 tool: &str,
697 text: &str,
698) -> std::result::Result<Vec<f32>, RpcError> {
699 let embedder = db.embedder().ok_or_else(|| {
700 RpcError::new(
701 INVALID_PARAMS,
702 format!(
703 "{tool}: 'text' requires an embedder on the server. \
704 Start with: mnemo serve --mcp <file> --embed-cmd '<cmd>' --embed-name <name>"
705 ),
706 )
707 })?;
708 let mut vs = embedder.embed(&[text]).map_err(mnemo_err)?;
709 vs.pop().ok_or_else(|| {
710 RpcError::new(
711 INTERNAL_ERROR,
712 format!("{tool}: embedder returned no vector for the input text"),
713 )
714 })
715}
716
717fn mnemo_err(e: MnemoError) -> RpcError {
718 RpcError::new(INTERNAL_ERROR, e.to_string())
719}
720
721fn tool_about(db: &mut Mnemo) -> std::result::Result<String, RpcError> {
722 let mems = db.about().map_err(mnemo_err)?;
723 let mut arr: Vec<Value> = mems.iter().map(memory_to_json).collect();
724 if let Some(render) = db.about_essence().map_err(mnemo_err)? {
727 arr.push(json!({ "kind": "essence", "scope": "project", "render": render }));
728 }
729 Ok(serde_json::to_string_pretty(&arr).unwrap_or_else(|_| "[]".into()))
730}
731
732fn tool_thread_export(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
733 let thread = args
734 .get("thread")
735 .and_then(|v| v.as_str())
736 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_export: missing 'thread'"))?;
737 let turns = db.thread_export_turns(thread).map_err(mnemo_err)?;
738 let arr: Vec<Value> = turns.iter().map(|t| t.to_json()).collect();
739 Ok(Value::Array(arr).to_string())
740}
741
742fn tool_remember(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
743 let content = args
744 .get("content")
745 .and_then(|v| v.as_str())
746 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "remember: missing 'content'"))?;
747 let mtype_str = args
748 .get("memory_type")
749 .and_then(|v| v.as_str())
750 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "remember: missing 'memory_type'"))?;
751 let mtype = MemoryType::parse(mtype_str).ok_or_else(|| {
752 RpcError::new(
753 INVALID_PARAMS,
754 format!("remember: unknown memory_type '{mtype_str}'"),
755 )
756 })?;
757 let vector = match (args.get("vector"), args.get("text")) {
760 (Some(_), Some(_)) => {
761 return Err(RpcError::new(
762 INVALID_PARAMS,
763 "remember: pass EITHER 'vector' or 'text', not both",
764 ));
765 }
766 (Some(v), None) => v
767 .as_array()
768 .ok_or_else(|| {
769 RpcError::new(INVALID_PARAMS, "remember: 'vector' must be a JSON array")
770 })?
771 .iter()
772 .map(|v| v.as_f64().map(|f| f as f32))
773 .collect::<Option<Vec<f32>>>()
774 .ok_or_else(|| {
775 RpcError::new(
776 INVALID_PARAMS,
777 "remember: 'vector' must be an array of numbers",
778 )
779 })?,
780 (None, Some(t)) => {
781 let text = t.as_str().ok_or_else(|| {
782 RpcError::new(INVALID_PARAMS, "remember: 'text' must be a string")
783 })?;
784 embed_text_via_server(db, "remember", text)?
785 }
786 (None, None) => {
787 return Err(RpcError::new(
788 INVALID_PARAMS,
789 "remember: missing 'vector' or 'text'",
790 ))
791 }
792 };
793
794 let mut mem = Memory::new(content, mtype, vector);
795 if let Some(agent) = args.get("agent_id").and_then(|v| v.as_str()) {
796 mem = mem.with_agent(agent);
797 }
798 if let Some(sess) = args.get("session_id").and_then(|v| v.as_str()) {
799 mem = mem.with_session(sess);
800 }
801 if let Some(imp) = args.get("importance").and_then(|v| v.as_f64()) {
802 mem = mem.with_importance(imp as f32);
803 }
804 if let Some(ttl) = args.get("ttl_secs").and_then(|v| v.as_i64()) {
805 mem = mem.with_ttl(ttl);
806 }
807 if let Some(shared) = args.get("shared").and_then(|v| v.as_bool()) {
808 if shared {
809 mem = mem.with_scope(Scope::Shared);
810 }
811 }
812 if let Some(meta) = args.get("metadata").and_then(|v| v.as_object()) {
813 for (k, v) in meta {
814 mem.metadata.insert(k.clone(), v.clone());
815 }
816 }
817
818 match args.get("key").and_then(|v| v.as_str()) {
823 None => {
824 let id = db.remember(mem).map_err(mnemo_err)?;
825 db.flush().map_err(mnemo_err)?;
826 Ok(json!({ "id": id.to_string(), "outcome": "inserted" }).to_string())
827 }
828 Some(k) => {
829 let (id, outcome) = db.remember_keyed(mem, k).map_err(mnemo_err)?;
830 db.flush().map_err(mnemo_err)?;
831 let outcome_str = match outcome {
832 crate::WriteOutcome::Added => "added",
833 crate::WriteOutcome::Superseded => "superseded",
834 crate::WriteOutcome::Unchanged => "unchanged",
835 crate::WriteOutcome::Updated => "superseded",
836 };
837 Ok(json!({ "id": id.to_string(), "outcome": outcome_str }).to_string())
838 }
839 }
840}
841
842fn tool_update(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
843 use crate::Ulid;
844 use crate::{MemoryPatch, WriteOutcome};
845 let id_str = args
846 .get("id")
847 .and_then(|v| v.as_str())
848 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: missing 'id'"))?;
849 let id = Ulid::from_string(id_str)
850 .map_err(|_| RpcError::new(INVALID_PARAMS, format!("update: invalid ULID '{id_str}'")))?;
851 let has_text = args.get("text").is_some();
855 let has_content = args.get("content").is_some();
856 let has_vector = args.get("vector").is_some();
857 if has_text && (has_content || has_vector) {
858 return Err(RpcError::new(
859 INVALID_PARAMS,
860 "update: pass EITHER 'text' or 'content'+'vector', not both",
861 ));
862 }
863 let mut patch = MemoryPatch::default();
864 if has_content {
865 let c = args
866 .get("content")
867 .and_then(|v| v.as_str())
868 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: 'content' must be a string"))?;
869 patch.content = Some(c.to_string());
870 }
871 if has_vector {
872 let v = args
873 .get("vector")
874 .and_then(|v| v.as_array())
875 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: 'vector' must be a JSON array"))?
876 .iter()
877 .map(|v| v.as_f64().map(|f| f as f32))
878 .collect::<Option<Vec<f32>>>()
879 .ok_or_else(|| {
880 RpcError::new(
881 INVALID_PARAMS,
882 "update: 'vector' must be an array of numbers",
883 )
884 })?;
885 patch.vector = Some(v);
886 }
887 if let Some(imp) = args.get("importance").and_then(|v| v.as_f64()) {
888 patch.importance = Some(imp as f32);
889 }
890 if let Some(meta) = args.get("metadata").and_then(|v| v.as_object()) {
891 patch.metadata = Some(meta.clone());
892 }
893 let clear_ttl = args
894 .get("clear_ttl")
895 .and_then(|v| v.as_bool())
896 .unwrap_or(false);
897 if clear_ttl {
898 patch.ttl_secs = Some(None);
899 } else if let Some(ttl) = args.get("ttl_secs").and_then(|v| v.as_i64()) {
900 patch.ttl_secs = Some(Some(ttl));
901 }
902
903 let outcome = if has_text {
904 let t = args
905 .get("text")
906 .and_then(|v| v.as_str())
907 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "update: 'text' must be a string"))?;
908 db.update_text(&id, t, patch).map_err(mnemo_err)?
909 } else {
910 db.update(&id, patch).map_err(mnemo_err)?
911 };
912 db.flush().map_err(mnemo_err)?;
913 let outcome_str = match outcome {
914 WriteOutcome::Updated => "updated",
915 WriteOutcome::Unchanged => "unchanged",
916 WriteOutcome::Added => "updated",
917 WriteOutcome::Superseded => "updated",
918 };
919 Ok(json!({ "id": id.to_string(), "outcome": outcome_str }).to_string())
920}
921
922fn tool_thread_start(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
923 let new = args
924 .get("new")
925 .and_then(|v| v.as_str())
926 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_start: missing 'new'"))?;
927 let from = args
928 .get("from")
929 .and_then(|v| v.as_str())
930 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_start: missing 'from'"))?;
931 let id = db.thread_start(new, from).map_err(mnemo_err)?;
932 db.flush().map_err(mnemo_err)?;
933 Ok(json!({ "id": id.to_string() }).to_string())
934}
935
936fn tool_commit_deltas(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
937 use crate::{
938 CommitBatch, EssenceOp, EssenceOpKind, EssenceScope, EssenceSection, KeyedMemory,
939 ThreadTurn, TurnPolicy, Ulid,
940 };
941 let mut batch = CommitBatch::default();
942
943 if let Some(arr) = args.get("facts").and_then(|v| v.as_array()) {
945 for (i, f) in arr.iter().enumerate() {
946 let obj = f.as_object().ok_or_else(|| {
947 RpcError::new(
948 INVALID_PARAMS,
949 format!("commit_deltas: facts[{i}] must be an object"),
950 )
951 })?;
952 let content = obj.get("content").and_then(|v| v.as_str()).ok_or_else(|| {
953 RpcError::new(
954 INVALID_PARAMS,
955 format!("commit_deltas: facts[{i}] missing 'content'"),
956 )
957 })?;
958 let mtype_str = obj
959 .get("memory_type")
960 .and_then(|v| v.as_str())
961 .ok_or_else(|| {
962 RpcError::new(
963 INVALID_PARAMS,
964 format!("commit_deltas: facts[{i}] missing 'memory_type'"),
965 )
966 })?;
967 let mtype = MemoryType::parse(mtype_str).ok_or_else(|| {
968 RpcError::new(
969 INVALID_PARAMS,
970 format!("commit_deltas: facts[{i}] unknown memory_type '{mtype_str}'"),
971 )
972 })?;
973 let vector = match obj.get("vector") {
977 None | Some(Value::Null) => Vec::new(),
978 Some(v) => v
979 .as_array()
980 .and_then(|a| {
981 a.iter()
982 .map(|x| x.as_f64().map(|f| f as f32))
983 .collect::<Option<Vec<f32>>>()
984 })
985 .ok_or_else(|| {
986 RpcError::new(
987 INVALID_PARAMS,
988 format!("commit_deltas: facts[{i}] vector must be an array of numbers"),
989 )
990 })?,
991 };
992 let mut mem = Memory::new(content, mtype, vector);
993 if let Some(a) = obj.get("agent_id").and_then(|v| v.as_str()) {
994 mem = mem.with_agent(a);
995 }
996 if let Some(s) = obj.get("session_id").and_then(|v| v.as_str()) {
997 mem = mem.with_session(s);
998 }
999 if let Some(imp) = obj.get("importance").and_then(|v| v.as_f64()) {
1000 mem = mem.with_importance(imp as f32);
1001 }
1002 if let Some(shared) = obj.get("shared").and_then(|v| v.as_bool()) {
1003 if shared {
1004 mem = mem.with_scope(Scope::Shared);
1005 }
1006 }
1007 if let Some(meta) = obj.get("metadata").and_then(|v| v.as_object()) {
1008 for (k, v) in meta {
1009 mem.metadata.insert(k.clone(), v.clone());
1010 }
1011 }
1012 let key = obj.get("key").and_then(|v| v.as_str()).map(String::from);
1013 batch.facts.push(KeyedMemory { memory: mem, key });
1014 }
1015 }
1016
1017 if let Some(arr) = args.get("essence_ops").and_then(|v| v.as_array()) {
1019 for (i, o) in arr.iter().enumerate() {
1020 let obj = o.as_object().ok_or_else(|| {
1021 RpcError::new(
1022 INVALID_PARAMS,
1023 format!("commit_deltas: essence_ops[{i}] must be an object"),
1024 )
1025 })?;
1026 let kind_str = obj.get("kind").and_then(|v| v.as_str()).ok_or_else(|| {
1027 RpcError::new(
1028 INVALID_PARAMS,
1029 format!("commit_deltas: essence_ops[{i}] missing 'kind'"),
1030 )
1031 })?;
1032 let kind = match kind_str {
1033 "add" => EssenceOpKind::Add,
1034 "append_timeline" => EssenceOpKind::AppendTimeline,
1035 "update" => EssenceOpKind::Update,
1036 "resolve" => EssenceOpKind::Resolve,
1037 "remove" => EssenceOpKind::Remove,
1038 other => {
1039 return Err(RpcError::new(
1040 INVALID_PARAMS,
1041 format!("commit_deltas: essence_ops[{i}] unknown kind '{other}'"),
1042 ));
1043 }
1044 };
1045 let thread = obj.get("thread").and_then(|v| v.as_str()).map(String::from);
1046 let project = obj
1047 .get("project")
1048 .and_then(|v| v.as_bool())
1049 .unwrap_or(false);
1050 let scope = match (thread, project) {
1051 (Some(t), false) => Some(EssenceScope::Thread(t)),
1052 (None, true) => Some(EssenceScope::Project),
1053 (None, false) => None,
1054 (Some(_), true) => {
1055 return Err(RpcError::new(
1056 INVALID_PARAMS,
1057 format!(
1058 "commit_deltas: essence_ops[{i}] can't set both 'thread' and 'project'"
1059 ),
1060 ));
1061 }
1062 };
1063 let section = obj
1064 .get("section")
1065 .and_then(|v| v.as_str())
1066 .and_then(EssenceSection::parse);
1067 let id = obj
1068 .get("id")
1069 .and_then(|v| v.as_str())
1070 .map(Ulid::from_string)
1071 .transpose()
1072 .map_err(|_| {
1073 RpcError::new(
1074 INVALID_PARAMS,
1075 format!("commit_deltas: essence_ops[{i}] 'id' is not a valid ULID"),
1076 )
1077 })?;
1078 let text = obj.get("text").and_then(|v| v.as_str()).map(String::from);
1079 let refs = obj.get("refs").and_then(|v| v.as_array()).map(|a| {
1080 a.iter()
1081 .filter_map(|x| x.as_str().map(String::from))
1082 .collect()
1083 });
1084 batch.essence_ops.push(EssenceOp {
1085 kind,
1086 scope,
1087 section,
1088 id,
1089 text,
1090 refs,
1091 });
1092 }
1093 }
1094
1095 if let Some(arr) = args.get("turns").and_then(|v| v.as_array()) {
1097 for (i, t) in arr.iter().enumerate() {
1098 let obj = t.as_object().ok_or_else(|| {
1099 RpcError::new(
1100 INVALID_PARAMS,
1101 format!("commit_deltas: turns[{i}] must be an object"),
1102 )
1103 })?;
1104 let turn_id = obj
1105 .get("turn_id")
1106 .and_then(|v| v.as_str())
1107 .ok_or_else(|| {
1108 RpcError::new(
1109 INVALID_PARAMS,
1110 format!("commit_deltas: turns[{i}] missing 'turn_id'"),
1111 )
1112 })?
1113 .to_string();
1114 let role = obj
1115 .get("role")
1116 .and_then(|v| v.as_str())
1117 .ok_or_else(|| {
1118 RpcError::new(
1119 INVALID_PARAMS,
1120 format!("commit_deltas: turns[{i}] missing 'role'"),
1121 )
1122 })?
1123 .to_string();
1124 let content = obj
1125 .get("content")
1126 .and_then(|v| v.as_str())
1127 .ok_or_else(|| {
1128 RpcError::new(
1129 INVALID_PARAMS,
1130 format!("commit_deltas: turns[{i}] missing 'content'"),
1131 )
1132 })?
1133 .to_string();
1134 let timestamp = obj.get("timestamp").and_then(|v| v.as_i64());
1135 batch.turns.push(ThreadTurn {
1136 turn_id,
1137 role,
1138 content,
1139 timestamp,
1140 });
1141 }
1142 }
1143 batch.thread_id = args
1144 .get("thread_id")
1145 .and_then(|v| v.as_str())
1146 .map(String::from);
1147 let policy_str = args
1148 .get("turn_policy")
1149 .and_then(|v| v.as_str())
1150 .unwrap_or("archive");
1151 batch.turn_policy = match policy_str {
1152 "archive" => TurnPolicy::Archive,
1153 "drop" => TurnPolicy::Drop,
1154 "expire_after" => {
1155 let ttl = args
1156 .get("turn_policy_ttl")
1157 .and_then(|v| v.as_i64())
1158 .ok_or_else(|| {
1159 RpcError::new(
1160 INVALID_PARAMS,
1161 "commit_deltas: turn_policy='expire_after' requires 'turn_policy_ttl' (seconds)",
1162 )
1163 })?;
1164 TurnPolicy::ExpireAfter(ttl)
1165 }
1166 other => {
1167 return Err(RpcError::new(
1168 INVALID_PARAMS,
1169 format!("commit_deltas: unknown turn_policy '{other}'"),
1170 ));
1171 }
1172 };
1173
1174 let report = db.commit_deltas(batch).map_err(mnemo_err)?;
1175 Ok(json!({
1176 "facts_added": report.facts_added,
1177 "facts_superseded": report.facts_superseded,
1178 "facts_unchanged": report.facts_unchanged,
1179 "essence_ops_applied": report.essence_ops_applied,
1180 "essence_ops_unchanged": report.essence_ops_unchanged,
1181 "turns_added": report.turns_added,
1182 "turns_skipped": report.turns_skipped,
1183 "turns_truncated": report.turns_truncated,
1184 "turns_chunked": report.turns_chunked,
1185 })
1186 .to_string())
1187}
1188
1189fn tool_essence(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1190 use crate::{EssenceScope, EssenceSection, Ulid, WriteOutcome};
1191 let op = args
1192 .get("op")
1193 .and_then(|v| v.as_str())
1194 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence: missing 'op'"))?;
1195 let thread = args
1196 .get("thread")
1197 .and_then(|v| v.as_str())
1198 .map(String::from);
1199 let project = args
1200 .get("project")
1201 .and_then(|v| v.as_bool())
1202 .unwrap_or(false);
1203 let scope_from = |t: Option<String>, p: bool| -> std::result::Result<EssenceScope, RpcError> {
1204 match (t, p) {
1205 (Some(s), false) => Ok(EssenceScope::Thread(s)),
1206 (None, true) => Ok(EssenceScope::Project),
1207 (None, false) => Err(RpcError::new(
1208 INVALID_PARAMS,
1209 "essence: pass either `thread` or `project: true`",
1210 )),
1211 (Some(_), true) => Err(RpcError::new(
1212 INVALID_PARAMS,
1213 "essence: `thread` and `project` are mutually exclusive",
1214 )),
1215 }
1216 };
1217 let refs: Option<Vec<String>> = args.get("refs").and_then(|v| v.as_array()).map(|arr| {
1218 arr.iter()
1219 .filter_map(|x| x.as_str().map(String::from))
1220 .collect()
1221 });
1222 let text = args.get("text").and_then(|v| v.as_str()).map(String::from);
1223 let id_str = args.get("id").and_then(|v| v.as_str());
1224 let outcome_to_str = |o: WriteOutcome| -> &'static str {
1225 match o {
1226 WriteOutcome::Updated => "updated",
1227 WriteOutcome::Unchanged => "unchanged",
1228 WriteOutcome::Added => "updated",
1229 WriteOutcome::Superseded => "updated",
1230 }
1231 };
1232 match op {
1233 "add" => {
1234 let section_str = args
1235 .get("section")
1236 .and_then(|v| v.as_str())
1237 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence add: missing 'section'"))?;
1238 let section = EssenceSection::parse(section_str).ok_or_else(|| {
1239 RpcError::new(
1240 INVALID_PARAMS,
1241 format!("essence add: unknown section '{section_str}'"),
1242 )
1243 })?;
1244 if section == EssenceSection::Timeline {
1245 return Err(RpcError::new(
1246 INVALID_PARAMS,
1247 "essence add: timeline uses op: append, not add",
1248 ));
1249 }
1250 let text =
1251 text.ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence add: missing 'text'"))?;
1252 let scope = scope_from(thread, project)?;
1253 let id = db
1254 .essence_add(scope, section, &text, refs)
1255 .map_err(mnemo_err)?;
1256 db.flush().map_err(mnemo_err)?;
1257 Ok(json!({ "id": id.to_string(), "outcome": "added" }).to_string())
1258 }
1259 "append" => {
1260 let text = text
1261 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence append: missing 'text'"))?;
1262 let scope = scope_from(thread, project)?;
1263 let id = db
1264 .essence_append_timeline(scope, &text, refs)
1265 .map_err(mnemo_err)?;
1266 db.flush().map_err(mnemo_err)?;
1267 Ok(json!({ "id": id.to_string(), "outcome": "appended" }).to_string())
1268 }
1269 "update" => {
1270 let id = id_str
1271 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence update: missing 'id'"))?;
1272 let ulid = Ulid::from_string(id).map_err(|_| {
1273 RpcError::new(
1274 INVALID_PARAMS,
1275 format!("essence update: invalid ULID '{id}'"),
1276 )
1277 })?;
1278 let text = text
1279 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence update: missing 'text'"))?;
1280 let outcome = db.essence_update(&ulid, &text, refs).map_err(mnemo_err)?;
1281 db.flush().map_err(mnemo_err)?;
1282 Ok(json!({ "id": id, "outcome": outcome_to_str(outcome) }).to_string())
1283 }
1284 "resolve" => {
1285 let id = id_str
1286 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence resolve: missing 'id'"))?;
1287 let ulid = Ulid::from_string(id).map_err(|_| {
1288 RpcError::new(
1289 INVALID_PARAMS,
1290 format!("essence resolve: invalid ULID '{id}'"),
1291 )
1292 })?;
1293 let outcome = db.essence_resolve(&ulid).map_err(mnemo_err)?;
1294 db.flush().map_err(mnemo_err)?;
1295 let s = match outcome {
1296 WriteOutcome::Updated => "resolved",
1297 WriteOutcome::Unchanged => "already-resolved",
1298 _ => "resolved",
1299 };
1300 Ok(json!({ "id": id, "outcome": s }).to_string())
1301 }
1302 "remove" => {
1303 let id = id_str
1304 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "essence remove: missing 'id'"))?;
1305 let ulid = Ulid::from_string(id).map_err(|_| {
1306 RpcError::new(
1307 INVALID_PARAMS,
1308 format!("essence remove: invalid ULID '{id}'"),
1309 )
1310 })?;
1311 db.essence_remove(&ulid).map_err(mnemo_err)?;
1312 db.flush().map_err(mnemo_err)?;
1313 Ok(json!({ "id": id, "outcome": "removed" }).to_string())
1314 }
1315 "render" => {
1316 let budget = args.get("budget").and_then(|v| v.as_u64()).unwrap_or(6144) as usize;
1317 let scope = scope_from(thread, project)?;
1318 let rendered = db.essence(scope, budget).map_err(mnemo_err)?;
1319 Ok(rendered)
1320 }
1321 other => Err(RpcError::new(
1322 INVALID_PARAMS,
1323 format!("essence: unknown op '{other}'"),
1324 )),
1325 }
1326}
1327
1328fn tool_thread_ingest(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1329 let thread = args
1330 .get("thread")
1331 .and_then(|v| v.as_str())
1332 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_ingest: missing 'thread'"))?;
1333 let path = args
1334 .get("transcript_path")
1335 .and_then(|v| v.as_str())
1336 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "thread_ingest: missing 'transcript_path'"))?;
1337 let format_str = args
1338 .get("format")
1339 .and_then(|v| v.as_str())
1340 .unwrap_or("jsonl");
1341 let format = crate::TranscriptFormat::parse(format_str).ok_or_else(|| {
1342 RpcError::new(
1343 INVALID_PARAMS,
1344 format!(
1345 "thread_ingest: unknown format '{format_str}' (want 'jsonl' or 'claude-code-jsonl')"
1346 ),
1347 )
1348 })?;
1349 let file = std::fs::File::open(path).map_err(|e| {
1350 RpcError::new(
1351 INVALID_PARAMS,
1352 format!("thread_ingest: cannot open {path}: {e}"),
1353 )
1354 })?;
1355 let turns = crate::read_transcript(file, format).map_err(mnemo_err)?;
1356 let report = db.ingest_thread(thread, &turns).map_err(mnemo_err)?;
1357 db.flush().map_err(mnemo_err)?;
1358 Ok(json!({
1359 "added_records": report.added_records,
1360 "skipped_existing_turns": report.skipped_existing_turns,
1361 "chunked_turns": report.chunked_turns,
1362 "truncated_turns": report.truncated_turns,
1363 })
1364 .to_string())
1365}
1366
1367fn tool_history(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1368 use crate::KeyScope;
1369 let key = args
1370 .get("key")
1371 .and_then(|v| v.as_str())
1372 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "history: missing 'key'"))?;
1373 let agent = args.get("agent").and_then(|v| v.as_str());
1374 let shared = args
1375 .get("shared")
1376 .and_then(|v| v.as_bool())
1377 .unwrap_or(false);
1378 if shared && agent.is_some() {
1379 return Err(RpcError::new(
1380 INVALID_PARAMS,
1381 "history: 'agent' and 'shared' are mutually exclusive",
1382 ));
1383 }
1384 let scope = if shared {
1386 KeyScope::Shared
1387 } else {
1388 match agent {
1389 Some(a) => KeyScope::Private(a.to_string()),
1390 None => KeyScope::private_default(),
1391 }
1392 };
1393 let chain = db.history_in(key, scope).map_err(mnemo_err)?;
1394 let out: Vec<Value> = chain
1395 .iter()
1396 .enumerate()
1397 .map(|(i, m)| {
1398 let mut v = memory_to_json(m);
1399 if let Some(obj) = v.as_object_mut() {
1400 obj.insert(
1401 "state".into(),
1402 Value::String(if i == 0 {
1403 "live".into()
1404 } else {
1405 "superseded".into()
1406 }),
1407 );
1408 }
1409 v
1410 })
1411 .collect();
1412 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1413}
1414
1415fn tool_recall(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1416 let query = match (args.get("query"), args.get("text")) {
1418 (Some(_), Some(_)) => {
1419 return Err(RpcError::new(
1420 INVALID_PARAMS,
1421 "recall: pass EITHER 'query' or 'text', not both",
1422 ))
1423 }
1424 (Some(v), None) => v
1425 .as_array()
1426 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "recall: 'query' must be a JSON array"))?
1427 .iter()
1428 .map(|v| v.as_f64().map(|f| f as f32))
1429 .collect::<Option<Vec<f32>>>()
1430 .ok_or_else(|| {
1431 RpcError::new(
1432 INVALID_PARAMS,
1433 "recall: 'query' must be an array of numbers",
1434 )
1435 })?,
1436 (None, Some(t)) => {
1437 let text = t
1438 .as_str()
1439 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "recall: 'text' must be a string"))?;
1440 embed_text_via_server(db, "recall", text)?
1441 }
1442 (None, None) => {
1443 return Err(RpcError::new(
1444 INVALID_PARAMS,
1445 "recall: missing 'query' or 'text'",
1446 ))
1447 }
1448 };
1449 let top_k = args.get("top_k").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
1450 let metric = match args
1451 .get("metric")
1452 .and_then(|v| v.as_str())
1453 .unwrap_or("cosine")
1454 {
1455 "cosine" => Metric::Cosine,
1456 "l2" => Metric::L2,
1457 "dot" => Metric::Dot,
1458 other => {
1459 return Err(RpcError::new(
1460 INVALID_PARAMS,
1461 format!("recall: unknown metric '{other}'"),
1462 ))
1463 }
1464 };
1465 let mut req = RecallRequest::new(query)
1478 .top_k(top_k)
1479 .metric(metric)
1480 .track_access(true);
1481 if let Some(types) = args.get("types").and_then(|v| v.as_array()) {
1482 let parsed: Vec<MemoryType> = types
1483 .iter()
1484 .filter_map(|t| t.as_str().and_then(MemoryType::parse))
1485 .collect();
1486 if !parsed.is_empty() {
1487 req = req.types(parsed);
1488 }
1489 }
1490 if let Some(agent) = args.get("agent").and_then(|v| v.as_str()) {
1491 req = req.agent(agent.to_string());
1492 }
1493 if let Some(filter_obj) = args.get("filter").and_then(|v| v.as_object()) {
1500 req = req.filter(parse_mcp_filter(filter_obj)?);
1501 }
1502 let has_max_bytes = args.get("max_bytes").is_some();
1504 if let Some(n) = args.get("max_bytes").and_then(|v| v.as_u64()) {
1505 req = req.max_bytes(n as usize);
1506 }
1507
1508 let resp = db.recall_budgeted(&req).map_err(mnemo_err)?;
1509 let out: Vec<Value> = resp
1510 .hits
1511 .into_iter()
1512 .map(|h| {
1513 let mut obj = memory_to_json(&h.memory);
1514 if let Some(map) = obj.as_object_mut() {
1515 map.insert("score".into(), json!(h.score));
1516 map.insert("similarity".into(), json!(h.similarity));
1517 }
1518 obj
1519 })
1520 .collect();
1521 if has_max_bytes {
1522 let wrapped = json!({
1525 "hits": out,
1526 "truncated": resp.truncated,
1527 });
1528 Ok(serde_json::to_string_pretty(&wrapped).unwrap_or_else(|_| "{}".into()))
1529 } else {
1530 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1531 }
1532}
1533
1534fn parse_mcp_filter(
1544 obj: &serde_json::Map<String, Value>,
1545) -> std::result::Result<crate::Filter, RpcError> {
1546 let mut f = crate::Filter::default();
1547 if let Some(sid) = obj.get("session_id").and_then(|v| v.as_str()) {
1548 f = f.with_session(sid.to_string());
1549 }
1550 if let Some(meta) = obj.get("metadata").and_then(|v| v.as_object()) {
1551 for (k, v) in meta {
1552 match v {
1553 Value::Object(ops) => {
1555 for (op_key, op_val) in ops {
1556 match op_key.as_str() {
1557 "$exists" => match op_val.as_bool() {
1558 Some(true) => {
1559 f = f.metadata_exists(k.clone());
1560 }
1561 Some(false) => {
1562 return Err(RpcError::new(
1563 INVALID_PARAMS,
1564 format!(
1565 "filter.metadata.{k}.$exists=false isn't supported; \
1566 drop the filter term instead"
1567 ),
1568 ));
1569 }
1570 None => {
1571 return Err(RpcError::new(
1572 INVALID_PARAMS,
1573 format!("filter.metadata.{k}.$exists must be a boolean"),
1574 ));
1575 }
1576 },
1577 "$in" => {
1578 let arr = op_val.as_array().ok_or_else(|| {
1579 RpcError::new(
1580 INVALID_PARAMS,
1581 format!("filter.metadata.{k}.$in must be an array"),
1582 )
1583 })?;
1584 f = f.metadata_in(k.clone(), arr.clone());
1585 }
1586 "$gte" | "$lte" => {
1587 let mut min = None;
1588 let mut max = None;
1589 for (bound, bound_val) in ops {
1590 let n = bound_val.as_f64().ok_or_else(|| {
1591 RpcError::new(
1592 INVALID_PARAMS,
1593 format!("filter.metadata.{k}.{bound} must be a number"),
1594 )
1595 })?;
1596 match bound.as_str() {
1597 "$gte" => min = Some(n),
1598 "$lte" => max = Some(n),
1599 other => {
1600 return Err(RpcError::new(
1601 INVALID_PARAMS,
1602 format!(
1603 "filter.metadata.{k}: unsupported range \
1604 operator '{other}' (want $gte, $lte)"
1605 ),
1606 ));
1607 }
1608 }
1609 }
1610 f = f.metadata_range(k.clone(), min, max);
1611 break; }
1613 other => {
1614 return Err(RpcError::new(
1615 INVALID_PARAMS,
1616 format!(
1617 "filter.metadata.{k}: unknown operator '{other}' \
1618 (want $exists, $in, $gte, $lte)"
1619 ),
1620 ));
1621 }
1622 }
1623 }
1624 }
1625 other => {
1627 f = f.metadata_eq(k.clone(), other.clone());
1628 }
1629 }
1630 }
1631 }
1632 Ok(f)
1633}
1634
1635fn tool_forget(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1636 let id_str = args
1637 .get("id")
1638 .and_then(|v| v.as_str())
1639 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "forget: missing 'id'"))?;
1640 let ulid = crate::Ulid::from_string(id_str)
1641 .map_err(|_| RpcError::new(INVALID_PARAMS, format!("forget: invalid ULID '{id_str}'")))?;
1642 db.delete(&ulid).map_err(mnemo_err)?;
1643 db.flush().map_err(mnemo_err)?;
1644 Ok(json!({ "deleted": id_str }).to_string())
1645}
1646
1647fn tool_list(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1648 let include_vectors = args
1649 .get("include_vectors")
1650 .and_then(|v| v.as_bool())
1651 .unwrap_or(false);
1652 let mems = db.memories().map_err(mnemo_err)?;
1653 let out: Vec<Value> = mems
1654 .iter()
1655 .map(|m| {
1656 let mut v = memory_to_json(m);
1657 if !include_vectors {
1658 if let Some(o) = v.as_object_mut() {
1659 o.remove("vector");
1660 }
1661 }
1662 v
1663 })
1664 .collect();
1665 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1666}
1667
1668fn tool_snapshot_list(db: &mut Mnemo) -> std::result::Result<String, RpcError> {
1669 let snaps = db.snapshots();
1670 let out: Vec<Value> = snaps
1671 .into_iter()
1672 .map(|s| {
1673 json!({
1674 "txn_id": s.txn_id,
1675 "created_at": s.created_at,
1676 "memory_count": s.memory_count,
1677 })
1678 })
1679 .collect();
1680 Ok(serde_json::to_string_pretty(&out).unwrap_or_else(|_| "[]".into()))
1681}
1682
1683fn tool_stats(db: &mut Mnemo) -> std::result::Result<String, RpcError> {
1684 let s = db.stats().map_err(mnemo_err)?;
1685 let (cache_pages, cache_capacity) = db.page_cache_stats();
1686 let snapshot_count = db.snapshots().len();
1687 Ok(json!({
1690 "memories": s.memories,
1691 "deleted": s.deleted,
1692 "dimensions": s.dimensions,
1693 "file_bytes": s.file_bytes,
1694 "agents": s.agents,
1695 "encrypted": s.encrypted,
1696 "created_at": s.created_at,
1697 "wal_pages": s.wal_pages,
1698 "index": s.index,
1699 "snapshot_count": snapshot_count,
1700 "cache_pages": cache_pages,
1701 "cache_capacity": cache_capacity,
1702 })
1703 .to_string())
1704}
1705
1706fn tool_cache_put(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1709 let ns = args
1710 .get("namespace")
1711 .and_then(|v| v.as_str())
1712 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put: missing 'namespace'"))?;
1713 let key = args
1714 .get("key")
1715 .and_then(|v| v.as_str())
1716 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put: missing 'key'"))?;
1717 let value = args
1718 .get("value")
1719 .and_then(|v| v.as_str())
1720 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put: missing 'value'"))?;
1721 let content_type = args
1722 .get("content_type")
1723 .and_then(|v| v.as_str())
1724 .unwrap_or("text")
1725 .to_string();
1726 let ttl_secs = args.get("ttl_secs").and_then(|v| v.as_u64());
1727 let opts = CachePutOpts {
1728 content_type,
1729 ttl_secs,
1730 cost_hint_ms: None,
1731 };
1732 db.cache_put(ns, key, value.as_bytes(), opts)
1733 .map_err(mnemo_err)?;
1734 db.flush().map_err(mnemo_err)?;
1735 Ok(json!({ "cached": true, "namespace": ns, "key": key, "bytes": value.len() }).to_string())
1736}
1737
1738fn tool_cache_get(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1739 let ns = args
1740 .get("namespace")
1741 .and_then(|v| v.as_str())
1742 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_get: missing 'namespace'"))?;
1743 let key = args
1744 .get("key")
1745 .and_then(|v| v.as_str())
1746 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_get: missing 'key'"))?;
1747 match db.cache_get(ns, key).map_err(mnemo_err)? {
1748 Some(hit) => Ok(json!({
1749 "hit": true,
1750 "value": String::from_utf8_lossy(&hit.value),
1751 "content_type": hit.content_type,
1752 "created_at": hit.created_at,
1753 "accessed_at": hit.accessed_at,
1754 "access_count": hit.access_count,
1755 })
1756 .to_string()),
1757 None => Ok(json!({ "hit": false }).to_string()),
1758 }
1759}
1760
1761fn tool_cache_stats(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1762 let ns = args.get("namespace").and_then(|v| v.as_str());
1763 let s = db.cache_stats(ns);
1764 Ok(json!({
1765 "namespace": ns,
1766 "entries": s.entries,
1767 "bytes": s.bytes,
1768 "hits": s.hits,
1769 "misses": s.misses,
1770 "hit_rate": s.hit_rate,
1771 "evictions": s.evictions,
1772 })
1773 .to_string())
1774}
1775
1776fn tool_cache_put_semantic(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1777 let ns = args
1778 .get("namespace")
1779 .and_then(|v| v.as_str())
1780 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'namespace'"))?;
1781 let key = args
1782 .get("key")
1783 .and_then(|v| v.as_str())
1784 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'key'"))?;
1785 let value = args
1786 .get("value")
1787 .and_then(|v| v.as_str())
1788 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'value'"))?;
1789 let (vector, model) = match (args.get("vector"), args.get("text")) {
1792 (Some(_), Some(_)) => {
1793 return Err(RpcError::new(
1794 INVALID_PARAMS,
1795 "cache_put_semantic: pass EITHER 'vector' or 'text', not both",
1796 ));
1797 }
1798 (Some(v), None) => {
1799 let vec = v
1800 .as_array()
1801 .ok_or_else(|| {
1802 RpcError::new(
1803 INVALID_PARAMS,
1804 "cache_put_semantic: 'vector' must be a JSON array",
1805 )
1806 })?
1807 .iter()
1808 .map(|v| v.as_f64().map(|f| f as f32))
1809 .collect::<Option<Vec<f32>>>()
1810 .ok_or_else(|| {
1811 RpcError::new(
1812 INVALID_PARAMS,
1813 "cache_put_semantic: 'vector' must be an array of numbers",
1814 )
1815 })?;
1816 let model = args
1817 .get("model")
1818 .and_then(|v| v.as_str())
1819 .ok_or_else(|| {
1820 RpcError::new(INVALID_PARAMS, "cache_put_semantic: missing 'model'")
1821 })?
1822 .to_string();
1823 (vec, model)
1824 }
1825 (None, Some(t)) => {
1826 let text = t.as_str().ok_or_else(|| {
1827 RpcError::new(
1828 INVALID_PARAMS,
1829 "cache_put_semantic: 'text' must be a string",
1830 )
1831 })?;
1832 let vec = embed_text_via_server(db, "cache_put_semantic", text)?;
1833 let model = match args.get("model").and_then(|v| v.as_str()) {
1835 Some(m) => m.to_string(),
1836 None => {
1837 db.embedder()
1838 .expect("embed_text_via_server would have errored otherwise")
1839 .identity()
1840 .name
1841 }
1842 };
1843 (vec, model)
1844 }
1845 (None, None) => {
1846 return Err(RpcError::new(
1847 INVALID_PARAMS,
1848 "cache_put_semantic: missing 'vector' or 'text'",
1849 ))
1850 }
1851 };
1852 let content_type = args
1853 .get("content_type")
1854 .and_then(|v| v.as_str())
1855 .unwrap_or("text")
1856 .to_string();
1857 let ttl_secs = args.get("ttl_secs").and_then(|v| v.as_u64());
1858 let opts = SemanticCachePutOpts {
1859 content_type,
1860 ttl_secs,
1861 cost_hint_ms: None,
1862 model,
1863 };
1864 db.cache_put_semantic(ns, key, vector, value.as_bytes(), opts)
1865 .map_err(mnemo_err)?;
1866 db.flush().map_err(mnemo_err)?;
1867 Ok(json!({ "cached": true, "namespace": ns, "key": key, "bytes": value.len() }).to_string())
1868}
1869
1870fn tool_cache_get_semantic(db: &mut Mnemo, args: &Value) -> std::result::Result<String, RpcError> {
1871 let ns = args
1872 .get("namespace")
1873 .and_then(|v| v.as_str())
1874 .ok_or_else(|| RpcError::new(INVALID_PARAMS, "cache_get_semantic: missing 'namespace'"))?;
1875 let (query, model_owned) = match (args.get("query"), args.get("text")) {
1878 (Some(_), Some(_)) => {
1879 return Err(RpcError::new(
1880 INVALID_PARAMS,
1881 "cache_get_semantic: pass EITHER 'query' or 'text', not both",
1882 ));
1883 }
1884 (Some(q), None) => {
1885 let vec = q
1886 .as_array()
1887 .ok_or_else(|| {
1888 RpcError::new(
1889 INVALID_PARAMS,
1890 "cache_get_semantic: 'query' must be a JSON array",
1891 )
1892 })?
1893 .iter()
1894 .map(|v| v.as_f64().map(|f| f as f32))
1895 .collect::<Option<Vec<f32>>>()
1896 .ok_or_else(|| {
1897 RpcError::new(
1898 INVALID_PARAMS,
1899 "cache_get_semantic: 'query' must be an array of numbers",
1900 )
1901 })?;
1902 let model = args
1903 .get("model")
1904 .and_then(|v| v.as_str())
1905 .ok_or_else(|| {
1906 RpcError::new(INVALID_PARAMS, "cache_get_semantic: missing 'model'")
1907 })?
1908 .to_string();
1909 (vec, model)
1910 }
1911 (None, Some(t)) => {
1912 let text = t.as_str().ok_or_else(|| {
1913 RpcError::new(
1914 INVALID_PARAMS,
1915 "cache_get_semantic: 'text' must be a string",
1916 )
1917 })?;
1918 let vec = embed_text_via_server(db, "cache_get_semantic", text)?;
1919 let model = match args.get("model").and_then(|v| v.as_str()) {
1920 Some(m) => m.to_string(),
1921 None => {
1922 db.embedder()
1923 .expect("embed_text_via_server would have errored otherwise")
1924 .identity()
1925 .name
1926 }
1927 };
1928 (vec, model)
1929 }
1930 (None, None) => {
1931 return Err(RpcError::new(
1932 INVALID_PARAMS,
1933 "cache_get_semantic: missing 'query' or 'text'",
1934 ))
1935 }
1936 };
1937 let model = model_owned.as_str();
1938 let threshold = args
1939 .get("threshold")
1940 .and_then(|v| v.as_f64())
1941 .map(|f| f as f32)
1942 .unwrap_or(DEFAULT_SEMANTIC_THRESHOLD);
1943 match db
1944 .cache_get_semantic(ns, &query, threshold, model)
1945 .map_err(mnemo_err)?
1946 {
1947 Some((hit, similarity)) => Ok(json!({
1948 "hit": true,
1949 "similarity": similarity,
1950 "value": String::from_utf8_lossy(&hit.value),
1951 "content_type": hit.content_type,
1952 "created_at": hit.created_at,
1953 "accessed_at": hit.accessed_at,
1954 "access_count": hit.access_count,
1955 })
1956 .to_string()),
1957 None => Ok(json!({ "hit": false }).to_string()),
1958 }
1959}
1960
1961fn memory_to_json(m: &Memory) -> Value {
1966 serde_json::to_value(m).unwrap_or_else(|_| {
1967 let mut map = Map::new();
1968 map.insert("id".into(), json!(m.id.to_string()));
1969 Value::Object(map)
1970 })
1971}