1use std::{
2 collections::VecDeque,
3 path::{Path, PathBuf},
4 sync::{Arc, OnceLock},
5 time::{Duration, Instant, SystemTime, UNIX_EPOCH},
6};
7
8use tokio::sync::{Semaphore, oneshot};
9
10use crate::{
11 api::{
12 ApiError, ApiMetadata, FileIndexFreshnessState, FileIndexRequest, FileIndexResponse,
13 FileQueryRequest, FileQueryResponse, RequestContext,
14 },
15 domain::{FreshnessPolicy, GraphVersion},
16 storage::{
17 FileIndexEntry, FileIndexRoot, FileIndexRootUpdate, FileIndexScanSummary,
18 FileSearchRequest, StorageError,
19 },
20};
21
22use crate::application::{FileIndexRootConfig, service::RelayKnowledgeService};
23
24use super::file_freshness::{FileFreshnessContext, file_freshness_diagnostics};
25
26pub const DEFAULT_FILE_QUERY_LIMIT: usize = 20;
27const MAX_FILE_QUERY_LIMIT: usize = 500;
28const MAX_CONCURRENT_FILE_SCANS: usize = 4;
29static FILE_SCAN_LIMITER: OnceLock<Arc<Semaphore>> = OnceLock::new();
30
31#[derive(Clone)]
32struct ScanBudget {
33 max_depth: usize,
34 max_file_bytes: u64,
35 max_files_per_root: usize,
36 excludes: Vec<String>,
37}
38
39impl RelayKnowledgeService {
40 pub async fn index_files(
42 &self,
43 request: FileIndexRequest,
44 context: RequestContext,
45 ) -> Result<FileIndexResponse, ApiError> {
46 let configured_scan = request.roots.is_empty();
47 let roots = self
48 .file_index_roots_from_request(request)
49 .map_err(ApiError::invalid_argument)?;
50 let active_roots = roots
51 .iter()
52 .map(file_index_root_from_config)
53 .collect::<Vec<_>>();
54 let store = self.storage.get().await.map_err(storage_api_error)?;
55 let now_ms = current_time_millis();
56 let updates = scan_roots(
57 roots,
58 ScanBudget {
59 max_depth: self.runtime.file_index.max_depth,
60 max_file_bytes: self.runtime.file_index.max_file_bytes,
61 max_files_per_root: self.runtime.file_index.max_files_per_root,
62 excludes: self.runtime.file_index.excludes.clone(),
63 },
64 now_ms,
65 self.runtime.file_index.scan_timeout,
66 )
67 .await
68 .map_err(storage_api_error)?;
69 let mut summary = FileIndexScanSummary::default();
70 for update in updates {
71 let status = store
72 .replace_file_index_root(update)
73 .await
74 .map_err(storage_api_error)?;
75 summary.root_count = summary.root_count.saturating_add(1);
76 summary.indexed_file_count = summary
77 .indexed_file_count
78 .saturating_add(status.indexed_file_count);
79 summary.missing_file_count = summary
80 .missing_file_count
81 .saturating_add(status.missing_file_count);
82 summary.scan_error_count = summary
83 .scan_error_count
84 .saturating_add(status.scan_error_count);
85 if status.truncated {
86 summary.truncated_root_count = summary.truncated_root_count.saturating_add(1);
87 }
88 summary.roots.push(status);
89 }
90 if configured_scan {
91 let diagnostics = store
92 .mark_file_index_roots_unconfigured(active_roots, now_ms)
93 .await
94 .map_err(storage_api_error)?;
95 summary = summary_from_diagnostics(diagnostics);
96 }
97
98 Ok(FileIndexResponse {
99 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
100 summary,
101 })
102 }
103
104 pub async fn index_configured_files_once(&self) -> Result<FileIndexResponse, ApiError> {
106 if self.runtime.file_index.roots.is_empty() {
107 let store = self.storage.get().await.map_err(storage_api_error)?;
108 let diagnostics = store
109 .mark_file_index_roots_unconfigured(Vec::new(), current_time_millis())
110 .await
111 .map_err(storage_api_error)?;
112 return Ok(FileIndexResponse {
113 metadata: ApiMetadata::graph_only(
114 &RequestContext::for_interface(crate::api::InterfaceKind::Cli),
115 GraphVersion::ZERO,
116 ),
117 summary: summary_from_diagnostics(diagnostics),
118 });
119 }
120
121 self.index_files(
122 FileIndexRequest {
123 source_scope: None,
124 roots: Vec::new(),
125 },
126 RequestContext::for_interface(crate::api::InterfaceKind::Cli),
127 )
128 .await
129 }
130
131 pub async fn query_files(
133 &self,
134 request: FileQueryRequest,
135 context: RequestContext,
136 ) -> Result<FileQueryResponse, ApiError> {
137 let query = required_query(request.query).map_err(ApiError::invalid_argument)?;
138 let limit = bounded_limit(request.limit).map_err(ApiError::invalid_argument)?;
139 let store = self.storage.get().await.map_err(storage_api_error)?;
140 let started = Instant::now();
141 let source_scope =
142 normalize_optional_text(request.source_scope).map_err(ApiError::invalid_argument)?;
143 let root_id =
144 normalize_optional_text(request.root_id).map_err(ApiError::invalid_argument)?;
145 let configured_roots = self
146 .runtime
147 .file_index
148 .roots
149 .iter()
150 .map(file_index_root_from_config)
151 .collect::<Vec<_>>();
152 let diagnostics = store
153 .file_index_diagnostics()
154 .await
155 .map_err(storage_api_error)?;
156 if request.freshness_policy == FreshnessPolicy::GraphOnly {
157 let degraded_reason = "graph_only freshness policy selected".to_owned();
158 let freshness = file_freshness_diagnostics(FileFreshnessContext {
159 file_index_enabled: self.runtime.file_index.enabled,
160 configured_roots: &configured_roots,
161 diagnostics: &diagnostics,
162 freshness_policy: request.freshness_policy,
163 source_scope: source_scope.clone(),
164 root_id: root_id.clone(),
165 graph_version: GraphVersion::ZERO.get(),
166 query_degraded_reason: Some(degraded_reason.clone()),
167 returned_hits: &[],
168 });
169 return Ok(FileQueryResponse {
170 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
171 query,
172 source_scope,
173 root_id,
174 freshness,
175 results: Vec::new(),
176 truncated: false,
177 duration_ms: elapsed_ms(started),
178 degraded_reason: Some(degraded_reason),
179 });
180 }
181 let freshness = file_freshness_diagnostics(FileFreshnessContext {
182 file_index_enabled: self.runtime.file_index.enabled,
183 configured_roots: &configured_roots,
184 diagnostics: &diagnostics,
185 freshness_policy: request.freshness_policy,
186 source_scope: source_scope.clone(),
187 root_id: root_id.clone(),
188 graph_version: GraphVersion::ZERO.get(),
189 query_degraded_reason: None,
190 returned_hits: &[],
191 });
192 if request.freshness_policy == FreshnessPolicy::WaitUntilFresh
193 && freshness.state != FileIndexFreshnessState::Fresh
194 {
195 return Err(ApiError::invalid_argument(format!(
196 "file index is {}; run files index before querying with wait_until_fresh",
197 file_freshness_state_label(freshness.state)
198 )));
199 }
200 let results = match store
201 .search_files(FileSearchRequest {
202 query: query.clone(),
203 source_scope: source_scope.clone(),
204 root_id: root_id.clone(),
205 limit: limit.saturating_add(1),
206 timeout_ms: query_timeout_ms(self.runtime.file_index.query_timeout),
207 })
208 .await
209 {
210 Ok(results) => results,
211 Err(error) if storage_error_timed_out(&error) => {
212 let degraded_reason = "file query timed out".to_owned();
213 let freshness = file_freshness_diagnostics(FileFreshnessContext {
214 file_index_enabled: self.runtime.file_index.enabled,
215 configured_roots: &configured_roots,
216 diagnostics: &diagnostics,
217 freshness_policy: request.freshness_policy,
218 source_scope: source_scope.clone(),
219 root_id: root_id.clone(),
220 graph_version: GraphVersion::ZERO.get(),
221 query_degraded_reason: Some(degraded_reason.clone()),
222 returned_hits: &[],
223 });
224 return Ok(FileQueryResponse {
225 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
226 query,
227 source_scope,
228 root_id,
229 freshness,
230 results: Vec::new(),
231 truncated: false,
232 duration_ms: elapsed_ms(started),
233 degraded_reason: Some(degraded_reason),
234 });
235 }
236 Err(error) => return Err(storage_api_error(error)),
237 };
238 let mut results = results;
239 let truncated = results.len() > limit;
240 results.truncate(limit);
241 let freshness = file_freshness_diagnostics(FileFreshnessContext {
242 file_index_enabled: self.runtime.file_index.enabled,
243 configured_roots: &configured_roots,
244 diagnostics: &diagnostics,
245 freshness_policy: request.freshness_policy,
246 source_scope: source_scope.clone(),
247 root_id: root_id.clone(),
248 graph_version: GraphVersion::ZERO.get(),
249 query_degraded_reason: None,
250 returned_hits: &results,
251 });
252
253 Ok(FileQueryResponse {
254 metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
255 query,
256 source_scope,
257 root_id,
258 freshness,
259 results,
260 truncated,
261 duration_ms: elapsed_ms(started),
262 degraded_reason: None,
263 })
264 }
265
266 fn file_index_roots_from_request(
267 &self,
268 request: FileIndexRequest,
269 ) -> Result<Vec<FileIndexRootConfig>, String> {
270 if request.roots.is_empty() {
271 if self.runtime.file_index.roots.is_empty() {
272 return Err("no file index roots are configured".to_owned());
273 }
274 return Ok(self.runtime.file_index.roots.clone());
275 }
276
277 let scope_id = normalize_optional_text(request.source_scope)?
278 .unwrap_or_else(|| "local-files".to_owned());
279 if self.runtime.file_index.roots.is_empty() {
280 return Err(
281 "file index roots must be configured before explicit roots can be scanned"
282 .to_owned(),
283 );
284 }
285 let mut roots = request
286 .roots
287 .into_iter()
288 .map(|root| {
289 let root = root.trim();
290 if root.is_empty() {
291 Err("file index root must not be empty".to_owned())
292 } else {
293 let root_path = PathBuf::from(root);
294 if !root_path.is_absolute() {
295 return Err("file index root must be an absolute path".to_owned());
296 }
297 let requested = FileIndexRootConfig::new(&scope_id, root_path);
298 self.runtime
299 .file_index
300 .roots
301 .iter()
302 .find(|authorized| {
303 authorized.scope_id == requested.scope_id
304 && authorized.root_id == requested.root_id
305 })
306 .cloned()
307 .ok_or_else(|| {
308 format!(
309 "file index root '{root}' is not configured for scope '{scope_id}'"
310 )
311 })
312 }
313 })
314 .collect::<Result<Vec<_>, _>>()?;
315 roots.sort_by(|left, right| {
316 left.scope_id
317 .cmp(&right.scope_id)
318 .then(left.root_id.cmp(&right.root_id))
319 });
320 roots.dedup_by(|left, right| {
321 left.scope_id == right.scope_id && left.root_id == right.root_id
322 });
323
324 Ok(roots)
325 }
326}
327
328async fn scan_roots(
329 roots: Vec<FileIndexRootConfig>,
330 budget: ScanBudget,
331 now_ms: u64,
332 scan_timeout: Duration,
333) -> Result<Vec<FileIndexRootUpdate>, StorageError> {
334 let mut updates = Vec::with_capacity(roots.len());
335 for root in roots {
336 updates.push(scan_root_with_timeout(root, budget.clone(), now_ms, scan_timeout).await?);
337 }
338
339 Ok(updates)
340}
341
342async fn scan_root_with_timeout(
343 root: FileIndexRootConfig,
344 budget: ScanBudget,
345 now_ms: u64,
346 scan_timeout: Duration,
347) -> Result<FileIndexRootUpdate, StorageError> {
348 if scan_timeout.is_zero() {
349 return Ok(timed_out_file_index_root_update(root, now_ms));
350 }
351 let permit = match file_scan_limiter().try_acquire_owned() {
352 Ok(permit) => permit,
353 Err(_) => return Ok(scan_worker_busy_file_index_root_update(root, now_ms)),
354 };
355 let timeout_root = root.clone();
356 let (sender, receiver) = oneshot::channel();
357 std::thread::Builder::new()
358 .name("relay-file-index-scan".to_owned())
359 .spawn(move || {
360 let _permit = permit;
361 let _ = sender.send(scan_root(root, &budget, now_ms));
362 })?;
363
364 match tokio::time::timeout(scan_timeout, receiver).await {
365 Ok(Ok(result)) => result,
366 Ok(Err(_)) => Err(StorageError::InvalidInput(
367 "file index scan worker stopped before reporting".to_owned(),
368 )),
369 Err(_) => Ok(timed_out_file_index_root_update(timeout_root, now_ms)),
370 }
371}
372
373fn file_scan_limiter() -> Arc<Semaphore> {
374 Arc::clone(
375 FILE_SCAN_LIMITER.get_or_init(|| Arc::new(Semaphore::new(MAX_CONCURRENT_FILE_SCANS))),
376 )
377}
378
379fn scan_worker_busy_file_index_root_update(
380 root: FileIndexRootConfig,
381 now_ms: u64,
382) -> FileIndexRootUpdate {
383 FileIndexRootUpdate {
384 root: storage_root(root.scope_id, root.root_id, &root.root_path),
385 entries: Vec::new(),
386 scan_error_count: 1,
387 truncated: true,
388 last_error: Some("file index scan worker is still busy".to_owned()),
389 now_ms,
390 }
391}
392
393fn timed_out_file_index_root_update(root: FileIndexRootConfig, now_ms: u64) -> FileIndexRootUpdate {
394 FileIndexRootUpdate {
395 root: storage_root(root.scope_id, root.root_id, &root.root_path),
396 entries: Vec::new(),
397 scan_error_count: 1,
398 truncated: true,
399 last_error: Some("file index scan timed out".to_owned()),
400 now_ms,
401 }
402}
403
404fn scan_root(
405 root: FileIndexRootConfig,
406 budget: &ScanBudget,
407 now_ms: u64,
408) -> Result<FileIndexRootUpdate, StorageError> {
409 let root_path = root.root_path;
410 let mut entries = Vec::new();
411 let mut scan_error_count = 0usize;
412 let mut truncated = false;
413 let mut last_error = None;
414 let canonical_root = match std::fs::canonicalize(&root_path) {
415 Ok(path) => path,
416 Err(error) => {
417 return Ok(FileIndexRootUpdate {
418 root: storage_root(root.scope_id, root.root_id, &root_path),
419 entries,
420 scan_error_count: 1,
421 truncated: false,
422 last_error: Some(error.to_string()),
423 now_ms,
424 });
425 }
426 };
427 let mut pending = VecDeque::from([(canonical_root.clone(), 0usize)]);
428
429 while let Some((directory, depth)) = pending.pop_front() {
430 if entries.len() >= budget.max_files_per_root {
431 truncated = true;
432 break;
433 }
434 if depth > budget.max_depth {
435 truncated = true;
436 continue;
437 }
438 let read_dir = match std::fs::read_dir(&directory) {
439 Ok(read_dir) => read_dir,
440 Err(error) => {
441 scan_error_count = scan_error_count.saturating_add(1);
442 last_error = Some(error.to_string());
443 continue;
444 }
445 };
446 for child in read_dir {
447 if entries.len() >= budget.max_files_per_root {
448 truncated = true;
449 pending.clear();
450 break;
451 }
452 let child = match child {
453 Ok(child) => child,
454 Err(error) => {
455 scan_error_count = scan_error_count.saturating_add(1);
456 last_error = Some(error.to_string());
457 continue;
458 }
459 };
460 let path = child.path();
461 if excluded(&path, &budget.excludes) {
462 continue;
463 }
464 let file_type = match child.file_type() {
465 Ok(file_type) => file_type,
466 Err(error) => {
467 scan_error_count = scan_error_count.saturating_add(1);
468 last_error = Some(error.to_string());
469 continue;
470 }
471 };
472 if file_type.is_symlink() {
473 continue;
474 }
475 if file_type.is_dir() {
476 pending.push_back((path, depth.saturating_add(1)));
477 continue;
478 }
479 let metadata = match child.metadata() {
480 Ok(metadata) => metadata,
481 Err(error) => {
482 scan_error_count = scan_error_count.saturating_add(1);
483 last_error = Some(error.to_string());
484 continue;
485 }
486 };
487 if file_type.is_file() && metadata.len() <= budget.max_file_bytes {
488 entries.push(file_entry(
489 &root.scope_id,
490 &root.root_id,
491 &canonical_root,
492 &path,
493 &metadata,
494 ));
495 }
496 }
497 }
498
499 Ok(FileIndexRootUpdate {
500 root: storage_root(root.scope_id, root.root_id, &canonical_root),
501 entries,
502 scan_error_count,
503 truncated,
504 last_error,
505 now_ms,
506 })
507}
508
509fn file_index_root_from_config(root: &FileIndexRootConfig) -> FileIndexRoot {
510 FileIndexRoot {
511 scope_id: root.scope_id.clone(),
512 root_id: root.root_id.clone(),
513 root_path: root.root_path.to_string_lossy().to_string(),
514 }
515}
516
517fn summary_from_diagnostics(
518 diagnostics: crate::storage::FileIndexDiagnostics,
519) -> FileIndexScanSummary {
520 FileIndexScanSummary {
521 root_count: diagnostics.root_count,
522 indexed_file_count: diagnostics.indexed_file_count,
523 missing_file_count: diagnostics.missing_file_count,
524 scan_error_count: diagnostics.scan_error_count,
525 truncated_root_count: diagnostics.truncated_root_count,
526 roots: diagnostics.roots,
527 }
528}
529
530fn file_entry(
531 scope_id: &str,
532 root_id: &str,
533 root: &Path,
534 path: &Path,
535 metadata: &std::fs::Metadata,
536) -> FileIndexEntry {
537 let relative_path = path.strip_prefix(root).unwrap_or(path);
538 let file_name = path
539 .file_name()
540 .map(|value| value.to_string_lossy().to_string())
541 .unwrap_or_default();
542 let extension = path
543 .extension()
544 .map(|value| value.to_string_lossy().to_ascii_lowercase());
545 let parent_dir = path
546 .parent()
547 .map(|value| value.to_string_lossy().to_string())
548 .unwrap_or_default();
549 let modified_at_ms = metadata
550 .modified()
551 .ok()
552 .and_then(system_time_millis)
553 .unwrap_or_default();
554
555 FileIndexEntry {
556 scope_id: scope_id.to_owned(),
557 root_id: root_id.to_owned(),
558 path: path.to_string_lossy().to_string(),
559 relative_path: relative_path.to_string_lossy().to_string(),
560 file_name,
561 extension,
562 parent_dir,
563 size_bytes: metadata.len(),
564 modified_at_ms,
565 fingerprint: format!("{}:{modified_at_ms}", metadata.len()),
566 }
567}
568
569fn storage_root(scope_id: String, root_id: String, root_path: &Path) -> FileIndexRoot {
570 FileIndexRoot {
571 scope_id,
572 root_id,
573 root_path: root_path.to_string_lossy().to_string(),
574 }
575}
576
577fn excluded(path: &Path, configured: &[String]) -> bool {
578 let Some(name) = path.file_name().map(|value| value.to_string_lossy()) else {
579 return false;
580 };
581 if name.starts_with('.') {
582 return true;
583 }
584 let lower = name.to_ascii_lowercase();
585 matches!(
586 lower.as_str(),
587 "target" | "node_modules" | ".git" | "__pycache__" | "tmp" | "temp" | "cache"
588 ) || configured
589 .iter()
590 .any(|pattern| lower.contains(&pattern.to_ascii_lowercase()))
591}
592
593fn required_query(query: String) -> Result<String, String> {
594 let query = query.trim().to_owned();
595 if query.is_empty() {
596 Err("file query must not be empty".to_owned())
597 } else {
598 Ok(query)
599 }
600}
601
602fn bounded_limit(limit: usize) -> Result<usize, String> {
603 match limit {
604 0 => Err("file query limit must be greater than zero".to_owned()),
605 value if value > MAX_FILE_QUERY_LIMIT => Err(format!(
606 "file query limit must not exceed {MAX_FILE_QUERY_LIMIT}"
607 )),
608 value => Ok(value),
609 }
610}
611
612fn normalize_optional_text(value: Option<String>) -> Result<Option<String>, String> {
613 value
614 .map(|value| {
615 let value = value.trim().to_owned();
616 if value.is_empty() {
617 Err("optional file query filter must not be empty".to_owned())
618 } else {
619 Ok(value)
620 }
621 })
622 .transpose()
623}
624
625fn current_time_millis() -> u64 {
626 SystemTime::now()
627 .duration_since(UNIX_EPOCH)
628 .map_or(0, |duration| {
629 u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
630 })
631}
632
633fn system_time_millis(time: SystemTime) -> Option<u64> {
634 time.duration_since(UNIX_EPOCH)
635 .ok()
636 .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
637}
638
639fn elapsed_ms(started: Instant) -> u64 {
640 u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
641}
642
643fn query_timeout_ms(timeout: std::time::Duration) -> u64 {
644 u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX)
645}
646
647fn storage_error_timed_out(error: &StorageError) -> bool {
648 matches!(error, StorageError::InvalidInput(message) if message.contains("file query timed out"))
649}
650
651fn file_freshness_state_label(state: FileIndexFreshnessState) -> &'static str {
652 match state {
653 FileIndexFreshnessState::Fresh => "fresh",
654 FileIndexFreshnessState::Pending => "pending",
655 FileIndexFreshnessState::Paused => "paused",
656 FileIndexFreshnessState::Stale => "stale",
657 FileIndexFreshnessState::Degraded => "degraded",
658 FileIndexFreshnessState::Overflow => "overflow",
659 }
660}
661
662fn storage_api_error(error: StorageError) -> ApiError {
663 ApiError::storage_unavailable(error.to_string())
664}
665
666#[cfg(test)]
667mod tests {
668 use std::{
669 fs,
670 path::{Path, PathBuf},
671 time::{SystemTime, UNIX_EPOCH},
672 };
673
674 use crate::{
675 application::RuntimeConfiguration,
676 env::{EnvironmentConfig, PlatformKind},
677 storage::{KnowledgeStore, SqliteGraphStore},
678 };
679
680 use super::*;
681
682 #[tokio::test]
683 async fn scan_roots_respects_budget_excludes_and_metadata() {
684 let fixture = TempFixture::new("scan-budget");
685 fixture.write("docs/report.pdf", "pdf");
686 fixture.write("target/generated.txt", "generated");
687 fixture.write(".hidden/secret.txt", "secret");
688 fixture.write("deep/a/b/c/too-deep.txt", "deep");
689 fixture.write("large.bin", "too large for budget");
690 fixture.write("notes/skipme.txt", "configured exclusion");
691 #[cfg(unix)]
692 std::os::unix::fs::symlink("/", fixture.path().join("escape"))
693 .expect("symlink fixture should be created");
694
695 let updates = scan_roots(
696 vec![FileIndexRootConfig::new(
697 "local-files",
698 fixture.path().to_path_buf(),
699 )],
700 ScanBudget {
701 max_depth: 2,
702 max_file_bytes: 8,
703 max_files_per_root: 10,
704 excludes: vec!["skipme".to_owned()],
705 },
706 42,
707 Duration::from_secs(30),
708 )
709 .await
710 .expect("scan should complete");
711
712 let update = updates.into_iter().next().expect("one root is scanned");
713 assert_eq!(update.root.scope_id, "local-files");
714 assert_eq!(update.now_ms, 42);
715 assert!(update.truncated);
716 assert_eq!(update.scan_error_count, 0);
717 assert_eq!(update.entries.len(), 1);
718 let entry = &update.entries[0];
719 assert_eq!(entry.file_name, "report.pdf");
720 assert_eq!(entry.extension.as_deref(), Some("pdf"));
721 assert!(entry.relative_path.ends_with("docs/report.pdf"));
722 assert!(entry.parent_dir.ends_with("docs"));
723 assert_eq!(entry.size_bytes, 3);
724 assert!(entry.fingerprint.starts_with("3:"));
725 }
726
727 #[tokio::test]
728 async fn scan_roots_reports_missing_roots_and_file_count_truncation() {
729 let fixture = TempFixture::new("scan-truncated");
730 fixture.write("first.txt", "one");
731 fixture.write("second.txt", "two");
732 let missing = fixture.path().join("missing");
733
734 let updates = scan_roots(
735 vec![
736 FileIndexRootConfig::new("local-files", fixture.path().to_path_buf()),
737 FileIndexRootConfig::new("local-files", missing),
738 ],
739 ScanBudget {
740 max_depth: 4,
741 max_file_bytes: 128,
742 max_files_per_root: 1,
743 excludes: Vec::new(),
744 },
745 7,
746 Duration::from_secs(30),
747 )
748 .await
749 .expect("scan should complete");
750
751 let truncated = updates
752 .iter()
753 .find(|update| update.root.root_path == fixture.path().to_string_lossy())
754 .expect("fixture root should be present");
755 assert!(truncated.truncated);
756 assert_eq!(truncated.entries.len(), 1);
757
758 let missing = updates
759 .iter()
760 .find(|update| update.root.root_path.ends_with("missing"))
761 .expect("missing root should be reported");
762 assert_eq!(missing.scan_error_count, 1);
763 assert!(missing.entries.is_empty());
764 assert!(missing.last_error.is_some());
765 }
766
767 #[tokio::test]
768 async fn scan_timeout_returns_degraded_root_update() {
769 let fixture = TempFixture::new("scan-timeout");
770
771 let update = scan_root_with_timeout(
772 FileIndexRootConfig::new("local-files", fixture.path().to_path_buf()),
773 ScanBudget {
774 max_depth: 4,
775 max_file_bytes: 128,
776 max_files_per_root: 1,
777 excludes: Vec::new(),
778 },
779 9,
780 Duration::ZERO,
781 )
782 .await
783 .expect("timeout update should be produced");
784
785 assert_eq!(update.scan_error_count, 1);
786 assert!(update.truncated);
787 assert!(update.entries.is_empty());
788 assert_eq!(
789 update.last_error.as_deref(),
790 Some("file index scan timed out")
791 );
792 }
793
794 #[test]
795 fn scan_worker_busy_update_reports_bounded_backpressure() {
796 let update = scan_worker_busy_file_index_root_update(
797 FileIndexRootConfig::new("local-files", PathBuf::from("/opt/docs")),
798 11,
799 );
800
801 assert_eq!(update.scan_error_count, 1);
802 assert!(update.truncated);
803 assert!(update.entries.is_empty());
804 assert_eq!(
805 update.last_error.as_deref(),
806 Some("file index scan worker is still busy")
807 );
808 assert_eq!(update.now_ms, 11);
809 }
810
811 #[test]
812 fn query_validation_helpers_reject_unbounded_inputs() {
813 assert_eq!(required_query(" quarter ".to_owned()).unwrap(), "quarter");
814 assert!(required_query(" \t ".to_owned()).is_err());
815 assert_eq!(bounded_limit(1).unwrap(), 1);
816 assert!(bounded_limit(0).is_err());
817 assert!(bounded_limit(MAX_FILE_QUERY_LIMIT + 1).is_err());
818 assert_eq!(
819 normalize_optional_text(Some(" root ".to_owned())).unwrap(),
820 Some("root".to_owned())
821 );
822 assert!(normalize_optional_text(Some(" ".to_owned())).is_err());
823 assert_eq!(normalize_optional_text(None).unwrap(), None);
824 }
825
826 #[test]
827 fn query_timeout_helpers_map_runtime_budget_and_storage_errors() {
828 assert_eq!(query_timeout_ms(std::time::Duration::from_millis(125)), 125);
829 assert!(storage_error_timed_out(&StorageError::InvalidInput(
830 "file query timed out waiting for storage lock".to_owned()
831 )));
832 assert!(!storage_error_timed_out(&StorageError::InvalidInput(
833 "different validation failure".to_owned()
834 )));
835 }
836
837 #[tokio::test]
838 async fn explicit_roots_must_match_authorized_runtime_roots() {
839 let fixture = TempFixture::new("authorized-roots");
840 let service = service_for_root(fixture.path()).await;
841 let authorized = service
842 .file_index_roots_from_request(FileIndexRequest {
843 source_scope: Some("local-files".to_owned()),
844 roots: vec![fixture.path().join(".").to_string_lossy().to_string()],
845 })
846 .expect("configured root spelling should be authorized");
847 assert_eq!(authorized.len(), 1);
848
849 let denied = service
850 .file_index_roots_from_request(FileIndexRequest {
851 source_scope: Some("local-files".to_owned()),
852 roots: vec![fixture.path().join("other").to_string_lossy().to_string()],
853 })
854 .expect_err("unconfigured root should be denied");
855 assert!(denied.contains("is not configured"));
856
857 let relative = service
858 .file_index_roots_from_request(FileIndexRequest {
859 source_scope: Some("local-files".to_owned()),
860 roots: vec!["relative/docs".to_owned()],
861 })
862 .expect_err("relative roots should be denied");
863 assert!(relative.contains("absolute path"));
864 }
865
866 #[tokio::test]
867 async fn same_path_roots_remain_distinct_across_scopes() {
868 let fixture = TempFixture::new("scope-roots");
869 let home = fixture.path().join("home");
870 let documents = home.join("Documents");
871 fs::create_dir_all(&documents).expect("documents directory should be created");
872 let environment = EnvironmentConfig::from_pairs(
873 PlatformKind::Unix,
874 [
875 ("HOME", home.to_string_lossy().to_string()),
876 ("TMPDIR", "/tmp".to_owned()),
877 (
878 "RELAY_KNOWLEDGE_FILE_INDEX_ROOTS",
879 documents.to_string_lossy().to_string(),
880 ),
881 (
882 "RELAY_KNOWLEDGE_FILE_INDEX_SCAN_TIMEOUT_MS",
883 "120000".to_owned(),
884 ),
885 ],
886 )
887 .expect("environment should parse");
888 let runtime = RuntimeConfiguration::from_environment(&environment)
889 .await
890 .expect("runtime should compose");
891 assert_eq!(runtime.file_index.scan_timeout, Duration::from_secs(120));
892
893 let matching_roots = runtime
894 .file_index
895 .roots
896 .iter()
897 .filter(|root| root.root_path.as_path() == documents.as_path())
898 .collect::<Vec<_>>();
899 assert_eq!(matching_roots.len(), 2);
900 assert_ne!(matching_roots[0].scope_id, matching_roots[1].scope_id);
901 }
902
903 struct TempFixture {
904 root: PathBuf,
905 }
906
907 impl TempFixture {
908 fn new(name: &str) -> Self {
909 let suffix = SystemTime::now()
910 .duration_since(UNIX_EPOCH)
911 .expect("time should be valid")
912 .as_nanos();
913 let root = std::env::temp_dir().join(format!(
914 "relay-knowledge-{name}-{}-{suffix}",
915 std::process::id()
916 ));
917 fs::create_dir_all(&root).expect("fixture root should be created");
918
919 Self { root }
920 }
921
922 fn path(&self) -> &Path {
923 &self.root
924 }
925
926 fn write(&self, relative: &str, content: &str) {
927 let path = self.root.join(relative);
928 if let Some(parent) = path.parent() {
929 fs::create_dir_all(parent).expect("fixture parent should be created");
930 }
931 fs::write(path, content).expect("fixture file should be written");
932 }
933 }
934
935 impl Drop for TempFixture {
936 fn drop(&mut self) {
937 let _ = fs::remove_dir_all(&self.root);
938 }
939 }
940
941 async fn service_for_root(root: &Path) -> RelayKnowledgeService {
942 let home = root.join("home");
943 fs::create_dir_all(&home).expect("home should be created");
944 let relay_home = root.join("relay");
945 let environment = EnvironmentConfig::from_pairs(
946 PlatformKind::Unix,
947 [
948 ("HOME", home.to_string_lossy().to_string()),
949 ("TMPDIR", "/tmp".to_owned()),
950 (
951 "RELAY_KNOWLEDGE_HOME",
952 relay_home.to_string_lossy().to_string(),
953 ),
954 (
955 "RELAY_KNOWLEDGE_FILE_INDEX_ROOTS",
956 root.to_string_lossy().to_string(),
957 ),
958 ],
959 )
960 .expect("environment should parse");
961 let runtime = RuntimeConfiguration::from_environment(&environment)
962 .await
963 .expect("runtime should compose");
964 let store = Arc::new(SqliteGraphStore::open_in_memory().expect("store should open"))
965 as Arc<dyn KnowledgeStore>;
966
967 RelayKnowledgeService::with_store(runtime, store)
968 }
969}