hallouminate 0.1.0

//! Daemon request dispatcher.
//!
//! Each handler:
//!   - resolves the target corpus via `effective_corpora` so derived
//!     `repo:{name}:wiki` / `repo:{name}:corpus` corpora are visible;
//!   - for mutating ops, takes the per-corpus mutex AND the global
//!     write-lane permit in that order via
//!     `DaemonState::acquire_mutation_guard` (the shared helper that
//!     replaces the open-coded `lock_corpus + acquire_owned` pattern every
//!     handler used to repeat);
//!   - returns a `DaemonResponse::Err(InvalidParams, ...)` for caller
//!     errors and `DaemonResponse::Err(Internal, ...)` for daemon faults.
//!
//! The corpus-selection / path-traversal / glob-validation / atomic-write
//! helpers live in `crate::domain::corpus::sandbox` and are shared with the
//! MCP tools transport, closing the maintenance liability of two divergent
//! forks (and the security asymmetry where the daemon was using
//! `tokio::fs::write` while MCP used `atomic_write_no_follow`).

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, 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,
    safe_relative_path,
};
use crate::domain::corpus::{MarkdownChunker, blake3_file};
use crate::domain::ground::{Format, GroundOpts, RenderOpts, ground, 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, DaemonRequest, DaemonRequestPayload,
    DaemonResponse, DeleteMarkdownRequest, DeleteMarkdownResult, GlobalizeMarkdownRequest,
    GlobalizeMarkdownResult, GroundRequest, GroundResult, IndexRequest, ListFilesRequest,
    ListTreeRequest, ListTreeResult, ReadMarkdownRequest, ReadMarkdownResult,
};
use super::state::DaemonState;

pub async fn dispatch(state: &DaemonState, req: DaemonRequest) -> DaemonResponse {
    // Resolve per-request config layering on every request: discover the
    // repo-layer `.hallouminate/config.toml` under `req.cwd` and merge with
    // the boot baseline. Discovery / merge failures surface to the client as
    // `InvalidParams` so a misconfigured workspace produces a clean error
    // instead of a silent fall-back to baseline-only.
    //
    // `state.baseline_xdg_path()` carries the baseline's source path (the
    // XDG path, or the `--config PATH` override) so scalar-conflict
    // messages name the file the user actually has to edit — AC #7.
    // Shutdown and Ping are config-independent control ops: handle them
    // before `resolve_for_cwd` so a stop request (or a liveness probe) works
    // even when the client's cwd has no discoverable repo config.
    if let DaemonRequestPayload::Shutdown = req.payload {
        state.shutdown_token().cancel();
        return DaemonResponse::ok(&"stopping");
    }
    let req_cwd = req.cwd.clone();
    let effective = match resolve_for_cwd(state.baseline(), &req.cwd, state.baseline_xdg_path()) {
        Ok((cfg, _layers)) => cfg,
        Err(e) => return DaemonResponse::invalid_params(e.to_string()),
    };
    match req.payload {
        DaemonRequestPayload::Ping => DaemonResponse::ok(&"pong"),
        DaemonRequestPayload::Ground(req) => handle_ground(state, &effective, &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).await,
        DaemonRequestPayload::DeleteMarkdown(req) => {
            handle_delete_markdown(state, &effective, req).await
        }
        DaemonRequestPayload::GlobalizeMarkdown(req) => {
            handle_globalize_markdown(state, &effective, req).await
        }
        DaemonRequestPayload::Shutdown => {
            // Handled before `resolve_for_cwd` above; unreachable here.
            DaemonResponse::ok(&"stopping")
        }
    }
}

fn effective_corpora(cfg: &Config) -> Result<Vec<CorpusConfig>, DaemonResponse> {
    cfg.effective_corpora()
        .map_err(|e| DaemonResponse::internal(e.to_string()))
}

