//! MCP transport: exposes the memory service as MCP tools over stdio.
//!
//! Only **memory semantics** are exposed (`remember / recall / relate / forget
//! / why`) — never raw database capabilities. See [`crate`] docs for the
//! license boundary this enforces.
use std::sync::Arc;
use rmcp::handler::server::router::tool::ToolRouter;
use rmcp::handler::server::wrapper::{Json, Parameters};
use rmcp::model::{ErrorCode, Implementation, ServerCapabilities, ServerInfo};
use rmcp::{tool, tool_handler, tool_router, ErrorData, ServerHandler};
use crate::limits::MAX_RECALL_LIMIT;
use crate::service::{LiveGenerationSlot, MemoryService};
/// Default number of memories returned by `recall`.
const DEFAULT_RECALL_LIMIT: usize = 10;
const UNREPORTED_MODEL: &str = "unreported";
// The boxed embedder and the shared, runtime-attached extraction backend the
// server stores — imported for internal use only. The canonical public paths are
// `velesdb_memory::DynEmbedder` / `velesdb_memory::DynExtractor` (crate root).
use crate::embedder::DynEmbedder;
use crate::extract::DynExtractor;
// --- Tool parameter / result DTOs ------------------------------------------
//
// The request envelopes, small id-results, and the id-echoing wire wrappers
// (`RecollectionDto`, `ExplanationDto` — the `id_str` twins of issue #1468)
// live in their own module so this file stays focused on the server and tool
// wiring; the domain types in `crate::model` are unchanged.
/// The context compiler's eight tools — a second `#[tool_router]` block whose
/// router is combined with the main one below, extending the ONE server.
#[cfg(feature = "context")]
mod context_tools;
mod advanced_tools;
mod dto;
mod extraction_job_model;
mod extraction_job_store;
mod extraction_jobs;
mod extractor_resolver;
mod migration_tools;
mod status;
mod wire;
use dto::{
EntityParams, EntityProfileDto, FeedbackParams, FeedbackResult, ForgetParams, ForgetResult,
RecallParams, RecallResult, RecallWhereParams, RelateParams, RelateResult, RememberParams,
RememberResult, UnrelateParams, UnrelateResult,
};
use extraction_jobs::{ExtractionJobs, JobError};
use extractor_resolver::ExtractorResolver;
/// Le constructeur de schema d'ENTREE, unique pour tous les outils :
/// [`crate::schema::wire_safe_input_schema`].
///
/// `keys` nomme les proprietes que CET outil accepte en chaine decimale
/// (`relate`/`unrelate` : `from`/`to` ; `forget`/`feedback` : `id` ;
/// `remember` : le `links[].target` imbrique) — la tolerance d'un id est une
/// connaissance de l'outil, jamais une regle globale :
/// `explain_compilation.fragment_id` est un `u64` STRICT et reste annonce
/// `integer`.
///
/// Il y avait deux constructeurs, celui-ci et un `wire_safe_input_schema`
/// gate sur `context` a cle `"id"` figee ; ils appliquaient la meme suite de
/// passes a une virgule pres. Il n'y en a plus qu'un.
use crate::schema::wire_safe_input_schema as id_wire_input_schema;
// --- The server ------------------------------------------------------------
/// MCP server wrapping a [`MemoryService`].
#[derive(Clone)]
pub struct McpServer {
service: Arc<LiveGenerationSlot<DynEmbedder>>,
/// Join guard of the background autograph worker (#1846) — present iff
/// an autograph extractor is configured. Held only for its `Drop`: the
/// server going down closes the queue and joins the worker — the job in
/// flight completes, still-queued ones are skipped and counted, so exit
/// waits for at most ONE generation.
_autograph_worker: Option<Arc<crate::service::AutographWorkerHandle>>,
/// Daemon-level default and per-call selection for `remember_extracted`.
extractors: Arc<parking_lot::RwLock<ExtractorResolver>>,
/// Durable receipt/status state machine for `remember_extracted`.
/// Configured by the native daemon once its store path is known.
extraction_jobs: Option<ExtractionJobs>,
/// Daemon-owned online embedding migration control plane.
online_migration: Option<Arc<crate::service::OnlineMigrationManager<DynEmbedder>>>,
/// Default time-to-live (seconds) applied to `remember`d facts that don't
/// specify their own `ttl_seconds`. `None` (the default) stores permanently.
/// Set from `VELESDB_MEMORY_DEFAULT_TTL` by the binary.
default_ttl: Option<u64>,
/// The store directory, for reading the embedding-provenance record
/// (#1751) in `memory_status`. `None` disables the provenance block
/// (reported as unrecorded — a store nobody can locate has no readable
/// record either way).
store_dir: Option<std::path::PathBuf>,
/// Allowlisted filesystem roots for `path`-referenced context fragments
/// (V2b-1). `None` (the default) disables path ingestion entirely — every
/// `path` fragment fails with an explicit error. Set from
/// `VELESDB_MEMORY_INGEST_ROOTS` by the binary via [`Self::with_ingest_roots`].
#[cfg(all(feature = "context", not(target_arch = "wasm32")))]
ingest_roots: Option<crate::context::IngestRoots>,
tool_router: ToolRouter<McpServer>,
}
#[tool_router]
impl McpServer {
/// Wrap a memory service as an MCP server.
#[must_use]
pub fn new(service: MemoryService<DynEmbedder>) -> Self {
let service = Arc::new(LiveGenerationSlot::new(service, UNREPORTED_MODEL));
// Autograph leaves the response path here (#1846): with an extractor
// configured, ONE background worker consumes a bounded queue and
// `remember` returns as soon as the fact is stored — measured 46-52 s
// inline against a 0.12 s embedding. The handle rides the server so
// shutdown finishes the job in flight and skips the rest, counted.
let autograph_worker = if matches!(service.inspect(MemoryService::has_autograph), Ok(true))
{
match service.spawn_autograph_worker(crate::limits::MAX_AUTOGRAPH_QUEUE) {
Ok(handle) => Some(Arc::new(handle)),
Err(error) => {
tracing::warn!(%error, "autograph worker not spawned; falling back inline");
None
}
}
} else {
None
};
Self {
service,
_autograph_worker: autograph_worker,
extractors: Arc::new(parking_lot::RwLock::new(ExtractorResolver::default())),
extraction_jobs: None,
online_migration: None,
default_ttl: None,
store_dir: None,
#[cfg(all(feature = "context", not(target_arch = "wasm32")))]
ingest_roots: None,
tool_router: Self::combined_router(),
}
}
/// Declare the running embedder's identity (model name + vector width)
/// so `memory_status` can name it — and say whether recall is semantic.
/// Undeclared, the status reports the embedder as unreported rather than
/// guessing from the service, which only ever sees `&[f32]`.
#[must_use]
pub fn with_embedder_identity(self, model: impl Into<String>, _dimension: usize) -> Self {
self.service.declare_model(model);
self
}
/// Point `memory_status` at the store directory so it can relay the
/// embedding-provenance record (#1751). Without it the provenance block
/// reports `recorded: false`.
#[must_use]
pub fn with_store_dir(mut self, dir: impl Into<std::path::PathBuf>) -> Self {
self.store_dir = Some(dir.into());
self
}
/// Attach the daemon-owned online migration control plane.
///
/// # Errors
/// Returns an error when its durable control directory cannot be created safely.
pub fn with_online_migration<F>(
mut self,
source: impl Into<std::path::PathBuf>,
factory: F,
) -> Result<Self, crate::MemoryError>
where
F: Fn(&str) -> Result<(DynEmbedder, String), crate::MemoryError> + Send + Sync + 'static,
{
let targets = Arc::new(move |backend: &str| {
factory(backend).map(|(embedder, model)| crate::service::JobTarget { embedder, model })
});
self.online_migration = Some(crate::service::OnlineMigrationManager::new(
Arc::clone(&self.service),
source,
targets,
)?);
Ok(self)
}
/// The full tool router: the memory tools, plus the context compiler's
/// tools when that feature is on. Combined here — rmcp routers add — so
/// there is exactly ONE server whichever features are enabled.
///
/// **Et le point de passage unique du durcissement d'entree.** Mesure du
/// 2026-07-29 : 10 outils sur 20 ne declaraient AUCUN `input_schema`
/// (`recall`, `recall_fused`, `entity`, `why`, `remember_extracted`,
/// `context_savings`, `retrieve_context_source`, `load_working_context`,
/// `list_working_contexts`, `suggest_budget`) — leur schema etait celui
/// derive par rmcp, que rien ne post-traitait. Un durcissement declare
/// outil par outil laisse donc chaque route future non protegee, et rien
/// ne le signale : c'est une omission, pas une erreur. Ici, une route
/// nouvelle est couverte parce qu'elle EXISTE.
///
/// L'attribut `#[tool(input_schema = …)]` garde ce qui, lui, est
/// vraiment per-outil : les cles d'id que l'outil accepte en chaine.
fn combined_router() -> ToolRouter<McpServer> {
#[cfg(feature = "context")]
let mut router = Self::tool_router()
+ Self::advanced_tool_router()
+ Self::status_tool_router()
+ Self::migration_tool_router()
+ Self::context_tool_router();
#[cfg(not(feature = "context"))]
let mut router = Self::tool_router()
+ Self::advanced_tool_router()
+ Self::status_tool_router()
+ Self::migration_tool_router();
// `reharden_tool_input` prend l'outil, pas un schema : `Tool` type
// ses deux schemas identiquement, donc c'est la signature — et non
// une convention de nommage — qui rend la sortie inatteignable ici.
for route in router.map.values_mut() {
crate::schema::reharden_tool_input(&mut route.attr);
}
assert_every_input_slot_is_typed(&router);
router
}
/// Attach an extraction backend, enabling the `remember_extracted` tool.
/// Without it the tool reports that extraction is not configured.
#[must_use]
pub fn with_extractor(self, extractor: DynExtractor) -> Self {
*self.extractors.write() = ExtractorResolver::unnamed(extractor);
self
}
/// Attach the named daemon-level default used when `remember_extracted`
/// omits its per-call `extractor` choice.
///
/// # Errors
/// Returns an error when `backend` is unknown or disables extraction.
pub fn with_named_extractor(
self,
backend: impl Into<String>,
extractor: DynExtractor,
) -> Result<Self, String> {
*self.extractors.write() = ExtractorResolver::named(backend.into(), extractor)?;
Ok(self)
}
/// Enable the durable extraction-job worker under the native store root.
///
/// Call this after opening the store and before serving requests. Accepted
/// and in-flight records are recovered immediately; corrupt durable state
/// fails startup instead of silently losing a receipt.
///
/// # Errors
/// Returns a descriptive error if the job directory cannot be created,
/// validated, read, or if the recovery worker cannot start.
pub fn with_extraction_jobs(
mut self,
store_root: impl AsRef<std::path::Path>,
) -> Result<Self, String> {
self.extraction_jobs = Some(
ExtractionJobs::open(
store_root.as_ref(),
Arc::clone(&self.service),
Arc::clone(&self.extractors),
)
.map_err(|error| error.to_string())?,
);
Ok(self)
}
/// Apply a default TTL (seconds) to `remember`d facts that don't carry their
/// own `ttl_seconds`. `0` is treated as "no default" (permanent).
#[must_use]
pub fn with_default_ttl(mut self, ttl_seconds: u64) -> Self {
self.default_ttl = (ttl_seconds > 0).then_some(ttl_seconds);
self
}
/// Enable path ingestion (V2b-1): `compile_context` and
/// `explain_compilation` fragments carrying `path` are resolved against
/// this allowlist before compilation. Without this (the default), every
/// `path` fragment fails with an explicit "ingestion disabled" error —
/// same pattern as [`Self::with_extractor`].
#[cfg(all(feature = "context", not(target_arch = "wasm32")))]
#[must_use]
pub fn with_ingest_roots(mut self, roots: crate::context::IngestRoots) -> Self {
self.ingest_roots = Some(roots);
self
}
#[tool(
name = "remember",
// Sans declaration explicite, rmcp derive un schema de sortie qui
// conserve des $ref qu'un client aveugle aux $defs ne resout pas —
// or les SDK MCP valident structuredContent contre ce schema.
output_schema = crate::schema::wire_safe_output_schema::<RememberResult>(),
description = "Store a fact in durable local memory. Optionally link it to existing memories (graph) and tag it with structured metadata like project/author/type/status/date (ColumnStore) for later filtering — metadata is capped at 64 KiB serialized. A fact is capped at 2048 bytes: that is roughly what the embedding model's context window holds, and a longer one is REFUSED with its size, not silently mangled — split a long passage into several atomic facts, or compile it with `compile_context` and remember a summary. Set `ttl_seconds` to make the fact expire after a delay (a durable TTL that survives restarts); omit it for a permanent memory — `ttl_seconds: 0` is refused, not read as \"never\". Returns the fact's stable id. With the async autograph worker active, edges derived from a remember land asynchronously: an `entity`/`why` read immediately after may not see them yet — the fact itself is always immediately readable. Ids exceed 2^53 — always relay them as strings (`id_str`); passing a JSON-number id read from a previous response will fail on float-lossy clients.",
input_schema = id_wire_input_schema::<RememberParams>(&["target"])
)]
async fn remember(
&self,
Parameters(params): Parameters<RememberParams>,
) -> Result<Json<RememberResult>, ErrorData> {
// No size pre-check here: `MemoryService::remember_with_ttl` refuses an
// over-long fact itself (`MAX_EMBEDDABLE_TEXT_BYTES`, far below
// `MAX_FACT_BYTES`), with a message naming the cap AND the received
// size — so every adapter reports the same thing, from one place.
let service = Arc::clone(&self.service);
let RememberParams {
fact,
links,
metadata,
ttl_seconds,
} = params;
let ttl = ttl_seconds.or(self.default_ttl);
let id = tokio::task::spawn_blocking(move || {
service.run(|current| current.remember_with_ttl(&fact, &links, metadata.as_ref(), ttl))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(RememberResult {
id,
id_str: id.to_string(),
}))
}
#[tool(
name = "recall",
// rmcp derives an output schema when none is given, and that
// derived form keeps `$ref`s a `$defs`-blind client cannot resolve.
output_schema = crate::schema::wire_safe_output_schema::<RecallResult>(),
description = "Recall memories semantically similar to a query (vector). Ranking blends similarity with each fact's learned confidence (see `feedback`), so the order is not pure similarity — the returned `score` is always the raw similarity, never the blended value. Optionally narrow to exact-match metadata via `filter` (ColumnStore), e.g. {\"project\":\"veles\",\"status\":\"resolved\"}. Ids exceed 2^53 — always relay them as strings (`id_str`); passing a JSON-number id read from a previous response will fail on float-lossy clients."
)]
async fn recall(
&self,
Parameters(params): Parameters<RecallParams>,
) -> Result<Json<RecallResult>, ErrorData> {
let limit = params
.limit
.unwrap_or(DEFAULT_RECALL_LIMIT)
.min(MAX_RECALL_LIMIT);
let service = Arc::clone(&self.service);
let RecallParams { query, filter, .. } = params;
let memories = tokio::task::spawn_blocking(move || {
service.run(|current| current.recall(&query, limit, filter.as_ref()))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(RecallResult::new(memories)))
}
#[tool(
name = "recall_where",
// rmcp derives an output schema when none is given, and that
// derived form keeps `$ref`s a `$defs`-blind client cannot resolve.
output_schema = crate::schema::wire_safe_output_schema::<RecallResult>(),
description = "Fused recall: semantically similar memories (vector) constrained by structured ColumnStore predicates over metadata — ranges and comparisons, not just equality. Each filter is {field, op (eq/ne/lt/le/gt/ge), value}, ANDed. Use for time-windowed or numeric-scoped recall, e.g. facts about a topic with `ts` in a date range. Comparisons are TYPE-STRICT, with no runtime coercion: a filter value of 20230601 (a JSON number) never matches a fact stored with metadata {\"ts\": \"20230601\"} (a JSON string) — same value, different JSON type, no match, no error. Store comparable values like dates NUMERICALLY at `remember` time (e.g. 20230601, not \"20230601\") so `recall_where` filters actually match them. Most similar first. Returns your own memories ONLY: entity hubs and the context compiler's artefacts (stored sources, compilation events, working contexts and their index) are internal scaffolding and never come back, whatever the predicate — including a `ne` one, which matches facts that lack the field entirely.",
// No id-named parameter here (hence the empty `keys`) — this goes
// through the shared helper purely for its `$ref` inlining, so
// `filters[]` advertises `ColumnFilter`'s own fields instead of a
// `$ref` a `$defs`-blind harness reads as "array of anything".
input_schema = id_wire_input_schema::<RecallWhereParams>(&[])
)]
async fn recall_where(
&self,
Parameters(params): Parameters<RecallWhereParams>,
) -> Result<Json<RecallResult>, ErrorData> {
let limit = params
.limit
.unwrap_or(DEFAULT_RECALL_LIMIT)
.min(MAX_RECALL_LIMIT);
let service = Arc::clone(&self.service);
let RecallWhereParams { query, filters, .. } = params;
let memories = tokio::task::spawn_blocking(move || {
service.run(|current| current.recall_where(&query, limit, &filters))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(RecallResult::new(memories)))
}
#[tool(
name = "feedback",
// Sans declaration explicite, rmcp derive un schema de sortie qui
// conserve des $ref qu'un client aveugle aux $defs ne resout pas —
// or les SDK MCP valident structuredContent contre ce schema.
output_schema = crate::schema::wire_safe_output_schema::<FeedbackResult>(),
description = "Reinforce a recalled memory with an outcome: `success=true` if the fact was useful, `false` if it was noise. This durably updates the fact's learned confidence, which `recall` uses to re-rank future results — over repeated feedback, useful facts drift up and noise drifts down, so the memory improves with use without retraining the model. Returns the fact's new confidence in [0,1].",
input_schema = id_wire_input_schema::<FeedbackParams>(&["id"])
)]
async fn feedback(
&self,
Parameters(params): Parameters<FeedbackParams>,
) -> Result<Json<FeedbackResult>, ErrorData> {
let service = Arc::clone(&self.service);
let FeedbackParams { id, success } = params;
let confidence = tokio::task::spawn_blocking(move || {
service.run(|current| current.feedback(id, success))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(FeedbackResult {
id,
id_str: id.to_string(),
confidence,
}))
}
#[tool(
name = "relate",
// Sans declaration explicite, rmcp derive un schema de sortie qui
// conserve des $ref qu'un client aveugle aux $defs ne resout pas —
// or les SDK MCP valident structuredContent contre ce schema.
output_schema = crate::schema::wire_safe_output_schema::<RelateResult>(),
description = "Create a typed, directional link between two memories (`from` → `to`) labeled by `relation`. These links are the graph edges that `why` and `recall_fused` later traverse to surface connected facts that share no words with the query — build the graph with `relate` so multi-hop reasoning works (e.g. link a decision to its cause, a fact to its source, a task to the person it concerns). Direction matters: traversal follows OUTGOING edges only, so point `from` at the memory you will later ask `why` about and `to` at its evidence (decision → cause, fact → source) — an edge pointing INTO a memory is invisible to `why(that memory)`. Idempotent per (from, relation, to); `from` and `to` must be DIFFERENT memories — a self-loop states nothing and only adds noise to `why`'s evidence trail, so it is refused. Returns the edge id as `edge_id`, plus `edge_id_str` for clients without u64-safe JSON number parsing — the one already there when this exact relation exists, since the call is idempotent. Ids exceed 2^53 — always relay them as strings (`edge_id_str`); passing a JSON-number id read from a previous response will fail on float-lossy clients.",
input_schema = id_wire_input_schema::<RelateParams>(&["from", "to"])
)]
async fn relate(
&self,
Parameters(params): Parameters<RelateParams>,
) -> Result<Json<RelateResult>, ErrorData> {
let service = Arc::clone(&self.service);
let RelateParams { from, to, relation } = params;
let edge_id = tokio::task::spawn_blocking(move || {
service.run(|current| current.relate(from, to, &relation))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(RelateResult {
edge_id,
edge_id_str: edge_id.to_string(),
}))
}
#[tool(
name = "unrelate",
// Sans declaration explicite, rmcp derive un schema de sortie qui
// conserve des $ref qu'un client aveugle aux $defs ne resout pas —
// or les SDK MCP valident structuredContent contre ce schema.
output_schema = crate::schema::wire_safe_output_schema::<UnrelateResult>(),
description = "Remove the typed link `from` -relation-> `to` — `relate`'s exact undo, so a mistaken edge no longer costs the facts at its endpoints. Only the edge is removed: the two memories, and any entity, are untouched. Idempotent: removing an absent edge answers `found: false` instead of erroring, so a cleanup can be replayed safely; `removed` counts the edges actually deleted. It refuses exactly what `relate` refuses (empty relation, `from` == `to`). Scope: the store does not distinguish a link you created with `relate` from one auto-derived from a passage, so `unrelate` removes both alike — to correct an auto-derived link, prefer `forget` + `remember` of the source fact, otherwise remembering the same passage again can rebuild the edge removed here. Same id wire contract as `relate`: pass ids as decimal strings (`id_str`) — a JSON-number id above 2^53 loses precision on float-lossy clients.",
input_schema = id_wire_input_schema::<UnrelateParams>(&["from", "to"])
)]
async fn unrelate(
&self,
Parameters(params): Parameters<UnrelateParams>,
) -> Result<Json<UnrelateResult>, ErrorData> {
let service = Arc::clone(&self.service);
let UnrelateParams { from, to, relation } = params;
let outcome = tokio::task::spawn_blocking(move || {
service.run(|current| current.unrelate(from, to, &relation))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(UnrelateResult {
found: outcome.found,
removed: outcome.removed,
}))
}
#[tool(
name = "forget",
// Sans declaration explicite, rmcp derive un schema de sortie qui
// conserve des $ref qu'un client aveugle aux $defs ne resout pas —
// or les SDK MCP valident structuredContent contre ce schema.
output_schema = crate::schema::wire_safe_output_schema::<ForgetResult>(),
description = "Permanently delete a memory by its `id` (as returned by `remember` or `recall`), removing the fact and its graph links. The deletion is durable and cannot be undone — use it to retract or correct stored knowledge. For automatic time-based expiry instead, set a TTL when calling `remember`. Returns the requested id plus `found`: `true` if a memory actually existed and was deleted, `false` if nothing was stored under that id (a stale id or a typo) — a no-op, not an error, but distinguishable from a real deletion.",
input_schema = id_wire_input_schema::<ForgetParams>(&["id"])
)]
async fn forget(
&self,
Parameters(params): Parameters<ForgetParams>,
) -> Result<Json<ForgetResult>, ErrorData> {
let service = Arc::clone(&self.service);
let id = params.id;
let found = tokio::task::spawn_blocking(move || service.run(|current| current.forget(id)))
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(ForgetResult {
id,
id_str: id.to_string(),
found,
}))
}
#[tool(
name = "entity",
// Sans declaration explicite, rmcp derive un schema de sortie qui
// conserve des $ref qu'un client aveugle aux $defs ne resout pas —
// or les SDK MCP valident structuredContent contre ce schema.
output_schema = crate::schema::wire_safe_output_schema::<EntityProfileDto>(),
description = "Look up everything the memory graph knows about a NAMED ENTITY (a person, a place, an organisation): the attributes it carries, the typed edges leaving it (`relations`) and the typed edges pointing AT it (`relations_in`). Both directions come back, because a question is only answerable from one side: with `camille --sister of--> theo` recorded, asking what Theo's OUTGOING edges say never finds Camille — she is in his `relations_in`. Use this for questions ABOUT a thing rather than about a sentence — \"how old is Theo\", \"who is Theo's father\", \"where does he live\" — where `recall` would only return sentences that happen to mention the name. Entities and their edges are built automatically by `remember_extracted`, which reads relationships (`X is the father of Y`) and properties (`Y is 15`) out of plain text; attributes land in ColumnStore metadata with their JSON type preserved, so a number stays a number. The name is matched case-insensitively, so `\"Theo Durand\"` and `\"theo durand\"` are the same entity — the id is content-addressed, so it is stable across sessions. Returns `found: false` when nothing has ever mentioned that name; `name` is echoed back in its canonical (trimmed, lowercased) form either way, so several lookups can be told apart. With the async autograph worker active, edges derived from a `remember` land asynchronously: an entity read immediately after that remember may not see them yet — the fact itself is always immediately readable. Ids exceed 2^53 — always relay them as strings (`id_str`)."
)]
async fn entity(
&self,
Parameters(params): Parameters<EntityParams>,
) -> Result<Json<EntityProfileDto>, ErrorData> {
let service = Arc::clone(&self.service);
let EntityParams { name } = params;
let looked_up = name.clone();
let profile = tokio::task::spawn_blocking(move || {
service.run(|current| current.entity_profile(&looked_up))
})
.await
.map_err(join_error)?
.map_err(to_error)?;
Ok(Json(EntityProfileDto::from_lookup(&name, profile)))
}
}
/// `#[tool_handler]` generates `list_tools` from the router — `call_tool` is
/// written by hand below (see its doc comment) and the macro skips what
/// already exists. `get_info` is overridden so the server identifies itself
/// as `velesdb-memory` (the macro default falls back to rmcp's own
/// identity). Per-tool guidance lives in each `#[tool(description = …)]`.
/// The server's one-shot vitrine to a connecting agent (V2a-1 quick win):
/// must cover every tool family, not just memory — a `#[cfg(feature =
/// "context")]` variant since the context-compiler tools only exist in that
/// build.
#[cfg(feature = "context")]
const SERVER_INSTRUCTIONS: &str = "Local-first memory and context engineering for AI agents, four tool families: (1) durable memory — remember, remember_extracted, extraction_status, recall, recall_fused, recall_where, relate, unrelate, forget, feedback, entity, and why — explainable (why returns the evidence trail) and self-improving (feedback re-ranks future recall); remember_extracted durably accepts a passage, then reads the entities, typed edges and attributes it STATES and wires them into the graph in the background: keep its request_id and poll extraction_status until committed or failed, reusing one idempotency_key across transport retries; entity(name) answers a question ABOUT a named thing rather than about the sentences mentioning it; memory_status reports the server's health — which embedder runs and whether recall is semantic, extraction wiring, and graph size; list_memories audits the store page by page — what recall cannot answer, because what resembles no query stays invisible; (2) online embedding migration — migration_start returns after durable acceptance, migration_status reports progress and recovery, migration_cancel is safe only while the source is authoritative, and migration_recover resumes a stopped pre-cutover job; (3) the deterministic context compiler — compile_context, compile_transcript, explain_compilation, retrieve_context_source, context_savings, and suggest_budget — token-budgets and audits prompt context with no LLM call, ever; (4) cross-session working-context resumption — save_working_context, load_working_context, and list_working_contexts. compile_context/explain_compilation fragments accept a `path` instead of inline `content` to ingest a file by reference — disabled unless the server is started with VELESDB_MEMORY_INGEST_ROOTS set to an allowlist of directories (compile_transcript's own `path` field uses the same allowlist). compile_transcript is a one-call shortcut over compile_context for a raw agent-session transcript: it segments plain or JSONL text into turns before compiling, so an agent no longer needs to segment a transcript by hand. Nothing ever leaves the machine.";
#[cfg(not(feature = "context"))]
const SERVER_INSTRUCTIONS: &str = "Local-first memory for AI agents: remember facts, recall them \
semantically, relate them, forget them, ask why a decision was made (connected subgraph), \
submit durable remember_extracted jobs and poll extraction_status to completion, \
read memory_status for the server's health — embedder semantics, extraction wiring, \
graph size — audit the store page by page with list_memories, and control daemon-owned \
online embedding migration with migration_start/status/cancel/recover.";
#[tool_handler(router = self.tool_router)]
impl ServerHandler for McpServer {
fn get_info(&self) -> ServerInfo {
let mut info = ServerInfo::default();
info.server_info = Implementation::new(env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"));
info.capabilities = ServerCapabilities::builder().enable_tools().build();
let mut instructions = SERVER_INSTRUCTIONS.to_owned();
// The one channel a client is REQUIRED to read. The stderr warning
// for the same fact is swallowed by every mainstream MCP harness, so
// a degraded server that only warned there was indistinguishable
// from a healthy one — the audit finding memory_status closes, and
// this note is its push half (the tool is the pull half).
if self
.service
.with_generation(|generation| generation.model() == "hash")
.unwrap_or(false)
{
instructions.push_str(
" NOTE: this server is running the offline 'hash' embedder — recall matches \
surface form, NOT meaning. If recall quality matters to the user, say so: a \
semantic embedder is an env-var switch away (call memory_status for details).",
);
}
info.instructions = Some(instructions);
info
}
/// One trace event per tool call (#1780): tool name, session id, verdict,
/// duration — never an argument or fact content
/// (`tests/daemon_logging.rs` holds a canary against that). Written by
/// hand so the event wraps the dispatch — `#[tool_handler]` sees the
/// method already exists and only generates `list_tools`; the dispatch
/// below is exactly what the macro would have generated.
async fn call_tool(
&self,
request: rmcp::model::CallToolRequestParams,
context: rmcp::service::RequestContext<rmcp::RoleServer>,
) -> Result<rmcp::model::CallToolResponse, ErrorData> {
let tool = request.name.clone();
let session = http_session_id(&context.extensions);
let started = std::time::Instant::now();
let tcc = rmcp::handler::server::tool::ToolCallContext::new(self, request, context);
let outcome = self.tool_router.call(tcc).await;
log_tool_call(&tool, session.as_deref(), &outcome, started);
outcome
}
}
/// The tool-level trace event (#1780), split from `call_tool` so the verdict
/// taxonomy is readable in one place. `Err` is a protocol-level failure, but
/// a *refused* tool call comes back as `Ok` with `is_error` set INSIDE a
/// valid result — reading only the outer `Result` would log every refusal as
/// a success, the exact misreading this event exists to prevent.
fn log_tool_call(
tool: &str,
session: Option<&str>,
outcome: &Result<rmcp::model::CallToolResponse, ErrorData>,
started: std::time::Instant,
) {
use rmcp::model::CallToolResponse;
let verdict = match outcome {
Err(_) => "error",
Ok(CallToolResponse::Complete(result)) if result.is_error == Some(true) => "tool_error",
Ok(CallToolResponse::Complete(_)) => "ok",
// rmcp 3's InputRequired/Task responses (SEP-2663) carry no verdict:
// the call has not completed. No tool of this server produces them —
// if one ever appears here, "pending" keeps the event truthful
// instead of misreporting an unfinished call as a success.
Ok(_) => "pending",
};
// `%` (Display) rather than the default Debug capture: Debug renders
// strings quoted (`tool="recall"`), and these lines exist to be grepped
// (`grep tool=recall`) by an operator mid-incident. The target is pinned
// explicitly for the same operators: a module refactor must not silently
// rename the lines their tooling matches on.
tracing::info!(
target: "velesdb_memory::mcp",
tool = %tool,
session = %session.unwrap_or(crate::logging::NO_SESSION),
verdict = %verdict,
elapsed_ms = crate::logging::elapsed_millis(started),
"mcp tool call"
);
}
/// The `mcp-session-id` this call arrived under, read from the HTTP request
/// parts rmcp injects into the request's extensions (its streamable-HTTP
/// tower service does this for every incoming `POST`). HTTP transport only —
/// there is no session id on stdio, and the event then reports `-`.
#[cfg(feature = "http")]
fn http_session_id(extensions: &rmcp::model::Extensions) -> Option<String> {
extensions
.get::<axum::http::request::Parts>()
.and_then(|parts| crate::http::session_from_headers(&parts.headers))
}
/// Without the HTTP transport nothing ever injects request parts, so there
/// is no session id to read — same cfg-pair shape as the binary's
/// transport-dependent helpers.
#[cfg(not(feature = "http"))]
fn http_session_id(_extensions: &rmcp::model::Extensions) -> Option<String> {
None
}
/// La post-condition du point de passage, verifiee SUR PLACE.
///
/// Un test lointain constate ; ici on refuse. Les schemas annonces sont
/// statiques — ils ne dependent ni du store, ni de l'horloge, ni d'une
/// entree — donc une violation est deterministe : elle ne peut pas se
/// produire chez un utilisateur sans se produire aussi au premier
/// `McpServer::new` de la suite de tests. Echouer a la construction dit ou
/// est le probleme ; laisser passer un slot intypable le fait ressortir
/// plusieurs jours plus tard, dans un aller-retour de deserialisation chez
/// un agent.
///
/// # Panics
/// Si une route publie un slot d'entree qui n'annonce ni `type`, ni `enum`,
/// ni `const`.
fn assert_every_input_slot_is_typed(router: &ToolRouter<McpServer>) {
let mut offenders: Vec<String> = Vec::new();
for route in router.map.values() {
for slot in crate::schema::untyped_input_slots(&route.attr.input_schema) {
offenders.push(format!(" {}: {slot}", route.attr.name));
}
}
assert!(
offenders.is_empty(),
"{} slot(s) d'entree n'annoncent aucun type — un harnais client les rend `{{}}`, le \
client envoie ce qu'il devine, et le serveur le refuse :\n{}",
offenders.len(),
offenders.join("\n")
);
}
#[allow(clippy::needless_pass_by_value)]
fn job_error(error: JobError) -> ErrorData {
let code = match error {
JobError::Invalid(_) | JobError::Conflict | JobError::NotFound(_) => {
ErrorCode::INVALID_PARAMS
}
JobError::AtCapacity | JobError::BackendNotConfigured | JobError::Storage(_) => {
ErrorCode::INTERNAL_ERROR
}
};
ErrorData::new(code, error.to_string(), None)
}
/// Map a `spawn_blocking` join failure (a panicked or cancelled tool task) to an
/// MCP error. Every tool body runs on the blocking pool, so they all funnel
/// through this on the (rare) task-failure path.
///
/// Takes the error by value so it can be used as `.map_err(join_error)`.
#[allow(clippy::needless_pass_by_value)]
fn join_error(join: tokio::task::JoinError) -> ErrorData {
ErrorData::new(
ErrorCode::INTERNAL_ERROR,
format!("memory task failed: {join}"),
None,
)
}
/// Map a domain error to an MCP error.
///
/// Map a [`MemoryError`](crate::error::MemoryError) onto a JSON-RPC error,
/// driven by its transport-neutral [`ErrorCategory`](crate::error::ErrorCategory)
/// so the MCP taxonomy can never drift from the bindings'. Client-input errors
/// become `invalid_params` (-32602); genuine faults `internal_error` (-32603).
/// JSON-RPC defines no "not found" code, so a missing id is reported as
/// `invalid_params` (a bad id is, from the protocol's view, a bad parameter).
///
/// Takes the error by value so it can be used as `.map_err(to_error)` at every
/// call site without a per-site closure.
#[allow(clippy::needless_pass_by_value)]
fn to_error(err: crate::error::MemoryError) -> ErrorData {
use crate::error::ErrorCategory;
let code = match err.category() {
ErrorCategory::InvalidInput | ErrorCategory::NotFound => ErrorCode::INVALID_PARAMS,
// A capability gap is the server's to own, not the caller's: the
// request was well-formed, this backend just cannot serve it.
// JSON-RPC has no "unsupported" code, so it rides internal_error
// with the message naming the missing capability.
ErrorCategory::Internal | ErrorCategory::Unsupported => ErrorCode::INTERNAL_ERROR,
};
ErrorData::new(code, err.to_string(), None)
}
#[cfg(all(test, feature = "persistence"))]
#[path = "mcp/generation_tests.rs"]
mod generation_tests;
#[cfg(test)]
#[path = "mcp/server_tests.rs"]
mod tests;