1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
//! Helper bodies for the cache admin MCP tools (`cache_stats`, `cache_gc`,
//! `cache_clear`). Kept out of `helpers.rs` so that file stays under the
//! 1000-line cap.
//!
//! ## In-process GC vs the offline CLI path
//!
//! `serve` holds the store's `.basemind/.lock` advisory flock for its entire
//! lifetime (acquired by `Store::open`). [`crate::store_gc::run_gc`] re-acquires
//! that flock, so calling it in-process would deadlock against ourselves. Instead
//! these helpers call the *unlocked* primitives
//! [`crate::store_gc::collect_referenced_hashes`] + [`crate::store_gc::gc_blobs`]
//! while holding a `state.store` `RwLock` guard as the mutual-exclusion mechanism:
//! a held read guard blocks the only in-process writer (`scan_and_refresh`, which
//! takes `write()`), and cross-process scans are already impossible because serve
//! holds the flock. `run_gc` remains the correct primitive for the offline CLI.
use std::sync::Arc;
use rmcp::ErrorData as McpError;
use rmcp::model::CallToolResult;
use super::ServerState;
use super::helpers::{json_result, scan_and_refresh};
use super::types::{
CacheClearParams, CacheClearResponse, CacheGcParams, CacheGcResponse, CacheStatsParams, CacheStatsResponse,
};
use crate::store_gc::{self, CacheComponent};
/// Body for the `cache_stats` MCP tool. Read-only: takes a `blocking_read()` store
/// guard inside `spawn_blocking` and gathers per-component sizes + blob accounting.
pub(super) async fn run_cache_stats(
state: Arc<ServerState>,
_params: CacheStatsParams,
) -> Result<CallToolResult, McpError> {
let state_for_stats = Arc::clone(&state);
let stats = tokio::task::spawn_blocking(move || {
// Read guard: blocks `scan_and_refresh` for the (cheap) stat walk; cross-process
// scans can't run because serve holds the flock.
let store = state_for_stats.store.blocking_read();
store_gc::cache_stats(&store.basemind_dir)
})
.await
.map_err(|e| McpError::internal_error(format!("cache_stats join: {e}"), None))?
.map_err(|e| McpError::internal_error(format!("cache_stats: {e}"), None))?;
json_result(&CacheStatsResponse::from(stats))
}
/// Body for the `cache_gc` MCP tool. In-process mark-and-sweep over orphaned blobs.
/// Uses the unlocked `collect_referenced_hashes` + `gc_blobs` primitives under a
/// `blocking_read()` guard — NEVER `run_gc` (which would deadlock on serve's flock).
pub(super) async fn run_cache_gc(state: Arc<ServerState>, _params: CacheGcParams) -> Result<CallToolResult, McpError> {
let state_for_gc = Arc::clone(&state);
let report = tokio::task::spawn_blocking(move || {
// The read guard blocks the only in-process writer (`scan_and_refresh`) for the
// mark+sweep, so a concurrent rescan can't write a blob we then orphan-reap.
let store = state_for_gc.store.blocking_read();
let referenced = store_gc::collect_referenced_hashes(&store.basemind_dir)?;
store_gc::gc_blobs(&store.basemind_dir, &referenced)
})
.await
.map_err(|e| McpError::internal_error(format!("cache_gc join: {e}"), None))?
.map_err(|e| McpError::internal_error(format!("cache_gc: {e}"), None))?;
json_result(&CacheGcResponse::from(report))
}
/// Body for the `cache_clear` MCP tool. Parses + validates the component token,
/// gates the destructive (live-index-backing) components behind `confirm=true`, and
/// rebuilds the live state after a destructive clear so queries recover.
pub(super) async fn run_cache_clear(
state: Arc<ServerState>,
params: CacheClearParams,
) -> Result<CallToolResult, McpError> {
// `views:<name>` clears a single view (bug #22). Safe in-process for any view EXCEPT the
// one this server currently has open (deleting its live Fjall handle's dir would break it
// — same hazard as `views`/`all`). Reclaims disk from stale rev/staged views while serving.
if let Some(name) = params.component.strip_prefix("views:") {
let name = name.to_string();
let active_view = state.store.read().await.view.clone();
if name == active_view {
return Err(McpError::invalid_request(
format!(
"view `{name}` is the one this server is serving; clearing it would break \
the live index. Stop the server and run `basemind cache clear --component \
views:{name}`, or serve a different view."
),
None,
));
}
let dir = state.store.read().await.basemind_dir.clone();
tokio::task::spawn_blocking(move || store_gc::clear_single_view(&dir, &name))
.await
.map_err(|e| McpError::internal_error(format!("cache_clear join: {e}"), None))?
.map_err(|e| McpError::invalid_request(format!("cache_clear: {e}"), None))?;
return json_result(&CacheClearResponse {
component: params.component.clone(),
cleared: true,
});
}
let component: CacheComponent = params.component.parse().map_err(|e: String| {
McpError::invalid_request(
format!("{e} (valid: blobs|views|lance|git-cache|telemetry|all, or views:<name>)"),
None,
)
})?;
match component {
// `all` removes the whole `.basemind/` dir, and `views` removes `index.fjall/` —
// the directory the live `IndexDb` handle has OPEN. Deleting either out from under
// the running server breaks the open Fjall tree (and, for `all`, the flock this
// server holds): a stale handle pointing at a deleted dir. Refuse in-process and
// point the operator at the offline CLI, which clears with no handles open.
CacheComponent::All | CacheComponent::Views => Err(McpError::invalid_request(
format!(
"clearing `{}` removes the live Fjall index out from under the running \
server; stop the server and run `basemind cache clear --component {}`",
component.as_str(),
component.as_str()
),
None,
)),
// Blobs are content-addressed files, not an open handle: safe to clear under a
// write guard, then a rescan re-extracts + rewrites them. Require confirm because
// queries briefly see missing L2 blobs until the rescan completes.
CacheComponent::Blobs => {
if !params.confirm {
return Err(McpError::invalid_request(
"clearing `blobs` drops cached extractions; pass confirm=true to proceed \
(a rescan runs afterwards to rebuild them)",
None,
));
}
clear_live_component(Arc::clone(&state), component).await?;
// Rebuild from source so subsequent queries don't hit missing blobs.
scan_and_refresh(state, None, crate::scanner::EmbedMode::Inline).await?;
json_result(&CacheClearResponse {
component: component.as_str().to_string(),
cleared: true,
})
}
// Non-live caches: clear freely, no confirm, no rebuild needed.
CacheComponent::Lance | CacheComponent::GitCache | CacheComponent::Telemetry => {
clear_live_component(Arc::clone(&state), component).await?;
json_result(&CacheClearResponse {
component: component.as_str().to_string(),
cleared: true,
})
}
}
}
/// Clear a single component under a `blocking_write()` store guard. The write guard
/// serializes against `scan_and_refresh` and the stats/GC read guards for the wipe.
async fn clear_live_component(state: Arc<ServerState>, component: CacheComponent) -> Result<(), McpError> {
tokio::task::spawn_blocking(move || {
let store = state.store.blocking_write();
store_gc::clear_component(&store.basemind_dir, component)
})
.await
.map_err(|e| McpError::internal_error(format!("cache_clear join: {e}"), None))?
.map_err(|e| McpError::internal_error(format!("cache_clear: {e}"), None))
}
/// Body for the `rescan` MCP tool. Re-indexes the working tree (or `paths`) in-process and,
/// because it is one of the few genuinely slow tools, emits MCP progress (when the client
/// supplies a token) and a completion logging notification. Lives here with the other admin-tool
/// bodies so `helpers.rs` stays under the line cap.
pub(super) async fn run_rescan(
state: Arc<ServerState>,
params: super::types::RescanParams,
peer: &rmcp::Peer<rmcp::RoleServer>,
progress_token: Option<rmcp::model::ProgressToken>,
) -> Result<CallToolResult, McpError> {
let started = std::time::Instant::now();
// `full` forces a complete working-tree scan even when `paths` is supplied (full wins);
// `None` scoped_paths is the full-scan signal in `scan_and_refresh`. Request paths are raw
// strings (not `RelPath`), so validate each through `normalize_query_path` BEFORE joining to
// the root: a `../..`-style path would otherwise `root.join` into a traversal that reads (and
// indexes) a file outside the repo when `scan.respect_gitignore = false`. `normalize_query_path`
// rejects any `..` that escapes the root (→ `None`), mirroring the `shell_spawn` cwd guard.
let scoped_paths: Option<Vec<std::path::PathBuf>> = match params.paths.filter(|_| !params.full) {
None => None,
Some(requested) => {
let mut out = Vec::with_capacity(requested.len());
for p in requested {
let normalized = crate::path::normalize_query_path(&p, &state.root).ok_or_else(|| {
McpError::invalid_params(format!("rescan: path {p:?} escapes the repository root"), None)
})?;
out.push(state.root.join(normalized));
}
Some(out)
}
};
let root = state.root.display().to_string();
// Tell the client the rescan has started (progress is indeterminate up front — the scanner
// discovers the file count as it walks).
if let Some(token) = progress_token.clone() {
super::notifications::emit_progress(peer, token, 0.0, None, "rescan: scanning working tree").await;
}
// Manual `rescan` embeds inline so the tool's return means the index is fully current.
let report = scan_and_refresh(Arc::clone(&state), scoped_paths, crate::scanner::EmbedMode::Inline).await?;
// Surface the outcome as a logging notification (gated on the client's level) and close out
// progress with the discovered file count as both value and total.
// `LoggingLevel` is deprecated by SEP-2577 with no replacement yet; allow until we migrate.
#[allow(deprecated)]
super::notifications::emit_log(
peer,
&state.log_level,
rmcp::model::LoggingLevel::Info,
"basemind.rescan",
serde_json::json!({
"event": "rescan_complete",
"scanned": report.stats.scanned,
"updated": report.stats.updated,
"removed": report.stats.removed,
"extract_failed": report.stats.extract_failed,
"elapsed_ms": started.elapsed().as_millis() as u64,
}),
)
.await;
if let Some(token) = progress_token {
let scanned = report.stats.scanned as f64;
super::notifications::emit_progress(
peer,
token,
scanned,
Some(scanned),
format!("rescan: done, {} files", report.stats.scanned),
)
.await;
}
json_result(&super::types::RescanResponse {
scanned: report.stats.scanned,
updated: report.stats.updated,
removed: report.stats.removed,
skipped_unchanged: report.stats.skipped_unchanged,
skipped_no_lang: report.stats.skipped_no_lang,
extract_failed: report.stats.extract_failed,
elapsed_ms: started.elapsed().as_millis(),
root,
})
}