/// Read-side corpus selection with wiki-defaulting.
///
/// When `requested` is `None`, try to default to `repo:{name}:wiki` for the
/// repository whose `path` contains `cwd` — that's the wiki the LLM is
/// actually working in. If no repository matches (or the derived corpus
/// isn't present), fall through to `pick_corpus`'s existing single-corpus /
/// ambiguity behavior. Mutating handlers do NOT use this — they require
/// an explicit corpus to avoid accidental writes to the wrong wiki.
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 wiki dir exists so an unindexed repository wiki doesn't error
    // out the first list call.
    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_ground(
    state: &DaemonState,
    cfg: &Config,
    cwd: &Path,
    req: GroundRequest,
) -> 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()),
        };
    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),
    };
    // Embeddings-OFF: skip the embedder entirely and let `ground` run the
    // lexical-only path. ON: borrow the shared embedder (lazy-loaded).
    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
    };
    // crossencoder is best-effort: if it's configured but failed to
    // load (e.g. model file vanished), log and ground without it
    // rather than refusing the request. Unconfigured paths return
    // Ok(None) and the rerank step is skipped entirely.
    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 = match ground(
        &req.query,
        &corpus.name,
        &corpus.paths,
        &store,
        embedder_dyn,
        crossencoder_dyn,
        opts,
    )
    .await
    {
        Ok(r) => r,
        Err(e) => return DaemonResponse::internal(e.to_string()),
    };
    drop(crossencoder);
    drop(embedder);
    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 {
    // Reject ad-hoc paths_from unconditionally: the daemon protocol does not
    // accept it yet, and silently ignoring the field when a corpus is also
    // selected would let MCP clients believe the path list landed.
    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() {
            // Pre-daemon CLI/MCP treated "no corpora configured" as invalid
            // input. Match that shape so a misconfigured daemon doesn't
            // silently report "index ok" with zero work done.
            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 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)
}

async fn handle_add_markdown(
    state: &DaemonState,
    cfg: &Config,
    req: AddMarkdownRequest,
) -> DaemonResponse {
    let corpora = match effective_corpora(cfg) {
        Ok(v) => v,
        Err(resp) => return resp,
    };
    let corpus = match pick_corpus(&corpora, Some(&req.corpus)) {
        Ok(c) => c,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let root = match first_corpus_root(&corpus) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let relative = match safe_relative_path(&req.path) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let dest = root.join(&relative);
    if let Err(e) = ensure_corpus_allows_file(&corpus, &dest) {
        return DaemonResponse::invalid_params(e.into_inner());
    }
    if let Err(msg) = ensure_wiki_root_safe(&corpus) {
        return DaemonResponse::invalid_params(msg);
    }

    let guard = match state.acquire_mutation_guard(&corpus.name).await {
        Ok(g) => g,
        Err(msg) => return DaemonResponse::internal(msg),
    };

    // Symlink-safe atomic write via the shared sandbox helper. Walks every
    // path component with `O_NOFOLLOW | O_DIRECTORY`, so a symlinked
    // intermediate dir bounces with `WriteErrorKind::Symlink` instead of
    // letting the writer punch through to whatever the symlink targets.
    let write_root = root.clone();
    let write_relative = relative.clone();
    let error_relative = relative.clone();
    let content_bytes = req.content.clone().into_bytes();
    let overwrite = req.overwrite;
    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) => {
            return DaemonResponse::internal(format!("write task panicked: {join_err}"));
        }
    };

    // Empty content produces zero chunks; the indexer would just count the
    // file as `files_skipped_empty` and burn an embedder call on a no-op.
    // Short-circuit so tests that exercise just the filesystem-mutation lane
    // don't need the embedding model active. When the request is overwriting
    // a previously-indexed file with empty content, also prune the existing
    // LanceDB rows so searches stop returning the deleted body — the full
    // `index_single_file` path does this via `files_skipped_empty > 0 &&
    // had_snapshot`, but the short-circuit below bypasses that loop.
    let mut stats = if req.content.trim().is_empty() {
        let mut stats = crate::domain::indexer::ApplyStats {
            files_skipped_empty: 1,
            ..Default::default()
        };
        if req.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()),
        }
    };

    // Auto-rebuild wiki indexes from the corpus root down to the parent of
    // the just-written file. Failures here surface as `Internal` and the
    // mutation guard is dropped — partial index regen would leave the wiki
    // in a less-coherent state than aborting outright.
    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,
        }],
    };
    DaemonResponse::ok(&AddMarkdownResult {
        corpus: corpus.name,
        path: relative.to_string_lossy().into_owned(),
        absolute_path: dest.to_string_lossy().into_owned(),
        indexed: report,
    })
}

