use std::sync::{
Arc,
atomic::{AtomicBool, Ordering},
};
use crate::{config::SearchConfig, embed::LazyEmbedder, sessions::Store};
#[derive(Clone)]
pub struct AppState {
pub store: Arc<Store>,
pub embedder: Arc<LazyEmbedder>,
pub search: SearchConfig,
activity: Arc<AtomicBool>,
}
impl AppState {
pub fn new(store: Arc<Store>, embedder: Arc<LazyEmbedder>, search: SearchConfig) -> Self {
Self {
store,
embedder,
search,
activity: Arc::new(AtomicBool::new(false)),
}
}
fn track_activity(&self) -> ActivityGuard {
ActivityGuard(Arc::clone(&self.activity))
}
pub fn take_completed_activity(&self) -> bool {
self.activity.swap(false, Ordering::AcqRel)
}
}
struct ActivityGuard(Arc<AtomicBool>);
impl Drop for ActivityGuard {
fn drop(&mut self) {
self.0.store(true, Ordering::Release);
}
}
pub mod http {
#[cfg(unix)]
use std::path::Path;
use std::{
future::Future,
net::{IpAddr, SocketAddr},
sync::Arc,
time::Duration,
};
use anyhow::Context;
use axum::{
Json, Router,
extract::{DefaultBodyLimit, Request, State},
http::{HeaderValue, StatusCode, header, uri::Authority},
middleware::{self, Next},
response::{IntoResponse, Response},
routing::post,
serve::Listener,
};
use rmcp::transport::streamable_http_server::{
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use super::AppState;
use crate::{
handlers::{pond_get_message, pond_get_session, pond_ingest, pond_search, pond_sql},
wire::{
ErrorCode, GetEnvelope, GetMessageRequest, GetSessionRequest, IngestEnvelope,
IngestRequest, SearchEnvelope, SearchRequest, SqlEnvelope, SqlRequest,
default_namespace, new_request_id,
},
};
pub const HTTP_BODY_LIMIT_BYTES: usize = 8 * 1024 * 1024;
fn mcp_allowed_hosts(extra: &[String]) -> Vec<String> {
let mut hosts = StreamableHttpServerConfig::default().allowed_hosts;
hosts.extend(extra.iter().cloned());
hosts
}
fn host_is_allowed(host: &Authority, allowed: &[String]) -> bool {
let normalize = |name: &str| name.trim_matches(['[', ']']).to_ascii_lowercase();
let wanted = normalize(host.host());
allowed.iter().any(|entry| {
let (name, port) = match Authority::try_from(entry.trim()) {
Ok(authority) => (normalize(authority.host()), authority.port_u16()),
Err(_) => (normalize(entry.trim()), None),
};
name == wanted && port.is_none_or(|port| host.port_u16() == Some(port))
})
}
async fn require_allowed_host(
State(allowed): State<Arc<[String]>>,
request: Request,
next: Next,
) -> Response {
let host = request
.headers()
.get(header::HOST)
.and_then(|value| value.to_str().ok())
.and_then(|value| Authority::try_from(value).ok())
.or_else(|| request.uri().authority().cloned());
match host {
Some(host) if host_is_allowed(&host, &allowed) => next.run(request).await,
_ => (
StatusCode::FORBIDDEN,
"Forbidden: Host header is not allowed; list this server's name with \
`pond serve --allowed-host <name>`",
)
.into_response(),
}
}
pub const SHUTDOWN_DRAIN: Duration = Duration::from_secs(5);
pub fn router(
state: AppState,
allowed_hosts: &[String],
mcp_shutdown: CancellationToken,
) -> Router {
let hosts = mcp_allowed_hosts(allowed_hosts);
let mcp_state = state.clone();
let mcp = StreamableHttpService::new(
move || Ok(super::mcp::PondMcp::new(mcp_state.clone())),
LocalSessionManager::default().into(),
StreamableHttpServerConfig::default()
.with_allowed_hosts(hosts.clone())
.with_cancellation_token(mcp_shutdown),
);
Router::new()
.route("/v1/search", post(search))
.route("/v1/get-session", post(get_session))
.route("/v1/get-message", post(get_message))
.route("/v1/ingest", post(ingest))
.route(
"/v1/x/sql",
post(sql).route_layer(middleware::from_fn_with_state(
Arc::<[String]>::from(hosts),
require_allowed_host,
)),
)
.layer(DefaultBodyLimit::max(HTTP_BODY_LIMIT_BYTES))
.with_state(state)
.nest_service("/mcp", mcp)
}
pub async fn bind(host: &str, port: u16) -> anyhow::Result<TcpListener> {
let ip: IpAddr = host
.parse()
.with_context(|| format!("invalid --host {host:?}"))?;
if ip.is_unspecified() {
tracing::warn!(
%host,
"binding to an unspecified address exposes pond on the LAN; \
the personal pond is single-user"
);
}
TcpListener::bind(SocketAddr::new(ip, port))
.await
.with_context(|| format!("failed to bind {host}:{port}"))
}
pub async fn serve(
listener: TcpListener,
state: AppState,
allowed_hosts: Vec<String>,
) -> anyhow::Result<()> {
let local = listener
.local_addr()
.context("failed to read bound address")?;
crate::output::line(&format!("serve: http listening on http://{local}"))?;
tracing::info!(%local, "pond serve listening (HTTP /v1/*, MCP /mcp)");
serve_with_shutdown(listener, state, &allowed_hosts, shutdown_signal()).await
}
#[cfg(unix)]
#[derive(Debug)]
pub struct SocketClaim {
path: std::path::PathBuf,
address: socket2::SockAddr,
_lock: std::fs::File,
bound: Option<(u64, u64)>,
}
#[cfg(unix)]
impl SocketClaim {
pub fn acquire(path: &Path) -> anyhow::Result<Self> {
use std::os::unix::fs::{FileTypeExt, OpenOptionsExt};
let address = socket2::SockAddr::unix(path).map_err(|_| {
anyhow::anyhow!(
"--socket {}: too long for a Unix socket path; pick a shorter path",
path.display()
)
})?;
let dir = path
.parent()
.filter(|dir| !dir.as_os_str().is_empty())
.unwrap_or(Path::new("."));
if !dir.is_dir() {
anyhow::bail!(
"--socket {}: directory {} does not exist; create it or pick another path",
path.display(),
dir.display()
);
}
let mut lock_path = path.as_os_str().to_owned();
lock_path.push(".lock");
let lock_path = std::path::PathBuf::from(lock_path);
let not_regular = || {
anyhow::anyhow!(
"--socket lock {} is not a regular file; remove it or pick another path",
lock_path.display()
)
};
let opened = std::fs::OpenOptions::new()
.write(true)
.create(true)
.truncate(false)
.mode(0o600)
.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK)
.open(&lock_path);
let lock = match opened {
Ok(file) => file,
Err(_)
if std::fs::symlink_metadata(&lock_path)
.is_ok_and(|metadata| !metadata.is_file()) =>
{
return Err(not_regular());
}
Err(error) => {
return Err(error).with_context(|| {
format!("failed to open the --socket lock {}", lock_path.display())
});
}
};
let lock_metadata = lock.metadata().with_context(|| {
format!(
"failed to inspect the --socket lock {}",
lock_path.display()
)
})?;
if !lock_metadata.is_file() {
return Err(not_regular());
}
match lock.try_lock() {
Ok(()) => {}
Err(std::fs::TryLockError::WouldBlock) => anyhow::bail!(
"--socket {}: another pond serve owns this path; stop it or pick another path",
path.display()
),
Err(std::fs::TryLockError::Error(error)) => {
return Err(error)
.with_context(|| format!("failed to lock {}", lock_path.display()));
}
}
let metadata =
match std::fs::symlink_metadata(path) {
Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
result => Some(result.with_context(|| {
format!("failed to inspect --socket {}", path.display())
})?),
};
if let Some(metadata) = metadata {
if !metadata.file_type().is_socket() {
anyhow::bail!(
"--socket {} exists and is not a socket; remove it or pick another path",
path.display()
);
}
std::fs::remove_file(path).with_context(|| {
format!("failed to remove stale --socket {}", path.display())
})?;
}
Ok(Self {
path: path.to_owned(),
address,
_lock: lock,
bound: None,
})
}
pub fn bind(&mut self) -> anyhow::Result<tokio::net::UnixListener> {
use socket2::{Domain, Socket, Type};
let socket = Socket::new(Domain::UNIX, Type::STREAM, None)
.with_context(|| self.failed("create"))?;
socket
.bind(&self.address)
.with_context(|| self.failed("bind"))?;
let listening = self.listen(socket);
if listening.is_err() {
self.remove_socket();
}
listening
}
fn listen(&mut self, socket: socket2::Socket) -> anyhow::Result<tokio::net::UnixListener> {
use std::{
fs::Permissions,
os::{fd::OwnedFd, unix::fs::PermissionsExt},
};
let path = &self.path;
let context = |step: &str| format!("failed to {step} --socket {}", path.display());
self.bound = Some(socket_identity(path).with_context(|| context("inspect"))?);
std::fs::set_permissions(path, Permissions::from_mode(0o600))
.with_context(|| context("restrict"))?;
socket.listen(1024).with_context(|| context("listen on"))?;
socket
.set_nonblocking(true)
.with_context(|| context("configure"))?;
let listener = std::os::unix::net::UnixListener::from(OwnedFd::from(socket));
tokio::net::UnixListener::from_std(listener).with_context(|| context("register"))
}
fn failed(&self, step: &str) -> String {
format!("failed to {step} --socket {}", self.path.display())
}
fn remove_socket(&self) {
let path = &self.path;
let Some(bound) = self.bound else {
return;
};
if socket_identity(path).ok() != Some(bound) {
return;
}
if let Err(error) = std::fs::remove_file(path)
&& error.kind() != std::io::ErrorKind::NotFound
{
tracing::warn!(%error, path = %path.display(), "failed to remove the socket file");
}
}
}
#[cfg(unix)]
fn socket_identity(path: &Path) -> std::io::Result<(u64, u64)> {
use std::os::unix::fs::MetadataExt;
let metadata = std::fs::symlink_metadata(path)?;
Ok((metadata.dev(), metadata.ino()))
}
#[cfg(unix)]
pub async fn serve_unix(
listener: tokio::net::UnixListener,
claim: SocketClaim,
state: AppState,
allowed_hosts: &[String],
stop: impl Future<Output = ()> + Send + 'static,
) -> anyhow::Result<()> {
let path = &claim.path;
let served = async {
crate::output::line(&format!("serve: http listening on unix:{}", path.display()))?;
tracing::info!(path = %path.display(), "pond serve listening (HTTP /v1/*, MCP /mcp)");
serve_with_shutdown(listener, state, allowed_hosts, stop).await
}
.await;
claim.remove_socket();
served
}
pub async fn serve_with_shutdown<L>(
listener: L,
state: AppState,
allowed_hosts: &[String],
stop: impl Future<Output = ()> + Send + 'static,
) -> anyhow::Result<()>
where
L: Listener,
L::Addr: std::fmt::Debug,
{
let mcp_shutdown = CancellationToken::new();
let server = axum::serve(listener, router(state, allowed_hosts, mcp_shutdown.clone()))
.with_graceful_shutdown({
let mcp_shutdown = mcp_shutdown.clone();
async move {
stop.await;
tracing::info!("shutdown: closing MCP sessions, then draining");
mcp_shutdown.cancel();
}
});
tokio::select! {
result = server => result.context("axum server error"),
() = async move {
mcp_shutdown.cancelled().await;
tokio::time::sleep(SHUTDOWN_DRAIN).await;
} => {
tracing::warn!(
seconds = SHUTDOWN_DRAIN.as_secs(),
"shutdown: connections still open after the drain window; exiting anyway"
);
Ok(())
}
}
}
pub async fn shutdown_signal() {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
match signal(SignalKind::terminate()) {
Ok(mut terminate) => {
tokio::select! {
_ = tokio::signal::ctrl_c() => {}
_ = terminate.recv() => {}
}
}
Err(error) => {
tracing::warn!(%error, "no SIGTERM handler; stopping on ctrl-c only");
let _ = tokio::signal::ctrl_c().await;
}
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
}
}
async fn search(
State(state): State<AppState>,
Json(mut request): Json<SearchRequest>,
) -> Response {
let _activity = state.track_activity();
request.namespace.get_or_insert_with(default_namespace);
let envelope = pond_search(&state.store, &state.embedder, request, &state.search).await;
let status = match &envelope {
SearchEnvelope::Success(_) => StatusCode::OK,
SearchEnvelope::Error(error) => status_for(&error.error.code),
};
with_request_id((status, Json(envelope)).into_response())
}
async fn get_session(
State(state): State<AppState>,
Json(mut request): Json<GetSessionRequest>,
) -> Response {
let _activity = state.track_activity();
request.namespace.get_or_insert_with(default_namespace);
let envelope = pond_get_session(&state.store, request).await;
let status = match &envelope {
GetEnvelope::Success(_) => StatusCode::OK,
GetEnvelope::Error(error) => status_for(&error.error.code),
};
with_request_id((status, Json(envelope)).into_response())
}
async fn get_message(
State(state): State<AppState>,
Json(mut request): Json<GetMessageRequest>,
) -> Response {
let _activity = state.track_activity();
request.namespace.get_or_insert_with(default_namespace);
let envelope = pond_get_message(&state.store, request).await;
let status = match &envelope {
GetEnvelope::Success(_) => StatusCode::OK,
GetEnvelope::Error(error) => status_for(&error.error.code),
};
with_request_id((status, Json(envelope)).into_response())
}
async fn ingest(
State(state): State<AppState>,
Json(mut request): Json<IngestRequest>,
) -> Response {
let _activity = state.track_activity();
request.namespace.get_or_insert_with(default_namespace);
let envelope = pond_ingest(&state.store, request).await;
let status = match &envelope {
IngestEnvelope::Success(_) => StatusCode::OK,
IngestEnvelope::Error(error) => status_for(&error.error.code),
};
with_request_id((status, Json(envelope)).into_response())
}
async fn sql(State(state): State<AppState>, Json(mut request): Json<SqlRequest>) -> Response {
let _activity = state.track_activity();
request.namespace.get_or_insert_with(default_namespace);
let envelope = pond_sql(&state.store, request).await;
let status = match &envelope {
SqlEnvelope::Success(_) => StatusCode::OK,
SqlEnvelope::Error(error) => status_for(&error.error.code),
};
with_request_id((status, Json(envelope)).into_response())
}
fn with_request_id(mut response: Response) -> Response {
if let Ok(value) = HeaderValue::from_str(&new_request_id()) {
response.headers_mut().insert("x-pond-request-id", value);
}
response
}
fn status_for(code: &ErrorCode) -> StatusCode {
match code {
ErrorCode::ValidationFailed
| ErrorCode::VersionUnsupported
| ErrorCode::NamespaceUnknown => StatusCode::BAD_REQUEST,
ErrorCode::NotFound => StatusCode::NOT_FOUND,
ErrorCode::Conflict => StatusCode::CONFLICT,
ErrorCode::StorageUnavailable => StatusCode::SERVICE_UNAVAILABLE,
ErrorCode::Internal => StatusCode::INTERNAL_SERVER_ERROR,
}
}
#[cfg(test)]
mod tests {
#![cfg_attr(
unix,
expect(
clippy::expect_used,
clippy::unwrap_used,
reason = "tests fail by panicking; the panicking ones are unix-only"
)
)]
use super::*;
#[test]
fn allowed_hosts_extend_the_loopback_defaults() {
let defaults = StreamableHttpServerConfig::default().allowed_hosts;
assert!(defaults.contains(&"localhost".to_owned()));
assert_eq!(mcp_allowed_hosts(&[]), defaults);
let hosted = mcp_allowed_hosts(&[
"pond.example.com".to_owned(),
"pond.internal:9797".to_owned(),
]);
assert_eq!(&hosted[..defaults.len()], &defaults[..]);
assert_eq!(
&hosted[defaults.len()..],
&[
"pond.example.com".to_owned(),
"pond.internal:9797".to_owned()
]
);
}
#[cfg(unix)]
#[test]
fn socket_claim_is_exclusive_and_clears_only_a_stale_socket() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("pond.sock");
let claim = SocketClaim::acquire(&path).unwrap();
let error = SocketClaim::acquire(&path).unwrap_err().to_string();
assert!(error.contains("another pond serve owns"), "{error}");
assert!(error.contains(&path.display().to_string()), "{error}");
drop(claim);
drop(std::os::unix::net::UnixListener::bind(&path).unwrap());
let claim = SocketClaim::acquire(&path).unwrap();
assert!(
!path.exists(),
"a dead run's socket is cleared under the lock"
);
drop(claim);
std::fs::write(&path, "not a socket").unwrap();
let error = SocketClaim::acquire(&path).unwrap_err().to_string();
assert!(error.contains("is not a socket"), "{error}");
assert!(path.exists(), "a non-socket file is refused, never deleted");
let orphan = temp.path().join("missing").join("pond.sock");
let error = SocketClaim::acquire(&orphan).unwrap_err().to_string();
assert!(error.contains(&orphan.display().to_string()), "{error}");
let too_long = temp.path().join("s".repeat(200));
let error = SocketClaim::acquire(&too_long).unwrap_err().to_string();
assert!(error.contains("too long for a Unix socket path"), "{error}");
}
#[cfg(unix)]
#[tokio::test]
async fn claimed_socket_binds_owner_only() {
use std::os::unix::fs::{FileTypeExt, PermissionsExt};
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("pond.sock");
let mut claim = SocketClaim::acquire(&path).unwrap();
let _listener = claim.bind().unwrap();
let metadata = std::fs::symlink_metadata(&path).unwrap();
assert!(metadata.file_type().is_socket());
assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
std::os::unix::net::UnixStream::connect(&path).expect("listening once bound");
claim.remove_socket();
assert!(!path.exists(), "the claim's own socket is removed on stop");
}
#[cfg(unix)]
#[test]
fn a_lock_path_that_is_not_a_regular_file_is_refused() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("pond.sock");
let lock = temp.path().join("pond.sock.lock");
std::os::unix::fs::symlink(temp.path().join("elsewhere"), &lock).unwrap();
let error = SocketClaim::acquire(&path).unwrap_err().to_string();
assert!(error.contains("not a regular file"), "{error}");
std::fs::remove_file(&lock).unwrap();
let made = std::process::Command::new("mkfifo").arg(&lock).status();
if !made.is_ok_and(|status| status.success()) {
return;
}
let error = SocketClaim::acquire(&path).unwrap_err().to_string();
assert!(error.contains("not a regular file"), "{error}");
}
#[cfg(unix)]
#[tokio::test]
async fn a_replaced_socket_is_not_removed_on_stop() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("pond.sock");
let mut claim = SocketClaim::acquire(&path).unwrap();
let _listener = claim.bind().unwrap();
std::fs::remove_file(&path).unwrap();
let _successor = std::os::unix::net::UnixListener::bind(&path).unwrap();
claim.remove_socket();
std::os::unix::net::UnixStream::connect(&path)
.expect("a socket bound at the path after this claim is left alone");
}
}
}
pub mod mcp {
use std::sync::Arc;
use anyhow::Context;
use base64::{Engine, engine::general_purpose::STANDARD};
use rmcp::{
ErrorData, RoleServer, ServerHandler, ServiceExt,
handler::server::{router::tool::ToolRouter, wrapper::Parameters},
model::{
CacheScope, CallToolResult, ContentBlock, ErrorCode as JsonRpcErrorCode,
Implementation, ListResourceTemplatesResult, ListResourcesResult, ListToolsResult,
MetaObject, PaginatedRequestParams, ProtocolVersion, ReadResourceRequestParams,
ReadResourceResponse, ReadResourceResult, Resource, ResourceContents,
ServerCapabilities, ServerConfig,
},
schemars,
service::RequestContext,
tool, tool_handler, tool_router,
transport::{async_rw::AsyncRwTransport, stdio},
};
use serde::Deserialize;
use uuid::Uuid;
use super::AppState;
use crate::{
PROTOCOL_VERSION,
handlers::{
pond_get_message as run_get_message, pond_get_session as run_get_session,
pond_search as run_search,
},
sql,
wire::{
ErrorCode as WireErrorCode, ErrorEnvelope, GetEnvelope, GetMessageRequest,
GetSessionRequest, ProjectFilter, SearchEnvelope, SearchFilters, SearchModeWire,
SearchRequest, SessionFrom, SortBy, default_namespace,
},
};
const SCHEMA_DOC: &str = "\
pond_search params: query (the distinctive words you expect in the \
conversation, not project names), mode (fts default - matches exact whole \
words, BM25 | vector - matches on meaning), \
sort_by (relevance default | recency), limit (returned sessions; default 10, \
max 200 - also the want-more knob, there is no pagination), project (path \
substring), session_id (exact session match - search within one session), \
source_agent (one source harness, exact-or-subpath: \"openclaw\" also covers \
\"openclaw/subagent\", not \"openclaw-x\"), from_date / to_date (YYYY-MM-DD). \
Subagents are excluded by default: a root source_agent like \"openclaw\" returns \
main sessions only, while naming a subpath (e.g. \"claude-code/general-purpose\") \
reaches those subagent sessions and is the opt-in that disables the default \
exclusion. Subagents are also reachable via pond_sql (parent_session_id).
pond_search response: a transcript. The first line states totals \
(`matched_total` is the message count before `limit` and byte-budget \
truncation), then results are grouped by session, best session first; within \
a session, matching messages are newest-first. Each hit is a `--- [n] score | \
role | time | message_id | project | agent | session ---` rule followed by its \
matched text (a ~600-char indexed window). `score` is in [0.0, 1.0] within one \
response (raw cosine for vector, normalized BM25 for fts). vector relevance \
ordering adds a gentle recency tiebreaker; sort_by=recency orders strictly \
newest-first and the response labels itself so. `has_more` warns the ranked \
set was cut by `limit` or the byte budget - raise `limit` to see the rest.
pond_search multilingual: pond's embedder (multilingual-e5-small) is trained \
for cross-lingual retrieval, so a vector query in language A can match indexed \
text in language B. fts is word-tokenized (the `simple` tokenizer with English \
stemming) and matches surface words only, so it stays within one language.
pond_get_session: id (a session id; a message id also works and resolves to \
its parent session with the page anchored at that message - the response notes \
the resolution). Output is a transcript - each message is a `--- [n] role | \
time | message_id ---` rule, then its text/content as real lines, then \
one-line tool refs (`-> name [call_id]` tool call, `<- name [call_id] \
(ok|failed)` result); full part bodies stay one pond_get_message away. limit \
defaults to 20; from=start|end picks which end the first page reads from (end \
= recent tail - the session's final state, e.g. post-compaction recovery). \
A partial page's header states the span and session total (`messages 1-130 \
of 234`); pages are bounded by limit and a size budget - page with \
after_message_id (forward) or before_message_id (backward) using the id a \
page marker shows. The first page \
also lists the session's subagents (each stored as its own session) in a \
footer; pass a listed id back to open it. Not for bulk export - use `pond \
copy --to <file>`.
pond_get_message: id (a message id - the target renders marked `>` with its \
full part bodies, reasoning included, plus context_before / context_after \
conversational siblings each side, both default 3, like grep -B/-A). A \
session id is rejected with a hint naming pond_get_session - it cannot pick \
one message.";
fn schema_doc_for(enabled: bool) -> String {
if enabled {
return SCHEMA_DOC.to_owned();
}
let rewrites = [
(
"mode (fts default - matches exact whole \
words, BM25 | vector - matches on meaning)",
"mode (fts, the only arm on this instance - matches exact whole words, BM25)",
),
(
"(raw cosine for vector, normalized BM25 for fts). vector relevance \
ordering adds a gentle recency tiebreaker;",
"(normalized BM25);",
),
(
"pond's embedder (multilingual-e5-small) is trained \
for cross-lingual retrieval, so a vector query in language A can match indexed \
text in language B. fts",
"fts",
),
];
let mut doc = SCHEMA_DOC.to_owned();
for (from, to) in rewrites {
debug_assert!(
doc.contains(from),
"schema doc drifted from rewrite: {from}"
);
doc = doc.replace(from, to);
}
doc
}
const SQL_SCHEMA_DOC: &str = "\
pond_sql runs ONE read-only SELECT (DataFusion SQL, PostgreSQL-compatible) \
over three registered tables. Read-only is hard-enforced: anything other than a \
single SELECT/WITH (or EXPLAIN of one) is rejected (no INSERT/UPDATE/DELETE/\
CREATE/DROP/COPY/SET).
Routing - pick the right surface before writing SQL:
- counts, group-by, time buckets, joins over metadata -> this tool, on \
messages/sessions.
- which tools ran / failed, tool params -> this tool, on parts (type = \
'tool_call' / 'tool_result'); worked example below.
- find text in conversations -> WHERE contains_tokens(search_text, '...') to \
filter, FROM fts('messages', ...) to rank (both below), or pond_search for \
meaning-based recall. Never LIKE over parts - tool bodies are JSON with no \
substring index (yet), and the conversational text is messages.search_text.
- read a transcript -> pond_get_session (a session) / pond_get_message (a \
message with context), not SQL.
Tables and columns:
- messages(session_id text, message_id text, timestamp timestamp(us, UTC), role \
text {user|assistant|system|tool}, source_agent text, project text, content text \
NULL [system-role messages only], search_text text NULL [the conversational text - \
null for system/tool messages], embedding_model text NULL, options json). The \
embedding `vector` column exists but is never returned (omitted from results) and \
explicit projection of it is rejected; you may still filter on it in WHERE, e.g. \
`vector IS NOT NULL`. For semantic search, use pond_search.
- sessions(session_id text, parent_session_id text NULL, parent_message_id text \
NULL, source_agent text, created_at timestamp(us, UTC), project text, options json).
- parts(session_id text, message_id text, id text, ordinal int, type text \
{text|reasoning|file|tool_call|tool_result|tool_approval_request|\
tool_approval_response - exact strings, underscores not hyphens}, provenance \
text {conversational|injected}, tool_name text NULL, call_id text NULL, \
is_failure boolean NULL, variant_data json, options json). `tool_name` / \
`call_id` / `is_failure` are narrow native columns materialized from the tool \
part bodies - ALWAYS prefer them over json_get_* on `variant_data` for tool \
analytics (name filters, GROUP BY, call->result joins, failure rates); they are \
NULL on non-tool parts (`is_failure` is non-NULL only on tool_result). The \
verbatim part body lives in `variant_data`; its fields follow the part type, \
e.g. tool_call carries {call_id, name, params}, tool_result carries {call_id, \
name, is_failure, result}, text/reasoning carry {text}. FilePart binary \
payloads are not exposed in SQL.
Enum literals matter: a wrong value (e.g. 'tool-call') is valid SQL and silently \
returns zero rows. Discovery from SQL works too: SELECT table_name, column_name, \
data_type FROM information_schema.columns.
Join keys: messages.session_id = sessions.session_id; parts.session_id = \
messages.session_id AND parts.message_id = messages.message_id. Subagents are \
sessions whose source_agent matches '%/%' (e.g. 'claude-code/general-purpose').
Indexed (fast) filter columns: messages.session_id / source_agent; \
parts.session_id / message_id / tool_name; sessions.session_id. \
Prefer equality/range predicates on these. Known limitation: prefix LIKE ('x%') and starts_with() FAIL \
on bitmap-indexed columns (messages.source_agent) with \"LIKE \
prefix queries are not supported for bitmap indexes\". Workarounds: equality, \
split_part(source_agent, '/', 1) = 'claude-code', or an infix pattern \
(LIKE '%/%' is fine - leading-wildcard patterns are not pushed to the index).
JSON columns (options, variant_data) are binary JSONB. Rules:
- NEVER CAST a JSON column (`variant_data::text` is rejected at plan time - the \
binary encoding can otherwise silently render as garbage). Stringify with \
json_extract(col, '$').
- A leading-wildcard LIKE over the whole document \
(`json_extract(variant_data, '$') LIKE '%...%'`) is rejected at plan time: it \
stringifies and scans every row and never finishes over parts. Match a single \
field (`json_extract(variant_data, '$.field') LIKE '...'`), scope to one session, \
or use contains_tokens for conversational text. (Substring search over tool \
bodies arrives with the FM-Index, #47.)
- json_extract(col, '$.a.b') takes a full JSONPath and returns JSON text of ANY \
value (objects/arrays serialize) - the right call for deeply nested or mixed-type \
fields, e.g. json_extract(variant_data, '$.params.command').
- json_get_string|json_get_int|json_get_float|json_get_bool(col, 'key', ...) walk \
a key path - json_get_string(options, 'anthropic', 'model') - array steps by \
numeric index. json_get_string serializes non-string values; the typed getters \
return NULL on a non-coercible value.
- json_get(col, 'key') returns JSONB for chaining: \
json_get_string(json_get(variant_data, 'params'), 'command').
- Also: json_array_contains(col, 'key', value), json_array_length(col, 'key').
Worked example - tool usage and failure rates over the last week, entirely on \
the narrow native columns (no JSON getters):
SELECT c.tool_name AS tool,
COUNT(*) AS calls,
SUM(CASE WHEN r.is_failure THEN 1 ELSE 0 END) AS failures
FROM parts c
JOIN messages m ON m.session_id = c.session_id AND m.message_id = c.message_id
LEFT JOIN parts r ON r.session_id = c.session_id
AND r.type = 'tool_result' AND r.call_id = c.call_id
WHERE c.type = 'tool_call' AND m.timestamp >= now() - INTERVAL '7 days'
GROUP BY tool ORDER BY calls DESC;
Worked example - counting tokens correctly (Claude Code sessions): pond is \
lossless - one source JSONL line = one row - so a multi-block assistant turn \
stores N rows sharing one provider message id \
(options.anthropic.id), EACH carrying that line's usage snapshot. Summing \
usage per row double-counts (~2-3x inflation). Usage is cumulative within a \
turn, so collapse to one value per provider message id with MAX (the final \
snapshot; DISTINCT over the usage tuple still double-counts when snapshots \
differ mid-turn), then aggregate. Always scope by session_id: usage lives \
inside the wide options column, which no index can serve (GROUP BY / MAX \
over JSON paths reads every candidate row's whole document), so an unscoped \
sweep is a full-column scan - minutes-to-timeout on a remote store. Even \
scoped, a remote store reads that session's options pages: raise \
timeout_seconds. For corpus-wide accounting, loop over session_ids or \
accept the full scan with timeout_seconds near the 600s cap.
WITH turns AS (
SELECT json_get_string(options, 'anthropic', 'id') AS turn_id,
MAX(json_get_int(options, 'anthropic', 'usage', 'input_tokens')) AS in_tok,
MAX(json_get_int(options, 'anthropic', 'usage', 'output_tokens')) AS out_tok,
MAX(json_get_int(options, 'anthropic', 'usage', 'cache_read_input_tokens')) AS cache_read
FROM messages
WHERE session_id = '<session-id>' AND role = 'assistant'
AND json_get_string(options, 'anthropic', 'id') IS NOT NULL
GROUP BY turn_id
)
SELECT COUNT(*) AS turns, SUM(in_tok) AS input_tokens,
SUM(out_tok) AS output_tokens, SUM(cache_read) AS cache_read_tokens
FROM turns;
Worked example - scope-then-scan, the pattern for substring-hunting tool \
bodies without a substring index: first collect candidate session_ids from \
the indexed conversational text, then run the field LIKE only inside them \
(the session_id btree makes the body scan tractable; unscoped it times out):
WITH hits AS (
SELECT DISTINCT session_id FROM messages
WHERE contains_tokens(search_text, 'cargo bench')
)
SELECT DISTINCT json_extract(p.variant_data, '$.params.command') AS cmd
FROM parts p JOIN hits h ON h.session_id = p.session_id
WHERE p.type = 'tool_call' AND p.tool_name = 'Bash'
AND json_extract(p.variant_data, '$.params.command') LIKE '%cargo bench%';
Full-text search in SQL is a pair - filter form and ranked form:
- Filtering (WHERE): contains_tokens(search_text, 'word1 word2') - true when the \
text contains ALL the words (split on punctuation/whitespace, case-sensitive \
tokens); accelerated by the FTS index. The right tool for exact strings, \
identifiers, and error messages - compose freely with other predicates: \
SELECT message_id FROM messages WHERE contains_tokens(search_text, 'OCC retry') \
AND project LIKE '%pond%'.
- Ranking (FROM): the fts() table function returns matches plus `_score` (BM25 \
relevance, a regular projectable column): SELECT message_id, _score, search_text \
FROM fts('messages', '{\"match\":{\"column\":\"search_text\",\"terms\":\"...\"}}') \
ORDER BY _score DESC - compose with WHERE/JOIN/GROUP BY around it. AND semantics: \
add \"operator\":\"And\" to the match; \"boolean\" queries (must/should/must_not \
over match clauses) also work. \"phrase\" queries are unavailable (index built \
without positions) - use contains_tokens or match + operator And, optionally with \
LIKE post-filters, for exact substrings.
fts() in WHERE is a plan-time error that points back here. Unlike pond_search, \
both forms cover subagent sessions (filter them out with WHERE NOT (source_agent \
LIKE '%/%') if unwanted). Vector/semantic search is NOT available in SQL; use \
pond_search for that.
Function quick-reference (exact DataFusion names so the model doesn't have to \
guess):
- aggregates: count, count(distinct ...), sum, avg, min, max, any_value, stddev, \
median, approx_distinct, approx_percentile_cont, array_agg, string_agg
- date/time: now(), date_trunc('day'|'hour'|'minute'|..., ts), date_part('year'|..., \
ts), date_bin(interval, ts, origin), to_char(ts, fmt), to_timestamp(text), \
extract(field FROM ts), age(t1, t2)
- intervals: `INTERVAL '7 days'`, `INTERVAL '1 hour'` (single-quoted, postgres-style)
- string: length, lower, upper, substr, position, split_part, regexp_like, \
regexp_match, regexp_replace, like, ilike, starts_with, ends_with, concat, \
concat_ws
- text search: contains_tokens(col, 'words') in WHERE; fts(table, query_json) in \
FROM (see above)
- numeric: round, floor, ceil, abs, sign, log, exp, power, sqrt
- conditional: CASE WHEN ... THEN ... ELSE ..., coalesce, nullif, greatest, least
- cast: CAST(x AS TYPE) or x::TYPE - but never on JSON columns (see the JSON \
rules above)
Quote identifiers with double quotes when they collide with keywords (e.g. \
\"timestamp\"); string literals use single quotes.
EXPLAIN is allowed: `EXPLAIN <query>` or `EXPLAIN ANALYZE <query>` returns the \
DataFusion plan (and per-operator timings for ANALYZE) so you can self-diagnose \
slow queries without leaving SQL.
Output modes (the `format` arg):
- text (default): a row-capped rendered ASCII table with a header showing \
`{total_rows} in {elapsed_ms} ms; showing {shown}` and, on truncation, a \
keyset-pagination hint. Long text cells clip at ~1000 chars with a \
`[+N chars ...]` marker - the full value is one format=ndjson call away, or \
project a narrower json_extract path.
- parquet | ndjson: write the FULL result set to a file and return a \
`pond-sql-export://<id>` resource link; read it via MCP resources/read. On a \
local/stdio install the response also names the on-disk path so you can open it \
directly with duckdb/polars.
Pagination - keyset (preferred):
Use ORDER BY on indexed columns plus a composite seek key for stable tie-breaking, \
spelled out as an OR - DataFusion rejects row-value comparisons like \
`(timestamp, message_id) < (...)`. The agent owns the cursor (the last sort value \
it saw); no server-side state.
-- page 1: most recent 100 messages in pond
SELECT message_id, timestamp, role, project
FROM messages
WHERE project LIKE '%pond%'
ORDER BY timestamp DESC, message_id DESC
LIMIT 100;
-- page 2: pass back the last (timestamp, message_id) the agent saw
SELECT message_id, timestamp, role, project
FROM messages
WHERE project LIKE '%pond%'
AND (timestamp < TIMESTAMP '2026-06-05T08:14:22.123456Z'
OR (timestamp = TIMESTAMP '2026-06-05T08:14:22.123456Z' AND message_id < 'last-id'))
ORDER BY timestamp DESC, message_id DESC
LIMIT 100;
Keyset stays stable across concurrent ingest (older rows don't shift) and uses \
the btree on `timestamp`/`message_id` directly. For known-bounded full results, skip \
pagination entirely: format=parquet writes everything in one call. OFFSET works \
but scans-and-discards prior rows and shifts pages under writes - prefer keyset.
Drilling from aggregates to content (instead of N round-trips of \
pond_get_message):
JOIN to messages/parts directly. Example - top 10 longest sessions with first \
user message:
WITH top_sessions AS (
SELECT session_id, COUNT(*) AS msgs
FROM messages
GROUP BY session_id
ORDER BY msgs DESC
LIMIT 10
)
SELECT ts.session_id, ts.msgs, s.project, s.source_agent,
m.search_text AS first_user_msg
FROM top_sessions ts
JOIN sessions s ON s.session_id = ts.session_id
LEFT JOIN messages m
ON m.session_id = ts.session_id
AND m.role = 'user'
AND m.timestamp = (
SELECT MIN(timestamp) FROM messages
WHERE session_id = ts.session_id AND role = 'user'
);
One call, agent picks exactly which columns to hydrate. When you want the \
rendered transcript (tool-call lines, subagent footer), call pond_get_session \
with the session_id - that's its job.
Examples (4 patterns the agent should recognize):
-- 1. Activity by project this week
SELECT project, COUNT(*) AS msgs, COUNT(DISTINCT session_id) AS sessions
FROM messages
WHERE timestamp >= now() - INTERVAL '7 days'
GROUP BY project
ORDER BY msgs DESC
LIMIT 20;
-- 2. Subagent breakdown
SELECT source_agent, COUNT(*) AS n
FROM sessions
WHERE source_agent LIKE '%/%'
GROUP BY source_agent
ORDER BY n DESC;
-- 3. Text filter in WHERE (all words must appear), composed with metadata
SELECT message_id, timestamp, project, substr(search_text, 1, 120) AS preview
FROM messages
WHERE contains_tokens(search_text, 'race condition')
AND timestamp >= now() - INTERVAL '30 days'
ORDER BY timestamp DESC
LIMIT 50;
-- 4. BM25 search in FROM, joined with metadata, relevance-ranked
SELECT m.session_id, m.timestamp, m.project, f._score, m.search_text
FROM fts('messages', \
'{\"match\":{\"column\":\"search_text\",\"terms\":\"race condition\"}}') f
JOIN messages m ON m.message_id = f.message_id
WHERE m.project LIKE '%pond%'
ORDER BY f._score DESC
LIMIT 50;";
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct McpSearchParams {
query: String,
#[serde(default)]
mode: Option<String>,
#[serde(default)]
sort_by: Option<String>,
#[serde(default)]
limit: Option<usize>,
#[serde(default)]
project: Option<String>,
#[serde(default)]
session_id: Option<String>,
#[serde(default)]
source_agent: Option<String>,
#[serde(default)]
from_date: Option<String>,
#[serde(default)]
to_date: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct McpGetSessionParams {
#[serde(alias = "session_id")]
id: String,
#[serde(default)]
limit: Option<usize>,
#[serde(default)]
from: Option<String>,
#[serde(default)]
after_message_id: Option<String>,
#[serde(default)]
before_message_id: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct McpGetMessageParams {
#[serde(alias = "message_id")]
id: String,
#[serde(default)]
context_before: Option<usize>,
#[serde(default)]
context_after: Option<usize>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct McpSqlParams {
#[serde(alias = "sql")]
query: String,
#[serde(default)]
format: Option<String>,
#[serde(default)]
timeout_seconds: Option<u64>,
}
fn parse_session_from(value: Option<String>) -> SessionFrom {
match value.as_deref() {
Some("end") => SessionFrom::End,
_ => SessionFrom::Start,
}
}
fn parse_search_mode(value: Option<&str>) -> Option<SearchModeWire> {
match value {
None | Some("fts") => Some(SearchModeWire::Fts),
Some("vector") => Some(SearchModeWire::Vector),
Some(_) => None,
}
}
fn parse_sort_by(value: Option<&str>) -> Option<SortBy> {
match value {
None | Some("relevance") => Some(SortBy::Relevance),
Some("recency") => Some(SortBy::Recency),
Some(_) => None,
}
}
#[derive(Clone)]
pub struct PondMcp {
state: AppState,
tool_router: ToolRouter<PondMcp>,
}
#[tool_router]
impl PondMcp {
pub fn new(state: AppState) -> Self {
Self {
state,
tool_router: Self::tool_router(),
}
}
#[tool(
description = "Find relevant messages in past sessions - the entry point for \
recall: \"have we worked on X\", \"what did we decide about Y\", \
\"find the session where...\". mode=\"fts\" (default) matches exact \
whole words (BM25); mode=\"vector\" matches meaning. Scope with \
project / session_id / source_agent / from_date / to_date; put the \
distinctive words you expect in the conversation in the query \
(error strings, symbol names, product names); project names go in \
the project filter. Returns scored hits grouped by \
session, best session first; pass a hit's session_id to \
pond_get_session or its message_id to pond_get_message to read \
it. Searches conversational text only (tool \
calls/results and reasoning are excluded by design - a gap there is \
expected, not a failure) and excludes subagent sessions; reach both \
via pond_sql. Response format details: resource schema://pond.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
async fn pond_search(
&self,
Parameters(params): Parameters<McpSearchParams>,
) -> Result<CallToolResult, ErrorData> {
let _activity = self.state.track_activity();
let Some(mode) = parse_search_mode(params.mode.as_deref()) else {
return Ok(CallToolResult::error(vec![ContentBlock::text(format!(
"unknown mode {:?}; use \"vector\" or \"fts\"",
params.mode.unwrap_or_default()
))]));
};
let Some(sort_by) = parse_sort_by(params.sort_by.as_deref()) else {
return Ok(CallToolResult::error(vec![ContentBlock::text(format!(
"unknown sort_by {:?}; use \"relevance\" or \"recency\"",
params.sort_by.unwrap_or_default()
))]));
};
let request = SearchRequest {
protocol_version: PROTOCOL_VERSION,
namespace: Some(default_namespace()),
query: params.query,
mode,
sort_by,
filters: SearchFilters {
project: params.project.map(ProjectFilter::Contains),
session_id: params.session_id,
source_agent: params.source_agent,
from_date: params.from_date,
to_date: params.to_date,
},
limit: params.limit.unwrap_or(10),
};
match run_search(
&self.state.store,
&self.state.embedder,
request.clone(),
&self.state.search,
)
.await
{
SearchEnvelope::Success(response) => {
Ok(tool_result(crate::render::render_search_transcript(
&response,
&request,
crate::render::Surface::Mcp,
)))
}
SearchEnvelope::Error(envelope) if is_semantic_disabled(&envelope) => Ok(
CallToolResult::error(vec![ContentBlock::text(envelope.error.message.clone())]),
),
SearchEnvelope::Error(envelope) => Err(to_error_data(&envelope)),
}
}
#[tool(
description = "Read a whole past session as a chronological transcript - the tool \
for analyzing, reviewing, or summarizing a session (user/assistant \
text plus one-line tool/file refs; tool bodies stay one \
pond_get_message away). Pass the id from pond_search or a subagent \
footer; a message_id also works - it resolves to its parent \
session with the page anchored at that message. Paging: limit \
(default 20), from=\"end\" reads the most recent turns first (the \
session's final state; late conclusions supersede early ones), \
after_message_id / before_message_id continue from a page marker. \
The first page lists subagent sessions in a footer - pass a listed \
id back to open one. Not for bulk export - use `pond copy --to \
<file>`. Response format details: resource schema://pond.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
async fn pond_get_session(
&self,
Parameters(params): Parameters<McpGetSessionParams>,
) -> Result<CallToolResult, ErrorData> {
let _activity = self.state.track_activity();
let first_page =
params.after_message_id.is_none() && params.before_message_id.is_none();
let request = GetSessionRequest {
protocol_version: PROTOCOL_VERSION,
namespace: Some(default_namespace()),
id: params.id,
limit: params.limit.unwrap_or(20),
from: parse_session_from(params.from),
after_message_id: params.after_message_id,
before_message_id: params.before_message_id,
};
match run_get_session(&self.state.store, request).await {
GetEnvelope::Success(response) => {
let mut transcript = crate::render::render_get_transcript(
&response,
crate::render::Surface::Mcp,
);
if first_page
&& let Ok(children) =
self.state.store.child_sessions(&response.session.id).await
&& !children.is_empty()
{
transcript.push_str(&crate::render::render_subagents_footer(
&children,
crate::render::Surface::Mcp,
));
}
Ok(tool_result(transcript))
}
GetEnvelope::Error(envelope) => Err(to_error_data(&envelope)),
}
}
#[tool(
description = "Expand one message with its full part bodies (tool_call / \
tool_result / reasoning / file), plus conversational neighbors \
for context. Pass a message_id from pond_search or a transcript \
line; context_before / context_after size the neighbor window \
(default 3, like grep -B/-A). For the whole session use \
pond_get_session. Response format details: resource schema://pond.",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
async fn pond_get_message(
&self,
Parameters(params): Parameters<McpGetMessageParams>,
) -> Result<CallToolResult, ErrorData> {
let _activity = self.state.track_activity();
let request = GetMessageRequest {
protocol_version: PROTOCOL_VERSION,
namespace: Some(default_namespace()),
id: params.id,
context_before: params.context_before.unwrap_or(3),
context_after: params.context_after.unwrap_or(3),
};
match run_get_message(&self.state.store, request).await {
GetEnvelope::Success(response) => Ok(tool_result(
crate::render::render_get_transcript(&response, crate::render::Surface::Mcp),
)),
GetEnvelope::Error(envelope) => Err(to_error_data(&envelope)),
}
}
#[tool(
description = "Advanced escape hatch: run ONE read-only SQL statement \
(SELECT/WITH, DataFusion / PostgreSQL-compatible) over the \
sessions / messages / parts tables. NOT for finding or reading \
conversations - pond_search and pond_get_session / \
pond_get_message cover almost all recall. \
Reach for SQL only for: aggregation (counts, group-by, joins, time \
buckets), exact strings or identifiers in conversational text \
(contains_tokens / fts), tool-call analytics and tool bodies, \
subagent sessions, bulk export (format=parquet|ndjson). Read \
resource schema://pond-sql FIRST - exact columns, indexed \
predicates, JSON access rules, worked examples; do not guess \
column names or JSON paths. Inline text output is row-capped \
and long cells clip with a +N chars marker (full values via \
format=parquet|ndjson); queries are wall-clock-capped (raise \
via timeout_seconds).",
annotations(read_only_hint = true, idempotent_hint = true, open_world_hint = false)
)]
async fn pond_sql(
&self,
Parameters(params): Parameters<McpSqlParams>,
) -> Result<CallToolResult, ErrorData> {
let _activity = self.state.track_activity();
let mode = match params.format.as_deref() {
None | Some("text") => sql::Mode::Inline,
Some("parquet") => sql::Mode::Export(sql::Format::Parquet),
Some("ndjson") => sql::Mode::Export(sql::Format::Ndjson),
Some(other) => {
return Ok(CallToolResult::error(vec![ContentBlock::text(format!(
"unknown format {other:?}; use \"text\", \"parquet\", or \"ndjson\""
))]));
}
};
let store = &self.state.store;
let outcome = async {
let tables = sql::open_tables(store, ¶ms.query, mode).await?;
sql::run(
&tables,
¶ms.query,
mode,
sql::DEFAULT_INLINE_ROWS,
params.timeout_seconds,
)
.await
}
.await;
match outcome {
Ok(sql::Outcome::Inline(text)) => Ok(tool_result(text)),
Ok(sql::Outcome::Export {
bytes,
format,
rows,
columns,
}) => {
let name = format!("{}.{}", Uuid::now_v7(), format.ext());
match store.export_write(&name, &bytes).await {
Ok(_) => Ok(export_result(
store,
&name,
format,
rows,
&columns,
bytes.len(),
)),
Err(error) => Err(ErrorData::internal_error(
format!("export write failed: {error}"),
None,
)),
}
}
Ok(sql::Outcome::Json(_)) => Err(ErrorData::internal_error(
"pond_sql never requests JSON rows".to_owned(),
None,
)),
Err(sql::SqlError::Query(message)) => {
Ok(CallToolResult::error(vec![ContentBlock::text(message)]))
}
Err(sql::SqlError::Storage(error)) => Err(ErrorData::internal_error(
format!(
"sql failed: {error:#}; the store failed transiently - retry the \
query, and run `pond status` to check the store if it persists"
),
None,
)),
Err(sql::SqlError::Infra(error)) => Err(ErrorData::internal_error(
format!("sql failed: {error:#}"),
None,
)),
}
}
}
#[tool_handler(router = self.tool_router)]
impl ServerHandler for PondMcp {
fn get_info(&self) -> ServerConfig {
ServerConfig::new(
ServerCapabilities::builder()
.enable_tools()
.enable_resources()
.build(),
)
.with_server_info(Implementation::new("pond", env!("CARGO_PKG_VERSION")))
.with_instructions(format!(
"pond recalls past agent sessions (Claude Code and others) - prior work, \
decisions, and context across sessions, not the live conversation. Three \
tools cover almost every request: pond_search finds relevant messages, \
pond_get_session reads a whole session, pond_get_message expands one \
message with its full tool bodies; all return readable transcripts. To \
analyze, review, or summarize a past session: pond_get_session(id) - one \
call, the whole transcript (any id works: a message id resolves to its \
session). For \"what did we decide / latest state\" questions, early \
conclusions in a long session are often superseded later - read from the \
end (pond_get_session from=\"end\") or re-sort with pond_search \
sort_by=\"recency\"; relevance rank favors the early, confident, possibly \
overturned phrasing. To recover context lost to compaction: pond_search (a \
distinctive recent topic + project + from_date=today), then \
pond_get_session(id, from=\"end\"). pond_sql is the advanced escape hatch \
for what search+get cannot express: corpus-wide aggregation (counts, \
group-by, joins), exact strings or identifiers inside tool bodies, \
bulk export (parquet/ndjson) - read resource \
schema://pond-sql before writing SQL. Scope with filters (project, \
session_id, source_agent, from_date / to_date), and phrase queries with \
the distinctive words you expect in the conversation (error strings, \
symbol names, product names) - project names go in the project filter. \
source_agent is exact-or-subpath: a root \
value (\"openclaw\", \"claude-code\") returns that harness's main \
sessions; subagents are stored as their own sessions and excluded by \
default - name a subpath (\"claude-code/general-purpose\") to search \
them, or reach them via pond_sql (parent_session_id). Search covers only \
user/assistant conversational text by \
design (tool calls/results and reasoning are excluded as low-signal \
noise, not a bug) - a zero/weak result is not proof of absence: verify \
exact strings with pond_sql WHERE contains_tokens(search_text, '...') \
before concluding something never happened. Deeper reference: resources \
schema://pond (search/get params + response format), schema://pond-sql \
(SQL columns, functions, worked examples), stats://pond (corpus stats).{}",
if crate::embed::embeddings_enabled() {
""
} else {
" Semantic (meaning-based) search is an opt-in on this instance - \
[embeddings].enabled in pond's config."
},
))
}
async fn list_resources(
&self,
_request: Option<PaginatedRequestParams>,
context: RequestContext<RoleServer>,
) -> Result<ListResourcesResult, ErrorData> {
let mut result = ListResourcesResult::with_all_items(vec![
Resource::new("schema://pond", "pond search schema"),
Resource::new("schema://pond-sql", "pond SQL table schema"),
Resource::new("stats://pond", "pond corpus stats"),
]);
(result.ttl_ms, result.cache_scope) = cache_hints(&context, CacheScope::Public);
Ok(result)
}
async fn list_resource_templates(
&self,
_request: Option<PaginatedRequestParams>,
context: RequestContext<RoleServer>,
) -> Result<ListResourceTemplatesResult, ErrorData> {
let mut result = ListResourceTemplatesResult::with_all_items(Vec::new());
(result.ttl_ms, result.cache_scope) = cache_hints(&context, CacheScope::Public);
Ok(result)
}
async fn read_resource(
&self,
request: ReadResourceRequestParams,
context: RequestContext<RoleServer>,
) -> Result<ReadResourceResponse, ErrorData> {
let _activity = self.state.track_activity();
let (mut result, scope) = match request.uri.as_str() {
"schema://pond" => (
ReadResourceResult::new(vec![ResourceContents::text(
schema_doc_for(crate::embed::embeddings_enabled()),
request.uri,
)]),
CacheScope::Public,
),
"schema://pond-sql" => (
ReadResourceResult::new(vec![ResourceContents::text(
SQL_SCHEMA_DOC,
request.uri,
)]),
CacheScope::Public,
),
uri if uri.starts_with("pond-sql-export://") => {
let name = uri.trim_start_matches("pond-sql-export://").to_owned();
if !valid_export_name(&name) {
return Err(ErrorData::resource_not_found(
format!("invalid export id: {name}"),
None,
));
}
let bytes = self.state.store.export_read(&name).await.map_err(|error| {
ErrorData::resource_not_found(format!("export not found: {error}"), None)
})?;
let contents = if name.ends_with(".ndjson") {
ResourceContents::text(
String::from_utf8_lossy(&bytes).into_owned(),
request.uri,
)
.with_mime_type("application/x-ndjson")
} else {
ResourceContents::blob(STANDARD.encode(&bytes), request.uri)
.with_mime_type("application/vnd.apache.parquet")
};
(ReadResourceResult::new(vec![contents]), CacheScope::Private)
}
"stats://pond" => {
let store = &self.state.store;
let map_err = |error: anyhow::Error| {
ErrorData::internal_error(format!("stats unavailable: {error}"), None)
};
let embedding_fut = async {
if crate::embed::embeddings_enabled() {
let (progress, stale) = tokio::try_join!(
store.embedding_progress(),
store.stale_embedding_count(),
)?;
let searchable = progress.total;
anyhow::Ok((Some((progress, stale)), searchable))
} else {
let searchable = store.searchable_message_count().await?;
anyhow::Ok((None, searchable))
}
};
let ((sessions, messages, parts), (embedding, searchable), indices) =
tokio::try_join!(store.row_counts(), embedding_fut, store.index_status())
.map_err(map_err)?;
let embedded_percent = embedding.as_ref().map(|(progress, _)| {
if progress.total == 0 {
0.0
} else {
let pct = (progress.embedded as f64 / progress.total as f64) * 100.0;
(pct * 10.0).round() / 10.0
}
});
let index_rows = indices
.iter()
.map(|status| {
serde_json::json!({
"table": status.table.as_str(),
"intent": status.intent_name,
"exists": status.exists,
"fragments_covered": status.fragments_covered,
"unindexed_rows": status.unindexed_rows,
})
})
.collect::<Vec<_>>();
let stats = serde_json::json!({
"corpus": {
"sessions": sessions,
"messages": messages,
"searchable_messages": searchable,
"parts": parts,
},
"embeddings": {
"model": embedding.as_ref().map(|(p, _)| p.model),
"embedded": embedding.as_ref().map(|(p, _)| p.embedded),
"searchable_total": embedding.as_ref().map(|(p, _)| p.total),
"embedded_percent": embedded_percent,
"stale_under_other_model": embedding.as_ref().map(|(_, s)| *s),
},
"indices": index_rows,
});
(
ReadResourceResult::new(vec![ResourceContents::text(
stats.to_string(),
request.uri,
)]),
CacheScope::Private,
)
}
other => {
return Err(ErrorData::resource_not_found(
format!("unknown resource: {other}"),
None,
));
}
};
(result.ttl_ms, result.cache_scope) = cache_hints(&context, scope);
Ok(result.into())
}
async fn list_tools(
&self,
_request: Option<PaginatedRequestParams>,
context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, ErrorData> {
let mut result = ListToolsResult::with_all_items(self.tool_router.list_all());
annotate_tool_limits(&mut result);
annotate_search_mode_with(&mut result, crate::embed::embeddings_enabled());
(result.ttl_ms, result.cache_scope) = cache_hints(&context, CacheScope::Public);
Ok(result)
}
}
const CACHE_HINT_TTL_MS: u64 = 0;
fn cache_hints(
context: &RequestContext<RoleServer>,
scope: CacheScope,
) -> (Option<u64>, Option<CacheScope>) {
let supports_cache_hints = context
.protocol_version()
.is_some_and(|version| version >= ProtocolVersion::V_2026_07_28);
(
supports_cache_hints.then_some(CACHE_HINT_TTL_MS),
supports_cache_hints.then_some(scope),
)
}
const SEARCH_DESCRIPTION_FTS_ONLY: &str = "Find relevant messages in past sessions - the \
entry point for recall: \"have we worked on X\", \"what did we decide about Y\", \"find \
the session where...\". Scope with project / session_id / source_agent / from_date / \
to_date; put the distinctive words you expect in the conversation in the query (error \
strings, symbol names, product names); project names go in the project filter. Returns \
scored hits grouped by session, best session first; pass a hit's session_id to \
pond_get_session or its message_id to pond_get_message to read it. Searches \
conversational text only (tool calls/results and reasoning are excluded by design - a \
gap there is expected, not a failure) and excludes subagent sessions; reach both via \
pond_sql. Response format details: resource schema://pond.";
const MODE_DOC_FTS_ONLY: &str =
"\"fts\" (default and only arm on this instance): exact whole words via BM25.";
fn is_semantic_disabled(envelope: &ErrorEnvelope) -> bool {
envelope
.error
.details
.get("config_key")
.and_then(serde_json::Value::as_str)
== Some(crate::handlers::SEMANTIC_DISABLED_CONFIG_KEY)
}
fn annotate_search_mode_with(result: &mut ListToolsResult, embeddings_enabled: bool) {
if embeddings_enabled {
return;
}
for tool in result
.tools
.iter_mut()
.filter(|tool| tool.name == "pond_search")
{
tool.description = Some(SEARCH_DESCRIPTION_FTS_ONLY.into());
if let Some(mode) = Arc::make_mut(&mut tool.input_schema)
.get_mut("properties")
.and_then(serde_json::Value::as_object_mut)
.and_then(|properties| properties.get_mut("mode"))
.and_then(serde_json::Value::as_object_mut)
{
mode.insert(
"description".to_owned(),
serde_json::Value::String(MODE_DOC_FTS_ONLY.to_owned()),
);
}
}
}
fn annotate_tool_limits(result: &mut ListToolsResult) {
for tool in &mut result.tools {
let chars = match tool.name.as_ref() {
"pond_search" => 80_000,
"pond_get_session" | "pond_get_message" => 200_000,
"pond_sql" => 80_000,
_ => continue,
};
let mut meta = serde_json::Map::new();
meta.insert(
"anthropic/maxResultSizeChars".to_owned(),
serde_json::json!(chars),
);
tool.meta = Some(MetaObject(meta));
}
}
pub async fn serve_stdio(state: AppState) -> anyhow::Result<()> {
let (stdin, stdout) = stdio();
serve_io(state, stdin, stdout).await
}
async fn serve_io<R, W>(state: AppState, read: R, write: W) -> anyhow::Result<()>
where
R: tokio::io::AsyncRead + Send + Unpin + 'static,
W: tokio::io::AsyncWrite + Send + Unpin + 'static,
{
let service = PondMcp::new(state)
.serve(AsyncRwTransport::new_server(read, write))
.await
.context("failed to start stdio MCP server")?;
service.waiting().await.context("stdio MCP server error")?;
Ok(())
}
fn tool_result(transcript: String) -> CallToolResult {
CallToolResult::success(vec![ContentBlock::text(transcript)])
}
fn export_result(
store: &crate::sessions::Store,
name: &str,
format: sql::Format,
rows: usize,
columns: &[String],
bytes: usize,
) -> CallToolResult {
let uri = format!("pond-sql-export://{name}");
let column_list = if columns.is_empty() {
"(none)".to_owned()
} else {
columns.join(", ")
};
let mut summary = format!(
"Exported {rows} row(s), {bytes} bytes ({}). Columns: {column_list}.\n\
Fetch via MCP resources/read on {uri}.",
format.ext()
);
if let Some(path) = store.export_local_path(name) {
summary.push_str(&format!(
"\nLocal file: {} - on this (stdio) install you can read it directly \
(e.g. duckdb, polars).",
path.display()
));
}
let link = Resource::new(uri, name.to_owned())
.with_description(format!("pond SQL export ({}, {rows} rows)", format.ext()))
.with_mime_type(format.mime().to_owned())
.with_size(u64::try_from(bytes).unwrap_or(u64::MAX));
CallToolResult::success(vec![
ContentBlock::text(summary),
ContentBlock::resource_link(link),
])
}
fn valid_export_name(name: &str) -> bool {
let Some((stem, ext)) = name.rsplit_once('.') else {
return false;
};
matches!(ext, "parquet" | "ndjson")
&& !stem.is_empty()
&& stem.bytes().all(|b| b.is_ascii_hexdigit() || b == b'-')
}
fn to_error_data(envelope: &ErrorEnvelope) -> ErrorData {
let (jsonrpc_code, pond_code, retryable) = match envelope.error.code {
WireErrorCode::ValidationFailed => (-32010, "validation_failed", false),
WireErrorCode::VersionUnsupported => (-32011, "version_unsupported", false),
WireErrorCode::NotFound => (-32012, "not_found", false),
WireErrorCode::NamespaceUnknown => (-32013, "namespace_unknown", false),
WireErrorCode::StorageUnavailable => (-32014, "storage_unavailable", true),
WireErrorCode::Conflict => (-32015, "conflict", true),
WireErrorCode::Internal => (-32016, "internal", false),
};
let mut data = match &envelope.error.details {
serde_json::Value::Object(map) => map.clone(),
_ => serde_json::Map::new(),
};
data.insert("pond_code".to_owned(), serde_json::json!(pond_code));
data.insert("retryable".to_owned(), serde_json::json!(retryable));
ErrorData::new(
JsonRpcErrorCode(jsonrpc_code),
envelope.error.message.clone(),
Some(serde_json::Value::Object(data)),
)
}
#[cfg(test)]
mod tests {
#![expect(
clippy::expect_used,
clippy::unwrap_used,
reason = "tests fail by panicking"
)]
use std::sync::Arc;
use rmcp::model::Tool;
use super::*;
use crate::wire::{ErrorBody, ErrorCode};
#[tokio::test]
async fn discovery_startup_exposes_pond_tools() -> anyhow::Result<()> {
tokio::time::timeout(std::time::Duration::from_secs(30), async {
let (_temp, client, server_task) = connect(true).await?;
let info = client.peer_info().unwrap();
assert_eq!(info.server_info.as_ref().unwrap().name, "pond");
let tools = client.list_all_tools().await?;
let mut names: Vec<_> = tools.iter().map(|tool| tool.name.as_ref()).collect();
names.sort_unstable();
assert_eq!(
names,
[
"pond_get_message",
"pond_get_session",
"pond_search",
"pond_sql"
]
);
client.cancel().await?;
server_task.await??;
anyhow::Ok(())
})
.await?
}
#[tokio::test]
async fn malformed_opener_leaves_initialize_fallback_open() -> anyhow::Result<()> {
for version in ["2025-06-18", "2026-07-28"] {
let replies = exchange(&[
serde_json::json!({
"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}
}),
initialize_frame(2, version),
serde_json::json!({"jsonrpc": "2.0", "method": "notifications/initialized"}),
serde_json::json!({
"jsonrpc": "2.0", "id": 3, "method": "tools/list", "params": {}
}),
])
.await?;
assert_eq!(
replies[&1]["error"]["code"],
JsonRpcErrorCode::INVALID_PARAMS.0,
"{version}"
);
assert!(
replies[&2]["result"]["protocolVersion"].is_string(),
"{version}"
);
assert!(has_tools(&replies[&3]), "{version}");
}
Ok(())
}
#[tokio::test]
async fn discover_probe_then_initialize_serves_classic_requests() -> anyhow::Result<()> {
let replies = exchange(&[
serde_json::json!({
"jsonrpc": "2.0", "id": 1, "method": "server/discover",
"params": {"_meta": {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {"name": "probe", "version": "0"},
"io.modelcontextprotocol/clientCapabilities": {}
}}
}),
initialize_frame(2, "2025-11-25"),
serde_json::json!({"jsonrpc": "2.0", "method": "notifications/initialized"}),
serde_json::json!({
"jsonrpc": "2.0", "id": 3, "method": "tools/list", "params": {}
}),
])
.await?;
assert!(replies[&1]["result"]["supportedVersions"].is_array());
assert_eq!(replies[&2]["result"]["protocolVersion"], "2025-11-25");
assert!(has_tools(&replies[&3]), "{}", replies[&3]);
Ok(())
}
fn initialize_frame(id: u64, version: &str) -> serde_json::Value {
serde_json::json!({
"jsonrpc": "2.0", "id": id, "method": "initialize",
"params": {
"protocolVersion": version,
"capabilities": {},
"clientInfo": {"name": "probe", "version": "0"}
}
})
}
fn has_tools(reply: &serde_json::Value) -> bool {
reply["result"]["tools"]
.as_array()
.is_some_and(|tools| !tools.is_empty())
}
async fn exchange(
frames: &[serde_json::Value],
) -> anyhow::Result<std::collections::HashMap<u64, serde_json::Value>> {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
tokio::time::timeout(std::time::Duration::from_secs(30), async {
let temp = tempfile::TempDir::new()?;
let (client_io, server_task) = spawn_server(&temp).await?;
let (read, mut write) = tokio::io::split(client_io);
for frame in frames {
write.write_all(format!("{frame}\n").as_bytes()).await?;
}
let mut lines = BufReader::new(read).lines();
let mut replies = std::collections::HashMap::new();
for _ in frames.iter().filter(|frame| frame.get("id").is_some()) {
let line = lines.next_line().await?.context("server hung up")?;
let reply: serde_json::Value = serde_json::from_str(&line)?;
replies.insert(reply["id"].as_u64().context("reply without id")?, reply);
}
drop((lines, write));
server_task.await??;
anyhow::Ok(replies)
})
.await?
}
async fn spawn_server(
temp: &tempfile::TempDir,
) -> anyhow::Result<(
tokio::io::DuplexStream,
tokio::task::JoinHandle<anyhow::Result<()>>,
)> {
let state = AppState::new(
Arc::new(crate::sessions::Store::open_local(temp.path()).await?),
Arc::new(crate::embed::LazyEmbedder::candle()),
crate::config::SearchConfig::default(),
);
let (client_io, server_io) = tokio::io::duplex(64 * 1024);
let (read, write) = tokio::io::split(server_io);
Ok((client_io, tokio::spawn(serve_io(state, read, write))))
}
async fn connect(
modern: bool,
) -> anyhow::Result<(
tempfile::TempDir,
rmcp::service::RunningService<rmcp::RoleClient, ()>,
tokio::task::JoinHandle<anyhow::Result<()>>,
)> {
use rmcp::{ClientLifecycleMode, ClientServiceExt};
let temp = tempfile::TempDir::new()?;
let (client_io, server_task) = spawn_server(&temp).await?;
let client = if modern {
().serve_with_lifecycle(
client_io,
ClientLifecycleMode::Discover {
preferred_versions: vec![ProtocolVersion::V_2026_07_28],
},
)
.await?
} else {
().serve(client_io).await?
};
Ok((temp, client, server_task))
}
async fn cache_hints_seen(
modern: bool,
) -> anyhow::Result<Vec<(&'static str, Option<u64>, Option<CacheScope>)>> {
let (_temp, client, server_task) = connect(modern).await?;
let tools = client.list_tools(None).await?;
let resources = client.list_resources(None).await?;
let templates = client.list_resource_templates(None).await?;
let schema = client
.read_resource(ReadResourceRequestParams::new("schema://pond"))
.await?;
let stats = client
.read_resource(ReadResourceRequestParams::new("stats://pond"))
.await?;
client.cancel().await?;
server_task.await??;
Ok(vec![
("tools/list", tools.ttl_ms, tools.cache_scope),
("resources/list", resources.ttl_ms, resources.cache_scope),
(
"resources/templates/list",
templates.ttl_ms,
templates.cache_scope,
),
(
"resources/read schema://pond",
schema.ttl_ms,
schema.cache_scope,
),
(
"resources/read stats://pond",
stats.ttl_ms,
stats.cache_scope,
),
])
}
#[tokio::test]
async fn modern_results_carry_the_required_cache_hints() -> anyhow::Result<()> {
tokio::time::timeout(std::time::Duration::from_secs(30), async {
let public = (Some(0), Some(CacheScope::Public));
let private = (Some(0), Some(CacheScope::Private));
let expected = [
("tools/list", public),
("resources/list", public),
("resources/templates/list", public),
("resources/read schema://pond", public),
("resources/read stats://pond", private),
]
.map(|(surface, (ttl_ms, scope))| (surface, ttl_ms, scope));
assert_eq!(cache_hints_seen(true).await?, expected);
anyhow::Ok(())
})
.await?
}
#[tokio::test]
async fn legacy_results_omit_the_cache_hints() -> anyhow::Result<()> {
tokio::time::timeout(std::time::Duration::from_secs(30), async {
for (surface, ttl_ms, scope) in cache_hints_seen(false).await? {
assert_eq!((ttl_ms, scope), (None, None), "{surface}");
}
anyhow::Ok(())
})
.await?
}
#[test]
fn error_data_carries_code_and_retryability() {
let cases = [
(
ErrorCode::ValidationFailed,
-32010,
"validation_failed",
false,
),
(
ErrorCode::VersionUnsupported,
-32011,
"version_unsupported",
false,
),
(ErrorCode::NotFound, -32012, "not_found", false),
(
ErrorCode::NamespaceUnknown,
-32013,
"namespace_unknown",
false,
),
(
ErrorCode::StorageUnavailable,
-32014,
"storage_unavailable",
true,
),
(ErrorCode::Conflict, -32015, "conflict", true),
(ErrorCode::Internal, -32016, "internal", false),
];
for (code, jsonrpc, pond_code, retryable) in cases {
let error = to_error_data(&ErrorEnvelope {
error: ErrorBody {
code,
message: "boom".to_owned(),
details: serde_json::json!({"detail": 1}),
},
});
assert_eq!(error.code, JsonRpcErrorCode(jsonrpc));
let data = error.data.unwrap();
assert_eq!(data["detail"], serde_json::json!(1));
assert_eq!(data["pond_code"], serde_json::json!(pond_code));
assert_eq!(data["retryable"], serde_json::json!(retryable));
assert!(
data.get("request_id").is_none(),
"MCP errors use JSON-RPC ids for correlation"
);
}
}
#[test]
fn search_params_deserialize_source_agent_and_map_to_filters() {
let params: McpSearchParams = serde_json::from_value(serde_json::json!({
"query": "vector store",
"source_agent": "openclaw/subagent",
}))
.expect("params deserialize");
assert_eq!(params.source_agent.as_deref(), Some("openclaw/subagent"));
let filters = crate::wire::SearchFilters {
source_agent: params.source_agent,
..Default::default()
};
assert_eq!(filters.source_agent.as_deref(), Some("openclaw/subagent"));
}
#[test]
fn annotate_tool_limits_sets_anthropic_meta() {
let schema = Arc::new(serde_json::Map::new());
let mut result = ListToolsResult::with_all_items(vec![
Tool::new("pond_search", "Search", Arc::clone(&schema)),
Tool::new("pond_get_session", "Get session", Arc::clone(&schema)),
Tool::new("pond_get_message", "Get message", Arc::clone(&schema)),
]);
annotate_tool_limits(&mut result);
let value = |name: &str| {
result
.tools
.iter()
.find(|tool| tool.name == name)
.and_then(|tool| tool.meta.as_ref())
.and_then(|meta| meta.0.get("anthropic/maxResultSizeChars"))
.and_then(serde_json::Value::as_i64)
};
assert_eq!(value("pond_search"), Some(80_000));
assert_eq!(value("pond_get_session"), Some(200_000));
assert_eq!(value("pond_get_message"), Some(200_000));
}
#[test]
fn parse_search_mode_defaults_to_fts() {
assert_eq!(parse_search_mode(None), Some(SearchModeWire::Fts));
assert_eq!(parse_search_mode(Some("fts")), Some(SearchModeWire::Fts));
assert_eq!(
parse_search_mode(Some("vector")),
Some(SearchModeWire::Vector)
);
assert_eq!(parse_search_mode(Some("hybrid")), None);
}
#[test]
fn semantic_disabled_predicate_matches_only_the_refusal() {
let envelope = |details: serde_json::Value| ErrorEnvelope {
error: ErrorBody {
code: ErrorCode::ValidationFailed,
message: "boom".to_owned(),
details,
},
};
assert!(is_semantic_disabled(&envelope(serde_json::json!({
"field": "mode",
"config_key": "embeddings.enabled",
"retryable": false,
}))));
assert!(!is_semantic_disabled(&envelope(
serde_json::json!({ "field": "query" })
)));
assert!(!is_semantic_disabled(&envelope(serde_json::json!({
"config_key": "search.nprobes"
}))));
}
fn listed_tools() -> ListToolsResult {
ListToolsResult::with_all_items(PondMcp::tool_router().list_all())
}
fn search_tool(result: &ListToolsResult) -> &Tool {
result
.tools
.iter()
.find(|tool| tool.name == "pond_search")
.expect("pond_search is registered")
}
fn mode_description(tool: &Tool) -> &str {
tool.input_schema["properties"]["mode"]["description"]
.as_str()
.expect("the mode property carries a description")
}
#[test]
fn search_surface_hides_the_vector_arm_when_embeddings_are_disabled() {
let mut disabled = listed_tools();
annotate_search_mode_with(&mut disabled, false);
let tool = search_tool(&disabled);
let description = tool.description.as_deref().unwrap_or_default();
assert!(
!description.contains("vector"),
"tool description must not offer an arm this instance refuses: {description}"
);
let mode = mode_description(tool);
assert!(
!mode.contains("vector"),
"mode description must not offer the vector arm: {mode}"
);
assert_eq!(
tool.input_schema["properties"]["mode"]["type"],
serde_json::json!(["string", "null"]),
);
let mut enabled = listed_tools();
annotate_search_mode_with(&mut enabled, true);
let tool = search_tool(&enabled);
assert!(
tool.description
.as_deref()
.unwrap_or_default()
.contains("vector"),
"an enabled instance advertises both arms"
);
assert!(mode_description(tool).contains("vector"));
}
#[test]
fn schema_doc_hides_the_vector_arm_when_embeddings_are_disabled() {
let disabled = schema_doc_for(false);
assert_ne!(disabled, SCHEMA_DOC, "no rewrite landed");
assert!(
!disabled.contains("vector"),
"schema doc must not offer an arm this instance refuses: {disabled}"
);
assert_eq!(schema_doc_for(true), SCHEMA_DOC);
}
}
}