use std::collections::HashMap;
use std::path::Path;
use std::time::UNIX_EPOCH;
use crate::adapters::lance::LanceStore;
use crate::app::cli::{CorpusReport, IndexReport};
use crate::app::config::{Config, ResolvedLayers, resolve_for_cwd};
use crate::domain::common::{CorpusConfig, FileRef, Mtime, canonicalize_or_passthrough};
#[cfg(test)]
use crate::domain::corpus::sandbox::FileEntry;
use crate::domain::corpus::sandbox::{
WriteError, WriteErrorKind, atomic_write_no_follow, delete_no_follow,
ensure_corpus_allows_file, first_corpus_root, list_corpus_files, pick_corpus, read_no_follow,
resolve_read_root, safe_relative_path,
};
use crate::domain::corpus::{MarkdownChunker, blake3_file};
use crate::domain::ground::{
Format, GroundOpts, RenderOpts, Warning, ground, ground_union, render, trim_snippets,
};
use crate::domain::indexer::{DEFAULT_BATCH_SIZE, FileSnapshot, apply, index_corpus, plan};
#[cfg(test)]
use crate::domain::repository::{RepoCorpusKind, repo_corpus_name};
use crate::domain::repository::{RepositoryConfig, default_wiki_for_cwd};
use super::ipc::{
AddMarkdownRequest, AddMarkdownResult, CorpusEntry, CorpusStatsResult, DaemonRequest,
DaemonRequestPayload, DaemonResponse, DeleteMarkdownRequest, DeleteMarkdownResult,
GroundRequest, GroundResult, IndexRequest, LineRange, ListFilesRequest, ListTreeRequest,
ListTreeResult, PongResult, Position, ReadMarkdownRequest, ReadMarkdownResult,
};
use super::state::DaemonState;
pub async fn dispatch(state: &DaemonState, req: DaemonRequest) -> DaemonResponse {
if let DaemonRequestPayload::Shutdown = req.payload {
state.shutdown_token().cancel();
return DaemonResponse::ok(&"stopping");
}
if let DaemonRequestPayload::Ping = req.payload {
return DaemonResponse::ok(&PongResult {
version: env!("CARGO_PKG_VERSION").to_string(),
});
}
let req_cwd = req.cwd.clone();
let (effective, layers) =
match resolve_for_cwd(state.baseline(), &req.cwd, state.baseline_xdg_path()) {
Ok(resolved) => resolved,
Err(e) => return DaemonResponse::invalid_params(e.to_string()),
};
match req.payload {
DaemonRequestPayload::Ping => {
DaemonResponse::ok(&PongResult {
version: env!("CARGO_PKG_VERSION").to_string(),
})
}
DaemonRequestPayload::Ground(req) => {
handle_ground(state, &effective, &layers, &req_cwd, req).await
}
DaemonRequestPayload::Index(req) => handle_index(state, &effective, req).await,
DaemonRequestPayload::ListCorpora => handle_list_corpora(&effective),
DaemonRequestPayload::ListFiles(req) => handle_list_files(&effective, &req_cwd, req),
DaemonRequestPayload::ListTree(req) => handle_list_tree(&effective, &req_cwd, req),
DaemonRequestPayload::AddMarkdown(req) => handle_add_markdown(state, &effective, req).await,
DaemonRequestPayload::ReadMarkdown(req) => {
handle_read_markdown(&effective, &req_cwd, req).await
}
DaemonRequestPayload::DeleteMarkdown(req) => {
handle_delete_markdown(state, &effective, req).await
}
DaemonRequestPayload::CorpusStats { corpus } => {
handle_corpus_stats(state, &effective, &req_cwd, corpus).await
}
DaemonRequestPayload::Shutdown => {
DaemonResponse::ok(&"stopping")
}
}
}
fn effective_corpora(cfg: &Config) -> Result<Vec<CorpusConfig>, DaemonResponse> {
cfg.effective_corpora()
.map_err(|e| DaemonResponse::internal(e.to_string()))
}
fn validate_wiki_path(
corpora: &[CorpusConfig],
corpus_name: &str,
path: &str,
) -> Result<(CorpusConfig, std::path::PathBuf, std::path::PathBuf), DaemonResponse> {
let corpus = pick_corpus(corpora, Some(corpus_name))
.map_err(|e| DaemonResponse::invalid_params(e.into_inner()))?;
let root = require_single_root(&corpus)?;
let relative =
safe_relative_path(path).map_err(|e| DaemonResponse::invalid_params(e.into_inner()))?;
let dest = root.join(&relative);
ensure_corpus_allows_file(&corpus, &dest)
.map_err(|e| DaemonResponse::invalid_params(e.into_inner()))?;
ensure_wiki_root_safe(&corpus).map_err(DaemonResponse::invalid_params)?;
Ok((corpus, root, relative))
}
fn validate_wiki_read_path(
corpora: &[CorpusConfig],
corpus_name: &str,
path: &str,
) -> Result<(CorpusConfig, std::path::PathBuf, std::path::PathBuf), DaemonResponse> {
let corpus = pick_corpus(corpora, Some(corpus_name))
.map_err(|e| DaemonResponse::invalid_params(e.into_inner()))?;
let relative =
safe_relative_path(path).map_err(|e| DaemonResponse::invalid_params(e.into_inner()))?;
let root = resolve_read_root(&corpus, &relative)
.map_err(|WriteError { kind, source }| map_read_error(kind, source, &relative))?;
let dest = root.join(&relative);
ensure_corpus_allows_file(&corpus, &dest)
.map_err(|e| DaemonResponse::invalid_params(e.into_inner()))?;
ensure_wiki_root_safe(&corpus).map_err(DaemonResponse::invalid_params)?;
Ok((corpus, root, relative))
}
fn require_single_root(corpus: &CorpusConfig) -> Result<std::path::PathBuf, DaemonResponse> {
if corpus.paths.len() == 1 {
return first_corpus_root(corpus)
.map_err(|e| DaemonResponse::invalid_params(e.into_inner()));
}
Err(DaemonResponse::invalid_params(format!(
"corpus {:?} has {} roots; mutations (add/delete) require a \
single-root corpus — multi-root corpora are read- and search-only",
corpus.name,
corpus.paths.len(),
)))
}
fn pick_corpus_or_default(
corpora: &[CorpusConfig],
repositories: &[RepositoryConfig],
cwd: &Path,
requested: Option<&str>,
) -> Result<CorpusConfig, crate::domain::corpus::sandbox::SandboxError> {
if requested.is_none()
&& let Some(name) = default_wiki_for_cwd(repositories, cwd)
&& let Some(found) = corpora.iter().find(|c| c.name == name).cloned()
{
return Ok(found);
}
pick_corpus(corpora, requested)
}
fn handle_list_corpora(cfg: &Config) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let entries: Vec<CorpusEntry> = corpora
.into_iter()
.map(|c| CorpusEntry {
name: c.name,
paths: c.paths,
})
.collect();
DaemonResponse::ok(&entries)
}
fn handle_list_files(cfg: &Config, cwd: &Path, req: ListFilesRequest) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let corpus =
match pick_corpus_or_default(&corpora, &cfg.repositories, cwd, req.corpus.as_deref()) {
Ok(c) => c,
Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
};
ensure_paths_exist(&corpus);
match list_corpus_files(&corpus) {
Ok(entries) => DaemonResponse::ok(&entries),
Err(e) => DaemonResponse::internal(e.to_string()),
}
}
fn handle_list_tree(cfg: &Config, cwd: &Path, req: ListTreeRequest) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let corpus =
match pick_corpus_or_default(&corpora, &cfg.repositories, cwd, req.corpus.as_deref()) {
Ok(c) => c,
Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
};
ensure_paths_exist(&corpus);
let root = match crate::domain::corpus::sandbox::build_corpus_tree(&corpus) {
Ok(node) => node,
Err(e) => return DaemonResponse::internal(e.to_string()),
};
DaemonResponse::ok(&ListTreeResult {
corpus: corpus.name,
root,
})
}
async fn handle_corpus_stats(
state: &DaemonState,
cfg: &Config,
cwd: &Path,
corpus: Option<String>,
) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let corpus_cfg =
match pick_corpus_or_default(&corpora, &cfg.repositories, cwd, corpus.as_deref()) {
Ok(c) => c,
Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
};
let store = state.store();
let chunk_stats = match store.corpus_chunk_stats(&corpus_cfg.name).await {
Ok(s) => s,
Err(e) => return DaemonResponse::internal(e.to_string()),
};
ensure_paths_exist(&corpus_cfg);
let disk_files = match list_corpus_files(&corpus_cfg) {
Ok(f) => f,
Err(e) => return DaemonResponse::internal(e.to_string()),
};
let indexed_map = match store.list_files(&corpus_cfg.name).await {
Ok(m) => m,
Err(e) => return DaemonResponse::internal(e.to_string()),
};
let indexed_paths: std::collections::HashSet<String> = indexed_map
.keys()
.map(|r| r.as_path().to_string_lossy().into_owned())
.collect();
let unindexed_files = disk_files
.iter()
.filter(|e| !indexed_paths.contains(&e.absolute_path))
.count() as u64;
DaemonResponse::ok(&CorpusStatsResult {
corpus: corpus_cfg.name,
indexed_files: chunk_stats.indexed_files,
total_chunks: chunk_stats.total_chunks,
last_indexed_ms: chunk_stats.last_indexed_ms,
unindexed_files,
})
}
async fn handle_ground(
state: &DaemonState,
cfg: &Config,
layers: &ResolvedLayers,
cwd: &Path,
req: GroundRequest,
) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let store = state.store();
let opts = GroundOpts {
top_files: req.top_files.unwrap_or(cfg.search.top_files_default),
chunks_per_file: req
.chunks_per_file
.unwrap_or(cfg.search.chunks_per_file_default),
limit: req.limit.unwrap_or(50),
};
let union = req.corpus.is_none() && default_wiki_for_cwd(&cfg.repositories, cwd).is_none();
let single_corpus = if union {
None
} else {
match pick_corpus_or_default(&corpora, &cfg.repositories, cwd, req.corpus.as_deref()) {
Ok(c) => Some(c),
Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
}
};
let mut embedder = if state.embeddings_enabled() {
match state.embedder().await {
Ok(g) => Some(g),
Err(e) => return DaemonResponse::internal(e.to_string()),
}
} else {
None
};
let mut crossencoder = match state.crossencoder(cfg.search.crossencoder.as_deref()).await {
Ok(g) => g,
Err(e) => {
tracing::warn!(
target: "hallouminate::daemon",
error = %e,
"crossencoder unavailable for this request; falling back to fusion-only ranking",
);
None
}
};
let crossencoder_dyn: Option<&mut dyn crate::domain::search::Crossencoder> = crossencoder
.as_mut()
.map(|g| &mut **g as &mut dyn crate::domain::search::Crossencoder);
let embedder_dyn: Option<&mut dyn crate::domain::embeddings::EmbedBatch> = embedder
.as_mut()
.map(|g| &mut **g as &mut dyn crate::domain::embeddings::EmbedBatch);
let response = if let Some(corpus) = &single_corpus {
ground(
&req.query,
&corpus.name,
&corpus.paths,
&store,
embedder_dyn,
crossencoder_dyn,
opts,
)
.await
} else {
let targets: Vec<(String, Vec<String>)> = corpora
.iter()
.map(|c| (c.name.clone(), c.paths.clone()))
.collect();
ground_union(
&req.query,
&targets,
&store,
embedder_dyn,
crossencoder_dyn,
opts,
)
.await
};
let mut response = match response {
Ok(r) => r,
Err(e) => return DaemonResponse::internal(e.to_string()),
};
drop(crossencoder);
drop(embedder);
mark_stale(&mut response).await;
if union {
for w in &layers.warnings {
response.warnings.push(Warning {
code: "cross-repo-union".to_string(),
message: w.clone(),
});
}
}
let response = if let Some(limit) = req.snippet_chars {
trim_snippets(&response, limit)
} else {
response
};
let outline = render(
&response,
Format::Outline,
&RenderOpts {
snippet_chars: None,
path_prefix_strip: None,
},
);
DaemonResponse::ok(&GroundResult { outline, response })
}
async fn handle_index(state: &DaemonState, cfg: &Config, req: IndexRequest) -> DaemonResponse {
if req.paths_from.is_some() {
return DaemonResponse::invalid_params("paths_from is not supported via the daemon yet");
}
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let selected: Vec<CorpusConfig> = if let Some(name) = req.corpus.as_deref() {
match corpora.iter().find(|c| c.name == name) {
Some(c) => vec![c.clone()],
None => {
return DaemonResponse::invalid_params(format!(
"corpus {name:?} not found in config"
));
}
}
} else {
if corpora.is_empty() {
return DaemonResponse::invalid_params(
"no corpora configured; add [[corpus]] or [[repository]] to config",
);
}
corpora.clone()
};
let store = state.store();
let chunker = state.make_chunker();
let mut report = IndexReport::default();
for corpus in selected {
let guard = match state.acquire_mutation_guard(&corpus.name).await {
Ok(g) => g,
Err(msg) => return DaemonResponse::internal(msg),
};
ensure_paths_exist(&corpus);
let missing = crate::domain::corpus::missing_roots(&corpus);
if !missing.is_empty() {
let roots = missing
.iter()
.map(|p| p.display().to_string())
.collect::<Vec<_>>()
.join(", ");
if req.strict {
return DaemonResponse::invalid_params(format!(
"corpus {:?}: root {roots} does not exist",
corpus.name
));
}
report.warnings.push(format!(
"corpus {:?}: root {roots} does not exist; skipped",
corpus.name
));
continue;
}
let mut embedder = if state.embeddings_enabled() {
match state.embedder().await {
Ok(g) => Some(g),
Err(e) => return DaemonResponse::internal(e.to_string()),
}
} else {
None
};
let embedder_dyn: Option<&mut dyn crate::domain::embeddings::EmbedBatch> = embedder
.as_mut()
.map(|g| &mut **g as &mut dyn crate::domain::embeddings::EmbedBatch);
let stats = match index_corpus(&corpus, &store, embedder_dyn, &chunker).await {
Ok(s) => s,
Err(e) => return DaemonResponse::internal(e.to_string()),
};
drop(embedder);
drop(guard);
report.corpora.push(CorpusReport {
name: corpus.name.clone(),
files_upserted: stats.files_upserted,
files_touched: stats.files_touched,
files_deleted: stats.files_deleted,
files_skipped_empty: stats.files_skipped_empty,
chunks_inserted: stats.chunks_inserted,
embeddings_inserted: stats.embeddings_inserted,
});
}
DaemonResponse::ok(&report)
}
enum EditMode {
WholeFile,
UnderHeading(String, Position),
ReplaceLines(LineRange),
ReplaceMatch(String),
}
fn classify_edit_mode(req: &AddMarkdownRequest) -> Result<EditMode, DaemonResponse> {
let count = req.under_heading.is_some() as u8
+ req.replace_lines.is_some() as u8
+ req.replace_match.is_some() as u8;
if count > 1 {
return Err(DaemonResponse::invalid_params(
"set at most one of under_heading / replace_lines / replace_match",
));
}
if let Some(h) = &req.under_heading {
return Ok(EditMode::UnderHeading(h.clone(), req.position));
}
if let Some(r) = req.replace_lines {
return Ok(EditMode::ReplaceLines(r));
}
if let Some(n) = &req.replace_match {
return Ok(EditMode::ReplaceMatch(n.clone()));
}
Ok(EditMode::WholeFile)
}
async fn read_existing_text(
root: std::path::PathBuf,
rel: std::path::PathBuf,
mode_label: &str,
) -> Result<String, DaemonResponse> {
let rel_disp = rel.display().to_string();
let raw = match tokio::task::spawn_blocking(move || read_no_follow(&root, &rel)).await {
Ok(Ok(b)) => b,
Ok(Err(WriteError { kind, source })) => {
return Err(match kind {
WriteErrorKind::Io => {
if source.kind() == std::io::ErrorKind::NotFound {
DaemonResponse::invalid_params(format!(
"{mode_label} requires an existing file; {rel_disp} not found"
))
} else {
tracing::error!(
target: "hallouminate::daemon",
error = %source,
path = %rel_disp,
"read_existing_text io error",
);
DaemonResponse::internal(format!("failed to read {rel_disp}: {source}"))
}
}
WriteErrorKind::Symlink | WriteErrorKind::InvalidPath => {
DaemonResponse::invalid_params(format!("refusing unsafe path {rel_disp}"))
}
WriteErrorKind::Exists => {
tracing::error!(
target: "hallouminate::daemon",
path = %rel_disp,
"read_existing_text: unexpected Exists variant on read path",
);
DaemonResponse::internal("unexpected Exists on read")
}
});
}
Err(e) => {
tracing::error!(
target: "hallouminate::daemon",
error = %e,
"read_existing_text read task panicked",
);
return Err(DaemonResponse::internal(format!("read task panicked: {e}")));
}
};
String::from_utf8(raw)
.map_err(|_| DaemonResponse::invalid_params("existing file is not valid UTF-8".to_string()))
}
async fn handle_add_markdown(
state: &DaemonState,
cfg: &Config,
mut req: AddMarkdownRequest,
) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let (corpus, root, relative) = match validate_wiki_path(&corpora, &req.corpus, &req.path) {
Ok(t) => t,
Err(resp) => return resp,
};
let mode = match classify_edit_mode(&req) {
Ok(m) => m,
Err(resp) => return resp,
};
let guard = match state.acquire_mutation_guard(&corpus.name).await {
Ok(g) => g,
Err(msg) => return DaemonResponse::internal(msg),
};
let force_overwrite: bool;
match mode {
EditMode::WholeFile => {
force_overwrite = req.overwrite;
}
EditMode::UnderHeading(heading, position) => {
let existing =
match read_existing_text(root.clone(), relative.clone(), "under_heading").await {
Ok(s) => s,
Err(resp) => return resp,
};
req.content = match crate::domain::corpus::splice_under_heading(
&existing,
&heading,
position,
&req.content,
) {
Ok(s) => s,
Err(crate::domain::corpus::SectionError::NotFound) => {
return DaemonResponse::invalid_params(format!(
"heading '{heading}' not found in {}",
relative.display()
));
}
Err(crate::domain::corpus::SectionError::Duplicate) => {
return DaemonResponse::invalid_params(format!(
"heading '{heading}' is ambiguous in {}",
relative.display()
));
}
};
force_overwrite = true;
}
EditMode::ReplaceLines(range) => {
let existing =
match read_existing_text(root.clone(), relative.clone(), "replace_lines").await {
Ok(s) => s,
Err(resp) => return resp,
};
req.content =
match crate::domain::corpus::replace_line_range(&existing, range, &req.content) {
Ok(s) => s,
Err(crate::domain::corpus::RangeError::OutOfRange) => {
return DaemonResponse::invalid_params(
"line range out of range".to_string(),
);
}
Err(crate::domain::corpus::RangeError::Inverted) => {
return DaemonResponse::invalid_params("start > end".to_string());
}
};
force_overwrite = true;
}
EditMode::ReplaceMatch(needle) => {
let existing =
match read_existing_text(root.clone(), relative.clone(), "replace_match").await {
Ok(s) => s,
Err(resp) => return resp,
};
req.content =
match crate::domain::corpus::replace_unique_match(&existing, &needle, &req.content)
{
Ok(s) => s,
Err(crate::domain::corpus::MatchError::NotFound) => {
return DaemonResponse::invalid_params("match not found".to_string());
}
Err(crate::domain::corpus::MatchError::Ambiguous(n)) => {
return DaemonResponse::invalid_params(format!(
"match ambiguous \u{2014} {n} occurrences"
));
}
};
force_overwrite = true;
}
}
let mut warnings = crate::domain::corpus::lint_markdown(&req.content);
warnings.extend(crate::domain::corpus::lint_frontmatter(&req.content));
warnings.extend(crate::domain::corpus::lint_claim_marks(&req.content));
let write_root = root.clone();
let write_relative = relative.clone();
let error_relative = relative.clone();
let overwrite = force_overwrite;
let content_is_empty = req.content.trim().is_empty();
let content_bytes = req.content.into_bytes();
let written = tokio::task::spawn_blocking(move || {
atomic_write_no_follow(&write_root, &write_relative, &content_bytes, overwrite)
})
.await;
let dest = match written {
Ok(Ok(p)) => p,
Ok(Err(WriteError { kind, source })) => {
let resp = match kind {
WriteErrorKind::Exists => DaemonResponse::invalid_params(format!(
"{} already exists; pass overwrite=true to replace it",
error_relative.display()
)),
WriteErrorKind::Symlink | WriteErrorKind::InvalidPath => {
DaemonResponse::invalid_params(format!(
"refusing unsafe path {}: {source}",
error_relative.display()
))
}
WriteErrorKind::Io => DaemonResponse::internal(source.to_string()),
};
return resp;
}
Err(join_err) => {
tracing::error!(
target: "hallouminate::daemon",
error = %join_err,
"add_markdown write task panicked",
);
return DaemonResponse::internal(format!("write task panicked: {join_err}"));
}
};
let mut stats = if content_is_empty {
let mut stats = crate::domain::indexer::ApplyStats {
files_skipped_empty: 1,
..Default::default()
};
if overwrite {
let file_ref = canonicalize_or_passthrough(&dest);
if let Some(file_ref_str) = file_ref.as_path().to_str() {
let store = state.store();
match store.get_file_snapshot(&corpus.name, file_ref_str).await {
Ok(Some(_)) => match store.delete_file(&corpus.name, file_ref_str).await {
Ok(()) => stats.files_deleted = 1,
Err(e) => return DaemonResponse::internal(e.to_string()),
},
Ok(None) => {}
Err(e) => return DaemonResponse::internal(e.to_string()),
}
}
}
stats
} else {
let store = state.store();
let chunker = state.make_chunker();
let mut embedder = if state.embeddings_enabled() {
match state.embedder().await {
Ok(g) => Some(g),
Err(e) => return DaemonResponse::internal(e.to_string()),
}
} else {
None
};
let embedder_dyn: Option<&mut dyn crate::domain::embeddings::EmbedBatch> = embedder
.as_mut()
.map(|g| &mut **g as &mut dyn crate::domain::embeddings::EmbedBatch);
match index_single_file(&store, embedder_dyn, &chunker, &corpus, &dest).await {
Ok(s) => s,
Err(e) => return DaemonResponse::internal(e.to_string()),
}
};
if is_wiki_corpus(&corpus) {
match rebuild_wiki_indexes(state, &corpus, &root, &relative).await {
Ok(extra) => fold_apply_stats(&mut stats, &extra),
Err(msg) => {
drop(guard);
return DaemonResponse::internal(msg);
}
}
}
drop(guard);
let report = IndexReport {
corpora: vec![CorpusReport {
name: corpus.name.clone(),
files_upserted: stats.files_upserted,
files_touched: stats.files_touched,
files_deleted: stats.files_deleted,
files_skipped_empty: stats.files_skipped_empty,
chunks_inserted: stats.chunks_inserted,
embeddings_inserted: stats.embeddings_inserted,
}],
warnings: Vec::new(),
};
DaemonResponse::ok(&AddMarkdownResult {
corpus: corpus.name,
path: relative.to_string_lossy().into_owned(),
absolute_path: dest.to_string_lossy().into_owned(),
indexed: report,
warnings,
})
}
async fn handle_read_markdown(
cfg: &Config,
cwd: &Path,
req: ReadMarkdownRequest,
) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let corpus_name =
match pick_corpus_or_default(&corpora, &cfg.repositories, cwd, req.corpus.as_deref()) {
Ok(c) => c.name,
Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
};
let req_path = req.path;
let resolved = tokio::task::spawn_blocking(move || {
let (corpus, root, relative) = validate_wiki_read_path(&corpora, &corpus_name, &req_path)?;
let bytes = read_no_follow(&root, &relative)
.map_err(|WriteError { kind, source }| map_read_error(kind, source, &relative))?;
Ok::<_, DaemonResponse>((corpus, root, relative, bytes))
})
.await;
let (corpus, root, relative, bytes) = match resolved {
Ok(Ok(t)) => t,
Ok(Err(resp)) => return resp,
Err(join_err) => {
tracing::error!(
target: "hallouminate::daemon",
error = %join_err,
"read_markdown read task panicked",
);
return DaemonResponse::internal(format!("read task panicked: {join_err}"));
}
};
let dest = root.join(&relative);
let content = match String::from_utf8(bytes) {
Ok(s) => s,
Err(e) => {
return DaemonResponse::invalid_params(format!(
"{} is not valid UTF-8: {e}",
relative.display()
));
}
};
DaemonResponse::ok(&ReadMarkdownResult {
corpus: corpus.name,
path: relative.to_string_lossy().into_owned(),
absolute_path: dest.to_string_lossy().into_owned(),
bytes: content.len() as u64,
content,
})
}
async fn handle_delete_markdown(
state: &DaemonState,
cfg: &Config,
req: DeleteMarkdownRequest,
) -> DaemonResponse {
let corpora = match effective_corpora(cfg) {
Ok(v) => v,
Err(resp) => return resp,
};
let (corpus, root, relative) = match validate_wiki_path(&corpora, &req.corpus, &req.path) {
Ok(t) => t,
Err(resp) => return resp,
};
let dest = root.join(&relative);
let guard = match state.acquire_mutation_guard(&corpus.name).await {
Ok(g) => g,
Err(msg) => return DaemonResponse::internal(msg),
};
let meta = match tokio::fs::symlink_metadata(&dest).await {
Ok(m) => m,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
return DaemonResponse::invalid_params(format!(
"{} does not exist",
relative.display()
));
}
Err(e) => {
return DaemonResponse::internal(format!("stat {}: {e}", dest.display()));
}
};
if meta.file_type().is_symlink() {
return DaemonResponse::invalid_params(format!(
"refusing to delete symlink {}",
relative.display()
));
}
if !meta.file_type().is_file() {
return DaemonResponse::invalid_params(format!(
"{} is not a regular file",
relative.display()
));
}
let file_ref = canonicalize_or_passthrough(&dest);
let file_ref_str = match file_ref.as_path().to_str() {
Some(s) => s.to_string(),
None => {
return DaemonResponse::internal(format!(
"non-utf8 path: {}",
file_ref.as_path().display()
));
}
};
let delete_root = root.clone();
let delete_relative = relative.clone();
let error_relative = relative.clone();
let deleted =
tokio::task::spawn_blocking(move || delete_no_follow(&delete_root, &delete_relative)).await;
match deleted {
Ok(Ok(())) => {}
Ok(Err(WriteError { kind, source })) => {
return map_delete_error(kind, source, &error_relative);
}
Err(join_err) => {
tracing::error!(
target: "hallouminate::daemon",
error = %join_err,
"delete_markdown unlink task panicked",
);
return DaemonResponse::internal(format!("unlink task panicked: {join_err}"));
}
}
if let Err(e) = state.store().delete_file(&corpus.name, &file_ref_str).await {
return DaemonResponse::internal(e.to_string());
}
if is_wiki_corpus(&corpus)
&& let Err(msg) = rebuild_wiki_indexes(state, &corpus, &root, &relative).await
{
drop(guard);
return DaemonResponse::internal(msg);
}
drop(guard);
DaemonResponse::ok(&DeleteMarkdownResult {
corpus: corpus.name,
path: relative.to_string_lossy().into_owned(),
absolute_path: dest.to_string_lossy().into_owned(),
file_ref: file_ref_str,
})
}
fn mtime_ms_from_duration(dur: std::time::Duration, file: &Path) -> anyhow::Result<i64> {
i64::try_from(dur.as_millis())
.map_err(|_| anyhow::anyhow!("mtime overflows i64 on {}", file.display()))
}
fn mtime_ms_from_rfc3339(rfc: &str) -> i64 {
chrono::DateTime::parse_from_rfc3339(rfc)
.ok()
.map(|dt| dt.timestamp_millis())
.unwrap_or(i64::MIN)
}
async fn mark_stale(response: &mut crate::domain::ground::GroundResponse) {
let paths: Vec<String> = response.docs.keys().cloned().collect();
let paths_for_join_err = paths.clone();
let indexed_mtimes: Vec<i64> = response
.docs
.values()
.map(|doc| mtime_ms_from_rfc3339(&doc.mtime))
.collect();
let stale_flags: Vec<(String, bool)> = tokio::task::spawn_blocking(move || {
paths
.into_iter()
.zip(indexed_mtimes)
.map(|(path, indexed_ms)| {
let canonical = canonicalize_or_passthrough(std::path::Path::new(&path));
let stale = match std::fs::metadata(canonical.as_path()) {
Err(_) => true, Ok(meta) => {
let disk_s = meta
.modified()
.ok()
.and_then(|t| t.duration_since(UNIX_EPOCH).ok())
.map(|d| d.as_secs() as i64)
.unwrap_or(i64::MAX);
disk_s > indexed_ms / 1000
}
};
(path, stale)
})
.collect::<Vec<_>>()
})
.await
.unwrap_or_else(|e| {
tracing::warn!(
target: "hallouminate::daemon",
error = %e,
"mark_stale: spawn_blocking task failed; marking all docs stale",
);
paths_for_join_err.into_iter().map(|p| (p, true)).collect()
});
for (path, stale) in stale_flags {
if let Some(doc) = response.docs.get_mut(&path) {
doc.stale = stale;
}
}
}
pub(super) async fn index_single_file(
store: &LanceStore,
embedder: Option<&mut dyn crate::domain::embeddings::EmbedBatch>,
chunker: &MarkdownChunker<tokenizers::Tokenizer>,
corpus: &CorpusConfig,
file: &Path,
) -> anyhow::Result<crate::domain::indexer::ApplyStats> {
let meta = tokio::fs::metadata(file).await?;
let modified = meta.modified()?;
let dur = modified
.duration_since(UNIX_EPOCH)
.map_err(|_| anyhow::anyhow!("pre-epoch mtime on {}", file.display()))?;
let mtime_ms = mtime_ms_from_duration(dur, file)?;
let file_ref = canonicalize_or_passthrough(file);
let file_ref_str = file_ref
.as_path()
.to_str()
.ok_or_else(|| anyhow::anyhow!("non-utf8 path: {}", file_ref.as_path().display()))?
.to_string();
let existing = store.get_file_snapshot(&corpus.name, &file_ref_str).await?;
let had_snapshot = existing.is_some();
let mut db: HashMap<FileRef, FileSnapshot> = HashMap::new();
if let Some(snap) = existing {
let hash_changed_without_mtime = if snap.mtime_ms == mtime_ms {
blake3_file(file)? != snap.content_hash.as_str()
} else {
false
};
if !hash_changed_without_mtime {
db.insert(file_ref.clone(), snap);
}
}
let p = plan(vec![(file_ref, Mtime(mtime_ms))], db);
let mut stats = apply(p, store, embedder, chunker, corpus, DEFAULT_BATCH_SIZE).await?;
if stats.files_skipped_empty > 0 && had_snapshot {
store.delete_file(&corpus.name, &file_ref_str).await?;
stats.files_deleted += 1;
}
Ok(stats)
}
fn ensure_paths_exist(corpus: &CorpusConfig) {
if !is_wiki_corpus(corpus) {
return;
}
for raw in &corpus.paths {
let path = crate::domain::common::expand_tilde(raw);
let _ = std::fs::create_dir_all(&path);
}
}
fn fold_apply_stats(
into: &mut crate::domain::indexer::ApplyStats,
extra: &crate::domain::indexer::ApplyStats,
) {
into.files_upserted += extra.files_upserted;
into.files_touched += extra.files_touched;
into.files_deleted += extra.files_deleted;
into.files_skipped_empty += extra.files_skipped_empty;
into.chunks_inserted += extra.chunks_inserted;
into.embeddings_inserted += extra.embeddings_inserted;
}
async fn rebuild_wiki_indexes(
state: &DaemonState,
corpus: &CorpusConfig,
root: &Path,
file_relative: &Path,
) -> Result<crate::domain::indexer::ApplyStats, String> {
use crate::domain::corpus::index_md::{
INDEX_FILENAME, ancestor_dirs, compose_index_md, is_index_md,
};
let written_is_index = is_index_md(file_relative);
let mut totals = crate::domain::indexer::ApplyStats::default();
let dirs = ancestor_dirs(root, file_relative);
let store = state.store();
let chunker = state.make_chunker();
for dir in &dirs {
let index_path = dir.join(INDEX_FILENAME);
if written_is_index
&& let Some(parent) = file_relative.parent()
&& dir == &root.join(parent)
{
continue;
}
let existing = match std::fs::read_to_string(&index_path) {
Ok(s) => Some(s),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
Err(e) => return Err(format!("read {}: {e}", index_path.display())),
};
let is_root = dir == root;
let (new_content, outcome) = compose_index_md(dir, is_root, existing.as_deref())
.map_err(|e| format!("compose index {}: {e}", dir.display()))?;
match outcome {
crate::domain::corpus::index_md::RewriteOutcome::NoMarkers
| crate::domain::corpus::index_md::RewriteOutcome::Unchanged => continue,
crate::domain::corpus::index_md::RewriteOutcome::Created
| crate::domain::corpus::index_md::RewriteOutcome::Updated => {}
}
let rel = match index_path.strip_prefix(root) {
Ok(p) => p.to_path_buf(),
Err(_) => {
return Err(format!(
"index path {} not under root",
index_path.display()
));
}
};
let write_root = root.to_path_buf();
let write_rel = rel.clone();
let bytes = new_content.into_bytes();
let written = tokio::task::spawn_blocking(move || {
atomic_write_no_follow(&write_root, &write_rel, &bytes, true)
})
.await
.map_err(|e| {
tracing::error!(
target: "hallouminate::daemon",
error = %e,
"rebuild_wiki_indexes write task panicked",
);
format!("index write task panicked: {e}")
})?;
let dest = match written {
Ok(p) => p,
Err(WriteError { kind, source }) => {
return Err(format!(
"writing index {} failed ({:?}): {source}",
index_path.display(),
kind,
));
}
};
let mut embedder = if state.embeddings_enabled() {
match state.embedder().await {
Ok(g) => Some(g),
Err(e) => return Err(format!("embedder: {e}")),
}
} else {
None
};
let embedder_dyn: Option<&mut dyn crate::domain::embeddings::EmbedBatch> = embedder
.as_mut()
.map(|g| &mut **g as &mut dyn crate::domain::embeddings::EmbedBatch);
let stats = index_single_file(&store, embedder_dyn, &chunker, corpus, &dest)
.await
.map_err(|e| format!("reindex {}: {e}", dest.display()))?;
drop(embedder);
fold_apply_stats(&mut totals, &stats);
}
Ok(totals)
}
fn is_wiki_corpus(corpus: &CorpusConfig) -> bool {
corpus.name.starts_with("repo:") && corpus.name.ends_with(":wiki")
}
fn ensure_wiki_root_safe(corpus: &CorpusConfig) -> Result<(), String> {
if !is_wiki_corpus(corpus) {
return Ok(());
}
let Some(raw) = corpus.paths.first() else {
return Ok(());
};
let root = crate::domain::common::expand_tilde(raw);
if let Some(parent) = root.parent()
&& let Ok(meta) = std::fs::symlink_metadata(parent)
&& meta.file_type().is_symlink()
{
return Err(format!(
"wiki corpus {} is unsafe: parent {} is a symlink",
corpus.name,
parent.display(),
));
}
if let Ok(meta) = std::fs::symlink_metadata(&root)
&& meta.file_type().is_symlink()
{
return Err(format!(
"wiki corpus {} is unsafe: wiki root is a symlink",
corpus.name,
));
}
Ok(())
}
fn map_read_error(kind: WriteErrorKind, source: std::io::Error, relative: &Path) -> DaemonResponse {
match kind {
WriteErrorKind::Symlink => DaemonResponse::invalid_params(format!(
"refusing to read symlink {}: {source}",
relative.display()
)),
WriteErrorKind::InvalidPath => {
if source.kind() == std::io::ErrorKind::NotFound {
DaemonResponse::invalid_params(format!("{} does not exist", relative.display()))
} else {
DaemonResponse::invalid_params(format!(
"refusing unsafe path {}: {source}",
relative.display()
))
}
}
WriteErrorKind::Io => {
if source.kind() == std::io::ErrorKind::NotFound {
DaemonResponse::invalid_params(format!("{} does not exist", relative.display()))
} else {
DaemonResponse::internal(format!("read {}: {source}", relative.display()))
}
}
WriteErrorKind::Exists => DaemonResponse::internal(source.to_string()),
}
}
fn map_delete_error(
kind: WriteErrorKind,
source: std::io::Error,
relative: &Path,
) -> DaemonResponse {
match kind {
WriteErrorKind::Symlink => DaemonResponse::invalid_params(format!(
"refusing to delete symlink {}: {source}",
relative.display()
)),
WriteErrorKind::InvalidPath => {
if source.kind() == std::io::ErrorKind::NotFound {
DaemonResponse::invalid_params(format!("{} does not exist", relative.display()))
} else {
DaemonResponse::invalid_params(format!(
"refusing unsafe path {}: {source}",
relative.display()
))
}
}
WriteErrorKind::Io => {
if source.kind() == std::io::ErrorKind::NotFound {
DaemonResponse::invalid_params(format!("{} does not exist", relative.display()))
} else {
DaemonResponse::internal(format!("unlink {}: {source}", relative.display()))
}
}
WriteErrorKind::Exists => DaemonResponse::internal(source.to_string()),
}
}
#[cfg(test)]
use serde_json::Value;
#[cfg(test)]
fn derived_corpus_name(repo_name: &str, kind: RepoCorpusKind) -> Result<String, String> {
repo_corpus_name(repo_name, kind).map_err(|e| e.to_string())
}
#[cfg(test)]
fn pong_value() -> Value {
serde_json::json!({ "version": env!("CARGO_PKG_VERSION") })
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn derived_corpus_name_emits_canonical_string_for_valid_inputs() {
let name = derived_corpus_name("tern", RepoCorpusKind::Wiki)
.expect("valid repo name must succeed");
assert_eq!(name, "repo:tern:wiki");
}
#[test]
fn derived_corpus_name_surfaces_underlying_error_as_string() {
let err =
derived_corpus_name("", RepoCorpusKind::Wiki).expect_err("empty repo name must fail");
assert!(err.contains("empty"), "got: {err}");
}
#[test]
fn pong_value_carries_the_daemon_binary_version() {
assert_eq!(pong_value()["version"], env!("CARGO_PKG_VERSION"));
}
#[test]
fn mtime_ms_from_duration_passes_through_normal_value() {
let dur = std::time::Duration::from_millis(1_700_000_000_000);
let got =
mtime_ms_from_duration(dur, Path::new("/tmp/a.md")).expect("a sane mtime must convert");
assert_eq!(got, 1_700_000_000_000_i64);
}
#[test]
fn mtime_ms_from_duration_accepts_i64_max_milliseconds() {
let max_ms = u64::try_from(i64::MAX).unwrap();
let dur = std::time::Duration::from_millis(max_ms);
let got = mtime_ms_from_duration(dur, Path::new("/tmp/max.md"))
.expect("i64::MAX ms is representable and must convert");
assert_eq!(got, i64::MAX);
}
#[test]
fn mtime_ms_from_duration_errors_one_past_i64_max() {
let overflow_ms = u64::try_from(i64::MAX).unwrap() + 1;
let dur = std::time::Duration::from_millis(overflow_ms);
let err = mtime_ms_from_duration(dur, Path::new("/tmp/huge.md"))
.expect_err("an mtime past i64::MAX ms must error, not truncate");
let msg = err.to_string();
assert!(
msg.contains("overflows i64") && msg.contains("huge.md"),
"overflow error must name the cause and file: {msg}",
);
}
#[test]
fn file_entry_re_export_keeps_field_names() {
let entry = FileEntry {
path: "a.md".to_string(),
absolute_path: "/r/a.md".to_string(),
};
let json = serde_json::to_value(&entry).unwrap();
assert_eq!(json["path"], "a.md");
assert_eq!(json["absolute_path"], "/r/a.md");
}
fn wiki_corpus_at(root: &Path) -> CorpusConfig {
CorpusConfig {
name: "repo:tern:wiki".into(),
paths: vec![root.to_string_lossy().into_owned()],
globs: vec!["**/*.md".into()],
exclude: vec![],
global: false,
}
}
fn assert_invalid_params(resp: DaemonResponse, needle: &str) {
match resp {
DaemonResponse::Err { kind, message } => {
assert_eq!(
kind,
ErrorKind::InvalidParams,
"a validation failure must surface as InvalidParams, not a \
server fault: {message}",
);
assert!(
message.contains(needle),
"error must explain the failing step (want {needle:?}): {message}",
);
}
DaemonResponse::Ok { result } => {
panic!("expected an InvalidParams error, got Ok({result:?})");
}
}
}
#[test]
fn validate_wiki_path_returns_corpus_root_and_relative_on_valid_input() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path().to_path_buf();
let corpus = wiki_corpus_at(&root);
let (got_corpus, got_root, got_relative) = validate_wiki_path(
std::slice::from_ref(&corpus),
"repo:tern:wiki",
"notes/a.md",
)
.expect("valid corpus + path must resolve");
assert_eq!(got_corpus.name, "repo:tern:wiki");
assert_eq!(got_root, root);
assert_eq!(got_relative, std::path::PathBuf::from("notes/a.md"));
}
#[test]
fn validate_wiki_path_maps_unknown_corpus_to_invalid_params() {
let tmp = tempfile::tempdir().unwrap();
let corpus = wiki_corpus_at(tmp.path());
let resp = validate_wiki_path(std::slice::from_ref(&corpus), "repo:nope:wiki", "a.md")
.expect_err("unknown corpus must fail validation");
assert_invalid_params(resp, "not found");
}
#[test]
fn validate_wiki_path_rejects_path_traversal_as_invalid_params() {
let tmp = tempfile::tempdir().unwrap();
let corpus = wiki_corpus_at(tmp.path());
let resp = validate_wiki_path(
std::slice::from_ref(&corpus),
"repo:tern:wiki",
"../../etc/passwd",
)
.expect_err("path traversal must fail validation");
assert_invalid_params(resp, "normal file components");
}
#[test]
fn validate_wiki_path_enforces_corpus_globs_as_invalid_params() {
let tmp = tempfile::tempdir().unwrap();
let corpus = wiki_corpus_at(tmp.path());
let resp = validate_wiki_path(
std::slice::from_ref(&corpus),
"repo:tern:wiki",
"notes/a.txt",
)
.expect_err("a non-markdown path must be rejected by corpus globs");
assert_invalid_params(resp, "not included by corpus globs");
}
use std::path::Path;
use crate::app::daemon::ErrorKind;
async fn state_with_ground(ground_dir: &Path, baseline_toml: &str) -> DaemonState {
let toml = format!(
"{baseline_toml}\n[storage]\nground_dir = \"{}\"\n",
ground_dir.display(),
);
let cfg: Config = toml::from_str(&toml).expect("baseline toml parses");
DaemonState::open(cfg, None)
.await
.expect("open daemon state")
}
fn write_repo_layer(repo_root: &Path, body: &str) {
let cfg_dir = repo_root.join(".hallouminate");
std::fs::create_dir_all(&cfg_dir).expect("mkdir .hallouminate");
std::fs::write(cfg_dir.join("config.toml"), body).expect("write repo config");
}
#[tokio::test]
async fn dispatch_ping_is_config_independent_and_reports_version() {
let tmp = tempfile::tempdir().expect("tempdir");
let cwd = tmp.path().to_path_buf();
let ground = tmp.path().join("ground");
let state = state_with_ground(&ground, "").await;
let req = DaemonRequest {
cwd,
payload: DaemonRequestPayload::Ping,
};
let resp = dispatch(&state, req).await;
match resp {
DaemonResponse::Ok { result } => {
assert_eq!(result, pong_value(), "ping must return the versioned pong");
}
DaemonResponse::Err { kind, message } => {
panic!("ping must succeed regardless of cwd; got {kind:?}: {message}");
}
}
}
#[tokio::test]
async fn dispatch_above_all_repos_falls_back_to_baseline_corpora() {
let tmp = tempfile::tempdir().expect("tempdir");
let cwd = tmp.path().to_path_buf();
let ground = tmp.path().join("ground");
let state = state_with_ground(
&ground,
"[[corpus]]\nname = \"cheese-global\"\npaths = [\"/srv/cheese-global\"]\n",
)
.await;
let req = DaemonRequest {
cwd: cwd.clone(),
payload: DaemonRequestPayload::ListCorpora,
};
let resp = dispatch(&state, req).await;
match resp {
DaemonResponse::Ok { result } => {
let entries = result.as_array().expect("ListCorpora returns an array");
let names: Vec<&str> = entries
.iter()
.filter_map(|e| e.get("name").and_then(serde_json::Value::as_str))
.collect();
assert!(
names.contains(&"cheese-global"),
"baseline corpus must be reachable from above all repos; got {names:?}",
);
}
DaemonResponse::Err { kind, message } => {
assert_eq!(kind, ErrorKind::InvalidParams, "{message}");
assert!(message.contains("stopped at repo root"), "{message}");
}
}
}
#[tokio::test]
async fn dispatch_inside_repo_without_config_still_hard_errors() {
let tmp = tempfile::tempdir().expect("tempdir");
let repo_root = tmp.path().to_path_buf();
std::fs::create_dir(repo_root.join(".git")).expect("mkdir .git");
let cwd = repo_root.join("src");
std::fs::create_dir_all(&cwd).expect("mkdir nested");
let ground = tmp.path().join("ground");
let state = state_with_ground(&ground, "").await;
let req = DaemonRequest {
cwd,
payload: DaemonRequestPayload::ListCorpora,
};
let resp = dispatch(&state, req).await;
match resp {
DaemonResponse::Err { kind, message } => {
assert_eq!(
kind,
ErrorKind::InvalidParams,
"discovery failure must map to InvalidParams: {message}",
);
assert!(
message.contains("stopped at repo root"),
"in-repo discovery error must explain the boundary: {message}",
);
}
DaemonResponse::Ok { result } => {
panic!("must not fall back to baseline-only inside a repo; got Ok({result:?})");
}
}
}
#[tokio::test]
async fn dispatch_with_scalar_conflict_returns_config_error() {
let tmp = tempfile::tempdir().expect("tempdir");
let cwd = tmp.path().to_path_buf();
write_repo_layer(&cwd, "[embeddings]\ncache_dir = \"/b\"\n");
let ground = tmp.path().join("ground");
let state = state_with_ground(&ground, "[embeddings]\ncache_dir = \"/a\"\n").await;
let req = DaemonRequest {
cwd,
payload: DaemonRequestPayload::ListCorpora,
};
let resp = dispatch(&state, req).await;
match resp {
DaemonResponse::Err { kind, message } => {
assert_eq!(
kind,
ErrorKind::InvalidParams,
"merge conflict must map to InvalidParams: {message}",
);
assert!(
message.contains("embeddings.cache_dir"),
"conflict error must name the field: {message}",
);
assert!(
message.contains("\"/a\"") && message.contains("\"/b\""),
"conflict error must show both values: {message}",
);
}
DaemonResponse::Ok { result } => {
panic!("scalar conflict must error; got Ok({result:?})");
}
}
}
#[tokio::test]
async fn dispatch_scalar_conflict_names_baseline_xdg_path_when_known() {
let tmp = tempfile::tempdir().expect("tempdir");
let cwd = tmp.path().to_path_buf();
write_repo_layer(&cwd, "[embeddings]\ncache_dir = \"/b\"\n");
let ground = tmp.path().join("ground");
let baseline_path = tmp.path().join("baseline.toml");
let baseline_toml = format!(
"[embeddings]\ncache_dir = \"/a\"\n[storage]\nground_dir = \"{}\"\n",
ground.display(),
);
let cfg: Config = toml::from_str(&baseline_toml).expect("baseline parses");
let state = DaemonState::open(cfg, Some(baseline_path.clone()))
.await
.expect("open with xdg_path");
let req = DaemonRequest {
cwd,
payload: DaemonRequestPayload::ListCorpora,
};
let resp = dispatch(&state, req).await;
let DaemonResponse::Err { message, .. } = resp else {
panic!("scalar conflict must error");
};
assert!(
message.contains(&baseline_path.display().to_string()),
"conflict message must name the baseline source path: {message}",
);
assert!(
!message.contains("(XDG baseline)"),
"must not fall back to the unsourced placeholder: {message}",
);
}
#[test]
fn mtime_ms_from_rfc3339_parses_known_timestamp() {
let ms = mtime_ms_from_rfc3339("2026-04-30T10:11:23Z");
assert_eq!(ms, 1_777_543_883_000_i64);
}
#[test]
fn mtime_ms_from_rfc3339_returns_i64_min_for_invalid_input() {
let ms = mtime_ms_from_rfc3339("not-a-date");
assert_eq!(ms, i64::MIN);
}
#[tokio::test]
async fn stale_false_when_file_unchanged_since_index() {
let dir = tempfile::tempdir().expect("tempdir");
let file = dir.path().join("doc.md");
std::fs::write(&file, "# Title\n\nBody text.\n").expect("write");
let meta = std::fs::metadata(&file).expect("stat");
let disk_ms = meta
.modified()
.unwrap()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let indexed_mtime = chrono::DateTime::<chrono::Utc>::from_timestamp_millis(disk_ms)
.unwrap()
.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
let abs = canonicalize_or_passthrough(&file);
let abs_str = abs.as_path().to_str().unwrap().to_string();
let mut docs = std::collections::BTreeMap::new();
docs.insert(
abs_str.clone(),
crate::domain::ground::DocFile {
summary: None,
keywords: vec![],
score: 0.5,
z_score: None,
mtime: indexed_mtime,
corpus: "test".into(),
path: None,
stale: false,
chunks: vec![],
},
);
let mut response = crate::domain::ground::GroundResponse {
query: "test".into(),
took_ms: 0,
stats: crate::domain::ground::Stats { hits: 1 },
docs,
code: std::collections::BTreeMap::new(),
warnings: vec![],
};
mark_stale(&mut response).await;
assert!(
!response.docs[&abs_str].stale,
"file unchanged since index must not be stale"
);
}
#[tokio::test]
async fn stale_true_when_file_modified_after_index() {
let dir = tempfile::tempdir().expect("tempdir");
let file = dir.path().join("doc.md");
std::fs::write(&file, "# Title\n\nOriginal.\n").expect("write");
let meta = std::fs::metadata(&file).expect("stat");
let disk_ms = meta
.modified()
.unwrap()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let past_ms = disk_ms - 1_000; let indexed_mtime = chrono::DateTime::<chrono::Utc>::from_timestamp_millis(past_ms)
.unwrap()
.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
let abs = canonicalize_or_passthrough(&file);
let abs_str = abs.as_path().to_str().unwrap().to_string();
let mut docs = std::collections::BTreeMap::new();
docs.insert(
abs_str.clone(),
crate::domain::ground::DocFile {
summary: None,
keywords: vec![],
score: 0.5,
z_score: None,
mtime: indexed_mtime,
corpus: "test".into(),
path: None,
stale: false,
chunks: vec![],
},
);
let mut response = crate::domain::ground::GroundResponse {
query: "test".into(),
took_ms: 0,
stats: crate::domain::ground::Stats { hits: 1 },
docs,
code: std::collections::BTreeMap::new(),
warnings: vec![],
};
mark_stale(&mut response).await;
assert!(
response.docs[&abs_str].stale,
"file modified after index must be stale"
);
}
#[tokio::test]
async fn corpus_stats_counts_indexed_and_unindexed_files() {
let tmp = tempfile::tempdir().expect("tempdir");
let corpus_dir = tmp.path().join("wiki");
std::fs::create_dir_all(&corpus_dir).expect("mkdir wiki");
let ground = tmp.path().join("ground");
let repo_config = format!(
"[[corpus]]\nname = \"test\"\npaths = [\"{}\"]\n",
corpus_dir.display()
);
write_repo_layer(tmp.path(), &repo_config);
let state = state_with_ground(&ground, "").await;
std::fs::write(corpus_dir.join("a.md"), "# Doc A\n\nContent A.\n").expect("write a");
std::fs::write(corpus_dir.join("b.md"), "# Doc B\n\nContent B.\n").expect("write b");
let index_resp = dispatch(
&state,
DaemonRequest {
cwd: tmp.path().to_path_buf(),
payload: DaemonRequestPayload::Index(IndexRequest {
corpus: Some("test".to_string()),
paths_from: None,
strict: false,
}),
},
)
.await;
assert!(
matches!(index_resp, DaemonResponse::Ok { .. }),
"index must succeed: {index_resp:?}"
);
std::fs::write(corpus_dir.join("c.md"), "# Doc C\n\nUnindexed.\n").expect("write c");
let resp = dispatch(
&state,
DaemonRequest {
cwd: tmp.path().to_path_buf(),
payload: DaemonRequestPayload::CorpusStats { corpus: None },
},
)
.await;
let DaemonResponse::Ok { result } = resp else {
panic!("corpus_stats must succeed: {resp:?}");
};
let stats: CorpusStatsResult =
serde_json::from_value(result).expect("parse CorpusStatsResult");
assert_eq!(stats.indexed_files, 2, "two files were indexed");
assert_eq!(stats.unindexed_files, 1, "one file added without re-index");
assert!(
stats.last_indexed_ms.is_some(),
"indexed corpus must carry a timestamp"
);
assert_eq!(stats.corpus, "test");
}
#[tokio::test]
async fn corpus_stats_excludes_glob_excluded_files_from_unindexed() {
let tmp = tempfile::tempdir().expect("tempdir");
let corpus_dir = tmp.path().join("wiki");
std::fs::create_dir_all(&corpus_dir).expect("mkdir wiki");
let ground = tmp.path().join("ground");
let repo_config = format!(
concat!(
"[[corpus]]\nname = \"test\"\n",
"paths = [\"{}\"]",
"\nglobs = [\"**/*.md\"]\nexclude = [\"**/excluded.md\"]\n"
),
corpus_dir.display()
);
write_repo_layer(tmp.path(), &repo_config);
let state = state_with_ground(&ground, "").await;
std::fs::write(corpus_dir.join("indexed.md"), "# Indexed\n\nContent.\n")
.expect("write indexed");
let index_resp = dispatch(
&state,
DaemonRequest {
cwd: tmp.path().to_path_buf(),
payload: DaemonRequestPayload::Index(IndexRequest {
corpus: Some("test".to_string()),
paths_from: None,
strict: false,
}),
},
)
.await;
assert!(
matches!(index_resp, DaemonResponse::Ok { .. }),
"index must succeed: {index_resp:?}"
);
std::fs::write(
corpus_dir.join("excluded.md"),
"# Excluded\n\nShould not be counted as unindexed.\n",
)
.expect("write excluded");
let resp = dispatch(
&state,
DaemonRequest {
cwd: tmp.path().to_path_buf(),
payload: DaemonRequestPayload::CorpusStats { corpus: None },
},
)
.await;
let DaemonResponse::Ok { result } = resp else {
panic!("corpus_stats must succeed: {resp:?}");
};
let stats: CorpusStatsResult =
serde_json::from_value(result).expect("parse CorpusStatsResult");
assert_eq!(stats.indexed_files, 1, "one file was indexed");
assert_eq!(
stats.unindexed_files, 0,
"excluded file must not count toward unindexed_files"
);
}
#[tokio::test]
async fn corpus_stats_returns_zeroed_result_for_never_created_wiki_corpus() {
let tmp = tempfile::tempdir().expect("tempdir");
let repo_root = tmp.path();
let ground = tmp.path().join("ground");
let repo_config = format!(
"[[repository]]\nname = \"myrepo\"\npath = \"{}\"\n",
repo_root.display()
);
write_repo_layer(repo_root, &repo_config);
let state = state_with_ground(&ground, "").await;
let resp = dispatch(
&state,
DaemonRequest {
cwd: repo_root.to_path_buf(),
payload: DaemonRequestPayload::CorpusStats {
corpus: Some("repo:myrepo:wiki".to_string()),
},
},
)
.await;
let DaemonResponse::Ok { result } = resp else {
panic!(
"corpus_stats on a never-created wiki corpus must return Ok, not an error: {resp:?}"
);
};
let stats: CorpusStatsResult =
serde_json::from_value(result).expect("parse CorpusStatsResult");
assert_eq!(stats.indexed_files, 0, "no files indexed yet");
assert_eq!(stats.total_chunks, 0, "no chunks yet");
assert_eq!(stats.unindexed_files, 0, "empty dir has no unindexed files");
assert!(
stats.last_indexed_ms.is_none(),
"never-indexed corpus must have null timestamp"
);
assert_eq!(stats.corpus, "repo:myrepo:wiki");
}
}