async fn handle_read_markdown(cfg: &Config, req: ReadMarkdownRequest) -> DaemonResponse {
    let corpora = match effective_corpora(cfg) {
        Ok(v) => v,
        Err(resp) => return resp,
    };
    let corpus = match pick_corpus(&corpora, Some(&req.corpus)) {
        Ok(c) => c,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let root = match first_corpus_root(&corpus) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let relative = match safe_relative_path(&req.path) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let dest = root.join(&relative);
    if let Err(e) = ensure_corpus_allows_file(&corpus, &dest) {
        return DaemonResponse::invalid_params(e.into_inner());
    }
    if let Err(msg) = ensure_wiki_root_safe(&corpus) {
        return DaemonResponse::invalid_params(msg);
    }

    // Symlink-safe read via the shared sandbox helper: walks every parent
    // component with `O_NOFOLLOW` and opens the leaf `O_RDONLY | O_NOFOLLOW`
    // so a symlinked intermediate dir or leaf bounces with the same
    // `WriteErrorKind::Symlink` shape `add_markdown` uses for writes.
    let read_root = root.clone();
    let read_relative = relative.clone();
    let error_relative = relative.clone();
    let read =
        tokio::task::spawn_blocking(move || read_no_follow(&read_root, &read_relative)).await;
    let bytes = match read {
        Ok(Ok(b)) => b,
        Ok(Err(WriteError { kind, source })) => {
            return map_read_error(kind, source, &error_relative);
        }
        Err(join_err) => {
            return DaemonResponse::internal(format!("read task panicked: {join_err}"));
        }
    };
    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 = match pick_corpus(&corpora, Some(&req.corpus)) {
        Ok(c) => c,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let root = match first_corpus_root(&corpus) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let relative = match safe_relative_path(&req.path) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let dest = root.join(&relative);
    if let Err(e) = ensure_corpus_allows_file(&corpus, &dest) {
        return DaemonResponse::invalid_params(e.into_inner());
    }
    if let Err(msg) = ensure_wiki_root_safe(&corpus) {
        return DaemonResponse::invalid_params(msg);
    }

    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()
        ));
    }

    // Compute canonical file_ref BEFORE unlinking so the LanceDB row we
    // prune matches what the indexer wrote.
    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()
            ));
        }
    };

    // Symlink-safe unlink via the shared sandbox helper: walks every parent
    // component with `O_NOFOLLOW` and refuses if the leaf is a symlink, so a
    // corpus-controlled symlinked directory cannot redirect the unlink
    // outside the corpus root.
    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) => {
            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());
    }

    // Auto-rebuild wiki indexes after the unlink so the parent index no
    // longer links to the deleted file. Same internal-error semantics as
    // the add_markdown path — partial regen would desync the wiki tree.
    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,
    })
}

/// Copy a single markdown entry from `source_corpus` into the corpus marked
/// `global = true`, reindexing the global corpus row synchronously. Mirrors
/// `add_markdown`'s contract: read via the no-follow reader, write via the
/// no-follow atomic writer, collision gated by `overwrite`. Source stays.
async fn handle_globalize_markdown(
    state: &DaemonState,
    cfg: &Config,
    req: GlobalizeMarkdownRequest,
) -> DaemonResponse {
    let corpora = match effective_corpora(cfg) {
        Ok(v) => v,
        Err(resp) => return resp,
    };

    // Resolve the single global corpus. None / more-than-one is a config
    // error the caller can't fix per-request, so surface it as InvalidParams.
    let globals: Vec<&CorpusConfig> = corpora.iter().filter(|c| c.global).collect();
    let global = match globals.as_slice() {
        [] => {
            return DaemonResponse::invalid_params(
                "no global corpus configured; mark one [[corpus]] with global = true",
            );
        }
        [only] => (*only).clone(),
        _ => {
            return DaemonResponse::internal(
                "more than one corpus marked global = true; config validation should have rejected this",
            );
        }
    };

    let source = match pick_corpus(&corpora, Some(&req.source_corpus)) {
        Ok(c) => c,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };

    // Reject the no-op self-copy (spec open question, leaning reject).
    if source.name == global.name {
        return DaemonResponse::invalid_params(format!(
            "source corpus {:?} is already the global corpus; nothing to globalize",
            source.name,
        ));
    }

    // Read the source entry (symlink-safe).
    let source_root = match first_corpus_root(&source) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let source_rel = match safe_relative_path(&req.path) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let source_abs = source_root.join(&source_rel);
    if let Err(e) = ensure_corpus_allows_file(&source, &source_abs) {
        return DaemonResponse::invalid_params(e.into_inner());
    }

    let read_root = source_root.clone();
    let read_rel = source_rel.clone();
    let read_err_rel = source_rel.clone();
    let read = tokio::task::spawn_blocking(move || read_no_follow(&read_root, &read_rel)).await;
    let bytes = match read {
        Ok(Ok(b)) => b,
        Ok(Err(WriteError { kind, source })) => {
            return map_read_error(kind, source, &read_err_rel);
        }
        Err(join_err) => {
            return DaemonResponse::internal(format!("read task panicked: {join_err}"));
        }
    };

    // Destination path defaults to the source path.
    let dest_raw = req.dest_path.as_deref().unwrap_or(req.path.as_str());
    let dest_root = match first_corpus_root(&global) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let dest_rel = match safe_relative_path(dest_raw) {
        Ok(r) => r,
        Err(e) => return DaemonResponse::invalid_params(e.into_inner()),
    };
    let dest_check = dest_root.join(&dest_rel);
    if let Err(e) = ensure_corpus_allows_file(&global, &dest_check) {
        return DaemonResponse::invalid_params(e.into_inner());
    }

    let guard = match state.acquire_mutation_guard(&global.name).await {
        Ok(g) => g,
        Err(msg) => return DaemonResponse::internal(msg),
    };

    let write_root = dest_root.clone();
    let write_rel = dest_rel.clone();
    let write_err_rel = dest_rel.clone();
    let overwrite = req.overwrite;
    let written = tokio::task::spawn_blocking(move || {
        atomic_write_no_follow(&write_root, &write_rel, &bytes, overwrite)
    })
    .await;
    let dest_abs = match written {
        Ok(Ok(p)) => p,
        Ok(Err(WriteError { kind, source })) => {
            let resp = match kind {
                WriteErrorKind::Exists => DaemonResponse::invalid_params(format!(
                    "{} already exists in the global corpus; pass overwrite=true to replace it",
                    write_err_rel.display()
                )),
                WriteErrorKind::Symlink | WriteErrorKind::InvalidPath => {
                    DaemonResponse::invalid_params(format!(
                        "refusing unsafe path {}: {source}",
                        write_err_rel.display()
                    ))
                }
                WriteErrorKind::Io => DaemonResponse::internal(source.to_string()),
            };
            drop(guard);
            return resp;
        }
        Err(join_err) => {
            drop(guard);
            return DaemonResponse::internal(format!("write task panicked: {join_err}"));
        }
    };

    // Reindex the global corpus row for the just-written file. Empty content
    // produces zero chunks; the indexer counts it as files_skipped_empty.
    let stats = {
        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) => {
                    drop(guard);
                    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, &global, &dest_abs).await {
            Ok(s) => s,
            Err(e) => {
                drop(guard);
                return DaemonResponse::internal(e.to_string());
            }
        }
    };
    drop(guard);

    let report = IndexReport {
        corpora: vec![CorpusReport {
            name: global.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(&GlobalizeMarkdownResult {
        source_corpus: source.name,
        source_path: source_rel.to_string_lossy().into_owned(),
        global_corpus: global.name,
        dest_path: dest_rel.to_string_lossy().into_owned(),
        absolute_path: dest_abs.to_string_lossy().into_owned(),
        indexed: report,
    })
}

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 mtime_ms = modified
        .duration_since(UNIX_EPOCH)
        .map_err(|_| anyhow::anyhow!("pre-epoch mtime on {}", file.display()))?
        .as_millis() as i64;
    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)
}

/// Best-effort `mkdir -p` on daemon-managed corpus roots so a fresh
/// repository wiki (which only exists logically until the first write)
/// doesn't blow up the first `list_files` / `index` call. Restricted to
/// `repo:*:wiki` corpora so a typo'd `[[corpus]] paths = ...` surfaces as a
/// clear scan error instead of silently creating an empty directory and
/// reporting success.
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);
    }
}

/// Sum `extra` into `into` so the daemon's IndexReport reflects both the
/// initial single-file write and the cascade of index.md rewrites that
/// followed it. Without this, the auto-built indexes would be silently
/// re-embedded but the report would still claim `files_upserted = 1`.
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;
}

/// Walk from `root` down to the parent of `file_relative`, rewriting each
/// directory's `index.md` between INDEX-START / INDEX-END markers. Returns
/// the aggregate of `index_single_file` stats for every regenerated index
/// file so the caller can roll them into the response. The dir that owns
/// `file_relative` is skipped when the file itself IS that dir's `index.md`
/// — the LLM's verbatim write is the final word for the leaf file, and
/// regenerating would clobber it.
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);
        // Skip the dir that owns the file we just wrote if that file IS
        // its index.md — the author's verbatim write wins. `Path::parent`
        // returns `Some(Path::new(""))` for a top-level filename, so this
        // covers root-level `index.md` writes too via `root.join("") == root`.
        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 => {}
        }

        // Use the same atomic-write-no-follow path that AddMarkdown uses so
        // the auto-index inherits its symlink safety.
        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| 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,
                ));
            }
        };

        // Refresh LanceDB rows for the just-rewritten index.md. Embeddings
        // are opt-in: when enabled, the embedder must load (a cold-cache or
        // network failure fails the mutation, same shape as the primary
        // add_markdown path); when disabled, reindex lexical-only.
        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)
}

/// True when `corpus.name` is a derived `repo:{name}:wiki` corpus produced
/// by `effective_corpora()`. The daemon owns those directories (creates
/// them on demand, enforces no-symlink safety), so a few helpers behave
/// differently for them than for user-declared `[[corpus]]` entries.
fn is_wiki_corpus(corpus: &CorpusConfig) -> bool {
    corpus.name.starts_with("repo:") && corpus.name.ends_with(":wiki")
}

/// Refuse to operate on a wiki corpus whose `.hallouminate` parent or
/// `wiki` leaf is a symlink. The repository's `path` is user-configured (so
/// trusted), but `.hallouminate` and `wiki` are daemon-managed names that a
/// malicious repository payload could swap for symlinks before the daemon
/// runs the no-follow component walk inside the sandbox. Best-effort
/// pre-flight — TOCTOU-vulnerable between this check and the subsequent
/// open, but the daemon serializes wiki mutations behind the per-corpus
/// mutex so a swap during the narrow window would race the daemon's own
/// consistency model.
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(())
}

/// Shared error mapping for `read_no_follow` failures — keeps
/// `handle_read_markdown` flat while preserving the distinct
/// NotFound / Symlink / IO shapes callers depend on.
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()),
    }
}

/// Shared error mapping for `delete_no_follow` failures — mirrors
/// `map_read_error` so the two handlers share one error vocabulary.
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;

/// Test helper for the corpus-name vocabulary the daemon dispatcher resolves
/// through. Gated behind `#[cfg(test)]` because production handlers reach
/// straight into `effective_corpora()` and never call this name constructor.
#[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())
}

/// Canonical `Ping` reply payload — dispatch() encodes `&"pong"` directly,
/// this helper lets tests match against the same literal without
/// hand-rolling the JSON.
#[cfg(test)]
fn pong_value() -> Value {
    Value::String("pong".into())
}

#[cfg(test)]
mod tests {
    //! Dispatch-level tests. The corpus-boundary helpers
    //! (`safe_relative_path`, `pick_corpus`, `ensure_corpus_allows_file`,
    //! `first_corpus_root`, `atomic_write_no_follow`, `list_corpus_files`)
    //! moved to `crate::domain::corpus::sandbox` and are tested there once,
    //! against a single contract. These tests only cover the daemon-only
    //! helpers (`derived_corpus_name`, `pong_value`).

    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_returns_pong_string_literal() {
        assert_eq!(pong_value(), Value::String("pong".to_string()));
    }

    /// FileEntry is re-exported from the shared sandbox module to keep the
    /// daemon's response struct serializing the same shape as before the
    /// extract — list_files clients depend on the `{ path, absolute_path }`
    /// field names.
    #[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");
    }

    // ── dispatch + resolve_for_cwd integration ──────────────────────────
    //
    // These tests cover AC #2 from .cheese/specs/repo-config-discovery.md:
    // dispatch consumes `req.cwd` via `resolve_for_cwd` on every request,
    // and a discovery / merge failure surfaces to the client as
    // `InvalidParams` (not a silent fall-back to baseline-only).

    use std::path::Path;

    use crate::app::daemon::ErrorKind;

    /// Build a `DaemonState` with a baseline `Config` that points its
    /// ground_dir at a tempdir-local subdir. Embedder load is tolerated
    /// failure on first run (see `DaemonState::open`), so a cold cache
    /// doesn't break tests that don't exercise the embedder.
    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_with_valid_cwd_returns_ok_for_ping() {
        // Discovery walks `req.cwd` upward and finds the tempdir's
        // `.hallouminate/config.toml`; the merge succeeds and `Ping` returns
        // the canonical `Ok { result: "pong" }` payload.
        let tmp = tempfile::tempdir().expect("tempdir");
        let cwd = tmp.path().to_path_buf();
        write_repo_layer(&cwd, "");
        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 echo `pong`");
            }
            DaemonResponse::Err { kind, message } => {
                panic!("ping must succeed; got {kind:?}: {message}");
            }
        }
    }

    #[tokio::test]
    async fn dispatch_with_no_repo_config_returns_config_error() {
        // A `cwd` whose ancestry contains neither `.hallouminate/` nor `.git`
        // until the filesystem root must surface as
        // `DaemonResponse::Err { kind: InvalidParams, .. }` carrying the
        // "filesystem root" discovery-walk error. The dispatcher must NOT
        // fall back to baseline-only here.
        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: cwd.clone(),
            payload: DaemonRequestPayload::Ping,
        };
        let resp = dispatch(&state, req).await;
        match resp {
            DaemonResponse::Err { kind, message } => {
                assert_eq!(
                    kind,
                    ErrorKind::InvalidParams,
                    "discovery failure must map to InvalidParams: {message}",
                );
                // The discovery walk either reaches the filesystem root
                // (most environments) or trips a `.git` boundary in unusual
                // CI sandboxes whose tmp tree sits inside a checkout. Both
                // outcomes prove dispatch refused to silently fall back.
                assert!(
                    message.contains("filesystem root") || message.contains("stopped at repo root"),
                    "discovery error must explain the boundary: {message}",
                );
            }
            DaemonResponse::Ok { result } => {
                panic!("must not fall back to baseline-only; got Ok({result:?})");
            }
        }
    }

    #[tokio::test]
    async fn dispatch_with_scalar_conflict_returns_config_error() {
        // Baseline explicitly sets `embeddings.cache_dir = "/a"`; repo layer
        // explicitly sets `embeddings.cache_dir = "/b"`. `merge_layers`
        // refuses the conflict, and dispatch must propagate that as
        // `InvalidParams` with the offending field named.
        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::Ping,
        };
        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:?})");
            }
        }
    }

    /// AC #7 regression: when the daemon was booted with a known baseline
    /// source path (XDG or `--config PATH`), scalar-conflict messages must
    /// name that path so the user knows which file holds the offending
    /// baseline value. Before this curd, dispatch passed `xdg_path: None`
    /// hard-coded and conflict messages said `"(XDG baseline)"` instead.
    #[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");
        // Construct state with an explicit baseline source path. The path
        // itself doesn't have to be the file the toml came from (the daemon
        // already has the parsed Config in memory by this point); we just
        // need a sentinel string to verify it lands in the diagnostic.
        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::Ping,
        };
        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}",
        );
    }
}