yams/tests/unit/daemon/daemon_metrics_status_test.cpp

// Daemon metrics and status test suite (Catch2)
// Consolidates status/metrics tests: embedding status, plugin degradation, WAL metrics, FSM states
// Covers: DaemonMetrics, StatusResponse serialization, plugin degradation, WAL metrics

#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>

#include <algorithm>
#include <array>
#include <chrono>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <future>
#include <mutex>
#include <random>
#include <span>
#include <stdexcept>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include "../../common/metadata_test_db.h"
#include "../../common/test_helpers_catch2.h"

#include <spdlog/spdlog.h>

#include <yams/app/services/graph_query_service.hpp>
#include <yams/app/services/session_service.hpp>
#include <yams/crypto/hasher.h>
#include <yams/daemon/components/DaemonLifecycleFsm.h>
#include <yams/daemon/components/DaemonMetrics.h>
#include <yams/daemon/components/DatabaseManager.h>
#include <yams/daemon/components/RequestDispatcher.h>
#include <yams/daemon/components/ResourceGovernor.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/WorkCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/daemon_lifecycle.h>
#include <yams/daemon/ipc/fsm_metrics_registry.h>
#include <yams/daemon/ipc/ipc_protocol.h>
#include <yams/daemon/ipc/proto_serializer.h>
#include <yams/daemon/ipc/request_context.h>
#include <yams/daemon/ipc/request_context_registry.h>
#include <yams/daemon/metric_keys.h>
#include <yams/daemon/resource/external_plugin_host.h>
#include <yams/daemon/resource/model_provider.h>
#include <yams/daemon/resource/plugin_host.h>
#include <yams/metadata/knowledge_graph_store.h>
#include <yams/metadata/metadata_repository.h>
#include <yams/metadata/path_utils.h>
#include <yams/vector/vector_database.h>
#include <yams/version.hpp>

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_future.hpp>

using namespace yams;
using namespace yams::daemon;

// =============================================================================
// Test Helpers
// =============================================================================

namespace {
class StubLifecycle : public IDaemonLifecycle {
public:
    explicit StubLifecycle(LifecycleState state = LifecycleState::Ready) {
        snapshot_.state = state;
    }

    LifecycleSnapshot getLifecycleSnapshot() const override {
        ++getLifecycleSnapshotCalls_;
        if (throwOnGetLifecycleSnapshotCall_ > 0 &&
            getLifecycleSnapshotCalls_ == throwOnGetLifecycleSnapshotCall_) {
            throw std::runtime_error("getLifecycleSnapshot boom");
        }
        if (throwOnGetLifecycleSnapshotCount_ > 0) {
            --throwOnGetLifecycleSnapshotCount_;
            throw std::runtime_error("getLifecycleSnapshot boom");
        }
        return snapshot_;
    }
    void setSubsystemDegraded(const std::string& subsystem, bool degraded,
                              const std::string& reason) override {
        if (throwOnSetSubsystemDegraded_) {
            throw std::runtime_error("setSubsystemDegraded boom");
        }
        lastSubsystem_ = subsystem;
        lastDegraded_ = degraded;
        lastReason_ = reason;
        setSubsystemDegradedCalls_++;
    }
    void onDocumentRemoved(const std::string& hash) override { removedHashes_.push_back(hash); }
    void requestShutdown(bool, bool) override {}

    const std::string& lastSubsystem() const { return lastSubsystem_; }
    bool lastDegraded() const { return lastDegraded_; }
    const std::string& lastReason() const { return lastReason_; }
    int setSubsystemDegradedCalls() const { return setSubsystemDegradedCalls_; }
    void resetDegradedState() {
        lastSubsystem_.clear();
        lastDegraded_ = false;
        lastReason_.clear();
        setSubsystemDegradedCalls_ = 0;
    }
    void setThrowOnSetSubsystemDegraded(bool enabled) { throwOnSetSubsystemDegraded_ = enabled; }
    void setThrowOnGetLifecycleSnapshot(bool enabled) {
        throwOnGetLifecycleSnapshotCount_ = enabled ? 1 : 0;
    }
    void setThrowOnGetLifecycleSnapshotCount(int count) {
        throwOnGetLifecycleSnapshotCount_ = count;
    }
    void setThrowOnGetLifecycleSnapshotCall(int call) { throwOnGetLifecycleSnapshotCall_ = call; }
    const std::vector<std::string>& removedHashes() const { return removedHashes_; }

private:
    LifecycleSnapshot snapshot_{};
    std::string lastSubsystem_;
    bool lastDegraded_{false};
    std::string lastReason_;
    int setSubsystemDegradedCalls_{0};
    bool throwOnSetSubsystemDegraded_{false};
    mutable int getLifecycleSnapshotCalls_{0};
    mutable int throwOnGetLifecycleSnapshotCount_{0};
    int throwOnGetLifecycleSnapshotCall_{0};
    std::vector<std::string> removedHashes_;
};

// Minimal stub provider for embedding tests
class StubModelProvider : public IModelProvider {
public:
    explicit StubModelProvider(size_t dim, std::string path)
        : dim_(dim), path_(std::move(path)), available_(true) {}

    Result<std::vector<float>> generateEmbedding(const std::string& text) override {
        lastTextArg_ = text;
        if (auto it = singleTextErrors_.find(text); it != singleTextErrors_.end()) {
            return it->second;
        }
        if (generateEmbeddingError_) {
            return *generateEmbeddingError_;
        }
        if (!singleEmbeddingResult_.empty()) {
            return singleEmbeddingResult_;
        }
        return std::vector<float>(dim_, 0.25f);
    }
    Result<std::vector<std::vector<float>>>
    generateBatchEmbeddings(const std::vector<std::string>& texts) override {
        lastBatchSizeArg_ = texts.size();
        if (!texts.empty()) {
            lastTextArg_ = texts.front();
        }
        if (generateBatchError_ && !isWarmupBatch(texts)) {
            return *generateBatchError_;
        }
        if (!batchEmbeddingResult_.empty()) {
            return batchEmbeddingResult_;
        }
        return std::vector<std::vector<float>>(lastBatchSizeArg_, std::vector<float>(dim_, 0.5f));
    }
    Result<std::vector<float>> generateEmbeddingFor(const std::string&,
                                                    const std::string& text) override {
        lastTextArg_ = text;
        return generateEmbedding(text);
    }
    Result<std::vector<std::vector<float>>>
    generateBatchEmbeddingsFor(const std::string&, const std::vector<std::string>& texts) override {
        lastBatchSizeArg_ = texts.size();
        if (!texts.empty()) {
            lastTextArg_ = texts.front();
        }
        return generateBatchEmbeddings(texts);
    }
    Result<void> loadModel(const std::string& modelName) override {
        if (loadError_) {
            return *loadError_;
        }
        lastOptionsModel_.clear();
        lastOptionsJson_.clear();
        loaded_.push_back(modelName);
        return Result<void>();
    }
    Result<void> loadModelWithOptions(const std::string& modelName,
                                      const std::string& optionsJson) override {
        if (loadError_) {
            return *loadError_;
        }
        lastOptionsModel_ = modelName;
        lastOptionsJson_ = optionsJson;
        loaded_.push_back(modelName);
        return Result<void>();
    }
    Result<void> unloadModel(const std::string& modelName) override {
        auto it = std::find(loaded_.begin(), loaded_.end(), modelName);
        if (it == loaded_.end()) {
            return ErrorCode::NotFound;
        }
        loaded_.erase(it);
        return Result<void>();
    }
    bool isModelLoaded(const std::string& modelName) const override {
        if (throwOnIsModelLoadedCount_ > 0) {
            --throwOnIsModelLoadedCount_;
            throw std::runtime_error("isModelLoaded boom");
        }
        return std::find(loaded_.begin(), loaded_.end(), modelName) != loaded_.end();
    }
    std::vector<std::string> getLoadedModels() const override { return loaded_; }
    size_t getLoadedModelCount() const override { return loaded_.size(); }
    Result<ModelInfo> getModelInfo(const std::string& modelName) const override {
        if (failModelInfo_) {
            return ErrorCode::NotFound;
        }
        if (!isModelLoaded(modelName))
            return ErrorCode::NotFound;
        ModelInfo mi;
        mi.name = modelName;
        mi.path = path_;
        mi.embeddingDim = dim_;
        return mi;
    }
    size_t getEmbeddingDim(const std::string&) const override { return dim_; }
    std::shared_ptr<vector::EmbeddingGenerator>
    getEmbeddingGenerator(const std::string& = "") override {
        return nullptr;
    }
    std::string getProviderName() const override { return "StubProvider"; }
    std::string getProviderVersion() const override { return "vtest"; }
    bool isAvailable() const override { return available_; }
    size_t getMemoryUsage() const override {
        if (throwOnMemoryUsage_) {
            throw std::runtime_error("memory boom");
        }
        return 0;
    }
    void releaseUnusedResources() override {}
    void shutdown() override {}
    void setAvailable(bool available) { available_ = available; }
    const std::string& lastOptionsModel() const { return lastOptionsModel_; }
    const std::string& lastOptionsJson() const { return lastOptionsJson_; }
    void setLoadError(Error error) { loadError_ = std::move(error); }
    void clearLoadError() { loadError_.reset(); }
    void setThrowOnMemoryUsage(bool enabled) { throwOnMemoryUsage_ = enabled; }
    void setFailModelInfo(bool enabled) { failModelInfo_ = enabled; }
    void setThrowOnIsModelLoaded(bool enabled) { throwOnIsModelLoadedCount_ = enabled ? 1 : 0; }
    void setThrowOnIsModelLoadedCount(int count) { throwOnIsModelLoadedCount_ = count; }
    void setSingleEmbeddingResult(std::vector<float> embedding) {
        singleEmbeddingResult_ = std::move(embedding);
    }
    void setBatchEmbeddingResult(std::vector<std::vector<float>> embeddings) {
        batchEmbeddingResult_ = std::move(embeddings);
    }
    void setGenerateEmbeddingError(Error error) { generateEmbeddingError_ = std::move(error); }
    void setGenerateBatchError(Error error) { generateBatchError_ = std::move(error); }
    void setSingleTextError(const std::string& text, Error error) {
        singleTextErrors_[text] = std::move(error);
    }
    void clearSingleTextErrors() { singleTextErrors_.clear(); }

private:
    static bool isWarmupBatch(const std::vector<std::string>& texts) {
        static const std::array<std::string, 4> kWarmupTexts = {
            "warmup text one", "warmup text two", "warmup text three", "warmup text four"};
        return texts.size() == kWarmupTexts.size() &&
               std::equal(texts.begin(), texts.end(), kWarmupTexts.begin());
    }

    size_t dim_;
    std::string path_;
    bool available_;
    std::vector<std::string> loaded_;
    std::string lastOptionsModel_;
    std::string lastOptionsJson_;
    std::optional<Error> loadError_;
    std::optional<Error> generateEmbeddingError_;
    std::optional<Error> generateBatchError_;
    bool throwOnMemoryUsage_{false};
    bool failModelInfo_{false};
    mutable int throwOnIsModelLoadedCount_{0};
    std::vector<float> singleEmbeddingResult_;
    std::vector<std::vector<float>> batchEmbeddingResult_;
    std::unordered_map<std::string, Error> singleTextErrors_;
    mutable std::string lastTextArg_;
    mutable std::size_t lastBatchSizeArg_{0};
};

using yams::test::ScopedEnvVar;

class TuneAdvisorListGuard {
public:
    TuneAdvisorListGuard(uint32_t inflightLimit, uint32_t admissionWaitMs)
        : originalInflightLimit_(TuneAdvisor::listInflightLimit()),
          originalAdmissionWaitMs_(TuneAdvisor::listAdmissionWaitMs()) {
        TuneAdvisor::setListInflightLimit(inflightLimit);
        TuneAdvisor::setListAdmissionWaitMs(admissionWaitMs);
    }

    ~TuneAdvisorListGuard() {
        TuneAdvisor::setListInflightLimit(originalInflightLimit_);
        TuneAdvisor::setListAdmissionWaitMs(originalAdmissionWaitMs_);
    }

private:
    uint32_t originalInflightLimit_;
    uint32_t originalAdmissionWaitMs_;
};

template <typename Fn> class ScopeExit {
public:
    explicit ScopeExit(Fn fn) : fn_(std::move(fn)) {}
    ~ScopeExit() { fn_(); }

private:
    Fn fn_;
};

std::shared_ptr<metadata::ConnectionPool> getPruneTestPool() {
    static std::shared_ptr<metadata::ConnectionPool> pool = [] {
        metadata::ConnectionPoolConfig cfg{};
        cfg.enableWAL = false;
        cfg.enableForeignKeys = false;
        auto p = std::make_shared<metadata::ConnectionPool>(":memory:", cfg);
        (void)p->initialize();
        return p;
    }();
    return pool;
}

std::filesystem::path makeTempMetadataDbPath(const std::string& prefix) {
    return yams::test::migrated_metadata_db_template().clone(prefix);
}

std::shared_ptr<metadata::MetadataRepository>
makeReadyMetadataRepo(const std::shared_ptr<metadata::ConnectionPool>& pool) {
    return std::make_shared<metadata::MetadataRepository>(
        *pool, nullptr, metadata::MetadataRepository::SchemaBootstrapMode::AssumeReady);
}

class StubPruneMetadataRepository : public metadata::MetadataRepository {
public:
    StubPruneMetadataRepository() : metadata::MetadataRepository(*getPruneTestPool()) {}

    void addDocument(const metadata::DocumentInfo& doc) {
        std::lock_guard<std::mutex> lk(mu_);
        docs_.push_back(doc);
        docsByHash_[doc.sha256Hash] = doc;
        docsById_[doc.id] = doc;
    }

    void setQueryError(Error error) { queryError_ = std::move(error); }
    void setThrowOnQuery(std::string message) { throwOnQuery_ = std::move(message); }
    void setMissingHash(const std::string& hash) { missingHashes_.insert(hash); }
    void setThrowOnLookupHash(const std::string& hash, std::string message) {
        throwOnLookup_[hash] = std::move(message);
    }
    void setDeleteError(int64_t id, Error error) { deleteErrors_[id] = std::move(error); }
    void setSnapshots(std::vector<std::string> snapshots) { snapshots_ = std::move(snapshots); }
    void setSnapshotError(Error error) { snapshotError_ = std::move(error); }
    void setSnapshotDocuments(std::vector<metadata::DocumentInfo> docs) {
        snapshotDocs_ = std::move(docs);
    }
    void setSnapshotDocumentsError(Error error) { snapshotDocsError_ = std::move(error); }
    void setExactPathDocument(std::optional<metadata::DocumentInfo> doc) {
        std::lock_guard<std::mutex> lk(mu_);
        exactPathDocument_ = std::move(doc);
    }
    void setExactPathError(Error error) {
        std::lock_guard<std::mutex> lk(mu_);
        exactPathError_ = std::move(error);
    }
    void setAllMetadata(int64_t id,
                        std::unordered_map<std::string, metadata::MetadataValue> values) {
        std::lock_guard<std::mutex> lk(mu_);
        metadataById_[id] = std::move(values);
    }
    void setAllMetadataError(int64_t id, Error error) {
        std::lock_guard<std::mutex> lk(mu_);
        metadataByIdErrors_[id] = std::move(error);
    }
    void setPathTreeDocuments(std::vector<metadata::DocumentInfo> docs) {
        std::lock_guard<std::mutex> lk(mu_);
        pathTreeDocs_ = std::move(docs);
    }
    void setMetadataValueCounts(
        std::unordered_map<std::string, std::vector<metadata::MetadataValueCount>> counts) {
        metadataValueCounts_ = std::move(counts);
    }
    void setMetadataValueCountsError(Error error) { metadataValueCountsError_ = std::move(error); }

    std::size_t deleteCallCount() const {
        std::lock_guard<std::mutex> lk(mu_);
        return deleteCallCount_;
    }

    std::vector<int64_t> deletedIds() const {
        std::lock_guard<std::mutex> lk(mu_);
        return deletedIds_;
    }

    Result<std::vector<metadata::DocumentInfo>>
    queryDocuments(const metadata::DocumentQueryOptions&) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (throwOnQuery_) {
            throw std::runtime_error(*throwOnQuery_);
        }
        if (queryError_) {
            return *queryError_;
        }
        return docs_;
    }

    Result<std::vector<metadata::DocumentInfo>>
    findDocumentsBySnapshot(const std::string&) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (snapshotDocsError_) {
            return *snapshotDocsError_;
        }
        return snapshotDocs_;
    }

    Result<std::vector<std::string>> getSnapshots() override {
        std::lock_guard<std::mutex> lk(mu_);
        if (snapshotError_) {
            return *snapshotError_;
        }
        return snapshots_;
    }

    Result<std::unordered_map<std::string, std::vector<metadata::MetadataValueCount>>>
    getMetadataValueCounts(const std::vector<std::string>&,
                           const metadata::DocumentQueryOptions&) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (metadataValueCountsError_) {
            return *metadataValueCountsError_;
        }
        return metadataValueCounts_;
    }

    Result<std::optional<metadata::DocumentInfo>>
    getDocumentByHash(const std::string& hash) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (auto it = throwOnLookup_.find(hash); it != throwOnLookup_.end()) {
            throw std::runtime_error(it->second);
        }
        if (missingHashes_.count(hash) > 0) {
            return std::optional<metadata::DocumentInfo>(std::nullopt);
        }
        if (auto it = docsByHash_.find(hash); it != docsByHash_.end()) {
            return std::optional<metadata::DocumentInfo>(it->second);
        }
        return std::optional<metadata::DocumentInfo>(std::nullopt);
    }

    Result<std::unordered_map<std::string, metadata::MetadataValue>>
    getAllMetadata(int64_t documentId) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (auto it = metadataByIdErrors_.find(documentId); it != metadataByIdErrors_.end()) {
            return it->second;
        }
        if (auto it = metadataById_.find(documentId); it != metadataById_.end()) {
            return it->second;
        }
        return std::unordered_map<std::string, metadata::MetadataValue>{};
    }

    Result<std::optional<metadata::DocumentInfo>>
    findDocumentByExactPath(const std::string&) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (exactPathError_) {
            return *exactPathError_;
        }
        if (exactPathDocument_.has_value()) {
            return exactPathDocument_;
        }
        return std::optional<metadata::DocumentInfo>(std::nullopt);
    }

    Result<std::vector<metadata::DocumentInfo>>
    findDocumentsByPathTreePrefix(std::string_view, bool = true, int limit = 0) override {
        std::lock_guard<std::mutex> lk(mu_);
        if (limit > 0 && static_cast<std::size_t>(limit) < pathTreeDocs_.size()) {
            return std::vector<metadata::DocumentInfo>(pathTreeDocs_.begin(),
                                                       pathTreeDocs_.begin() + limit);
        }
        return pathTreeDocs_;
    }

    Result<void> deleteDocument(int64_t id) override {
        std::lock_guard<std::mutex> lk(mu_);
        deleteCallCount_++;
        if (auto it = deleteErrors_.find(id); it != deleteErrors_.end()) {
            return it->second;
        }
        deletedIds_.push_back(id);
        if (auto it = docsById_.find(id); it != docsById_.end()) {
            docsByHash_.erase(it->second.sha256Hash);
            docsById_.erase(it);
        }
        docs_.erase(
            std::remove_if(docs_.begin(), docs_.end(),
                           [id](const metadata::DocumentInfo& doc) { return doc.id == id; }),
            docs_.end());
        return Result<void>();
    }

private:
    mutable std::mutex mu_;
    std::vector<metadata::DocumentInfo> docs_;
    std::unordered_map<std::string, metadata::DocumentInfo> docsByHash_;
    std::unordered_map<int64_t, metadata::DocumentInfo> docsById_;
    std::optional<Error> queryError_;
    std::optional<Error> snapshotError_;
    std::optional<Error> snapshotDocsError_;
    std::optional<Error> metadataValueCountsError_;
    std::optional<std::string> throwOnQuery_;
    std::unordered_set<std::string> missingHashes_;
    std::unordered_map<std::string, std::string> throwOnLookup_;
    std::unordered_map<int64_t, Error> deleteErrors_;
    std::optional<metadata::DocumentInfo> exactPathDocument_;
    std::optional<Error> exactPathError_;
    std::unordered_map<int64_t, std::unordered_map<std::string, metadata::MetadataValue>>
        metadataById_;
    std::unordered_map<int64_t, Error> metadataByIdErrors_;
    std::vector<metadata::DocumentInfo> pathTreeDocs_;
    std::vector<std::string> snapshots_;
    std::vector<metadata::DocumentInfo> snapshotDocs_;
    std::unordered_map<std::string, std::vector<metadata::MetadataValueCount>> metadataValueCounts_;
    std::size_t deleteCallCount_{0};
    std::vector<int64_t> deletedIds_;
};

class StubContentStore : public api::IContentStore {
public:
    void setBlob(const std::string& hash, const std::string& content) {
        blobs_[hash] = std::vector<std::byte>(reinterpret_cast<const std::byte*>(content.data()),
                                              reinterpret_cast<const std::byte*>(content.data()) +
                                                  content.size());
    }

    void setRemoveResult(const std::string& hash, Result<bool> result) {
        removeResults_[hash] = std::move(result);
    }

    bool hasBlob(const std::string& hash) const { return blobs_.contains(hash); }

    std::vector<std::string> removedHashes() const { return removedHashes_; }

    Result<api::StoreResult> storeBytes(std::span<const std::byte> data,
                                        const api::ContentMetadata&) override {
        std::string hash = "stub-hash-" + std::to_string(blobs_.size() + 1);
        blobs_[hash] = std::vector<std::byte>(data.begin(), data.end());
        api::StoreResult result;
        result.contentHash = hash;
        result.bytesStored = blobs_[hash].size();
        return result;
    }

    Result<std::vector<std::byte>> retrieveBytes(const std::string& hash) override {
        auto it = blobs_.find(hash);
        if (it == blobs_.end()) {
            return Error{ErrorCode::NotFound, "content not found"};
        }
        return it->second;
    }

    Result<std::vector<std::byte>> retrieveBytesPrefix(const std::string& hash,
                                                       std::size_t maxBytes) override {
        auto bytes = retrieveBytes(hash);
        if (!bytes) {
            return bytes.error();
        }
        const auto prefixSize = std::min(maxBytes, bytes.value().size());
        bytes.value().resize(prefixSize);
        return bytes;
    }

    Result<api::IContentStore::RawContent> retrieveRaw(const std::string& hash) override {
        auto bytes = retrieveBytes(hash);
        if (!bytes) {
            return bytes.error();
        }
        api::IContentStore::RawContent raw;
        raw.data = std::move(bytes.value());
        return raw;
    }

    std::future<Result<api::IContentStore::RawContent>>
    retrieveRawAsync(const std::string& hash) override {
        return std::async(std::launch::deferred, [this, hash]() { return retrieveRaw(hash); });
    }

    Result<api::StoreResult> store(const std::filesystem::path& path, const api::ContentMetadata&,
                                   api::ProgressCallback) override {
        std::ifstream ifs(path, std::ios::binary);
        if (!ifs.good()) {
            return Error{ErrorCode::FileNotFound, "stub file not found"};
        }
        std::vector<std::byte> bytes;
        char ch = 0;
        while (ifs.get(ch)) {
            bytes.push_back(static_cast<std::byte>(ch));
        }
        if (!ifs.eof()) {
            return Error{ErrorCode::IOError, "stub read failed"};
        }
        return storeBytes(bytes, {});
    }
    Result<api::RetrieveResult> retrieve(const std::string&, const std::filesystem::path&,
                                         api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }
    Result<api::StoreResult> storeStream(std::istream&, const api::ContentMetadata&,
                                         api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }
    Result<api::RetrieveResult> retrieveStream(const std::string&, std::ostream&,
                                               api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }
    Result<bool> exists(const std::string&) const override { return ErrorCode::NotImplemented; }
    Result<bool> remove(const std::string& hash) override {
        if (auto it = removeResults_.find(hash); it != removeResults_.end()) {
            return it->second;
        }
        removedHashes_.push_back(hash);
        blobs_.erase(hash);
        return true;
    }
    Result<api::ContentMetadata> getMetadata(const std::string&) const override {
        return ErrorCode::NotImplemented;
    }
    Result<void> updateMetadata(const std::string&, const api::ContentMetadata&) override {
        return ErrorCode::NotImplemented;
    }
    std::vector<Result<api::StoreResult>>
    storeBatch(const std::vector<std::filesystem::path>&,
               const std::vector<api::ContentMetadata>&) override {
        return {};
    }
    std::vector<Result<bool>> removeBatch(const std::vector<std::string>&) override { return {}; }
    api::ContentStoreStats getStats() const override { return {}; }
    api::HealthStatus checkHealth() const override { return {}; }
    Result<void> verify(api::ProgressCallback) override { return ErrorCode::NotImplemented; }
    Result<void> compact(api::ProgressCallback) override { return ErrorCode::NotImplemented; }
    Result<void> garbageCollect(api::ProgressCallback) override {
        return ErrorCode::NotImplemented;
    }

private:
    std::unordered_map<std::string, std::vector<std::byte>> blobs_;
    std::unordered_map<std::string, Result<bool>> removeResults_;
    std::vector<std::string> removedHashes_;
};

class StubGraphQueryService : public app::services::IGraphQueryService {
public:
    void addConnectedDocument(std::string hash, std::string relation, int distance = 1,
                              double relevance = 0.0) {
        app::services::GraphConnectedNode node{};
        node.nodeMetadata.documentHash = std::move(hash);
        node.distance = distance;
        node.relevanceScore = relevance;
        app::services::GraphEdgeDescriptor edge{};
        edge.relation = std::move(relation);
        node.connectingEdges.push_back(std::move(edge));
        response_.allConnectedNodes.push_back(std::move(node));
    }

    Result<app::services::GraphQueryResponse>
    query(const app::services::GraphQueryRequest&) override {
        return response_;
    }

    Result<app::services::ListSnapshotsResponse>
    listSnapshots(const app::services::ListSnapshotsRequest&) override {
        return app::services::ListSnapshotsResponse{};
    }

    Result<app::services::PathHistoryResponse>
    getPathHistory(const app::services::PathHistoryRequest&) override {
        return app::services::PathHistoryResponse{};
    }

    Result<std::optional<std::int64_t>> resolveToNodeId(const std::string&) override {
        return std::optional<std::int64_t>(std::nullopt);
    }

private:
    app::services::GraphQueryResponse response_{};
};

std::filesystem::path makeTempDir(const std::string& prefix);

metadata::DocumentInfo makeGraphDispatcherDocument(const std::filesystem::path& path,
                                                   const std::string& hash) {
    metadata::DocumentInfo info;
    info.filePath = path.string();
    info.fileName = path.filename().string();
    info.fileExtension = path.extension().string();
    info.fileSize = static_cast<std::int64_t>(std::filesystem::file_size(path));
    info.sha256Hash = hash;
    info.mimeType = "text/plain";
    const auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
    info.createdTime = now;
    info.modifiedTime = now;
    info.indexedTime = now;
    info.contentExtracted = true;
    info.extractionStatus = metadata::ExtractionStatus::Success;
    metadata::populatePathDerivedFields(info);
    return info;
}

struct GraphDispatcherFixture {
    GraphDispatcherFixture() {
        static const bool kSuppressInfoLogs = [] {
            spdlog::set_level(spdlog::level::warn);
            return true;
        }();
        (void)kSuppressInfoLogs;
        testDir = makeTempDir("yams_graph_dispatcher_");
        dbPath = makeTempMetadataDbPath("yams_graph_dispatcher_db_");
        cfg.dataDir = testDir;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        svc = std::make_unique<ServiceManager>(cfg, state, lifecycleFsm);
        dispatcher = std::make_unique<RequestDispatcher>(&lifecycle, svc.get(), &state);
    }

    ~GraphDispatcherFixture() {
        dispatcher.reset();
        svc.reset();
        if (writeWork) {
            writeWork.reset();
        }
        writeIo.stop();
        if (writeThread.joinable()) {
            writeThread.join();
        }
        kgStore.reset();
        repo.reset();
        pool.reset();
        std::error_code ec;
        std::filesystem::remove_all(testDir, ec);
        yams::test::remove_sqlite_artifacts(dbPath);
    }

    void initMetadata(bool withKg = true) {
        metadata::ConnectionPoolConfig poolConfig{};
        poolConfig.minConnections = 1;
        poolConfig.maxConnections = 2;
        pool = std::make_shared<metadata::ConnectionPool>(dbPath.string(), poolConfig);
        REQUIRE(pool->initialize().has_value());

        repo = makeReadyMetadataRepo(pool);
        svc->__test_setMetadataRepo(repo);

        if (withKg) {
            metadata::KnowledgeGraphStoreConfig kgConfig{};
            kgConfig.enable_alias_fts = true;
            kgConfig.enable_wal = false;
            auto kgResult = metadata::makeSqliteKnowledgeGraphStore(*pool, kgConfig);
            REQUIRE(kgResult.has_value());
            kgStore = std::shared_ptr<metadata::KnowledgeGraphStore>(std::move(kgResult).value());
            repo->setKnowledgeGraphStore(kgStore);

            writeIo.restart();
            writeWork = std::make_unique<
                boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
                writeIo.get_executor());
            writeThread = std::thread([this] { writeIo.run(); });
            auto writeCoord = std::make_unique<WriteCoordinator>(writeIo, kgStore, repo);
            writeCoord->start();
            svc->__test_setWriteCoordinator(std::move(writeCoord));
        }
    }

    int64_t upsertNode(const std::string& key, const std::string& label, const std::string& type,
                       const std::string& properties = {}) {
        REQUIRE(kgStore);
        metadata::KGNode node;
        node.nodeKey = key;
        node.label = label;
        node.type = type;
        if (!properties.empty()) {
            node.properties = properties;
        }
        auto result = kgStore->upsertNode(node);
        REQUIRE(result.has_value());
        return result.value();
    }

    void addEdge(int64_t srcNodeId, int64_t dstNodeId, const std::string& relation,
                 float weight = 1.0f, const std::string& properties = {}) {
        REQUIRE(kgStore);
        metadata::KGEdge edge;
        edge.srcNodeId = srcNodeId;
        edge.dstNodeId = dstNodeId;
        edge.relation = relation;
        edge.weight = weight;
        if (!properties.empty()) {
            edge.properties = properties;
        }
        REQUIRE(kgStore->addEdge(edge).has_value());
    }

    void execSql(const std::string& sql) {
        REQUIRE(pool);
        auto result = pool->withConnection(
            [&](metadata::Database& db) -> Result<void> { return db.execute(sql); });
        REQUIRE(result.has_value());
    }

    std::filesystem::path testDir;
    std::filesystem::path dbPath;
    DaemonConfig cfg;
    StubLifecycle lifecycle;
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    std::unique_ptr<ServiceManager> svc;
    std::unique_ptr<RequestDispatcher> dispatcher;
    boost::asio::io_context writeIo;
    std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>
        writeWork;
    std::thread writeThread;
    std::shared_ptr<metadata::ConnectionPool> pool;
    std::shared_ptr<metadata::MetadataRepository> repo;
    std::shared_ptr<metadata::KnowledgeGraphStore> kgStore;
};

std::filesystem::path makeTempDir(const std::string& prefix) {
    auto base = std::filesystem::temp_directory_path();
    auto dir = base / (prefix + std::to_string(std::random_device{}()));
    std::filesystem::create_directories(dir);
    return dir;
}

std::filesystem::path writeTempFile(const std::string& prefix, const std::string& name,
                                    const std::string& content) {
    auto path = makeTempDir(prefix) / name;
    std::ofstream out(path, std::ios::binary);
    REQUIRE(out.good());
    out << content;
    out.close();
    REQUIRE(out.good());
    return path;
}

std::filesystem::path createMockExternalPlugin(const std::filesystem::path& baseDir,
                                               const std::string& name) {
    auto pluginDir = baseDir / name;
    std::filesystem::create_directories(pluginDir);

    {
        std::ofstream pluginFile(pluginDir / "plugin.py");
        pluginFile << R"PY(#!/usr/bin/env python3
import json
import sys


def handle_request(req):
    method = req.get("method", "")
    if method == "handshake.manifest":
        return {
            "name": "dispatcher_external_plugin",
            "version": "1.0.0",
            "interfaces": ["content_extractor_v1"]
        }
    if method == "plugin.init":
        return {"status": "initialized"}
    if method == "plugin.health":
        return {"status": "ok"}
    if method == "plugin.shutdown":
        return {"status": "ok"}
    raise ValueError(f"Method not found: {method}")


for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    req_id = None
    try:
        req = json.loads(line)
        req_id = req.get("id")
        response = {"jsonrpc": "2.0", "id": req_id, "result": handle_request(req)}
    except Exception as exc:
        response = {
            "jsonrpc": "2.0",
            "id": req_id,
            "error": {"code": -32603, "message": str(exc)}
        }
    print(json.dumps(response), flush=True)
)PY";
    }

    {
        std::ofstream manifestFile(pluginDir / "yams-plugin.json");
        manifestFile << R"JSON({
  "name": "dispatcher_external_plugin",
  "version": "1.0.0",
  "interfaces": ["content_extractor_v1"],
  "entry": {
    "fallback_cmd": )JSON";
#ifdef _WIN32
        manifestFile << R"JSON(["python", "-u", "${plugin_dir}/plugin.py"])JSON";
#else
        manifestFile << R"JSON(["/usr/bin/env", "python3", "-u", "${plugin_dir}/plugin.py"])JSON";
#endif
        manifestFile << R"JSON(,
    "env": {
      "PYTHONUNBUFFERED": "1"
    }
  }
})JSON";
    }

    return pluginDir;
}

std::string makePythonShimPath() {
    auto shimDir = makeTempDir("yams_python_shim_");
    auto shim = shimDir / "python";

    {
        std::ofstream shimFile(shim);
        shimFile << "#!/bin/sh\n"
                    "exec /usr/bin/env python3 \"$@\"\n";
    }

    std::filesystem::permissions(shim,
                                 std::filesystem::perms::owner_read |
                                     std::filesystem::perms::owner_write |
                                     std::filesystem::perms::owner_exec,
                                 std::filesystem::perm_options::replace);

    if (const char* currentPath = std::getenv("PATH"); currentPath && *currentPath) {
        return shimDir.string() + ":" + currentPath;
    }
    return shimDir.string() + ":/usr/bin:/bin";
}

Response dispatchRequest(RequestDispatcher& dispatcher, const Request& req) {
    boost::asio::io_context ioc;
    auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(req), boost::asio::use_future);
    ioc.run();
    return fut.get();
}

std::shared_ptr<SpscQueue<InternalEventBus::StoreDocumentTask>>
getStoreDocumentTaskChannel(std::size_t capacity = 0) {
    auto existing = InternalEventBus::instance().get_channel<InternalEventBus::StoreDocumentTask>(
        "store_document_tasks");
    if (existing) {
        return existing;
    }
    if (capacity == 0) {
        capacity = static_cast<std::size_t>(TuneAdvisor::storeDocumentChannelCapacity());
    }
    return InternalEventBus::instance().get_or_create_channel<InternalEventBus::StoreDocumentTask>(
        "store_document_tasks", capacity);
}

void drainStoreDocumentTasks() {
    auto channel = InternalEventBus::instance().get_channel<InternalEventBus::StoreDocumentTask>(
        "store_document_tasks");
    if (!channel) {
        return;
    }
    InternalEventBus::StoreDocumentTask task;
    while (channel->try_pop(task)) {
    }
}

template <typename Predicate> bool waitForCondition(Predicate&& predicate, int attempts = 100) {
    for (int i = 0; i < attempts; ++i) {
        if (predicate()) {
            return true;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(20));
    }
    return predicate();
}
} // namespace

// =============================================================================
// Embedding Status Tests
// =============================================================================

TEST_CASE("DaemonMetrics: Embedding provider status", "[daemon][metrics][embedding]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_metrics_embed_");
    ServiceManager svc(cfg, state, lifecycleFsm);

    SECTION("Snapshot includes embedding backend information") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/test-model.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setAdoptedProviderPluginName("stub");

        // Model provider is now directly used - no need to ensure generator

        DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
        auto snap = metrics.getSnapshot();

        REQUIRE(snap != nullptr);

        // Backend should be "plugin:<name>" format or "unknown"
        bool validBackend =
            (snap->embeddingBackend == "unknown" || snap->embeddingBackend.starts_with("plugin:"));
        REQUIRE(validBackend);
    }

    SECTION("Metrics snapshot doesn't crash without provider") {
        // No provider set
        DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
        auto snap = metrics.getSnapshot();

        REQUIRE(snap != nullptr);
        REQUIRE(!snap->embeddingBackend.empty());
    }
}

// =============================================================================
// WAL Metrics Tests
// =============================================================================

TEST_CASE("DaemonMetrics: WAL metrics in GetStats", "[daemon][metrics][wal]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_metrics_wal_");

    YamsDaemon daemon(cfg);
    DaemonLifecycleAdapter lifecycleAdapter(&daemon);
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);

    SECTION("GetStats includes WAL metric keys") {
        GetStatsRequest req;
        Request r = req;

        boost::asio::io_context ioc;
        auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
        ioc.run();
        auto resp = fut.get();

        REQUIRE(std::holds_alternative<GetStatsResponse>(resp));

        auto stats = std::get<GetStatsResponse>(resp);

        // WAL metrics should be present
        REQUIRE(stats.additionalStats.count("wal_active_transactions") > 0);
        REQUIRE(stats.additionalStats.count("wal_pending_entries") > 0);
    }
}

TEST_CASE("RequestDispatcher: status includes canonical readiness flags",
          "[daemon][status][readiness]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_readiness_flags_");

    YamsDaemon daemon(cfg);
    DaemonLifecycleAdapter lifecycleAdapter(&daemon);
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);

    StatusRequest req;
    req.detailed = false;
    Request r = req;

    boost::asio::io_context ioc;
    auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
    ioc.run();
    auto resp = fut.get();

    REQUIRE(std::holds_alternative<StatusResponse>(resp));
    const auto& status = std::get<StatusResponse>(resp);

    const std::vector<std::string_view> requiredCoreReadinessKeys = {
        readiness::kIpcServer,
        readiness::kContentStore,
        readiness::kDatabase,
        readiness::kMetadataRepo,
        readiness::kSearchEngine,
        readiness::kModelProvider,
        readiness::kVectorIndex,
        readiness::kVectorDb,
        readiness::kPlugins,
        readiness::kVectorDbInitAttempted,
        readiness::kVectorDbReady,
        readiness::kVectorDbDim,
        readiness::kEmbeddingReady,
        readiness::kEmbeddingDegraded,
        readiness::kPluginsReady,
        readiness::kPluginsDegraded,
        readiness::kContentExtractorsReady,
        readiness::kSymbolExtractorsReady,
        readiness::kEntityExtractorsReady,
        readiness::kTitleExtractorReady,
        readiness::kRepairService,
        readiness::kSearchEngineBuildReasonInitial,
        readiness::kSearchEngineBuildReasonRebuild,
        readiness::kSearchEngineBuildReasonDegraded,
        readiness::kSearchEngineLexicalReady,
        readiness::kSearchEngineVectorUsable,
        readiness::kSearchEngineHybridUsable,
        readiness::kSearchEngineLexicalEnhancementConfigured,
        readiness::kSearchEngineLexicalEnhancementReady,
        readiness::kSearchEngineLexicalEnhancementBuilding,
        readiness::kSearchEngineFragmentGeometryReady};

    for (const auto key : requiredCoreReadinessKeys) {
        INFO("missing readiness key: " << key);
        REQUIRE(status.readinessStates.count(std::string(key)) > 0);
    }

    const std::array<std::string_view, 5> requiredCapabilityMetricKeys = {
        metrics::kContentExtractorsLoaded, metrics::kSymbolExtractorsLoaded,
        metrics::kEntityExtractorsLoaded,  metrics::kTitleExtractorEnabled,
        metrics::kPluginSkippedCount,
    };

    for (const auto key : requiredCapabilityMetricKeys) {
        INFO("missing capability metric key: " << key);
        REQUIRE(status.requestCounts.count(std::string(key)) > 0);
    }
}

TEST_CASE("RequestDispatcher: compact and detailed status agree on FSM readiness",
          "[daemon][status][readiness]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_fsm_readiness_parity_");

    YamsDaemon daemon(cfg);
    DaemonLifecycleAdapter lifecycleAdapter(&daemon);
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);

    auto fetch = [&](bool detailed) {
        StatusRequest req;
        req.detailed = detailed;
        Request r = req;
        boost::asio::io_context ioc;
        auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
        ioc.run();
        auto resp = fut.get();
        REQUIRE(std::holds_alternative<StatusResponse>(resp));
        return std::get<StatusResponse>(resp);
    };

    const auto compact = fetch(false);
    const auto detailed = fetch(true);

    for (const auto key : {readiness::kEmbeddingReady, readiness::kEmbeddingDegraded,
                           readiness::kPluginsReady, readiness::kPluginsDegraded}) {
        const std::string k(key);
        INFO("readiness key must agree between compact and detailed: " << k);
        REQUIRE(compact.readinessStates.count(k) > 0);
        REQUIRE(detailed.readinessStates.count(k) > 0);
        REQUIRE(compact.readinessStates.at(k) == detailed.readinessStates.at(k));
    }
}

TEST_CASE("RequestDispatcher: status includes repair metrics and flags",
          "[daemon][status][repair]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_repair_metrics_");

    YamsDaemon daemon(cfg);
    DaemonLifecycleAdapter lifecycleAdapter(&daemon);
    StateComponent state;
    state.stats.repairQueueDepth.store(7);
    state.stats.repairBatchesAttempted.store(11);
    state.stats.repairEmbeddingsGenerated.store(5);
    state.stats.repairEmbeddingsSkipped.store(2);
    state.stats.repairFailedOperations.store(1);
    state.stats.repairIdleTicks.store(99);
    state.stats.repairBusyTicks.store(33);
    state.stats.repairCurrentOperationCode.store(10);
    state.stats.repairCurrentOperationStartedMs.store(
        static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(
                                  std::chrono::steady_clock::now().time_since_epoch())
                                  .count()) -
        1500);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);

    StatusRequest req;
    req.detailed = true;
    Request r = req;

    boost::asio::io_context ioc;
    auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
    ioc.run();
    auto resp = fut.get();

    REQUIRE(std::holds_alternative<StatusResponse>(resp));
    const auto& status = std::get<StatusResponse>(resp);

    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairQueueDepth)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairBatchesAttempted)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairEmbeddingsGenerated)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairEmbeddingsSkipped)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairFailedOperations)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairIdleTicks)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairBusyTicks)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairRunning)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairInProgress)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairCurrentOperationCode)) > 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairCurrentOperationElapsedMs)) > 0);
    CHECK(status.requestCounts.at(std::string(metrics::kRepairCurrentOperationCode)) == 10);
    CHECK(status.requestCounts.at(std::string(metrics::kRepairCurrentOperationElapsedMs)) >= 1);
}

TEST_CASE("RequestDispatcher: status includes the current request in requestsProcessed",
          "[daemon][status][requests]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_request_count_");

    YamsDaemon daemon(cfg);
    DaemonLifecycleAdapter lifecycleAdapter(&daemon);
    StateComponent state;
    state.stats.requestsProcessed.store(41, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);

    auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});

    REQUIRE(std::holds_alternative<StatusResponse>(resp));
    const auto& status = std::get<StatusResponse>(resp);
    CHECK(status.requestsProcessed == 42);
    CHECK(state.stats.requestsProcessed.load(std::memory_order_relaxed) == 42);
}

TEST_CASE("RequestDispatcher: status responds even when lifecycle not ready",
          "[daemon][status][readiness]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_not_ready_");

    YamsDaemon daemon(cfg);
    DaemonLifecycleAdapter lifecycleAdapter(&daemon);
    // Daemon is not started; lifecycle state is expected to be non-ready.
    {
        const auto snap = daemon.getLifecycle().snapshot();
        REQUIRE(snap.state != LifecycleState::Ready);
        REQUIRE(snap.state != LifecycleState::Degraded);
    }

    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);

    StatusRequest req;
    req.detailed = true;
    Request r = req;

    boost::asio::io_context ioc;
    auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
    ioc.run();
    auto resp = fut.get();

    REQUIRE(std::holds_alternative<StatusResponse>(resp));
}

TEST_CASE("RequestDispatcher: tree diff handler validates inputs and missing metadata repo",
          "[daemon][tree-diff][dispatcher]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_tree_diff_dispatcher_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    SECTION("returns invalid argument when snapshot ids are missing") {
        ListTreeDiffRequest req;
        req.targetSnapshotId = "target-only";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message.find("baseSnapshotId") != std::string::npos);
    }

    SECTION("returns internal error when metadata repository is unavailable") {
        ListTreeDiffRequest req;
        req.baseSnapshotId = "base";
        req.targetSnapshotId = "target";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Metadata repository not available") != std::string::npos);
    }
}

TEST_CASE("RequestDispatcher: prune handler reports missing metadata repo",
          "[daemon][prune][dispatcher]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_prune_dispatcher_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    PruneRequest req;
    req.dryRun = true;
    req.extensions = {"log"};

    auto resp = dispatchRequest(dispatcher, Request{req});

    REQUIRE(std::holds_alternative<ErrorResponse>(resp));
    const auto& err = std::get<ErrorResponse>(resp);
    CHECK(err.code == ErrorCode::InternalError);
    CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
}

TEST_CASE("RequestDispatcher: prune handler covers parsing and execution branches",
          "[daemon][prune][dispatcher]") {
    auto makeDoc = [](int64_t id, std::string filePath, std::string ext, int64_t size,
                      std::string hash, int64_t ageDays) {
        metadata::DocumentInfo doc{};
        doc.id = id;
        doc.filePath = std::move(filePath);
        doc.fileName = std::filesystem::path(doc.filePath).filename().string();
        doc.fileExtension = std::move(ext);
        doc.fileSize = size;
        doc.sha256Hash = std::move(hash);
        doc.mimeType = "text/plain";
        const auto now =
            std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());
        doc.modifiedTime = now - std::chrono::hours(ageDays * 24);
        doc.indexedTime = now;
        return doc;
    };

    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_prune_dispatcher_live_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    auto repo = std::make_shared<StubPruneMetadataRepository>();
    auto store = std::make_shared<StubContentStore>();
    svc.__test_setMetadataRepo(repo);
    svc.__test_setContentStore(store);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    repo->addDocument(makeDoc(1, "build/output.log", "log", 2048, std::string(64, 'a'), 10));
    repo->addDocument(
        makeDoc(2, "logs/server.log", ".log", 3 * 1024 * 1024, std::string(64, 'b'), 14));
    repo->addDocument(makeDoc(3, "notes/readme.txt", "txt", 64, std::string(64, 'c'), 2));
    repo->addDocument(makeDoc(4, "cache/old.tmp", "tmp", 4096, std::string(64, 'd'), 500));

    SECTION("dry run parses age and size filters across units") {
        PruneRequest req;
        req.dryRun = true;
        req.olderThan = "1w";
        req.largerThan = "1KB";
        req.smallerThan = "5MB";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.filesDeleted == 0);
        CHECK(pruneResp.filesFailed == 0);
        CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 4096));
        REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
        REQUIRE(pruneResp.categoryCounts.count("logs") == 1);
        REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
        CHECK(pruneResp.categoryCounts.at("build-object") == 1);
        CHECK(pruneResp.categoryCounts.at("logs") == 1);
        CHECK(pruneResp.categoryCounts.at("temp") == 1);
    }

    SECTION("dry run accepts month and year age units") {
        PruneRequest monthReq;
        monthReq.dryRun = true;
        monthReq.olderThan = "1m";

        auto monthResp = dispatchRequest(dispatcher, Request{monthReq});

        REQUIRE(std::holds_alternative<PruneResponse>(monthResp));
        const auto& monthPruneResp = std::get<PruneResponse>(monthResp);
        REQUIRE(monthPruneResp.categoryCounts.count("temp") == 1);
        CHECK(monthPruneResp.categoryCounts.at("temp") == 1);
        CHECK(monthPruneResp.categoryCounts.count("logs") == 0);

        PruneRequest yearReq;
        yearReq.dryRun = true;
        yearReq.olderThan = "1y";

        auto yearResp = dispatchRequest(dispatcher, Request{yearReq});

        REQUIRE(std::holds_alternative<PruneResponse>(yearResp));
        const auto& yearPruneResp = std::get<PruneResponse>(yearResp);
        CHECK(yearPruneResp.filesDeleted == 0);
        CHECK(yearPruneResp.totalBytesFreed == 4096);
        REQUIRE(yearPruneResp.categoryCounts.count("temp") == 1);
        CHECK(yearPruneResp.categoryCounts.at("temp") == 1);
    }

    SECTION("dry run accepts gigabyte size units") {
        PruneRequest req;
        req.dryRun = true;
        req.largerThan = "1GB";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.totalBytesFreed == 0);
        CHECK(pruneResp.categoryCounts.empty());
    }

    SECTION("dry run falls back on unknown age unit and raw byte sizes") {
        PruneRequest req;
        req.dryRun = true;
        req.olderThan = "3q";
        req.largerThan = "2000";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 4096));
        REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
        REQUIRE(pruneResp.categoryCounts.count("logs") == 1);
        REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
        CHECK(pruneResp.categoryCounts.at("build-object") == 1);
        CHECK(pruneResp.categoryCounts.at("logs") == 1);
        CHECK(pruneResp.categoryCounts.at("temp") == 1);
    }

    SECTION("dry run filters by extension and categories") {
        PruneRequest req;
        req.dryRun = true;
        req.categories = {"build"};
        req.extensions = {"log"};

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.totalBytesFreed == 2048);
        CHECK(pruneResp.categoryCounts.size() == 1);
        REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
        CHECK(pruneResp.categoryCounts.at("build-object") == 1);
    }

    SECTION("dry run matches dotted extensions when category is logs") {
        PruneRequest req;
        req.dryRun = true;
        req.categories = {"logs"};
        req.extensions = {"log"};

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.totalBytesFreed == 3 * 1024 * 1024);
        CHECK(pruneResp.categoryCounts.size() == 1);
        REQUIRE(pruneResp.categoryCounts.count("logs") == 1);
        CHECK(pruneResp.categoryCounts.at("logs") == 1);
    }

    SECTION("dry run handles query failures from repository") {
        repo->setQueryError(Error{ErrorCode::InternalError, "query boom"});

        auto resp = dispatchRequest(dispatcher, Request{PruneRequest{}});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.errorMessage.empty());
        CHECK(pruneResp.totalBytesFreed == 0);
        CHECK(pruneResp.categoryCounts.empty());
    }

    SECTION("dry run handles repository exceptions") {
        repo->setThrowOnQuery("query exception");

        auto resp = dispatchRequest(dispatcher, Request{PruneRequest{}});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.errorMessage == "Failed to query candidates: query exception");
    }

    SECTION("apply mode counts deleted missing failed and exception cases") {
        repo->setMissingHash(std::string(64, 'b'));
        repo->setDeleteError(4, Error{ErrorCode::InternalError, "delete failed"});
        repo->setThrowOnLookupHash(std::string(64, 'c'), "lookup boom");

        PruneRequest req;
        req.dryRun = false;
        req.olderThan = "1d";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.filesDeleted == 0);
        CHECK(pruneResp.filesFailed == 0);
        CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 64 + 4096));
        REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
        REQUIRE(pruneResp.categoryCounts.count("logs") == 1);
        REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
        CHECK(pruneResp.categoryCounts.at("build-object") == 1);
        CHECK(pruneResp.categoryCounts.at("logs") == 1);
        CHECK(pruneResp.categoryCounts.at("temp") == 1);
        CHECK_FALSE(pruneResp.statusMessage.empty());
        REQUIRE(waitForCondition([&] { return repo->deleteCallCount() == 2; }));
        CHECK(repo->deleteCallCount() == 2);
        CHECK(repo->deletedIds() == std::vector<int64_t>{1});
        const auto removed = store->removedHashes();
        CHECK(std::find(removed.begin(), removed.end(), std::string(64, 'd')) != removed.end());
    }

    SECTION("apply mode removes content through the document deletion service") {
        const auto targetHash = std::string(64, 'a');
        store->setBlob(targetHash, "indexed log bytes");

        PruneRequest req;
        req.dryRun = false;
        req.categories = {"build"};
        req.extensions = {"log"};

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.filesDeleted == 0);
        CHECK(pruneResp.filesFailed == 0);
        CHECK(pruneResp.totalBytesFreed == 2048);
        CHECK_FALSE(pruneResp.statusMessage.empty());
        REQUIRE(waitForCondition([&] { return repo->deleteCallCount() == 1; }));
        CHECK_FALSE(store->hasBlob(targetHash));
        CHECK(store->removedHashes() == std::vector<std::string>{targetHash});
        CHECK(repo->deletedIds() == std::vector<int64_t>{1});
    }

    SECTION("apply mode yields during large prune batches") {
        auto fullHashFor = [](int i) {
            static constexpr char kHex[] = "0123456789abcdef";
            std::string hash(64, 'e');
            hash[62] = kHex[(i / 16) & 0x0f];
            hash[63] = kHex[i & 0x0f];
            return hash;
        };
        for (int i = 0; i < 105; ++i) {
            repo->addDocument(makeDoc(1000 + i, "cache/bulk_" + std::to_string(i) + ".tmp", "tmp",
                                      2048, fullHashFor(i), 30));
        }

        PruneRequest req;
        req.dryRun = false;
        req.categories = {"temp"};
        req.extensions = {"tmp"};
        req.olderThan = "1d";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.filesDeleted == 0);
        CHECK(pruneResp.filesFailed == 0);
        CHECK(pruneResp.totalBytesFreed == (4096 + 105 * 2048));
        REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
        CHECK(pruneResp.categoryCounts.at("temp") == 106);
        CHECK_FALSE(pruneResp.statusMessage.empty());
        REQUIRE(waitForCondition([&] { return repo->deleteCallCount() == 106; }));
        CHECK(repo->deleteCallCount() == 106);
        CHECK(repo->deletedIds().size() == 106);
    }

    SECTION("dry run leaves empty size filters as zero") {
        PruneRequest req;
        req.dryRun = true;
        req.largerThan = "";
        req.smallerThan = "";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PruneResponse>(resp));
        const auto& pruneResp = std::get<PruneResponse>(resp);
        CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 64 + 4096));
        REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
        REQUIRE(pruneResp.categoryCounts.count("logs") == 1);
        REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
        CHECK(pruneResp.categoryCounts.at("build-object") == 1);
        CHECK(pruneResp.categoryCounts.at("logs") == 1);
        CHECK(pruneResp.categoryCounts.at("temp") == 1);
    }
}

TEST_CASE("RequestDispatcher: model handlers cover validation and status branches",
          "[daemon][model][dispatcher]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_model_dispatcher_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    SECTION("load model rejects missing name when provider exists") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
        svc.__test_setModelProvider(provider);

        LoadModelRequest req;
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidData);
        CHECK(err.message == "modelName is required");
    }

    SECTION("load model with options succeeds and records provider state") {
        auto provider = std::make_shared<StubModelProvider>(768, "/tmp/provider-model.onnx");
        svc.__test_setModelProvider(provider);

        LoadModelRequest req;
        req.modelName = "dispatcher-model";
        req.optionsJson = R"({"mode":"offline"})";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        const auto& loadResp = std::get<ModelLoadResponse>(resp);
        CHECK(loadResp.success);
        CHECK(loadResp.modelName == "dispatcher-model");
        CHECK(provider->isModelLoaded("dispatcher-model"));
        CHECK(provider->lastOptionsModel() == "dispatcher-model");
        CHECK(provider->lastOptionsJson() == req.optionsJson);
    }

    SECTION("load model failure marks embedding degraded") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/failing-model.onnx");
        provider->setLoadError(Error{ErrorCode::InvalidState, "provider refused load"});
        svc.__test_setModelProvider(provider);
        lifecycle.resetDegradedState();

        LoadModelRequest req;
        req.modelName = "failing-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == "provider refused load");
        CHECK(lifecycle.setSubsystemDegradedCalls() == 1);
        CHECK(lifecycle.lastSubsystem() == "embedding");
        CHECK(lifecycle.lastDegraded());
        CHECK(lifecycle.lastReason() == "provider_load_failed");
    }

    SECTION("load model sanitizes invalid utf8 in provider errors") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/utf8-model.onnx");
        std::string invalidMessage = "prefix ";
        invalidMessage.push_back(static_cast<char>(0xC2));
        invalidMessage.push_back(static_cast<char>(0xA9));
        invalidMessage.push_back(static_cast<char>(0xE2));
        invalidMessage.push_back(static_cast<char>(0x82));
        invalidMessage.push_back(static_cast<char>(0xAC));
        invalidMessage.push_back(static_cast<char>(0xF0));
        invalidMessage.push_back(static_cast<char>(0x9F));
        invalidMessage.push_back(static_cast<char>(0x92));
        invalidMessage.push_back(static_cast<char>(0xA9));
        invalidMessage.push_back(static_cast<char>(0xC0));
        invalidMessage.push_back(static_cast<char>(0xFF));
        provider->setLoadError(Error{ErrorCode::InvalidState, invalidMessage});
        svc.__test_setModelProvider(provider);

        LoadModelRequest req;
        req.modelName = "utf8-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        std::string expected = "prefix ";
        expected.push_back(static_cast<char>(0xC2));
        expected.push_back(static_cast<char>(0xA9));
        expected.push_back(static_cast<char>(0xE2));
        expected.push_back(static_cast<char>(0x82));
        expected.push_back(static_cast<char>(0xAC));
        expected.push_back(static_cast<char>(0xF0));
        expected.push_back(static_cast<char>(0x9F));
        expected.push_back(static_cast<char>(0x92));
        expected.push_back(static_cast<char>(0xA9));
        expected += "\xEF\xBF\xBD\xEF\xBF\xBD";
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == expected);
    }

    SECTION("load model returns internal error when provider throws after load") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/throwing-model.onnx");
        provider->setThrowOnMemoryUsage(true);
        svc.__test_setModelProvider(provider);

        LoadModelRequest req;
        req.modelName = "throwing-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Load model failed: memory boom") != std::string::npos);
    }

    SECTION("load model skips rebuild when model is already loaded") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/already-loaded-model.onnx");
        REQUIRE(provider->loadModel("already-loaded-model").has_value());
        svc.__test_setModelProvider(provider);

        LoadModelRequest req;
        req.modelName = "already-loaded-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        const auto& loadResp = std::get<ModelLoadResponse>(resp);
        CHECK(loadResp.success);
        CHECK(loadResp.modelName == "already-loaded-model");
        CHECK(provider->getLoadedModelCount() == 1);
    }

    SECTION("load model honors timeout env floor") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/timeout-model.onnx");
        svc.__test_setModelProvider(provider);
        ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "5");
        struct ShutdownBeforeEnvRestore {
            ServiceManager& svc;
            ~ShutdownBeforeEnvRestore() { svc.shutdown(); }
        } shutdownBeforeEnvRestore{svc};

        LoadModelRequest req;
        req.modelName = "timeout-clamped-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        CHECK(std::get<ModelLoadResponse>(resp).success);
        CHECK(provider->isModelLoaded("timeout-clamped-model"));
    }

    SECTION("load model survives invalid timeout env value") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/timeout-invalid.onnx");
        svc.__test_setModelProvider(provider);
        ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "not-a-number");
        struct ShutdownBeforeEnvRestore {
            ServiceManager& svc;
            ~ShutdownBeforeEnvRestore() { svc.shutdown(); }
        } shutdownBeforeEnvRestore{svc};

        LoadModelRequest req;
        req.modelName = "timeout-invalid-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        CHECK(std::get<ModelLoadResponse>(resp).success);
        CHECK(provider->isModelLoaded("timeout-invalid-model"));
    }

    SECTION("load model rebuilds when cached search engine reports missing embedding generator") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/rebuild-health.onnx");
        REQUIRE(provider->loadModel("already-loaded-health-model").has_value());
        svc.__test_setModelProvider(provider);

        metadata::ConnectionPoolConfig poolCfg{};
        poolCfg.enableWAL = false;
        poolCfg.enableForeignKeys = false;
        auto pool = std::make_shared<metadata::ConnectionPool>(":memory:", poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = std::make_shared<metadata::MetadataRepository>(*pool);

        search::SearchEngineConfig config;
        config.vectorWeight = 1.0f;
        auto unhealthyEngine =
            std::make_shared<search::SearchEngine>(repo, nullptr, nullptr, nullptr, config);
        svc.__test_setCachedSearchEngine(unhealthyEngine, true);

        LoadModelRequest req;
        req.modelName = "already-loaded-health-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        CHECK(std::get<ModelLoadResponse>(resp).success);
        CHECK(provider->getLoadedModelCount() == 1);
    }

    SECTION("load model ignores isModelLoaded exceptions before loading") {
        auto provider = std::make_shared<StubModelProvider>(4, "/tmp/isloaded-throw.onnx");
        svc.__test_setModelProvider(provider);

        metadata::ConnectionPoolConfig poolCfg{};
        poolCfg.enableWAL = false;
        poolCfg.enableForeignKeys = false;
        auto pool = std::make_shared<metadata::ConnectionPool>(":memory:", poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = std::make_shared<metadata::MetadataRepository>(*pool);

        yams::vector::VectorDatabaseConfig vectorCfg{};
        vectorCfg.database_path = ":memory:";
        vectorCfg.embedding_dim = 4;
        vectorCfg.create_if_missing = true;
        vectorCfg.use_in_memory = true;
        auto vectorDb = std::make_shared<yams::vector::VectorDatabase>(vectorCfg);
        REQUIRE(vectorDb->initialize());

        search::SearchEngineConfig config;
        config.vectorWeight = 1.0f;
        auto healthyEngine = std::make_shared<search::SearchEngine>(
            repo, vectorDb, provider->getEmbeddingGenerator(), nullptr, config);
        svc.__test_setCachedSearchEngine(healthyEngine, true);
        provider->setThrowOnIsModelLoadedCount(1);

        LoadModelRequest req;
        req.modelName = "isloaded-throw-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        const auto& loadResp = std::get<ModelLoadResponse>(resp);
        CHECK(loadResp.success);
        CHECK(loadResp.modelName == "isloaded-throw-model");
        provider->setThrowOnIsModelLoaded(false);
        CHECK(provider->isModelLoaded("isloaded-throw-model"));
    }

    SECTION("load model survives lifecycle degradation callback exceptions") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/lifecycle-throw.onnx");
        svc.__test_setModelProvider(provider);
        lifecycle.setThrowOnSetSubsystemDegraded(true);

        LoadModelRequest req;
        req.modelName = "lifecycle-throw-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
        const auto& loadResp = std::get<ModelLoadResponse>(resp);
        CHECK(loadResp.success);
        CHECK(loadResp.modelName == "lifecycle-throw-model");
        CHECK(provider->isModelLoaded("lifecycle-throw-model"));
        lifecycle.setThrowOnSetSubsystemDegraded(false);
    }

    SECTION("unload model rejects missing name") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
        svc.__test_setModelProvider(provider);

        UnloadModelRequest req;
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidData);
        CHECK(err.message == "modelName is required");
    }

    SECTION("unload model propagates provider missing-model error") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
        svc.__test_setModelProvider(provider);

        UnloadModelRequest req;
        req.modelName = "missing-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
    }

    SECTION("unload model reports provider not ready when no provider is available") {
        auto resp = dispatchRequest(dispatcher, Request{UnloadModelRequest{"missing-provider"}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message.find("Plugin system") != std::string::npos);
    }

    SECTION("model status returns loaded model details and filters by name") {
        auto provider = std::make_shared<StubModelProvider>(512, "/tmp/status-model.onnx");
        svc.__test_setModelProvider(provider);
        REQUIRE(provider->loadModel("status-model").has_value());

        ModelStatusRequest req;
        req.modelName = "status-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ModelStatusResponse>(resp));
        const auto& statusResp = std::get<ModelStatusResponse>(resp);
        REQUIRE(statusResp.models.size() == 1);
        CHECK(statusResp.models.front().name == "status-model");
        CHECK(statusResp.models.front().loaded);
        CHECK(statusResp.models.front().embeddingDim == 512);
    }

    SECTION("model status handles model info lookup failure") {
        auto provider = std::make_shared<StubModelProvider>(256, "/tmp/status-fallback.onnx");
        REQUIRE(provider->loadModel("status-fallback").has_value());
        provider->setFailModelInfo(true);
        svc.__test_setModelProvider(provider);

        auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}});

        REQUIRE(std::holds_alternative<ModelStatusResponse>(resp));
        const auto& statusResp = std::get<ModelStatusResponse>(resp);
        REQUIRE(statusResp.models.size() == 1);
        CHECK(statusResp.models.front().name == "status-fallback");
        CHECK(statusResp.models.front().memoryMb == 0);
        CHECK(statusResp.models.front().maxSequenceLength == 0);
        CHECK(statusResp.models.front().embeddingDim == 256);
    }

    SECTION("model status returns empty response when provider unavailable") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
        provider->setAvailable(false);
        svc.__test_setModelProvider(provider);

        auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}});

        REQUIRE(std::holds_alternative<ModelStatusResponse>(resp));
        const auto& statusResp = std::get<ModelStatusResponse>(resp);
        CHECK(statusResp.models.empty());
        CHECK(statusResp.totalMemoryMb == 0);
    }

    SECTION("model status converts provider exceptions into internal errors") {
        auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
        REQUIRE(provider->loadModel("status-throw").has_value());
        provider->setThrowOnMemoryUsage(true);
        svc.__test_setModelProvider(provider);

        auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("memory boom") != std::string::npos);
    }
}

TEST_CASE("RequestDispatcher: document handlers cover direct helper and error branches",
          "[daemon][documents][dispatcher]") {
    auto makeDoc = [](int64_t id, std::string filePath, std::string hash) {
        metadata::DocumentInfo doc{};
        doc.id = id;
        doc.filePath = std::move(filePath);
        doc.fileName = std::filesystem::path(doc.filePath).filename().string();
        doc.fileExtension = std::filesystem::path(doc.filePath).extension().string();
        doc.fileSize = 12;
        doc.sha256Hash = std::move(hash);
        doc.mimeType = "text/plain";
        doc.setIndexedTime(1710000000 + id);
        return doc;
    };

    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_documents_dispatcher_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    SECTION("prepareSession returns missing-session error for unknown session") {
        ScopedEnvVar stateHome("XDG_STATE_HOME",
                               makeTempDir("yams_prepare_missing_session_").string().c_str());

        RequestDispatcher::PrepareSessionOptions opts;
        opts.sessionName = "missing";

        CHECK(dispatcher.prepareSession(opts) == -2);
    }

    SECTION("prepareSession warms current session documents") {
        auto stateRoot = makeTempDir("yams_prepare_session_");
        ScopedEnvVar stateHome("XDG_STATE_HOME", stateRoot.string().c_str());

        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(7, "src/demo.cpp", "warm-hash");
        repo->setPathTreeDocuments({doc});
        store->setBlob("warm-hash", "hello snippet world");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        auto sessionSvc = app::services::makeSessionService(nullptr);
        REQUIRE(sessionSvc != nullptr);
        sessionSvc->init("warm-session", "coverage");
        sessionSvc->addPathSelector("src/**/*.cpp", {}, {});

        RequestDispatcher::PrepareSessionOptions opts;
        opts.sessionName = "warm-session";
        opts.limit = 5;
        opts.snippetLen = 5;

        CHECK(dispatcher.prepareSession(opts) == 1);

        auto materialized = sessionSvc->listMaterialized("warm-session");
        REQUIRE(materialized.size() == 1);
        CHECK(materialized.front().hash == "warm-hash");
    }

    SECTION("cancel request succeeds when context is registered") {
        auto ctx = std::make_shared<RequestContext>();
        RequestContextRegistry::instance().register_context(4242, ctx);

        auto resp = dispatchRequest(dispatcher, Request{CancelRequest{4242}});

        RequestContextRegistry::instance().deregister_context(4242);
        REQUIRE(std::holds_alternative<SuccessResponse>(resp));
        CHECK(std::get<SuccessResponse>(resp).message == "Cancel accepted");
        CHECK(ctx->canceled.load(std::memory_order_relaxed));
    }

    SECTION("get init rejects missing hash and name") {
        auto resp = dispatchRequest(dispatcher, Request{GetInitRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message.find("hash or name required") != std::string::npos);
    }

    SECTION("get init reports missing content store") {
        GetInitRequest req;
        req.hash = "missing-store-hash";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("content store unavailable") != std::string::npos);
    }

    SECTION("get init reports name resolution failure") {
        GetInitRequest req;
        req.name = "missing-name";
        req.byName = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("Metadata repository not available") != std::string::npos);
    }

    SECTION("get init reports retrieve bytes failure") {
        auto store = std::make_shared<StubContentStore>();
        svc.__test_setContentStore(store);

        GetInitRequest req;
        req.hash = "missing-bytes";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("retrieveBytes failed") != std::string::npos);
    }

    SECTION("get init reports missing retrieval session manager") {
        auto store = std::make_shared<StubContentStore>();
        store->setBlob("needs-session-hash", "hello retrieval session");
        svc.__test_setContentStore(store);

        GetInitRequest req;
        req.hash = "needs-session-hash";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("retrieval session manager unavailable") != std::string::npos);
    }

    SECTION("get init chunk and end succeed across retrieval session lifecycle") {
        auto store = std::make_shared<StubContentStore>();
        store->setBlob("session-flow-hash", "hello retrieval flow");
        svc.__test_setContentStore(store);
        svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());

        GetInitRequest initReq;
        initReq.hash = "session-flow-hash";
        initReq.chunkSize = 5;

        auto initResp = dispatchRequest(dispatcher, Request{initReq});

        REQUIRE(std::holds_alternative<GetInitResponse>(initResp));
        const auto& init = std::get<GetInitResponse>(initResp);
        CHECK(init.transferId > 0);
        CHECK(init.totalSize == 20);
        CHECK(init.chunkSize == 5);
        CHECK(init.metadata.at("hash") == "session-flow-hash");

        auto chunkResp =
            dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 0, 5}});

        REQUIRE(std::holds_alternative<GetChunkResponse>(chunkResp));
        const auto& chunk = std::get<GetChunkResponse>(chunkResp);
        CHECK(chunk.data == "hello");
        CHECK(chunk.bytesRemaining == 15);

        auto endResp = dispatchRequest(dispatcher, Request{GetEndRequest{init.transferId}});

        REQUIRE(std::holds_alternative<SuccessResponse>(endResp));
        CHECK(std::get<SuccessResponse>(endResp).message == "OK");

        auto postEndChunk =
            dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 0, 5}});

        REQUIRE(std::holds_alternative<ErrorResponse>(postEndChunk));
        const auto& postEndErr = std::get<ErrorResponse>(postEndChunk);
        CHECK(postEndErr.code == ErrorCode::NotFound);
        CHECK(postEndErr.message.find("Invalid transferId") != std::string::npos);
    }

    SECTION("get chunk rejects offsets beyond total size") {
        auto store = std::make_shared<StubContentStore>();
        store->setBlob("offset-error-hash", "hello retrieval flow");
        svc.__test_setContentStore(store);
        svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());

        GetInitRequest initReq;
        initReq.hash = "offset-error-hash";

        auto initResp = dispatchRequest(dispatcher, Request{initReq});

        REQUIRE(std::holds_alternative<GetInitResponse>(initResp));
        const auto& init = std::get<GetInitResponse>(initResp);

        auto chunkResp =
            dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 99, 5}});

        REQUIRE(std::holds_alternative<ErrorResponse>(chunkResp));
        const auto& err = std::get<ErrorResponse>(chunkResp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message.find("Offset beyond total size") != std::string::npos);
    }

    SECTION("get chunk respects max bytes cap and can return empty chunk") {
        auto store = std::make_shared<StubContentStore>();
        store->setBlob("cap-flow-hash", "hello retrieval flow");
        svc.__test_setContentStore(store);
        svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());

        GetInitRequest initReq;
        initReq.hash = "cap-flow-hash";
        initReq.chunkSize = 10;
        initReq.maxBytes = 7;

        auto initResp = dispatchRequest(dispatcher, Request{initReq});

        REQUIRE(std::holds_alternative<GetInitResponse>(initResp));
        const auto& init = std::get<GetInitResponse>(initResp);

        auto chunkResp =
            dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 7, 5}});

        REQUIRE(std::holds_alternative<GetChunkResponse>(chunkResp));
        const auto& chunk = std::get<GetChunkResponse>(chunkResp);
        CHECK(chunk.data.empty());
        CHECK(chunk.bytesRemaining == 0);
    }

    SECTION("get chunk reports missing retrieval session manager") {
        auto resp = dispatchRequest(dispatcher, Request{GetChunkRequest{9999, 0, 4}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("retrieval session manager unavailable") != std::string::npos);
    }

    SECTION("get chunk reports invalid transfer id") {
        svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());

        auto resp = dispatchRequest(dispatcher, Request{GetChunkRequest{9999, 0, 4}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message.find("Invalid transferId") != std::string::npos);
    }

    SECTION("get end succeeds even without retrieval session manager") {
        auto resp = dispatchRequest(dispatcher, Request{GetEndRequest{777}});

        REQUIRE(std::holds_alternative<SuccessResponse>(resp));
        CHECK(std::get<SuccessResponse>(resp).message == "OK");
    }

    SECTION("delete rejects directory removal without recursive flag") {
        DeleteRequest req;
        req.directory = "cache";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message.find("recursive flag") != std::string::npos);
    }

    SECTION("delete recursive directory returns not found when no documents match") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto store = std::make_shared<StubContentStore>();
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        DeleteRequest req;
        req.directory = "cache";
        req.recursive = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message.find("No documents found matching criteria") != std::string::npos);
    }

    SECTION("legacy delete purge does not imply force for ambiguous names") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto store = std::make_shared<StubContentStore>();
        auto firstDoc = makeDoc(19, "/tmp/delete/a/shared.txt", "delete-shared-a");
        auto secondDoc = makeDoc(20, "/tmp/delete/b/shared.txt", "delete-shared-b");
        repo->addDocument(firstDoc);
        repo->addDocument(secondDoc);
        store->setBlob(firstDoc.sha256Hash, "first");
        store->setBlob(secondDoc.sha256Hash, "second");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        DeleteRequest req;
        req.name = "shared.txt";
        req.purge = true;
        req.force = false;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidOperation);
        CHECK(err.message.find("use --force") != std::string::npos);
        CHECK(store->removedHashes().empty());
    }

    SECTION("delete reports mixed success and failure results") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto store = std::make_shared<StubContentStore>();
        auto successDoc = makeDoc(21, "/tmp/delete/ok.txt", "delete-ok-hash");
        auto failedDoc = makeDoc(22, "/tmp/delete/fail.txt", "delete-fail-hash");
        repo->addDocument(successDoc);
        repo->addDocument(failedDoc);
        store->setBlob(successDoc.sha256Hash, "ok");
        store->setBlob(failedDoc.sha256Hash, "fail");
        store->setRemoveResult(failedDoc.sha256Hash,
                               Error{ErrorCode::IOError, "store remove failed"});
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        DeleteRequest req;
        req.pattern = "*.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<DeleteResponse>(resp));
        const auto& deleteResp = std::get<DeleteResponse>(resp);
        CHECK_FALSE(deleteResp.dryRun);
        CHECK(deleteResp.successCount == 1);
        CHECK(deleteResp.failureCount == 1);
        REQUIRE(deleteResp.results.size() == 2);
        CHECK(lifecycle.removedHashes().size() == 1);
        CHECK(lifecycle.removedHashes().front() == successDoc.sha256Hash);
    }

    SECTION("delete reports missing content store") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        svc.__test_setMetadataRepo(repo);

        DeleteRequest req;
        req.hash = "no-store-hash";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("Content store not available") != std::string::npos);
    }

    SECTION("get request maps related documents when graph view is enabled") {
        auto repoPath = makeTempMetadataDbPath("yams_get_graph_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto graphQuery = std::make_shared<StubGraphQueryService>();
        auto mainDoc = makeDoc(69, "/tmp/get-graph/main.txt",
                               "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff");
        auto relatedDoc =
            makeDoc(70, "/tmp/get-graph/related.txt",
                    "1111111111111111111111111111111111111111111111111111111111111111");
        auto mainId = repo->insertDocument(mainDoc);
        auto relatedId = repo->insertDocument(relatedDoc);
        REQUIRE(mainId.has_value());
        REQUIRE(relatedId.has_value());
        REQUIRE(repo->upsertPathTreeForDocument(mainDoc, mainId.value(), true, {}).has_value());
        REQUIRE(
            repo->upsertPathTreeForDocument(relatedDoc, relatedId.value(), true, {}).has_value());
        graphQuery->addConnectedDocument(relatedDoc.sha256Hash, "same_content", 1, 0.8);
        store->setBlob(mainDoc.sha256Hash, "main content");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);
        svc.__test_setGraphQueryService(graphQuery);

        GetRequest req;
        req.hash = mainDoc.sha256Hash;
        req.showGraph = true;
        req.graphDepth = 1;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GetResponse>(resp));
        const auto& getResp = std::get<GetResponse>(resp);
        CHECK(getResp.graphEnabled);
        CHECK(getResp.hash == mainDoc.sha256Hash);
        REQUIRE(getResp.related.size() == 1);
        CHECK(getResp.related.front().hash == relatedDoc.sha256Hash);
        CHECK(getResp.related.front().relationship == "same_content");
    }

    SECTION("get request still succeeds without query trace env") {
        auto repoPath = makeTempMetadataDbPath("yams_get_query_trace_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(71, "/tmp/get/query-trace.txt",
                           "9999999999999999999999999999999999999999999999999999999999999999");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "query trace content");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        GetRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GetResponse>(resp));
        const auto& getResp = std::get<GetResponse>(resp);
        CHECK(getResp.hash == doc.sha256Hash);
        CHECK(getResp.name == doc.fileName);
    }

    SECTION("get request maps fallback documents vector response") {
        auto repoPath = makeTempMetadataDbPath("yams_get_documents_vector_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(72, "/tmp/get/documents-vector.txt",
                           "abababababababababababababababababababababababababababababababab");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "documents vector content");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        RequestDispatcher::__test_forceGetResponseDocumentsVectorOnce();

        GetRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GetResponse>(resp));
        const auto& getResp = std::get<GetResponse>(resp);
        CHECK(getResp.hash == doc.sha256Hash);
        CHECK(getResp.name == doc.fileName);
        CHECK(getResp.path == doc.filePath);
    }

    SECTION("get request returns not found when service result is forced empty") {
        auto repoPath = makeTempMetadataDbPath("yams_get_empty_result_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(73, "/tmp/get/empty-result.txt",
                           "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "empty result content");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        RequestDispatcher::__test_forceGetEmptyResultOnce();

        GetRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "No documents found matching criteria");
    }

    SECTION("file history reports missing metadata repository") {
        FileHistoryRequest req;
        req.filepath = "missing.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("Metadata repository not available") != std::string::npos);
    }

    SECTION("file history falls back to filename query and reports no snapshot versions") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto doc = makeDoc(9, "/tmp/archive/report.txt", "history-hash");
        repo->addDocument(doc);
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = "report.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        CHECK_FALSE(history.found);
        CHECK(history.totalVersions == 0);
        CHECK(history.versions.empty());
        CHECK(history.message == "File found in index but not in any snapshot");
        CHECK(history.filepath.find("report.txt") != std::string::npos);
    }

    SECTION("file history reports file not found in index") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = "ghost.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        CHECK_FALSE(history.found);
        CHECK(history.totalVersions == 0);
        CHECK(history.message == "File not found in index");
    }

    SECTION("file history returns sorted snapshot versions from exact path") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto doc = makeDoc(11, "/tmp/archive/sorted-report.txt", "sorted-history-hash");
        repo->setExactPathDocument(doc);
        repo->setAllMetadata(
            doc.id, {{"snapshot_id", metadata::MetadataValue("snap-alpha")},
                     {"snapshot_id:snap-beta", metadata::MetadataValue("")},
                     {"snapshot_time:snap-alpha", metadata::MetadataValue("1710000000000000")}});
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        REQUIRE(history.found);
        REQUIRE(history.totalVersions == 2);
        REQUIRE(history.versions.size() == 2);
        CHECK(history.message == "Found 2 version(s) across snapshots");
        CHECK(history.versions[0].snapshotId == "snap-beta");
        CHECK(history.versions[1].snapshotId == "snap-alpha");
        CHECK(history.versions[0].indexedTimestamp > history.versions[1].indexedTimestamp);
    }

    SECTION(
        "file history falls back to document indexed time when snapshot-specific time is missing") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto doc = makeDoc(42, "/tmp/archive/fallback-time.txt", "fallback-time-hash");
        repo->setExactPathDocument(doc);
        repo->setAllMetadata(doc.id, {{"snapshot_id", metadata::MetadataValue("snap-fallback")}});
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        REQUIRE(history.found);
        REQUIRE(history.totalVersions == 1);
        REQUIRE(history.versions.size() == 1);
        CHECK(history.versions.front().snapshotId == "snap-fallback");
        CHECK(history.versions.front().indexedTimestamp == 1710000042);
    }

    SECTION("file history ignores malformed snapshot timestamps and falls back to indexed time") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto doc = makeDoc(64, "/tmp/archive/bad-snapshot-time.txt", "bad-snapshot-time-hash");
        repo->setExactPathDocument(doc);
        repo->setAllMetadata(doc.id, {{"snapshot_id", metadata::MetadataValue("snap-bad-time")},
                                      {"snapshot_time", metadata::MetadataValue("not-a-number")},
                                      {"snapshot_time:snap-bad-time",
                                       metadata::MetadataValue("still-not-a-number")}});
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        REQUIRE(history.found);
        REQUIRE(history.totalVersions == 1);
        REQUIRE(history.versions.size() == 1);
        CHECK(history.versions.front().snapshotId == "snap-bad-time");
        CHECK(history.versions.front().indexedTimestamp == 1710000064);
    }

    SECTION("list request paths-only mode trims snippets metadata and tags") {
        auto repoPath = makeTempMetadataDbPath("yams_list_dispatcher_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto doc = makeDoc(55, "/tmp/list/paths-only.md", "list-paths-hash");
        REQUIRE(repo->insertDocument(doc).has_value());
        svc.__test_setMetadataRepo(repo);

        ListRequest req;
        req.limit = 5;
        req.pathsOnly = true;
        req.showMetadata = true;
        req.showTags = true;
        req.showSnippets = true;
        req.recent = false;
        req.recentCount = 0;
        req.namePattern = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ListResponse>(resp));
        const auto& listResp = std::get<ListResponse>(resp);
        REQUIRE(listResp.items.size() == 1);
        CHECK(listResp.totalCount == 1);
        CHECK(listResp.items.front().path == doc.filePath);
        CHECK(listResp.items.front().snippet.empty());
        CHECK(listResp.items.front().metadata.empty());
        CHECK(listResp.items.front().tags.empty());
    }

    SECTION("list request emits dispatch timing stats by default") {
        auto repoPath = makeTempMetadataDbPath("yams_list_query_trace_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto doc = makeDoc(56, "/tmp/list/query-trace.md", "list-query-trace-hash");
        REQUIRE(repo->insertDocument(doc).has_value());
        svc.__test_setMetadataRepo(repo);

        ListRequest req;
        req.limit = 3;
        req.recent = false;
        req.recentCount = 0;
        req.namePattern = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ListResponse>(resp));
        const auto& listResp = std::get<ListResponse>(resp);
        REQUIRE(listResp.items.size() == 1);
        CHECK(listResp.listStats.contains("phase_dispatch_total_ms"));
        CHECK(listResp.listStats.contains("phase_dispatch_map_us"));
    }

    SECTION("list request parses filter tags and recent count") {
        auto repoPath = makeTempMetadataDbPath("yams_list_filter_tags_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto doc = makeDoc(67, "/tmp/list/filter-tags.md", "list-filter-tags-hash");
        REQUIRE(repo->insertDocument(doc).has_value());
        svc.__test_setMetadataRepo(repo);

        ListRequest req;
        req.limit = 5;
        req.offset = 0;
        req.recent = false;
        req.recentCount = 3;
        req.filterTags = " alpha, beta ";
        req.namePattern = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ListResponse>(resp));
        const auto& listResp = std::get<ListResponse>(resp);
        CHECK(listResp.totalCount >= 0);
    }

    SECTION("list request works when inflight limit is disabled") {
        TuneAdvisorListGuard tuneGuard(0, 1);

        auto repoPath = makeTempMetadataDbPath("yams_list_no_inflight_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto doc = makeDoc(68, "/tmp/list/no-inflight.md", "list-no-inflight-hash");
        REQUIRE(repo->insertDocument(doc).has_value());
        svc.__test_setMetadataRepo(repo);

        ListRequest req;
        req.limit = 5;
        req.offset = 0;
        req.recent = true;
        req.showSnippets = true;
        req.noSnippets = true;
        req.namePattern = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ListResponse>(resp));
        const auto& listResp = std::get<ListResponse>(resp);
        REQUIRE(listResp.items.size() == 1);
        CHECK(listResp.items.front().snippet.empty());
    }

    SECTION("list request reports missing metadata repository") {
        ListRequest req;
        req.limit = 2;
        req.recent = false;
        req.recentCount = 0;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message.find("Metadata repository not available") != std::string::npos);
    }

    SECTION("list request reports resource exhaustion when inflight limit is saturated") {
        TuneAdvisorListGuard tuneGuard(1, 1);

        auto repoPath = makeTempMetadataDbPath("yams_list_exhaustion_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);
        auto doc = makeDoc(60, "/tmp/list/exhausted.md", "list-exhausted-hash");
        REQUIRE(repo->insertDocument(doc).has_value());
        svc.__test_setMetadataRepo(repo);

        RequestDispatcher::__test_setListInflightRequests(1);
        auto resetGuard = ScopeExit([] { RequestDispatcher::__test_setListInflightRequests(0); });

        ListRequest req;
        req.limit = 1;
        req.recent = false;
        req.recentCount = 0;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::ResourceExhausted);
        CHECK(err.message.find("List concurrency limit reached") != std::string::npos);
    }

    SECTION("list request catches std exceptions") {
        RequestDispatcher::__test_forceListExceptionOnce("forced list exception");

        ListRequest req;
        req.limit = 2;
        req.recent = false;
        req.recentCount = 0;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "List failed: forced list exception");
    }

    SECTION("list request catches unknown exceptions") {
        RequestDispatcher::__test_forceListUnknownExceptionOnce();

        ListRequest req;
        req.limit = 2;
        req.recent = false;
        req.recentCount = 0;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "List failed: unknown error");
    }

    SECTION("add document queues deferred work when system is under pressure") {
        const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
        const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setEnableResourceGovernor(originalGovernor);
            TuneAdvisor::setEnableAdmissionControl(originalAdmission);
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
            drainStoreDocumentTasks();
        });

        TuneAdvisor::setEnableResourceGovernor(true);
        TuneAdvisor::setEnableAdmissionControl(true);
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
        drainStoreDocumentTasks();

        ScopedEnvVar retriesGuard("YAMS_INGEST_ENQUEUE_RETRIES", "not-a-number");
        ScopedEnvVar baseDelayGuard("YAMS_INGEST_ENQUEUE_BASE_DELAY_MS", "-5");
        ScopedEnvVar maxDelayGuard("YAMS_INGEST_ENQUEUE_MAX_DELAY_MS", "999999");

        AddDocumentRequest req;
        req.name = "pressure.txt";
        req.content = "queued under pressure";

        auto resp = dispatchRequest(dispatcher, Request{req});

        INFO([&] {
            if (std::holds_alternative<ErrorResponse>(resp)) {
                return std::get<ErrorResponse>(resp).message;
            }
            return std::string("not an error");
        }());
        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        auto hasher = yams::crypto::createSHA256Hasher();
        REQUIRE(hasher != nullptr);
        CHECK(addResp.path == "pressure.txt");
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash == hasher->hash(req.content));
        CHECK(addResp.message == "Queued for deferred processing (system under pressure).");

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask task;
        REQUIRE(channel->try_pop(task));
        CHECK(task.request.name == req.name);
        CHECK(task.request.content == req.content);
    }

    SECTION("add document clears deferred hash when pressure-path hashing fails") {
        const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
        const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setEnableResourceGovernor(originalGovernor);
            TuneAdvisor::setEnableAdmissionControl(originalAdmission);
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
            drainStoreDocumentTasks();
        });

        TuneAdvisor::setEnableResourceGovernor(true);
        TuneAdvisor::setEnableAdmissionControl(true);
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
        drainStoreDocumentTasks();

        auto filePath = writeTempFile("yams_documents_pressure_file_", "pressure-file.txt",
                                      "pressure file content");
        RequestDispatcher::__test_forceDocumentsHashFailureOnce();

        AddDocumentRequest req;
        req.path = filePath.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash.empty());
        CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
    }

    SECTION("add document clears deferred hash for recursive pressure path") {
        const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
        const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setEnableResourceGovernor(originalGovernor);
            TuneAdvisor::setEnableAdmissionControl(originalAdmission);
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
            drainStoreDocumentTasks();
        });

        TuneAdvisor::setEnableResourceGovernor(true);
        TuneAdvisor::setEnableAdmissionControl(true);
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
        drainStoreDocumentTasks();

        AddDocumentRequest req;
        req.path = makeTempDir("yams_documents_pressure_recursive_").string();
        req.recursive = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash.empty());
        CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
    }

    SECTION("add document reports resource exhaustion when pressure queue is full") {
        const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
        const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setEnableResourceGovernor(originalGovernor);
            TuneAdvisor::setEnableAdmissionControl(originalAdmission);
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
            drainStoreDocumentTasks();
        });

        TuneAdvisor::setEnableResourceGovernor(true);
        TuneAdvisor::setEnableAdmissionControl(true);
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
        drainStoreDocumentTasks();

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask filler;
        while (channel->try_push(filler)) {
        }

        AddDocumentRequest req;
        req.name = "queue-full.txt";
        req.content = "full";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::ResourceExhausted);
        CHECK((err.message == "Ingestion queue is full. Please try again later." ||
               err.message == "Ingestion queue is full; retry shortly"));
    }

    SECTION("add document queues async work while daemon is initializing") {
        DaemonConfig notReadyCfg;
        notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_");
        StubLifecycle notReadyLifecycle(LifecycleState::Starting);
        StateComponent notReadyState;
        notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm notReadyLifecycleFsm;
        ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
        RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);

        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        AddDocumentRequest req;
        req.name = "initializing.txt";
        req.content = "booting daemon";

        auto resp = dispatchRequest(notReadyDispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        auto hasher = yams::crypto::createSHA256Hasher();
        REQUIRE(hasher != nullptr);
        CHECK(addResp.path == "initializing.txt");
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash == hasher->hash(req.content));
        CHECK(addResp.message ==
              "Ingestion accepted for asynchronous processing (daemon initializing).");

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask task;
        REQUIRE(channel->try_pop(task));
        CHECK(task.request.name == req.name);
        CHECK(task.request.content == req.content);
    }

    SECTION("add document reports queue full while daemon is initializing") {
        DaemonConfig notReadyCfg;
        notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_full_");
        StubLifecycle notReadyLifecycle(LifecycleState::Starting);
        StateComponent notReadyState;
        notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm notReadyLifecycleFsm;
        ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
        RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);

        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            drainStoreDocumentTasks();
        });

        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        drainStoreDocumentTasks();

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask filler;
        while (channel->try_push(filler)) {
        }

        AddDocumentRequest req;
        req.name = "initializing-full.txt";
        req.content = "booting daemon full";

        auto resp = dispatchRequest(notReadyDispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::ResourceExhausted);
        CHECK((err.message == "Ingestion queue is full. Please try again later." ||
               err.message == "Ingestion queue is full; retry shortly"));
    }

    SECTION("add document clears async hash when daemon-initializing hashing fails") {
        DaemonConfig notReadyCfg;
        notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_hash_fail_");
        StubLifecycle notReadyLifecycle(LifecycleState::Starting);
        StateComponent notReadyState;
        notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm notReadyLifecycleFsm;
        ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
        RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);

        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        auto filePath =
            writeTempFile("yams_documents_init_hash_fail_", "init-file.txt", "init file content");
        RequestDispatcher::__test_forceDocumentsHashFailureOnce();

        AddDocumentRequest req;
        req.path = filePath.string();

        auto resp = dispatchRequest(notReadyDispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash.empty());
        CHECK(addResp.message ==
              "Ingestion accepted for asynchronous processing (daemon initializing).");
    }

    SECTION("add document hashes file path while daemon is initializing") {
        DaemonConfig notReadyCfg;
        notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_file_hash_");
        StubLifecycle notReadyLifecycle(LifecycleState::Starting);
        StateComponent notReadyState;
        notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm notReadyLifecycleFsm;
        ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
        RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);

        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        auto filePath = writeTempFile("yams_documents_init_file_hash_", "init-file-hash.txt",
                                      "init file hash content");
        AddDocumentRequest req;
        req.path = filePath.string();

        auto resp = dispatchRequest(notReadyDispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        auto hasher = yams::crypto::createSHA256Hasher();
        REQUIRE(hasher != nullptr);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash == hasher->hashFile(req.path));
        CHECK(addResp.message ==
              "Ingestion accepted for asynchronous processing (daemon initializing).");
    }

    SECTION("add document treats directory path as async directory ingestion") {
        auto dirPath = makeTempDir("yams_documents_directory_ingest_");
        drainStoreDocumentTasks();

        AddDocumentRequest req;
        req.path = dirPath.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.path == dirPath.string());
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash.empty());
        CHECK(addResp.message == "Directory ingestion accepted for asynchronous processing.");

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask task;
        REQUIRE(channel->try_pop(task));
        CHECK(task.request.path == req.path);
    }

    SECTION("add document rejects missing path and content") {
        AddDocumentRequest req;
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message == "Provide either 'path' or 'content' + 'name'");
    }

    SECTION("add document uses async single-file path by default when ready") {
        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        AddDocumentRequest req;
        req.name = "async-default.txt";
        req.content = "ready but async";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        auto hasher = yams::crypto::createSHA256Hasher();
        REQUIRE(hasher != nullptr);
        CHECK(addResp.path == "async-default.txt");
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash == hasher->hash(req.content));
        CHECK(addResp.extractionStatus == "pending");
        CHECK(addResp.message == "Ingestion accepted for asynchronous processing.");

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask task;
        REQUIRE(channel->try_pop(task));
        CHECK(task.request.name == req.name);
        CHECK(task.request.content == req.content);
    }

    SECTION("add document hashes file paths on async path when ready") {
        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        auto filePath =
            writeTempFile("yams_documents_async_file_", "async-file.txt", "async file content");
        AddDocumentRequest req;
        req.path = filePath.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        auto hasher = yams::crypto::createSHA256Hasher();
        REQUIRE(hasher != nullptr);
        CHECK(addResp.path == req.path);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash == hasher->hashFile(req.path));
        CHECK(addResp.message == "Ingestion accepted for asynchronous processing.");
    }

    SECTION("add document hashes file path on pressure path after one retry") {
        const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
        const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setEnableResourceGovernor(originalGovernor);
            TuneAdvisor::setEnableAdmissionControl(originalAdmission);
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
            RequestDispatcher::__test_setDocumentsEnqueueFailuresBeforeSuccess(0);
            drainStoreDocumentTasks();
        });

        TuneAdvisor::setEnableResourceGovernor(true);
        TuneAdvisor::setEnableAdmissionControl(true);
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
        drainStoreDocumentTasks();

        ScopedEnvVar retriesGuard("YAMS_INGEST_ENQUEUE_RETRIES", "3");
        ScopedEnvVar baseDelayGuard("YAMS_INGEST_ENQUEUE_BASE_DELAY_MS", "1");
        ScopedEnvVar maxDelayGuard("YAMS_INGEST_ENQUEUE_MAX_DELAY_MS", "1");
        RequestDispatcher::__test_setDocumentsEnqueueFailuresBeforeSuccess(1);

        auto filePath = writeTempFile("yams_documents_pressure_retry_", "pressure-retry.txt",
                                      "pressure retry content");
        AddDocumentRequest req;
        req.path = filePath.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        auto hasher = yams::crypto::createSHA256Hasher();
        REQUIRE(hasher != nullptr);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash == hasher->hashFile(req.path));
        CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
    }

    SECTION("add document falls back to initializing path when later lifecycle snapshot throws") {
        DaemonConfig notReadyCfg;
        notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_lifecycle_throw_late_");
        StubLifecycle notReadyLifecycle(LifecycleState::Ready);
        notReadyLifecycle.setThrowOnGetLifecycleSnapshotCall(2);
        StateComponent notReadyState;
        notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm notReadyLifecycleFsm;
        ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
        RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);

        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        AddDocumentRequest req;
        req.name = "late-throw.txt";
        req.content = "late throw";

        auto resp = dispatchRequest(notReadyDispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.message ==
              "Ingestion accepted for asynchronous processing (daemon initializing).");
    }

    SECTION("add document reports queue full on async single-file path") {
        const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
        auto restore = ScopeExit([&] {
            TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
            drainStoreDocumentTasks();
        });

        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        TuneAdvisor::setStoreDocumentChannelCapacity(64);
        drainStoreDocumentTasks();

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask filler;
        while (channel->try_push(filler)) {
        }

        AddDocumentRequest req;
        req.name = "async-full.txt";
        req.content = "full async queue";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::ResourceExhausted);
        CHECK((err.message == "Ingestion queue is full. Please try again later." ||
               err.message == "Ingestion queue is full; retry shortly"));
    }

    SECTION("add document clears async hash when hashing fails on ready path") {
        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
        drainStoreDocumentTasks();

        auto filePath = writeTempFile("yams_documents_async_hash_fail_", "async-hash-fail.txt",
                                      "async hash fail content");
        RequestDispatcher::__test_forceDocumentsHashFailureOnce();

        AddDocumentRequest req;
        req.path = filePath.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.documentsAdded == 0);
        CHECK(addResp.hash.empty());
        CHECK(addResp.extractionStatus == "pending");
        CHECK(addResp.message == "Ingestion accepted for asynchronous processing.");
    }

    SECTION("add document uses sync fallback when explicitly enabled") {
        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "1");
        drainStoreDocumentTasks();
        svc.__test_setContentStore(std::make_shared<StubContentStore>());

        AddDocumentRequest req;
        req.name = "sync-add.txt";
        req.content = "sync path content";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.path == "sync-add.txt");
        CHECK(addResp.documentsAdded == 1);
        CHECK(addResp.size == req.content.size());
        CHECK(addResp.extractionStatus == "pending");
        CHECK(addResp.message == "Document stored successfully.");
        CHECK_FALSE(addResp.hash.empty());

        auto channel = getStoreDocumentTaskChannel();
        InternalEventBus::StoreDocumentTask task;
        CHECK_FALSE(channel->try_pop(task));
    }

    SECTION("add document sync fallback forwards metadata and path inputs") {
        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "true");
        drainStoreDocumentTasks();
        svc.__test_setContentStore(std::make_shared<StubContentStore>());

        auto filePath =
            writeTempFile("yams_documents_sync_file_", "sync-file.txt", "sync file content");
        AddDocumentRequest req;
        req.path = filePath.string();
        req.mimeType = "text/custom";
        req.disableAutoMime = true;
        req.tags = {"alpha", "beta"};
        req.metadata["task"] = "documents";
        req.collection = "focus";
        req.snapshotId = "snap-1";
        req.snapshotLabel = "label-1";
        req.sessionId = "session-1";
        req.noEmbeddings = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
        const auto& addResp = std::get<AddDocumentResponse>(resp);
        CHECK(addResp.path == req.path);
        CHECK(addResp.documentsAdded == 1);
        CHECK(addResp.size == std::filesystem::file_size(filePath));
        CHECK_FALSE(addResp.hash.empty());
    }

    SECTION("add document sync fallback returns store errors") {
        ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "1");
        drainStoreDocumentTasks();
        svc.__test_setContentStore(std::shared_ptr<api::IContentStore>{});

        AddDocumentRequest req;
        req.name = "sync-error.txt";
        req.content = "sync failure";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message == "Content store not available");
    }

    SECTION("file history skips documents when metadata lookup fails") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto failingDoc = makeDoc(57, "/tmp/archive/history-mixed.txt", "metadata-fail-hash");
        auto okDoc = makeDoc(58, "/var/tmp/archive/history-mixed.txt", "metadata-ok-hash");
        repo->addDocument(failingDoc);
        repo->addDocument(okDoc);
        repo->setAllMetadataError(failingDoc.id,
                                  Error{ErrorCode::DatabaseError, "metadata lookup failed"});
        repo->setAllMetadata(okDoc.id,
                             {{"snapshot_id", metadata::MetadataValue("snap-ok")},
                              {"snapshot_time", metadata::MetadataValue("1710000000000000")}});
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = "history-mixed.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        REQUIRE(history.found);
        CHECK(history.totalVersions == 1);
        REQUIRE(history.versions.size() == 1);
        CHECK(history.versions.front().snapshotId == "snap-ok");
    }

    SECTION("file history falls back when exact path lookup errors") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto doc =
            makeDoc(63, "/tmp/archive/fallback-exact-error.txt", "fallback-exact-error-hash");
        repo->addDocument(doc);
        repo->setExactPathError(Error{ErrorCode::DatabaseError, "exact lookup failed"});
        repo->setAllMetadata(doc.id,
                             {{"snapshot_id", metadata::MetadataValue("snap-fallback-error")}});
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = doc.filePath;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        REQUIRE(history.found);
        CHECK(history.totalVersions == 1);
        REQUIRE(history.versions.size() == 1);
        CHECK(history.versions.front().snapshotId == "snap-fallback-error");
    }

    SECTION("file history caps oversized filename matches to first hundred documents") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        for (int i = 0; i < 101; ++i) {
            auto doc = makeDoc(1000 + i,
                               "/tmp/archive/subdir_" + std::to_string(i) + "/history-overflow.txt",
                               "history-overflow-hash-" + std::to_string(i));
            doc.fileName = "history-overflow.txt";
            repo->addDocument(doc);
            repo->setAllMetadata(
                doc.id, {{"snapshot_id", metadata::MetadataValue("snap-" + std::to_string(i))}});
        }
        svc.__test_setMetadataRepo(repo);

        FileHistoryRequest req;
        req.filepath = "history-overflow.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
        const auto& history = std::get<FileHistoryResponse>(resp);
        REQUIRE(history.found);
        CHECK(history.totalVersions == 100);
        CHECK(history.versions.size() == 100);
    }

    SECTION("file history reports invalid filepath when current directory is missing") {
#ifdef _WIN32
        SKIP("Windows keeps the current working directory locked, so missing-cwd simulation is "
             "unreliable");
#endif
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        svc.__test_setMetadataRepo(repo);

        auto originalCwd = std::filesystem::current_path();
        auto doomedDir = makeTempDir("yams_missing_cwd_");
        std::filesystem::current_path(doomedDir);
        std::filesystem::remove_all(doomedDir);

        FileHistoryRequest req;
        req.filepath = "history-from-missing-cwd.txt";

        auto resp = dispatchRequest(dispatcher, Request{req});

        std::filesystem::current_path(originalCwd);

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message.find("Invalid filepath") != std::string::npos);
    }

    SECTION("cat request reports missing content") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(58, "/tmp/cat/missing.txt", "cat-missing-hash");
        repo->addDocument(doc);
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        CatRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message.find("Document not found") != std::string::npos);
    }

    SECTION("cat request reports missing content payload for indexed document") {
        auto repoPath = makeTempMetadataDbPath("yams_cat_missing_content_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(59, "/tmp/cat/metadata-only.txt",
                           "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa");
        REQUIRE(repo->insertDocument(doc).has_value());
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        CatRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message.find("Document content not found") != std::string::npos);
    }

    SECTION("cat request hits missing document branch after successful retrieval") {
        auto repoPath = makeTempMetadataDbPath("yams_cat_force_missing_doc_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(61, "/tmp/cat/force-missing-doc.txt",
                           "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "payload");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        RequestDispatcher::__test_forceCatMissingDocumentOnce();

        CatRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "Document not found");
    }

    SECTION("cat request hits content unavailable branch after successful retrieval") {
        auto repoPath = makeTempMetadataDbPath("yams_cat_force_missing_content_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(62, "/tmp/cat/force-missing-content.txt",
                           "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "payload");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        RequestDispatcher::__test_forceCatMissingContentOnce();

        CatRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Content unavailable");
    }

    SECTION("cat request hits native missing document branch") {
        auto repoPath = makeTempMetadataDbPath("yams_cat_native_missing_doc_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(65, "/tmp/cat/native-missing-doc.txt",
                           "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "payload");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        RequestDispatcher::__test_forceCatNativeMissingDocumentOnce();

        CatRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "Document not found");
    }

    SECTION("cat request hits native missing content branch") {
        auto repoPath = makeTempMetadataDbPath("yams_cat_native_missing_content_repo_");
        metadata::ConnectionPoolConfig poolCfg{};
        auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
        REQUIRE(pool->initialize().has_value());
        auto repo = makeReadyMetadataRepo(pool);

        auto store = std::make_shared<StubContentStore>();
        auto doc = makeDoc(66, "/tmp/cat/native-missing-content.txt",
                           "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee");
        REQUIRE(repo->insertDocument(doc).has_value());
        store->setBlob(doc.sha256Hash, "payload");
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(store);

        RequestDispatcher::__test_forceCatNativeMissingContentOnce();

        CatRequest req;
        req.hash = doc.sha256Hash;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Content unavailable");
    }
}

TEST_CASE("RequestDispatcher: download handler enforces daemon policy",
          "[daemon][documents][dispatcher][download]") {
    ScopedEnvVar daemonDownloadEnv("YAMS_ENABLE_DAEMON_DOWNLOAD", "0");

    auto dispatchDownload = [](RequestDispatcher& dispatcher, const std::string& url,
                               std::string outputPath = {}, std::string checksum = {}) {
        DownloadRequest req;
        req.url = url;
        req.outputPath = std::move(outputPath);
        req.checksum = std::move(checksum);
        return dispatchRequest(dispatcher, Request{req});
    };
    auto waitForDownloadJob = [](RequestDispatcher& dispatcher, const std::string& jobId,
                                 std::chrono::milliseconds timeout = std::chrono::seconds(2)) {
        const auto deadline = std::chrono::steady_clock::now() + timeout;
        DownloadResponse latest;
        while (std::chrono::steady_clock::now() < deadline) {
            auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{jobId}});
            if (std::holds_alternative<DownloadResponse>(resp)) {
                latest = std::get<DownloadResponse>(resp);
                if (latest.state != "queued" && latest.state != "running") {
                    return latest;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        return latest;
    };
    auto waitForFollowRedirectsCapture = [](std::chrono::milliseconds timeout =
                                                std::chrono::seconds(2)) {
        const auto deadline = std::chrono::steady_clock::now() + timeout;
        std::optional<bool> latest;
        while (std::chrono::steady_clock::now() < deadline) {
            latest = RequestDispatcher::__test_lastDownloadServiceFollowRedirects();
            if (latest.has_value()) {
                return latest;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        return latest;
    };

    RequestDispatcher::__test_resetDownloadServiceRequestCapture();

    SECTION("disabled policy returns safe reminder") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_disabled_");

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("Daemon download is disabled") != std::string::npos);
    }

    SECTION("disallowed scheme is rejected before download starts") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_scheme_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "http://example.com/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("scheme 'http'") != std::string::npos);
    }

    SECTION("invalid URL shape is rejected before policy host checks") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_invalid_url_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "not-a-url");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error == "Invalid download URL");
    }

    SECTION("download URL rejects missing authority after scheme") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_missing_authority_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error == "Invalid download URL");
    }

    SECTION("download URL rejects empty host after credentials stripping") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_empty_host_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://user@/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error == "Invalid download URL");
    }

    SECTION("wildcard host policy accepts matching subdomains before checksum gate") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_wildcard_host_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = true;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"*.example.com"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://api.example.com/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("checksum is required") != std::string::npos);
    }

    SECTION("IPv6 authority parses and matches exact allowed host") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_ipv6_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = true;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"::1"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://[::1]/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("checksum is required") != std::string::npos);
    }

    SECTION("invalid IPv6 authority is rejected") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_ipv6_invalid_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://[]/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error == "Invalid download URL");
    }

    SECTION("disallowed host is rejected before download starts") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_host_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"downloads.example.com"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("host 'example.com'") != std::string::npos);
    }

    SECTION("store-only policy rejects output path") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_store_only_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.storeOnly = true;

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", "/tmp/export.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("store_only=true") != std::string::npos);
    }

    SECTION("invalid checksum format is rejected before download starts") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_bad_checksum_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp =
            dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha1:not-allowed");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("invalid checksum format") != std::string::npos);
    }

    SECTION("unsupported checksum algorithm with valid hex is rejected") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_bad_checksum_algo_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp =
            dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha1:abcdef12");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("invalid checksum format") != std::string::npos);
    }

    SECTION("malformed checksum missing separator is rejected") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_bad_checksum_shape_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("invalid checksum format") != std::string::npos);
    }

    SECTION("checksum-required policy rejects missing checksum") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_checksum_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = true;
        cfg.downloadPolicy.allowedSchemes = {"https"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.error.find("checksum is required") != std::string::npos);
    }

    SECTION("valid checksum enters job tracking and handles unavailable service") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_valid_checksum_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"*"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();

        auto resp =
            dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK_FALSE(dl.success);
        CHECK(dl.state == "queued");
        CHECK_FALSE(dl.jobId.empty());

        const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
        CHECK(finalStatus.jobId == dl.jobId);
        CHECK(finalStatus.state == "failed");
        CHECK(finalStatus.error == "Download service not available in daemon");
    }

    SECTION("sha512 checksum enters job tracking and handles unavailable service") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_sha512_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"*"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", {},
                                     "sha512:abcdef1234567890");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.state == "queued");
        CHECK_FALSE(dl.jobId.empty());

        const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
        CHECK(finalStatus.jobId == dl.jobId);
        CHECK(finalStatus.state == "failed");
        CHECK(finalStatus.error == "Download service not available in daemon");
    }

    SECTION("md5 checksum enters job tracking and handles unavailable service") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_md5_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"*"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();

        auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", {},
                                     "md5:abcdef1234567890");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.state == "queued");
        CHECK_FALSE(dl.jobId.empty());

        const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
        CHECK(finalStatus.jobId == dl.jobId);
        CHECK(finalStatus.state == "failed");
        CHECK(finalStatus.error == "Download service not available in daemon");
    }

    SECTION("forced successful download completes the tracked job") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_success_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"*"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        RequestDispatcher::__test_forceDownloadServiceSuccessOnce();

        auto resp =
            dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.state == "queued");
        CHECK_FALSE(dl.jobId.empty());

        const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
        CHECK(finalStatus.jobId == dl.jobId);
        CHECK(finalStatus.state == "completed");
        CHECK(finalStatus.success);
        CHECK(finalStatus.error.empty());
        CHECK(finalStatus.hash == "sha256:forced-download-success");
        CHECK(finalStatus.localPath.find("forced-download-success.bin") != std::string::npos);
        CHECK(finalStatus.size == 64);
    }

    SECTION("daemon download disables redirects and surfaces blocked redirect failures") {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_download_policy_redirect_blocked_");
        cfg.downloadPolicy.enable = true;
        cfg.downloadPolicy.requireChecksum = false;
        cfg.downloadPolicy.allowedSchemes = {"https"};
        cfg.downloadPolicy.allowedHosts = {"*"};

        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        RequestDispatcher::__test_forceDownloadServiceFailureOnce(
            "Redirect blocked by policy during probe (follow_redirects=false)");

        auto resp =
            dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.state == "queued");
        CHECK_FALSE(dl.jobId.empty());

        const auto lastFollowRedirects = waitForFollowRedirectsCapture();
        REQUIRE(lastFollowRedirects.has_value());
        CHECK_FALSE(*lastFollowRedirects);

        const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
        CHECK(finalStatus.jobId == dl.jobId);
        CHECK(finalStatus.state == "failed");
        CHECK(finalStatus.error ==
              "Redirect blocked by policy during probe (follow_redirects=false)");
    }
}

TEST_CASE("RequestDispatcher: download job status and list use in-memory registry",
          "[daemon][documents][dispatcher][download-jobs]") {
    RequestDispatcher::__test_resetDownloadJobs();

    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_download_job_registry_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    DownloadResponse first;
    first.url = "https://example.com/a.bin";
    first.jobId = "download-1";
    first.state = "completed";
    first.createdAtMs = 100;
    first.updatedAtMs = 200;
    first.success = true;

    DownloadResponse second;
    second.url = "https://example.com/b.bin";
    second.jobId = "download-2";
    second.state = "failed";
    second.createdAtMs = 300;
    second.updatedAtMs = 400;
    second.success = false;
    second.error = "network boom";

    RequestDispatcher::__test_seedDownloadJob(first);
    RequestDispatcher::__test_seedDownloadJob(second);

    SECTION("download status returns the seeded job") {
        auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{"download-2"}});

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.jobId == "download-2");
        CHECK(dl.state == "failed");
        CHECK(dl.error == "network boom");
    }

    SECTION("download status returns not found for unknown job") {
        auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{"missing-job"}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "Download job not found");
    }

    SECTION("list download jobs returns most recent first and honors limit") {
        auto resp = dispatchRequest(dispatcher, Request{ListDownloadJobsRequest{1}});

        REQUIRE(std::holds_alternative<ListDownloadJobsResponse>(resp));
        const auto& jobs = std::get<ListDownloadJobsResponse>(resp);
        REQUIRE(jobs.jobs.size() == 1);
        CHECK(jobs.jobs.front().jobId == "download-2");
        CHECK(jobs.jobs.front().state == "failed");
    }

    SECTION("list download jobs clamps zero limit to one") {
        auto resp = dispatchRequest(dispatcher, Request{ListDownloadJobsRequest{0}});

        REQUIRE(std::holds_alternative<ListDownloadJobsResponse>(resp));
        const auto& jobs = std::get<ListDownloadJobsResponse>(resp);
        REQUIRE(jobs.jobs.size() == 1);
        CHECK(jobs.jobs.front().jobId == "download-2");
    }

    SECTION("cancel download job marks running job as canceled") {
        DownloadResponse running;
        running.url = "https://example.com/c.bin";
        running.jobId = "download-3";
        running.state = "running";
        running.createdAtMs = 500;
        running.updatedAtMs = 500;
        running.success = false;
        RequestDispatcher::__test_seedDownloadJob(running);

        auto resp = dispatchRequest(dispatcher, Request{CancelDownloadJobRequest{"download-3"}});

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.jobId == "download-3");
        CHECK(dl.state == "canceled");
        CHECK(dl.error == "Canceled by client");
        CHECK_FALSE(dl.success);
    }

    SECTION("cancel download job keeps terminal jobs terminal") {
        auto resp = dispatchRequest(dispatcher, Request{CancelDownloadJobRequest{"download-2"}});

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        CHECK(dl.jobId == "download-2");
        CHECK(dl.state == "failed");
        CHECK(dl.error == "network boom");
    }

    SECTION("cancel download job returns not found for unknown job") {
        auto resp = dispatchRequest(dispatcher, Request{CancelDownloadJobRequest{"missing-job"}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "Download job not found");
    }
}

TEST_CASE("RequestDispatcher: download jobs reload from persisted registry",
          "[daemon][documents][dispatcher][download-jobs]") {
    RequestDispatcher::__test_resetDownloadJobs();
    auto dispatchDownload = [](RequestDispatcher& dispatcher, const std::string& url,
                               std::string outputPath = {}, std::string checksum = {}) {
        DownloadRequest req;
        req.url = url;
        req.outputPath = std::move(outputPath);
        req.checksum = std::move(checksum);
        return dispatchRequest(dispatcher, Request{req});
    };
    auto waitForDownloadJob = [](RequestDispatcher& dispatcher, const std::string& jobId,
                                 std::chrono::milliseconds timeout = std::chrono::seconds(2)) {
        const auto deadline = std::chrono::steady_clock::now() + timeout;
        DownloadResponse latest;
        while (std::chrono::steady_clock::now() < deadline) {
            auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{jobId}});
            if (std::holds_alternative<DownloadResponse>(resp)) {
                latest = std::get<DownloadResponse>(resp);
                if (latest.state != "queued" && latest.state != "running") {
                    return latest;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        return latest;
    };

    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_download_job_registry_persist_");
    cfg.downloadPolicy.enable = true;
    cfg.downloadPolicy.requireChecksum = false;
    cfg.downloadPolicy.allowedSchemes = {"https"};
    cfg.downloadPolicy.allowedHosts = {"*"};

    std::string jobId;

    {
        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();

        auto resp =
            dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");

        REQUIRE(std::holds_alternative<DownloadResponse>(resp));
        const auto& dl = std::get<DownloadResponse>(resp);
        REQUIRE_FALSE(dl.jobId.empty());
        CHECK_FALSE(dl.success);
        CHECK(dl.state == "queued");
        jobId = dl.jobId;

        const auto finalStatus = waitForDownloadJob(dispatcher, jobId);
        CHECK(finalStatus.state == "failed");
        CHECK(finalStatus.error == "Download service not available in daemon");
    }

    RequestDispatcher::__test_forgetDownloadJobsCache();

    {
        StubLifecycle lifecycle;
        StateComponent state;
        state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        DaemonLifecycleFsm lifecycleFsm;
        ServiceManager svc(cfg, state, lifecycleFsm);
        RequestDispatcher dispatcher(&lifecycle, &svc, &state);

        auto statusResp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{jobId}});
        REQUIRE(std::holds_alternative<DownloadResponse>(statusResp));
        const auto& status = std::get<DownloadResponse>(statusResp);
        CHECK(status.jobId == jobId);
        CHECK(status.state == "failed");
        CHECK(status.error == "Download service not available in daemon");

        auto listResp = dispatchRequest(dispatcher, Request{ListDownloadJobsRequest{1}});
        REQUIRE(std::holds_alternative<ListDownloadJobsResponse>(listResp));
        const auto& list = std::get<ListDownloadJobsResponse>(listResp);
        REQUIRE(list.jobs.size() == 1);
        CHECK(list.jobs.front().jobId == jobId);
        CHECK(list.jobs.front().state == "failed");
    }
}

TEST_CASE("RequestDispatcher: embedding handlers cover generation and repair branches",
          "[daemon][embedding][dispatcher]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_embedding_dispatcher_");
    cfg.configFilePath = cfg.dataDir / "missing-config.toml";
    ScopedEnvVar preferredModelGuard("YAMS_PREFERRED_MODEL", std::nullopt);
    ScopedEnvVar embedBackendGuard("YAMS_EMBED_BACKEND", std::nullopt);
    ScopedEnvVar configPathGuard("YAMS_CONFIG", cfg.configFilePath.string());
    ScopedEnvVar xdgConfigHomeGuard("XDG_CONFIG_HOME", cfg.dataDir.string());
    ScopedEnvVar homeGuard("HOME", cfg.dataDir.string());

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    SECTION("generate embedding marks provider unavailable as degraded") {
        auto resp = dispatchRequest(
            dispatcher, Request{GenerateEmbeddingRequest{"hello world", "embed-model", true}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message.find("Plugin system") != std::string::npos);
        CHECK(lifecycle.setSubsystemDegradedCalls() == 1);
        CHECK(lifecycle.lastSubsystem() == "embedding");
        CHECK(lifecycle.lastDegraded());
        CHECK(lifecycle.lastReason() == "provider_unavailable");
    }

    SECTION("generate embedding auto-loads unloaded model and returns vector") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-model.onnx");
        provider->setSingleEmbeddingResult({0.1f, 0.2f, 0.3f});
        svc.__test_setModelProvider(provider);

        GenerateEmbeddingRequest req{"hello world", "embed-model", true};
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<EmbeddingResponse>(resp));
        const auto& embedResp = std::get<EmbeddingResponse>(resp);
        CHECK(embedResp.embedding == std::vector<float>{0.1f, 0.2f, 0.3f});
        CHECK(embedResp.dimensions == 3);
        CHECK(embedResp.modelUsed == "embed-model");
        CHECK(provider->isModelLoaded("embed-model"));
    }

    SECTION("generate embedding forwards generator errors") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-error.onnx");
        provider->setGenerateEmbeddingError(
            Error{ErrorCode::InvalidState, "generator unavailable"});
        svc.__test_setModelProvider(provider);

        GenerateEmbeddingRequest req{"hello world", "embed-error", true};
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == "generator unavailable");
    }

    SECTION("generate embedding sanitizes utf8 in load failures") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-utf8.onnx");
        std::string invalidMessage = "prefix ";
        invalidMessage.push_back(static_cast<char>(0xC2));
        invalidMessage.push_back(static_cast<char>(0xA9));
        invalidMessage.push_back(static_cast<char>(0xE2));
        invalidMessage.push_back(static_cast<char>(0x82));
        invalidMessage.push_back(static_cast<char>(0xAC));
        invalidMessage.push_back(static_cast<char>(0xF0));
        invalidMessage.push_back(static_cast<char>(0x9F));
        invalidMessage.push_back(static_cast<char>(0x92));
        invalidMessage.push_back(static_cast<char>(0xA9));
        invalidMessage.push_back(static_cast<char>(0xC0));
        invalidMessage.push_back(static_cast<char>(0xFF));
        provider->setLoadError(Error{ErrorCode::InvalidState, invalidMessage});
        svc.__test_setModelProvider(provider);

        GenerateEmbeddingRequest req{"hello world", "embed-utf8", true};
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        std::string expected = "prefix ";
        expected.push_back(static_cast<char>(0xC2));
        expected.push_back(static_cast<char>(0xA9));
        expected.push_back(static_cast<char>(0xE2));
        expected.push_back(static_cast<char>(0x82));
        expected.push_back(static_cast<char>(0xAC));
        expected.push_back(static_cast<char>(0xF0));
        expected.push_back(static_cast<char>(0x9F));
        expected.push_back(static_cast<char>(0x92));
        expected.push_back(static_cast<char>(0xA9));
        expected += "\xEF\xBF\xBD\xEF\xBF\xBD";
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == expected);
    }

    SECTION("generate embedding honors timeout env floor while auto-loading") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-timeout.onnx");
        provider->setSingleEmbeddingResult({0.7f, 0.8f, 0.9f});
        svc.__test_setModelProvider(provider);
        ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "5");

        GenerateEmbeddingRequest req{"hello world", "embed-timeout", true};
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<EmbeddingResponse>(resp));
        const auto& embedResp = std::get<EmbeddingResponse>(resp);
        CHECK(embedResp.embedding == std::vector<float>{0.7f, 0.8f, 0.9f});
        CHECK(provider->isModelLoaded("embed-timeout"));
    }

    SECTION("generate embedding converts isModelLoaded exceptions into internal errors") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-throw.onnx");
        provider->setThrowOnIsModelLoadedCount(2);
        svc.__test_setModelProvider(provider);

        GenerateEmbeddingRequest req{"hello world", "embed-throw", true};
        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Embedding generation failed: isModelLoaded boom") !=
              std::string::npos);
    }

    SECTION("batch embedding falls back to per-item generation") {
        auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-fallback.onnx");
        provider->setGenerateBatchError(Error{ErrorCode::NotImplemented, "batch unsupported"});
        provider->setSingleEmbeddingResult({0.4f, 0.6f});
        svc.__test_setModelProvider(provider);

        BatchEmbeddingRequest req;
        req.texts = {"one", "two"};
        req.modelName = "batch-fallback";
        req.normalize = false;
        req.batchSize = 2;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<BatchEmbeddingResponse>(resp));
        const auto& batchResp = std::get<BatchEmbeddingResponse>(resp);
        REQUIRE(batchResp.embeddings.size() == 2);
        CHECK(batchResp.embeddings[0] == std::vector<float>{0.4f, 0.6f});
        CHECK(batchResp.embeddings[1] == std::vector<float>{0.4f, 0.6f});
        CHECK(batchResp.successCount == 2);
        CHECK(batchResp.failureCount == 0);
        CHECK(batchResp.dimensions == 2);
        CHECK(batchResp.modelUsed == "batch-fallback");
    }

    SECTION("batch embedding counts per-item failures during fallback") {
        auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-partial.onnx");
        provider->setGenerateBatchError(Error{ErrorCode::NotImplemented, "batch unsupported"});
        provider->setSingleEmbeddingResult({0.9f, 0.1f});
        provider->setSingleTextError("bad", Error{ErrorCode::InternalError, "single failed"});
        svc.__test_setModelProvider(provider);

        BatchEmbeddingRequest req;
        req.texts = {"good", "bad", "good-again"};
        req.modelName = "batch-partial";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<BatchEmbeddingResponse>(resp));
        const auto& batchResp = std::get<BatchEmbeddingResponse>(resp);
        REQUIRE(batchResp.embeddings.size() == 2);
        CHECK(batchResp.successCount == 2);
        CHECK(batchResp.failureCount == 1);
        CHECK(batchResp.dimensions == 2);
        CHECK(batchResp.modelUsed == "batch-partial");
    }

    SECTION("batch embedding forwards non-fallback errors") {
        auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-error.onnx");
        provider->setGenerateBatchError(Error{ErrorCode::InvalidState, "batch failed"});
        svc.__test_setModelProvider(provider);

        BatchEmbeddingRequest req;
        req.texts = {"one", "two"};
        req.modelName = "batch-error";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == "batch failed");
    }

    SECTION("batch embedding auto-loads model and returns direct batch results") {
        auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-direct.onnx");
        provider->setBatchEmbeddingResult({{0.2f, 0.8f}});
        svc.__test_setModelProvider(provider);
        ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "not-a-number");

        BatchEmbeddingRequest req;
        req.texts = {"one", "two"};
        req.modelName = "batch-direct";
        req.normalize = true;
        req.batchSize = 2;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<BatchEmbeddingResponse>(resp));
        const auto& batchResp = std::get<BatchEmbeddingResponse>(resp);
        REQUIRE(batchResp.embeddings.size() == 1);
        CHECK(batchResp.embeddings[0] == std::vector<float>{0.2f, 0.8f});
        CHECK(batchResp.successCount == 1);
        CHECK(batchResp.failureCount == 1);
        CHECK(batchResp.dimensions == 2);
        CHECK(batchResp.modelUsed == "batch-direct");
        CHECK(provider->isModelLoaded("batch-direct"));
    }

    SECTION("batch embedding returns load failures for unloaded models") {
        auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-load-fail.onnx");
        provider->setLoadError(Error{ErrorCode::InvalidState, "batch load refused"});
        svc.__test_setModelProvider(provider);
        ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "5");

        BatchEmbeddingRequest req;
        req.texts = {"one", "two"};
        req.modelName = "batch-load-fail";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == "batch load refused");
    }

    SECTION("batch embedding converts isModelLoaded exceptions into internal errors") {
        auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-throw.onnx");
        provider->setThrowOnIsModelLoadedCount(2);
        svc.__test_setModelProvider(provider);

        BatchEmbeddingRequest req;
        req.texts = {"one"};
        req.modelName = "batch-throw";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Batch embedding failed: isModelLoaded boom") != std::string::npos);
    }

    SECTION("embed documents rejects missing service manager") {
        RequestDispatcher noServiceDispatcher(&lifecycle, nullptr, &state);

        auto resp = dispatchRequest(noServiceDispatcher, Request{EmbedDocumentsRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message == "ServiceManager not available");
    }

    SECTION("embed documents reports unavailable model provider") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-provider.onnx");
        provider->setAvailable(false);
        svc.__test_setModelProvider(provider);

        EmbedDocumentsRequest req;
        req.modelName = "repair-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message == "Model provider not available");
    }

    SECTION("embed documents returns success when repair finds no work") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-empty.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());
        svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());

        EmbedDocumentsRequest req;
        req.modelName = "repair-model";
        req.batchSize = 4;
        req.skipExisting = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<EmbedDocumentsResponse>(resp));
        const auto& embedResp = std::get<EmbedDocumentsResponse>(resp);
        CHECK(embedResp.requested == 0);
        CHECK(embedResp.embedded == 0);
        CHECK(embedResp.skipped == 0);
        CHECK(embedResp.failed == 0);
    }

    SECTION("embed documents rejects non-empty hash lists that resolve to no daemon documents") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-missing-hashes.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());
        svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());

        EmbedDocumentsRequest req;
        req.modelName = "repair-model";
        req.documentHashes = {std::string(64, 'a')};
        req.batchSize = 4;
        req.skipExisting = false;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message.find("requested documents were found") != std::string::npos);
    }

    SECTION("embed documents rejects degraded providers") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-degraded.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setModelProviderDegraded(true, "simulated degraded state");

        auto resp = dispatchRequest(dispatcher, Request{EmbedDocumentsRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == "Embedding generation disabled: provider degraded");

        svc.__test_setModelProviderDegraded(false);
    }

    SECTION("embed documents reports missing content store") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-content.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());

        EmbedDocumentsRequest req;
        req.modelName = "repair-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message == "Content store not available");
    }

    SECTION("embed documents reports missing metadata repo") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-meta.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());

        EmbedDocumentsRequest req;
        req.modelName = "repair-model";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message == "Metadata repository not available");
    }

    SECTION("embed documents reports missing configured model name") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-empty-model.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());
        svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());

        EmbedDocumentsRequest req;
        req.modelName = "";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotInitialized);
        CHECK(err.message == "No embedding model configured");
    }

    SECTION("embed documents uses daemon config resolver when request model is empty") {
        auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-config-model.onnx");
        svc.__test_setModelProvider(provider);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());
        svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());

        const auto configPath = cfg.dataDir / "config.toml";
        {
            std::ofstream out(configPath);
            REQUIRE(out.is_open());
            out << "[embeddings]\n";
            out << "preferred_model = \"config-repair-model\"\n";
        }
        cfg.configFilePath = configPath;
        ScopedEnvVar sectionConfigPathGuard("YAMS_CONFIG", configPath.string());

        DaemonLifecycleFsm configLifecycleFsm;
        ServiceManager configSvc(cfg, state, configLifecycleFsm);
        configSvc.__test_setModelProvider(provider);
        configSvc.__test_setContentStore(std::make_shared<StubContentStore>());
        configSvc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
        RequestDispatcher configDispatcher(&lifecycle, &configSvc, &state);

        EmbedDocumentsRequest req;
        req.modelName.clear();
        req.batchSize = 4;
        req.skipExisting = true;

        auto resp = dispatchRequest(configDispatcher, Request{req});

        REQUIRE(std::holds_alternative<EmbedDocumentsResponse>(resp));
        CHECK(provider->isModelLoaded("config-repair-model"));
    }
}

TEST_CASE("RequestDispatcher: graph maintenance handlers report missing graph component",
          "[daemon][graph][dispatcher]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_graph_dispatcher_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    SECTION("graph repair returns internal error when graph component is unavailable") {
        GraphRepairRequest req;
        req.dryRun = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("GraphComponent not available") != std::string::npos);
    }

    SECTION("graph validate returns internal error when graph component is unavailable") {
        auto resp = dispatchRequest(dispatcher, Request{GraphValidateRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("GraphComponent not available") != std::string::npos);
    }
}

TEST_CASE("RequestDispatcher: graph query and ingest handlers cover dispatcher branches",
          "[daemon][graph][dispatcher][query]") {
    GraphDispatcherFixture fixture;

    SECTION("graph query reports unavailable metadata repository") {
        auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphQueryRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Metadata repository unavailable");
    }

    SECTION("graph query reports unavailable knowledge graph store") {
        fixture.initMetadata(false);

        auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphQueryRequest{}});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK_FALSE(graphResp.kgAvailable);
        CHECK(graphResp.warning == "Knowledge graph not available");
    }

    SECTION("graph path history reports unavailable metadata repository") {
        GraphPathHistoryRequest req;
        req.path = "/repo/file.cpp";

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Metadata repository unavailable");
    }

    SECTION("graph explore reports unavailable metadata repository") {
        auto resp =
            dispatchRequest(*fixture.dispatcher, Request{GraphExploreRequest{.query = "entry"}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Metadata repository unavailable");
    }

    SECTION("graph explore reports unavailable knowledge graph store") {
        fixture.initMetadata(false);

        auto resp =
            dispatchRequest(*fixture.dispatcher, Request{GraphExploreRequest{.query = "entry"}});

        REQUIRE(std::holds_alternative<GraphExploreResponse>(resp));
        const auto& graphResp = std::get<GraphExploreResponse>(resp);
        CHECK(graphResp.query == "entry");
        CHECK_FALSE(graphResp.kgAvailable);
        REQUIRE(graphResp.warnings.size() == 1);
        CHECK(graphResp.warnings.front() == "Knowledge graph not available");
    }

    SECTION("graph query traversal requires an origin") {
        fixture.initMetadata();

        auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphQueryRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message.find("No valid origin specified") != std::string::npos);
    }

    SECTION("graph explore returns full and omitted snippet modes through dispatcher") {
        fixture.initMetadata();

        const auto sourcePath = fixture.testDir / "src" / "fallback.cpp";
        std::filesystem::create_directories(sourcePath.parent_path());
        {
            std::ofstream out(sourcePath);
            REQUIRE(out.good());
            out << "int fallbackEntry() {\n    return 7;\n}\n";
        }

        REQUIRE(
            fixture.repo->insertDocument(makeGraphDispatcherDocument(sourcePath, "hash-fallback"))
                .has_value());

        metadata::KGNode node;
        node.nodeKey = "function:demo::fallbackEntry@" + sourcePath.string();
        node.label = "fallbackEntry";
        node.type = "function";
        REQUIRE(fixture.kgStore->upsertNode(node).has_value());

        GraphExploreRequest fullReq;
        fullReq.query = "fallbackEntry";

        auto fullRespRaw = dispatchRequest(*fixture.dispatcher, Request{fullReq});
        REQUIRE(std::holds_alternative<GraphExploreResponse>(fullRespRaw));
        const auto& fullResp = std::get<GraphExploreResponse>(fullRespRaw);
        REQUIRE(fullResp.files.size() == 1);
        CHECK(fullResp.files.front().mode == "full");
        CHECK(fullResp.files.front().content.find("fallbackEntry") != std::string::npos);
        CHECK(fullResp.entrySymbols.size() == 1);
        CHECK(fullResp.entrySymbols.front().label == "fallbackEntry");
        CHECK(fullResp.snippetsRendered == 1);

        GraphExploreRequest omittedReq;
        omittedReq.query = "fallbackEntry";
        omittedReq.includeCode = false;

        auto omittedRespRaw = dispatchRequest(*fixture.dispatcher, Request{omittedReq});
        REQUIRE(std::holds_alternative<GraphExploreResponse>(omittedRespRaw));
        const auto& omittedResp = std::get<GraphExploreResponse>(omittedRespRaw);
        REQUIRE(omittedResp.files.size() == 1);
        CHECK(omittedResp.files.front().mode == "omitted");
        CHECK(omittedResp.files.front().content.empty());
        CHECK(omittedResp.snippetRenderMicros == 0);
        CHECK(omittedResp.snippetsRendered == 0);
    }

    SECTION("graph query lists node types") {
        fixture.initMetadata();
        fixture.upsertNode("fn:list:1", "alpha", "function");
        fixture.upsertNode("fn:list:2", "beta", "function");
        fixture.upsertNode("path:list:1", "file.cpp", "file");

        GraphQueryRequest req;
        req.listTypes = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.kgAvailable);
        CHECK(graphResp.originNode.label == "listTypes");
        REQUIRE(graphResp.nodeTypeCounts.size() == 2);
        CHECK(graphResp.nodeTypeCounts[0] == std::pair<std::string, uint64_t>{"function", 2});
        CHECK(graphResp.nodeTypeCounts[1] == std::pair<std::string, uint64_t>{"file", 1});
    }

    SECTION("graph query list types reports store errors") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");

        GraphQueryRequest req;
        req.listTypes = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code != ErrorCode::Success);
        CHECK_FALSE(err.message.empty());
    }

    SECTION("graph query lists relation counts") {
        fixture.initMetadata();
        auto callerId = fixture.upsertNode("fn:rel:caller", "caller", "function");
        auto calleeId = fixture.upsertNode("fn:rel:callee", "callee", "function");
        auto fileId = fixture.upsertNode("path:rel:file", "file.cpp", "file");
        fixture.addEdge(callerId, calleeId, "calls");
        fixture.addEdge(callerId, fileId, "includes");

        GraphQueryRequest req;
        req.listRelations = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.originNode.label == "listRelations");
        REQUIRE(graphResp.relationTypeCounts.size() == 2);
        CHECK(graphResp.relationTypeCounts[0] == std::pair<std::string, uint64_t>{"calls", 1});
        CHECK(graphResp.relationTypeCounts[1] == std::pair<std::string, uint64_t>{"includes", 1});
    }

    SECTION("graph query list relations reports store errors") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_edges RENAME TO kg_edges_broken");

        GraphQueryRequest req;
        req.listRelations = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code != ErrorCode::Success);
        CHECK_FALSE(err.message.empty());
    }

    SECTION("graph query search mode requires a pattern") {
        fixture.initMetadata();

        GraphQueryRequest req;
        req.searchMode = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message == "searchPattern is required for search mode");
    }

    SECTION("graph query search mode returns matching nodes with properties") {
        fixture.initMetadata();
        fixture.upsertNode("fn:search:alpha", "AlphaNode", "function", R"({"rank":1})");
        fixture.upsertNode("fn:search:beta", "BetaNode", "function");

        GraphQueryRequest req;
        req.searchMode = true;
        req.searchPattern = "%Alpha%";
        req.includeNodeProperties = true;
        req.limit = 10;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.originNode.label == "search:%Alpha%");
        REQUIRE(graphResp.connectedNodes.size() == 1);
        CHECK(graphResp.connectedNodes[0].nodeKey == "fn:search:alpha");
        CHECK(graphResp.connectedNodes[0].properties == R"({"rank":1})");
    }

    SECTION("graph query search mode reports store errors") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");

        GraphQueryRequest req;
        req.searchMode = true;
        req.searchPattern = "%Alpha%";

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code != ErrorCode::Success);
        CHECK_FALSE(err.message.empty());
    }

    SECTION("graph query list by type requires node type") {
        fixture.initMetadata();

        GraphQueryRequest req;
        req.listByType = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message == "nodeType is required for listByType mode");
    }

    SECTION("graph query list by type returns paginated nodes") {
        fixture.initMetadata();
        fixture.upsertNode("fn:type:1", "TypeOne", "function", R"({"kind":"a"})");
        fixture.upsertNode("fn:type:2", "TypeTwo", "function", R"({"kind":"b"})");
        fixture.upsertNode("path:type:1", "file", "file");

        GraphQueryRequest req;
        req.listByType = true;
        req.nodeType = "function";
        req.limit = 1;
        req.offset = 0;
        req.includeNodeProperties = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.originNode.label == "listByType:function");
        CHECK(graphResp.totalNodesFound == 2);
        CHECK(graphResp.truncated);
        REQUIRE(graphResp.connectedNodes.size() == 1);
        CHECK(graphResp.connectedNodes[0].type == "function");
        CHECK_FALSE(graphResp.connectedNodes[0].properties.empty());
    }

    SECTION("graph query list by type scopes and paginates within current code roots") {
        fixture.initMetadata();

        const auto sourcePath = fixture.testDir / "src" / "inside.cpp";
        const auto includePath = fixture.testDir / "include" / "inside.hpp";
        const auto testPath = fixture.testDir / "tests" / "outside.cpp";
        for (const auto& path : {sourcePath, includePath, testPath}) {
            std::filesystem::create_directories(path.parent_path());
            std::ofstream out(path);
            REQUIRE(out.good());
            out << "int value;\n";
        }

        REQUIRE(fixture.repo->insertDocument(makeGraphDispatcherDocument(sourcePath, "hash-src"))
                    .has_value());
        REQUIRE(
            fixture.repo->insertDocument(makeGraphDispatcherDocument(includePath, "hash-include"))
                .has_value());
        REQUIRE(fixture.repo->insertDocument(makeGraphDispatcherDocument(testPath, "hash-test"))
                    .has_value());

        fixture.upsertNode("function:insideSource@" + sourcePath.string(), "insideSource",
                           "function");
        fixture.upsertNode("function:insideHeader@" + includePath.string(), "insideHeader",
                           "function");
        fixture.upsertNode("function:outsideTest@" + testPath.string(), "outsideTest", "function");

        GraphQueryRequest firstPage;
        firstPage.listByType = true;
        firstPage.nodeType = "function";
        firstPage.scopePathPrefix = fixture.testDir.string();
        firstPage.limit = 1;

        auto firstRaw = dispatchRequest(*fixture.dispatcher, Request{firstPage});
        REQUIRE(std::holds_alternative<GraphQueryResponse>(firstRaw));
        const auto& first = std::get<GraphQueryResponse>(firstRaw);
        CHECK(first.totalNodesFound == 2);
        CHECK(first.truncated);
        REQUIRE(first.connectedNodes.size() == 1);
        CHECK(first.connectedNodes.front().label != "outsideTest");

        firstPage.offset = 1;
        auto secondRaw = dispatchRequest(*fixture.dispatcher, Request{firstPage});
        REQUIRE(std::holds_alternative<GraphQueryResponse>(secondRaw));
        const auto& second = std::get<GraphQueryResponse>(secondRaw);
        CHECK(second.totalNodesFound == 2);
        CHECK_FALSE(second.truncated);
        REQUIRE(second.connectedNodes.size() == 1);
        CHECK(second.connectedNodes.front().label != "outsideTest");
        CHECK(second.connectedNodes.front().nodeId != first.connectedNodes.front().nodeId);
    }

    SECTION("graph query list by type reports store errors") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");

        GraphQueryRequest req;
        req.listByType = true;
        req.nodeType = "function";

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code != ErrorCode::Success);
        CHECK_FALSE(err.message.empty());
    }

    SECTION("graph query isolated mode canonicalizes relation name") {
        fixture.initMetadata();
        auto calledId = fixture.upsertNode("fn:isolated:called", "called", "function");
        fixture.upsertNode("fn:isolated:one", "isolated1", "function");
        fixture.upsertNode("fn:isolated:two", "isolated2", "function");
        auto callerId = fixture.upsertNode("fn:isolated:caller", "caller", "function");
        fixture.addEdge(callerId, calledId, "calls");

        GraphQueryRequest req;
        req.isolatedMode = true;
        req.nodeType = "function";
        req.isolatedRelation = " CALL ";
        req.limit = 10;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.originNode.label == "isolated:function:calls");
        CHECK(graphResp.totalNodesFound == 3);
        for (const auto& node : graphResp.connectedNodes) {
            CHECK(node.nodeKey != "fn:isolated:called");
        }
    }

    SECTION("graph query isolated mode uses default node type and relation") {
        fixture.initMetadata();
        auto calledId = fixture.upsertNode("fn:isolated:default:called", "called", "function");
        fixture.upsertNode("fn:isolated:default:free", "free", "function");
        auto callerId = fixture.upsertNode("fn:isolated:default:caller", "caller", "function");
        fixture.addEdge(callerId, calledId, "calls");

        GraphQueryRequest req;
        req.isolatedMode = true;
        req.limit = 10;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.originNode.label == "isolated:function:calls");
        REQUIRE(graphResp.connectedNodes.size() == 2);
    }

    SECTION("graph query isolated mode reports store errors") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");

        GraphQueryRequest req;
        req.isolatedMode = true;
        req.limit = 10;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code != ErrorCode::Success);
        CHECK_FALSE(err.message.empty());
    }

    SECTION("graph query node key lookup returns not found") {
        fixture.initMetadata();

        GraphQueryRequest req;
        req.nodeKey = "missing-node";

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "Node not found: missing-node");
    }

    SECTION("graph query traverses from node key with canonical relation filters") {
        fixture.initMetadata();
        auto callerId = fixture.upsertNode("fn:key:caller", "caller", "function");
        auto calleeId = fixture.upsertNode("fn:key:callee", "callee", "function");
        auto fileId = fixture.upsertNode("path:key:file", "header", "file");
        fixture.addEdge(callerId, calleeId, "calls", 0.8f, R"({"source":"unit"})");
        fixture.addEdge(callerId, fileId, "includes");

        GraphQueryRequest req;
        req.nodeKey = "fn:key:caller";
        req.maxDepth = 1;
        req.maxResults = 10;
        req.relationFilters = {" Call "};
        req.includeEdgeProperties = true;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        CHECK(graphResp.kgAvailable);
        CHECK(graphResp.originNode.nodeKey == "fn:key:caller");
        REQUIRE(graphResp.connectedNodes.size() == 1);
        CHECK(graphResp.connectedNodes[0].nodeKey == "fn:key:callee");
        REQUIRE(graphResp.edges.size() == 1);
        CHECK(graphResp.edges[0].relation == "calls");
        CHECK(graphResp.edges[0].properties == R"({"source":"unit"})");
    }

    SECTION("graph query canonicalizes relation aliases and skips blank filters") {
        fixture.initMetadata();
        auto originId = fixture.upsertNode("fn:key:origin", "origin", "function");
        auto includeId = fixture.upsertNode("path:key:include", "header", "file");
        auto inheritId = fixture.upsertNode("cls:key:inherit", "Base", "class");
        auto implementId = fixture.upsertNode("iface:key:impl", "Iface", "interface");
        auto referenceId = fixture.upsertNode("sym:key:ref", "Ref", "symbol");
        auto renameToId = fixture.upsertNode("path:key:rename_to", "new.cpp", "file");
        auto renameFromId = fixture.upsertNode("path:key:rename_from", "old.cpp", "file");
        auto customId = fixture.upsertNode("sym:key:custom", "Custom", "symbol");
        fixture.addEdge(originId, includeId, "includes");
        fixture.addEdge(originId, inheritId, "inherits");
        fixture.addEdge(originId, implementId, "implements");
        fixture.addEdge(originId, referenceId, "references");
        fixture.addEdge(originId, renameToId, "renamed_to");
        fixture.addEdge(originId, renameFromId, "renamed_from");
        fixture.addEdge(originId, customId, "custom_rel");

        GraphQueryRequest req;
        req.nodeKey = "fn:key:origin";
        req.maxDepth = 1;
        req.maxResults = 20;
        req.relationFilters = {"   ",       " Include ", "inherit",     "implement",
                               "reference", "rename_to", "rename_from", "custom-rel"};

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
        const auto& graphResp = std::get<GraphQueryResponse>(resp);
        REQUIRE(graphResp.edges.size() == 7);
        std::unordered_set<std::string> relations;
        for (const auto& edge : graphResp.edges) {
            relations.insert(edge.relation);
        }
        CHECK(relations.count("includes") == 1);
        CHECK(relations.count("inherits") == 1);
        CHECK(relations.count("implements") == 1);
        CHECK(relations.count("references") == 1);
        CHECK(relations.count("renamed_to") == 1);
        CHECK(relations.count("renamed_from") == 1);
        CHECK(relations.count("custom_rel") == 1);
    }

    SECTION("graph path history requires path") {
        fixture.initMetadata();

        auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphPathHistoryRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidArgument);
        CHECK(err.message == "Path is required for path history query");
    }

    SECTION("graph path history returns empty response when kg store is unavailable") {
        fixture.initMetadata(false);

        GraphPathHistoryRequest req;
        req.path = "/repo/file.cpp";

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphPathHistoryResponse>(resp));
        const auto& historyResp = std::get<GraphPathHistoryResponse>(resp);
        CHECK(historyResp.queryPath == "/repo/file.cpp");
        CHECK(historyResp.history.empty());
        CHECK_FALSE(historyResp.hasMore);
    }

    SECTION("graph path history returns snapshot entries") {
        fixture.initMetadata();
        fixture.upsertNode("path:snap1:/repo/file.cpp", "/repo/file.cpp", "path",
                           R"({"snapshot_id":"snap1","path":"/repo/file.cpp"})");
        fixture.upsertNode("path:snap2:/repo/file.cpp", "/repo/file.cpp", "path",
                           R"({"snapshot_id":"snap2","path":"/repo/file.cpp"})");

        GraphPathHistoryRequest req;
        req.path = "/repo/file.cpp";
        req.limit = 10;

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<GraphPathHistoryResponse>(resp));
        const auto& historyResp = std::get<GraphPathHistoryResponse>(resp);
        CHECK(historyResp.queryPath == "/repo/file.cpp");
        REQUIRE(historyResp.history.size() == 2);
        CHECK(historyResp.history[0].snapshotId == "snap2");
        CHECK(historyResp.history[0].blobHash.empty());
        CHECK(historyResp.history[0].changeType == "unknown");
        CHECK(historyResp.history[1].snapshotId == "snap1");
        CHECK_FALSE(historyResp.hasMore);
    }

    SECTION("graph path history reports store errors") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");

        GraphPathHistoryRequest req;
        req.path = "/repo/file.cpp";

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code != ErrorCode::Success);
        CHECK_FALSE(err.message.empty());
    }

    SECTION("kg ingest reports unavailable metadata repository") {
        auto resp = dispatchRequest(*fixture.dispatcher, Request{KgIngestRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Metadata repository unavailable");
    }

    SECTION("kg ingest reports unavailable knowledge graph store") {
        fixture.initMetadata(false);

        auto resp = dispatchRequest(*fixture.dispatcher, Request{KgIngestRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Knowledge graph store unavailable");
    }

    SECTION("kg ingest records node edge and alias resolution errors") {
        fixture.initMetadata();
        auto existingId = fixture.upsertNode("fn:existing", "existing", "function");
        (void)existingId;

        KgIngestRequest req;
        req.nodes.push_back(KgIngestNode{"fn:existing", "existing updated", "function", ""});
        req.edges.push_back(KgIngestEdge{"fn:existing", "fn:missing", "calls", 1.0f, ""});
        req.aliases.push_back(KgIngestAlias{"fn:missing", "missing-alias", "unit", 0.5f});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK(ingestResp.success);
        CHECK(ingestResp.nodesInserted == 1);
        CHECK(ingestResp.edgesSkipped == 1);
        CHECK(ingestResp.aliasesSkipped == 1);
        REQUIRE(ingestResp.errors.size() == 2);
        CHECK(ingestResp.errors[0] == "Edge destination node not found: fn:missing");
        CHECK(ingestResp.errors[1] == "Alias node not found: fn:missing");
    }

    SECTION("kg ingest inserts nodes edges and aliases") {
        fixture.initMetadata();

        KgIngestRequest req;
        req.nodes.push_back(
            KgIngestNode{"fn:ingest:src", "Source", "function", R"({"role":"src"})"});
        req.nodes.push_back(KgIngestNode{"fn:ingest:dst", "Dest", "function", ""});
        req.edges.push_back(KgIngestEdge{"fn:ingest:src", "fn:ingest:dst", "calls", 0.7f,
                                         R"({"origin":"dispatcher"})"});
        req.aliases.push_back(KgIngestAlias{"fn:ingest:src", "sourceAlias", "unit", 0.9f});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK(ingestResp.success);
        CHECK(ingestResp.nodesInserted == 2);
        CHECK(ingestResp.edgesInserted == 1);
        CHECK(ingestResp.aliasesInserted == 1);
        CHECK(ingestResp.errors.empty());

        auto sourceNode = fixture.kgStore->getNodeByKey("fn:ingest:src");
        REQUIRE(sourceNode.has_value());
        REQUIRE(sourceNode.value().has_value());
        auto edges =
            fixture.kgStore->getEdgesFrom(sourceNode.value()->id, std::string_view("calls"), 10, 0);
        REQUIRE(edges.has_value());
        REQUIRE(edges.value().size() == 1);
        auto aliases = fixture.kgStore->resolveAliasExact("sourceAlias", 10);
        REQUIRE(aliases.has_value());
        REQUIRE(aliases.value().size() == 1);
    }

    SECTION("kg ingest reports node upsert failures") {
        fixture.initMetadata();
        fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");

        KgIngestRequest req;
        req.nodes.push_back(KgIngestNode{"fn:broken:node", "Broken", "function", ""});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK_FALSE(ingestResp.success);
        CHECK(ingestResp.nodesInserted == 0);
        REQUIRE(ingestResp.errors.size() == 1);
        CHECK(ingestResp.errors[0].find("Node upsert failed:") == 0);
    }

    SECTION("kg ingest inserts edges without de-duplication when requested") {
        fixture.initMetadata();
        fixture.upsertNode("fn:addedges:src", "Source", "function");
        fixture.upsertNode("fn:addedges:dst", "Dest", "function");

        KgIngestRequest req;
        req.skipExistingEdges = false;
        req.edges.push_back(KgIngestEdge{"fn:addedges:src", "fn:addedges:dst", "calls", 1.0f, ""});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK(ingestResp.success);
        CHECK(ingestResp.edgesInserted == 1);
        CHECK(ingestResp.errors.empty());
    }

    SECTION("kg ingest reports edge insert failures") {
        fixture.initMetadata();
        fixture.upsertNode("fn:edgefail:src", "Source", "function");
        fixture.upsertNode("fn:edgefail:dst", "Dest", "function");
        fixture.execSql("ALTER TABLE kg_edges RENAME TO kg_edges_broken");

        KgIngestRequest req;
        req.skipExistingEdges = false;
        req.edges.push_back(KgIngestEdge{"fn:edgefail:src", "fn:edgefail:dst", "calls", 1.0f, ""});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK_FALSE(ingestResp.success);
        CHECK(ingestResp.edgesInserted == 0);
        REQUIRE(ingestResp.errors.size() == 1);
        CHECK(ingestResp.errors[0].find("Edge insert failed:") == 0);
    }

    SECTION("kg ingest reports alias insert failures") {
        fixture.initMetadata();
        fixture.upsertNode("fn:aliasfail:node", "Node", "function");
        fixture.execSql("ALTER TABLE kg_aliases RENAME TO kg_aliases_broken");

        KgIngestRequest req;
        req.aliases.push_back(KgIngestAlias{"fn:aliasfail:node", "brokenAlias", "unit", 0.4f});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK(ingestResp.success);
        CHECK(ingestResp.aliasesInserted == 0);
        REQUIRE(ingestResp.errors.size() == 1);
        CHECK(ingestResp.errors[0].find("Alias insert failed:") == 0);
    }

    SECTION("kg ingest resolves existing nodes for edges and aliases") {
        fixture.initMetadata();
        auto srcId = fixture.upsertNode("fn:existing:src", "Source", "function");
        auto dstId = fixture.upsertNode("fn:existing:dst", "Dest", "function");
        auto aliasNodeId = fixture.upsertNode("fn:existing:alias", "AliasTarget", "function");
        (void)srcId;
        (void)dstId;
        (void)aliasNodeId;

        KgIngestRequest req;
        req.edges.push_back(KgIngestEdge{"fn:existing:src", "fn:existing:dst", "calls", 1.0f, ""});
        req.aliases.push_back(KgIngestAlias{"fn:existing:alias", "existingAlias", "", 0.6f});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK(ingestResp.success);
        CHECK(ingestResp.nodesInserted == 0);
        CHECK(ingestResp.edgesInserted == 1);
        CHECK(ingestResp.aliasesInserted == 1);
        CHECK(ingestResp.errors.empty());

        auto aliases = fixture.kgStore->resolveAliasExact("existingAlias", 10);
        REQUIRE(aliases.has_value());
        REQUIRE(aliases.value().size() == 1);
    }

    SECTION("kg ingest reports missing edge source nodes") {
        fixture.initMetadata();
        fixture.upsertNode("fn:existing:dst_only", "Dest", "function");

        KgIngestRequest req;
        req.edges.push_back(
            KgIngestEdge{"fn:missing:src", "fn:existing:dst_only", "calls", 1.0f, ""});

        auto resp = dispatchRequest(*fixture.dispatcher, Request{req});

        REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
        const auto& ingestResp = std::get<KgIngestResponse>(resp);
        CHECK(ingestResp.success);
        CHECK(ingestResp.edgesInserted == 0);
        CHECK(ingestResp.edgesSkipped == 1);
        REQUIRE(ingestResp.errors.size() == 1);
        CHECK(ingestResp.errors[0] == "Edge source node not found: fn:missing:src");
    }
}

TEST_CASE("RequestDispatcher: plugin handlers cover readiness and error branches",
          "[daemon][plugin][dispatcher]") {
    auto makeReadyService = []() {
        DaemonConfig cfg;
        cfg.dataDir = makeTempDir("yams_plugin_dispatcher_");

        auto state = std::make_unique<StateComponent>();
        state->readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state->readiness.metadataRepoReady.store(true, std::memory_order_relaxed);

        auto lifecycleFsm = std::make_unique<DaemonLifecycleFsm>();
        auto svc = std::make_unique<ServiceManager>(cfg, *state, *lifecycleFsm);
        svc->__test_pluginScanComplete(0);
        REQUIRE(svc->getPluginHostFsmSnapshot().state == PluginHostState::Ready);

        return std::tuple{std::move(state), std::move(lifecycleFsm), std::move(svc)};
    };

    SECTION("plugin handlers reject non-ready host state") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

        svc->__test_pluginLoadFailed("simulated plugin init failure");

        auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InvalidState);
        CHECK(err.message == "Plugin subsystem not ready (state=failed)");
    }

    SECTION("plugin handlers reject missing service manager and non-ready FSM states") {
        auto state = std::make_unique<StateComponent>();
        state->readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        state->readiness.metadataRepoReady.store(true, std::memory_order_relaxed);

        StubLifecycle lifecycle;

        {
            RequestDispatcher dispatcher(&lifecycle, nullptr, state.get());
            auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});

            REQUIRE(std::holds_alternative<ErrorResponse>(resp));
            const auto& err = std::get<ErrorResponse>(resp);
            CHECK(err.code == ErrorCode::InvalidState);
            CHECK(err.message == "Plugin host unavailable");
        }

        {
            auto [readyState, lifecycleFsm, svc] = makeReadyService();
            RequestDispatcher dispatcher(&lifecycle, svc.get(), readyState.get());

            svc->__test_pluginScanStarted(1);

            auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});

            REQUIRE(std::holds_alternative<ErrorResponse>(resp));
            const auto& err = std::get<ErrorResponse>(resp);
            CHECK(err.code == ErrorCode::InvalidState);
            CHECK(err.message == "Plugin subsystem not ready (state=scanning)");
        }

        {
            auto [readyState, lifecycleFsm, svc] = makeReadyService();
            RequestDispatcher dispatcher(&lifecycle, svc.get(), readyState.get());

            svc->__test_pluginLoaded("mock-loaded-plugin");
            svc->__test_pluginScanComplete(1);
            svc->__test_pluginLoaded("mock-loaded-plugin");

            auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});

            REQUIRE(std::holds_alternative<ErrorResponse>(resp));
            const auto& err = std::get<ErrorResponse>(resp);
            CHECK(err.code == ErrorCode::InvalidState);
            CHECK(err.message == "Plugin subsystem not ready (state=loading)");
        }
    }

    SECTION("plugin handlers report missing hosts and trust failures") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

        svc->__test_setAbiHost({});
        svc->__test_setExternalPluginHost({});

        auto scanResp = dispatchRequest(dispatcher, Request{PluginScanRequest{}});
        REQUIRE(std::holds_alternative<ErrorResponse>(scanResp));
        const auto& scanErr = std::get<ErrorResponse>(scanResp);
        CHECK(scanErr.code == ErrorCode::NotImplemented);
        CHECK(scanErr.message == "No plugin host available");

        PluginLoadRequest loadReq;
        loadReq.pathOrName = "missing-plugin";
        auto loadResp = dispatchRequest(dispatcher, Request{loadReq});

        REQUIRE(std::holds_alternative<ErrorResponse>(loadResp));
        const auto& loadErr = std::get<ErrorResponse>(loadResp);
        CHECK(loadErr.code == ErrorCode::NotImplemented);
        CHECK(loadErr.message == "No plugin host available");

        auto unloadResp =
            dispatchRequest(dispatcher, Request{PluginUnloadRequest{"missing-plugin"}});
        REQUIRE(std::holds_alternative<ErrorResponse>(unloadResp));
        const auto& unloadErr = std::get<ErrorResponse>(unloadResp);
        CHECK(unloadErr.code == ErrorCode::NotFound);
        CHECK(unloadErr.message == "Plugin not found or unload failed");

        auto trustAddResp =
            dispatchRequest(dispatcher, Request{PluginTrustAddRequest{"/tmp/missing-host"}});
        REQUIRE(std::holds_alternative<ErrorResponse>(trustAddResp));
        const auto& trustAddErr = std::get<ErrorResponse>(trustAddResp);
        CHECK(trustAddErr.code == ErrorCode::Unknown);
        CHECK(trustAddErr.message == "Trust add failed");

        auto trustRemoveResp =
            dispatchRequest(dispatcher, Request{PluginTrustRemoveRequest{"/tmp/missing-host"}});
        REQUIRE(std::holds_alternative<ErrorResponse>(trustRemoveResp));
        const auto& trustRemoveErr = std::get<ErrorResponse>(trustRemoveResp);
        CHECK(trustRemoveErr.code == ErrorCode::Unknown);
        CHECK(trustRemoveErr.message == "Trust remove failed");
    }

    SECTION("plugin load dry run reports missing plugin") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

        PluginLoadRequest req;
        req.pathOrName = (svc->getResolvedDataDir() / "missing_plugin.py").string();
        req.dryRun = true;

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::NotFound);
        CHECK(err.message == "Plugin not found");
    }

    SECTION("plugin scan and load search default directories") {
        auto homeDir = makeTempDir("yams_plugin_home_");
        auto homeText = homeDir.string();
        ScopedEnvVar homeGuard("HOME", homeText.c_str());
#ifdef _WIN32
        ScopedEnvVar localAppDataGuard("LOCALAPPDATA", homeText.c_str());
#endif

        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

#ifdef _WIN32
        auto abiDir = homeDir / "yams" / "plugins";
#else
        auto abiDir = homeDir / ".local" / "lib" / "yams" / "plugins";
#endif
        std::filesystem::create_directories(abiDir);

        auto fakePlugin = abiDir /
#ifdef _WIN32
                          "yams_default_scan.dll";
#else
                          "yams_default_scan.so";
#endif
        {
            std::ofstream out(fakePlugin, std::ios::binary);
            out << "not a valid plugin";
        }

#ifdef _WIN32
        auto externalDir = homeDir / "yams" / "external-plugins";
#else
        auto externalDir = homeDir / ".local" / "lib" / "yams" / "external-plugins";
#endif
        std::filesystem::create_directories(externalDir);
        auto externalPluginDir =
            createMockExternalPlugin(externalDir, "dispatcher_external_default");

        auto missingResp = dispatchRequest(
            dispatcher, Request{PluginLoadRequest{"missing_default_plugin.so", "", false}});
        REQUIRE(std::holds_alternative<ErrorResponse>(missingResp));
        const auto& missingErr = std::get<ErrorResponse>(missingResp);
        CHECK(missingErr.code == ErrorCode::NotFound);
        CHECK(missingErr.message == "Plugin not found");

        auto scanResp = dispatchRequest(dispatcher, Request{PluginScanRequest{}});
        REQUIRE(std::holds_alternative<PluginScanResponse>(scanResp));
        const auto& scan = std::get<PluginScanResponse>(scanResp);

        bool foundDefaultPlugin = false;
        bool foundDefaultExternalPlugin = false;
        for (const auto& record : scan.plugins) {
            if (record.path == fakePlugin.string()) {
                foundDefaultPlugin = true;
                CHECK(record.name == "yams_default_scan");
            }
            if (record.path == externalPluginDir.string()) {
                foundDefaultExternalPlugin = true;
                CHECK(record.name == "dispatcher_external_plugin");
            }
        }
        CHECK(foundDefaultPlugin);
        CHECK(foundDefaultExternalPlugin);

        auto trustResp =
            dispatchRequest(dispatcher, Request{PluginTrustAddRequest{abiDir.string()}});
        REQUIRE(std::holds_alternative<SuccessResponse>(trustResp));

        auto* external = svc->getExternalPluginHost();
        REQUIRE(external != nullptr);
        REQUIRE(external->trustAdd(externalDir).has_value());

        PluginLoadRequest req;
        req.pathOrName = fakePlugin.filename().string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Plugin load failed for: " + fakePlugin.string());

        PluginLoadRequest externalReq;
        externalReq.pathOrName = externalPluginDir.filename().string();

        auto externalResp = dispatchRequest(dispatcher, Request{externalReq});

        REQUIRE(std::holds_alternative<PluginLoadResponse>(externalResp));
        const auto& externalLoad = std::get<PluginLoadResponse>(externalResp);
        CHECK(externalLoad.loaded);
        CHECK(externalLoad.record.name == "dispatcher_external_plugin");
        CHECK(externalLoad.record.path == externalPluginDir.string());

        auto externalUnloadResp =
            dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
        REQUIRE(std::holds_alternative<SuccessResponse>(externalUnloadResp));
    }

    SECTION("plugin handlers route external plugins when ABI host is absent") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

        svc->__test_setAbiHost({});

        auto pluginDir = createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external");

        PluginScanRequest scanReq;
        scanReq.target = pluginDir.string();
        auto scanResp = dispatchRequest(dispatcher, Request{scanReq});

        REQUIRE(std::holds_alternative<PluginScanResponse>(scanResp));
        const auto& scan = std::get<PluginScanResponse>(scanResp);
        REQUIRE(scan.plugins.size() == 1);
        CHECK(scan.plugins.front().name == "dispatcher_external_plugin");

        PluginLoadRequest dryRunReq;
        dryRunReq.pathOrName = pluginDir.string();
        dryRunReq.dryRun = true;
        auto dryRunResp = dispatchRequest(dispatcher, Request{dryRunReq});

        REQUIRE(std::holds_alternative<PluginLoadResponse>(dryRunResp));
        const auto& dryRun = std::get<PluginLoadResponse>(dryRunResp);
        CHECK_FALSE(dryRun.loaded);
        CHECK(dryRun.message == "dry-run");
        CHECK(dryRun.record.name == "dispatcher_external_plugin");

        auto* external = svc->getExternalPluginHost();
        REQUIRE(external != nullptr);
        REQUIRE(external->trustAdd(pluginDir).has_value());

        PluginLoadRequest loadReq;
        loadReq.pathOrName = pluginDir.string();
        auto loadResp = dispatchRequest(dispatcher, Request{loadReq});

        REQUIRE(std::holds_alternative<PluginLoadResponse>(loadResp));
        const auto& load = std::get<PluginLoadResponse>(loadResp);
        CHECK(load.loaded);
        CHECK(load.message == "loaded");
        CHECK(load.record.name == "dispatcher_external_plugin");
        CHECK(external->listLoaded().size() == 1);

        auto unloadResp =
            dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});

        REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
        CHECK(std::get<SuccessResponse>(unloadResp).message == "unloaded");
        CHECK(external->listLoaded().empty());
    }

    SECTION("plugin load routes direct external files before ABI fallback") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
        auto shimPath = makePythonShimPath();
        ScopedEnvVar pythonGuard("PATH", shimPath.c_str());

        auto pluginDir =
            createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_py");
        auto pluginFile = pluginDir / "plugin.py";

        auto* external = svc->getExternalPluginHost();
        REQUIRE(external != nullptr);
        REQUIRE(external->trustAdd(pluginFile).has_value());

        PluginLoadRequest req;
        req.pathOrName = pluginFile.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PluginLoadResponse>(resp));
        const auto& load = std::get<PluginLoadResponse>(resp);
        CHECK(load.loaded);
        CHECK(load.record.name == "dispatcher_external_plugin");
        CHECK(load.record.path == pluginFile.string());

        auto unloadResp =
            dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
        REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
    }

    SECTION("plugin load recognizes manifest-adjacent executable files") {
#ifdef _WIN32
        SKIP("manifest-adjacent executable file semantics are POSIX-specific");
#else
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

        auto pluginDir =
            createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_exec");
        auto executable = pluginDir / "plugin_runner";
        std::filesystem::copy_file(pluginDir / "plugin.py", executable,
                                   std::filesystem::copy_options::overwrite_existing);
        std::filesystem::permissions(executable,
                                     std::filesystem::perms::owner_read |
                                         std::filesystem::perms::owner_write |
                                         std::filesystem::perms::owner_exec,
                                     std::filesystem::perm_options::replace);

        auto* external = svc->getExternalPluginHost();
        REQUIRE(external != nullptr);
        REQUIRE(external->trustAdd(executable).has_value());

        PluginLoadRequest req;
        req.pathOrName = executable.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<PluginLoadResponse>(resp));
        const auto& load = std::get<PluginLoadResponse>(resp);
        CHECK(load.loaded);
        CHECK(load.record.name == "dispatcher_external_plugin");
        CHECK(load.record.path == executable.string());

        auto unloadResp =
            dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
        REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
#endif
    }

    SECTION("plugin load falls back after untrusted external file fails") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());

        auto pluginDir =
            createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_untrusted");
        auto pluginFile = pluginDir / "plugin.py";

        PluginLoadRequest req;
        req.pathOrName = pluginFile.string();

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message == "Plugin load failed for: " + pluginFile.string());
    }

    SECTION("plugin trust add loads regular files and reuses loaded path metadata") {
        auto [state, lifecycleFsm, svc] = makeReadyService();
        StubLifecycle lifecycle;
        RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
        auto shimPath = makePythonShimPath();
        ScopedEnvVar pythonGuard("PATH", shimPath.c_str());

        auto pluginDir =
            createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_trust_file");
        auto pluginFile = pluginDir / "plugin.py";

        auto trustAddResp =
            dispatchRequest(dispatcher, Request{PluginTrustAddRequest{pluginFile.string()}});
        REQUIRE(std::holds_alternative<SuccessResponse>(trustAddResp));

        auto* external = svc->getExternalPluginHost();
        REQUIRE(external != nullptr);
        REQUIRE(waitForCondition([&] { return external->listLoaded().size() == 1; }));
        CHECK(external->listLoaded().front().path == pluginFile);

        auto secondTrustResp =
            dispatchRequest(dispatcher, Request{PluginTrustAddRequest{pluginDir.string()}});
        REQUIRE(std::holds_alternative<SuccessResponse>(secondTrustResp));
        CHECK(waitForCondition([&] { return external->listLoaded().size() == 1; }));

        auto unloadResp =
            dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
        REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
    }
}

TEST_CASE("RequestDispatcher: collection handlers report missing dependencies and batch defaults",
          "[daemon][collections][dispatcher]") {
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_collections_dispatcher_");

    StubLifecycle lifecycle;
    StateComponent state;
    state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
    state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    ServiceManager svc(cfg, state, lifecycleFsm);
    RequestDispatcher dispatcher(&lifecycle, &svc, &state);

    struct SeededDocument {
        int64_t id{0};
        std::string hash;
        std::string relativePath;
    };

    struct ReadyCollectionService {
        std::shared_ptr<metadata::ConnectionPool> pool;
        std::shared_ptr<metadata::MetadataRepository> metadata;
        std::shared_ptr<StubContentStore> content;
        std::shared_ptr<ServiceManager> manager;
    };

    auto makeReadyService = [&](StateComponent& readyState, DaemonLifecycleFsm& readyLifecycleFsm,
                                const std::string& prefix) {
        DaemonConfig readyCfg = cfg;
        readyCfg.dataDir = makeTempDir(prefix);

        metadata::ConnectionPoolConfig poolConfig{};
        auto pool = std::make_shared<metadata::ConnectionPool>(
            makeTempMetadataDbPath(prefix + "db_").string(), poolConfig);
        REQUIRE(pool->initialize().has_value());

        ReadyCollectionService ready{
            .pool = pool,
            .metadata = makeReadyMetadataRepo(pool),
            .content = std::make_shared<StubContentStore>(),
            .manager = std::make_shared<ServiceManager>(readyCfg, readyState, readyLifecycleFsm),
        };
        ready.manager->__test_setMetadataRepo(ready.metadata);
        ready.manager->__test_setContentStore(ready.content);
        readyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
        readyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
        return ready;
    };

    auto seedDocument = [&](const ReadyCollectionService& ready, const std::string& relativePath,
                            const std::string& text, const std::string& collectionName,
                            const std::string& snapshotId) {
        const auto textBytes = std::as_bytes(std::span<const char>(text.data(), text.size()));
        auto storeRes = ready.content->storeBytes(textBytes, {});
        REQUIRE(storeRes.has_value());

        std::filesystem::path relativeDocPath(relativePath);
        std::string extension = relativeDocPath.extension().string();
        if (!extension.empty() && extension.front() == '.') {
            extension.erase(0, 1);
        }

        metadata::DocumentInfo doc{};
        doc.fileName = relativeDocPath.filename().string();
        doc.filePath = relativePath;
        doc.fileExtension = extension;
        doc.fileSize = static_cast<int64_t>(text.size());
        doc.sha256Hash = storeRes.value().contentHash;
        doc.mimeType = "text/plain";
        const auto now =
            std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());
        doc.createdTime = now;
        doc.modifiedTime = now;
        doc.indexedTime = now;

        auto idRes = ready.metadata->insertDocument(doc);
        REQUIRE(idRes.has_value());
        REQUIRE(
            ready.metadata
                ->setMetadata(idRes.value(), "collection", metadata::MetadataValue(collectionName))
                .has_value());
        REQUIRE(ready.metadata
                    ->setMetadata(idRes.value(), "snapshot_id", metadata::MetadataValue(snapshotId))
                    .has_value());

        return SeededDocument{idRes.value(), storeRes.value().contentHash, relativePath};
    };

    SECTION("list snapshots returns internal error when metadata repo is unavailable") {
        auto resp = dispatchRequest(dispatcher, Request{ListSnapshotsRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
    }

    SECTION("list snapshots forwards repository errors") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        repo->setSnapshotError(Error{ErrorCode::DatabaseError, "snapshot lookup failed"});
        svc.__test_setMetadataRepo(repo);

        auto resp = dispatchRequest(dispatcher, Request{ListSnapshotsRequest{}});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::DatabaseError);
        CHECK(err.message == "snapshot lookup failed");
    }

    SECTION("restore collection returns internal error before dependencies are initialized") {
        RestoreCollectionRequest req;
        req.collection = "dispatcher-collection";
        req.outputDirectory = "out";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
    }

    SECTION("restore collection returns internal error when content store is unavailable") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        svc.__test_setMetadataRepo(repo);

        RestoreCollectionRequest req;
        req.collection = "dispatcher-collection";
        req.outputDirectory = "out";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Content store unavailable") != std::string::npos);
    }

    SECTION("restore collection forwards repository query errors") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        repo->setQueryError(Error{ErrorCode::DatabaseError, "collection query failed"});
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());

        RestoreCollectionRequest req;
        req.collection = "dispatcher-collection";
        req.outputDirectory = "out";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::DatabaseError);
        CHECK(err.message == "Failed to find collection documents: collection query failed");
    }

    SECTION("restore snapshot returns internal error when metadata repo is unavailable") {
        RestoreSnapshotRequest req;
        req.snapshotId = "dispatcher-snapshot";
        req.outputDirectory = "out";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
    }

    SECTION("restore snapshot returns internal error when content store is unavailable") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        svc.__test_setMetadataRepo(repo);

        RestoreSnapshotRequest req;
        req.snapshotId = "dispatcher-snapshot";
        req.outputDirectory = "out";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Content store unavailable") != std::string::npos);
    }

    SECTION("restore snapshot forwards repository query errors") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        repo->setSnapshotDocumentsError(Error{ErrorCode::DatabaseError, "snapshot query failed"});
        svc.__test_setMetadataRepo(repo);
        svc.__test_setContentStore(std::make_shared<StubContentStore>());

        RestoreSnapshotRequest req;
        req.snapshotId = "dispatcher-snapshot";
        req.outputDirectory = "out";

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::DatabaseError);
        CHECK(err.message == "Failed to find snapshot documents: snapshot query failed");
    }

    SECTION("metadata value counts returns internal error when metadata repo is unavailable") {
        MetadataValueCountsRequest req;
        req.keys = {"collection"};

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::InternalError);
        CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
    }

    SECTION("metadata value counts forwards repository errors") {
        auto repo = std::make_shared<StubPruneMetadataRepository>();
        repo->setMetadataValueCountsError(Error{ErrorCode::DatabaseError, "counts failed"});
        svc.__test_setMetadataRepo(repo);

        MetadataValueCountsRequest req;
        req.keys = {"collection"};

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<ErrorResponse>(resp));
        const auto& err = std::get<ErrorResponse>(resp);
        CHECK(err.code == ErrorCode::DatabaseError);
        CHECK(err.message == "counts failed");
    }

    SECTION("batch request returns not implemented items for each sequence") {
        BatchRequest req;
        req.items.push_back(BatchItem{.sequenceId = 7, .request = SearchRequest{.query = "one"}});
        req.items.push_back(BatchItem{.sequenceId = 8, .request = ListRequest{}});

        auto resp = dispatchRequest(dispatcher, Request{req});

        REQUIRE(std::holds_alternative<BatchResponse>(resp));
        const auto& batchResp = std::get<BatchResponse>(resp);
        CHECK(batchResp.totalCount == 2);
        CHECK(batchResp.successCount == 0);
        CHECK(batchResp.errorCount == 2);
        REQUIRE(batchResp.items.size() == 2);
        CHECK(batchResp.items[0].sequenceId == 7);
        CHECK_FALSE(batchResp.items[0].success);
        REQUIRE(std::holds_alternative<ErrorResponse>(batchResp.items[0].response));
        CHECK(std::get<ErrorResponse>(batchResp.items[0].response).code ==
              ErrorCode::NotImplemented);
        CHECK(batchResp.items[1].sequenceId == 8);
        CHECK_FALSE(batchResp.items[1].success);
    }

    SECTION("restore collection exercises regex, layout, and io branches") {
        StubLifecycle readyLifecycle;
        StateComponent readyState;
        DaemonLifecycleFsm readyLifecycleFsm;
        auto ready =
            makeReadyService(readyState, readyLifecycleFsm, "yams_collections_live_dispatcher_");
        RequestDispatcher readyDispatcher(&readyLifecycle, ready.manager.get(), &readyState);

        const std::string collectionName = "dispatcher-live-collection";
        const std::string snapshotId = "dispatcher-live-snapshot";
        const auto seeded = seedDocument(ready, "nested/alpha.txt", "alpha collection text\n",
                                         collectionName, snapshotId);

        auto includeOut = makeTempDir("yams_collection_include_");
        RestoreCollectionRequest includeReq;
        includeReq.collection = collectionName;
        includeReq.outputDirectory = includeOut.string();
        includeReq.includePatterns = {"*alpha.txt"};
        includeReq.dryRun = true;

        auto includeResp = dispatchRequest(readyDispatcher, Request{includeReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(includeResp));
        const auto& included = std::get<RestoreCollectionResponse>(includeResp);
        REQUIRE(included.files.size() == 1);
        CHECK(included.filesRestored == 1);
        CHECK(included.files.front().path == (includeOut / seeded.relativePath).string());

        RestoreCollectionRequest invalidPatternReq;
        invalidPatternReq.collection = collectionName;
        invalidPatternReq.outputDirectory = includeOut.string();
        invalidPatternReq.includePatterns = {"["};
        invalidPatternReq.dryRun = true;

        auto invalidPatternResp = dispatchRequest(readyDispatcher, Request{invalidPatternReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(invalidPatternResp));
        const auto& invalidPattern = std::get<RestoreCollectionResponse>(invalidPatternResp);
        CHECK(invalidPattern.filesRestored == 0);
        CHECK(invalidPattern.files.empty());

        RestoreCollectionRequest excludeReq;
        excludeReq.collection = collectionName;
        excludeReq.outputDirectory = includeOut.string();
        excludeReq.excludePatterns = {"*alpha.txt"};
        excludeReq.dryRun = true;

        auto excludeResp = dispatchRequest(readyDispatcher, Request{excludeReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(excludeResp));
        const auto& excluded = std::get<RestoreCollectionResponse>(excludeResp);
        CHECK(excluded.filesRestored == 0);
        CHECK(excluded.files.empty());

        auto writeOut = makeTempDir("yams_collection_write_");
        RestoreCollectionRequest writeReq;
        writeReq.collection = collectionName;
        writeReq.outputDirectory = writeOut.string();
        writeReq.layoutTemplate = "deep/{hash}{ext}";
        writeReq.overwrite = true;
        writeReq.dryRun = false;

        auto writeResp = dispatchRequest(readyDispatcher, Request{writeReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(writeResp));
        const auto& written = std::get<RestoreCollectionResponse>(writeResp);
        REQUIRE(written.files.size() == 1);
        CHECK(written.filesRestored == 1);
        const auto writtenPath = writeOut / "deep" / (seeded.hash + ".txt");
        CHECK(std::filesystem::path(written.files.front().path).lexically_normal() ==
              writtenPath.lexically_normal());
        REQUIRE(std::filesystem::exists(writtenPath));
        std::ifstream writtenFile(writtenPath, std::ios::binary);
        REQUIRE(writtenFile.good());
        std::string writtenText((std::istreambuf_iterator<char>(writtenFile)),
                                std::istreambuf_iterator<char>());
        CHECK(writtenText == "alpha collection text\n");

        auto createDirBlocker = makeTempDir("yams_collection_create_dir_blocker_") / "blocked";
        {
            std::ofstream blocker(createDirBlocker);
            REQUIRE(blocker.good());
            blocker << "blocker";
        }
        RestoreCollectionRequest createDirFailReq;
        createDirFailReq.collection = collectionName;
        createDirFailReq.outputDirectory = createDirBlocker.string();
        createDirFailReq.createDirs = true;
        createDirFailReq.overwrite = true;
        createDirFailReq.dryRun = false;

        auto createDirFailResp = dispatchRequest(readyDispatcher, Request{createDirFailReq});

        REQUIRE(std::holds_alternative<ErrorResponse>(createDirFailResp));
        CHECK(std::get<ErrorResponse>(createDirFailResp).code == ErrorCode::IOError);
        CHECK(std::get<ErrorResponse>(createDirFailResp)
                  .message.find("Failed to create output directory") != std::string::npos);

        auto parentFailOut = makeTempDir("yams_collection_parent_fail_");
        auto occupiedParent = parentFailOut / "occupied";
        {
            std::ofstream blocker(occupiedParent);
            REQUIRE(blocker.good());
            blocker << "occupied";
        }
        RestoreCollectionRequest parentFailReq;
        parentFailReq.collection = collectionName;
        parentFailReq.outputDirectory = parentFailOut.string();
        parentFailReq.layoutTemplate = "occupied/restored.txt";
        parentFailReq.overwrite = true;
        parentFailReq.dryRun = false;

        auto parentFailResp = dispatchRequest(readyDispatcher, Request{parentFailReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(parentFailResp));
        const auto& parentFailed = std::get<RestoreCollectionResponse>(parentFailResp);
        REQUIRE(parentFailed.files.size() == 1);
        CHECK(parentFailed.filesRestored == 0);
        CHECK(parentFailed.files.front().skipped);
        CHECK(parentFailed.files.front().skipReason.find("Failed to create parent directory") !=
              std::string::npos);

        auto openFailOut = makeTempDir("yams_collection_open_fail_");
        RestoreCollectionRequest openFailReq;
        openFailReq.collection = collectionName;
        openFailReq.outputDirectory = openFailOut.string();
        openFailReq.layoutTemplate = ".";
        openFailReq.overwrite = true;
        openFailReq.dryRun = false;

        auto openFailResp = dispatchRequest(readyDispatcher, Request{openFailReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(openFailResp));
        const auto& openFailed = std::get<RestoreCollectionResponse>(openFailResp);
        REQUIRE(openFailed.files.size() == 1);
        CHECK(openFailed.filesRestored == 0);
        CHECK(openFailed.files.front().skipped);
        CHECK(openFailed.files.front().skipReason == "Failed to open output file");

        auto writeFailOut = makeTempDir("yams_collection_write_fail_");
        RestoreCollectionRequest writeFailReq;
        writeFailReq.collection = collectionName;
        writeFailReq.outputDirectory = writeFailOut.string();
        writeFailReq.layoutTemplate = "write-fail/{name}{ext}";
        writeFailReq.overwrite = true;
        writeFailReq.dryRun = false;

        RequestDispatcher::__test_forceRestoreWriteFailureOnce();
        auto writeFailResp = dispatchRequest(readyDispatcher, Request{writeFailReq});

        REQUIRE(std::holds_alternative<RestoreCollectionResponse>(writeFailResp));
        const auto& writeFailed = std::get<RestoreCollectionResponse>(writeFailResp);
        REQUIRE(writeFailed.files.size() == 1);
        CHECK(writeFailed.filesRestored == 0);
        CHECK(writeFailed.files.front().skipped);
        CHECK(writeFailed.files.front().skipReason == "Failed to write file content");
    }

    SECTION("restore snapshot exercises regex, layout, and io branches") {
        StubLifecycle readyLifecycle;
        StateComponent readyState;
        DaemonLifecycleFsm readyLifecycleFsm;
        auto ready =
            makeReadyService(readyState, readyLifecycleFsm, "yams_snapshots_live_dispatcher_");
        RequestDispatcher readyDispatcher(&readyLifecycle, ready.manager.get(), &readyState);

        const std::string collectionName = "dispatcher-live-collection";
        const std::string snapshotId = "dispatcher-live-snapshot";
        const auto seeded = seedDocument(ready, "nested/alpha.txt", "alpha snapshot text\n",
                                         collectionName, snapshotId);

        auto defaultOut = makeTempDir("yams_snapshot_default_");
        RestoreSnapshotRequest defaultReq;
        defaultReq.snapshotId = snapshotId;
        defaultReq.outputDirectory = defaultOut.string();
        defaultReq.includePatterns = {"*alpha.txt"};
        defaultReq.dryRun = true;

        auto defaultResp = dispatchRequest(readyDispatcher, Request{defaultReq});

        REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(defaultResp));
        const auto& defaultRestore = std::get<RestoreSnapshotResponse>(defaultResp);
        REQUIRE(defaultRestore.files.size() == 1);
        CHECK(defaultRestore.filesRestored == 1);
        CHECK(defaultRestore.files.front().path == (defaultOut / seeded.relativePath).string());

        RestoreSnapshotRequest invalidPatternReq;
        invalidPatternReq.snapshotId = snapshotId;
        invalidPatternReq.outputDirectory = defaultOut.string();
        invalidPatternReq.includePatterns = {"["};
        invalidPatternReq.dryRun = true;

        auto invalidPatternResp = dispatchRequest(readyDispatcher, Request{invalidPatternReq});

        REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(invalidPatternResp));
        const auto& invalidPattern = std::get<RestoreSnapshotResponse>(invalidPatternResp);
        CHECK(invalidPattern.filesRestored == 0);
        CHECK(invalidPattern.files.empty());

        auto writeOut = makeTempDir("yams_snapshot_write_");
        RestoreSnapshotRequest writeReq;
        writeReq.snapshotId = snapshotId;
        writeReq.outputDirectory = writeOut.string();
        writeReq.layoutTemplate = "deep/{name}{ext}";
        writeReq.overwrite = true;
        writeReq.dryRun = false;

        auto writeResp = dispatchRequest(readyDispatcher, Request{writeReq});

        REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(writeResp));
        const auto& written = std::get<RestoreSnapshotResponse>(writeResp);
        REQUIRE(written.files.size() == 1);
        CHECK(written.filesRestored == 1);
        const auto writtenPath = writeOut / "deep" / "alpha.txt";
        CHECK(std::filesystem::path(written.files.front().path).lexically_normal() ==
              writtenPath.lexically_normal());
        REQUIRE(std::filesystem::exists(writtenPath));
        std::ifstream writtenFile(writtenPath, std::ios::binary);
        REQUIRE(writtenFile.good());
        std::string writtenText((std::istreambuf_iterator<char>(writtenFile)),
                                std::istreambuf_iterator<char>());
        CHECK(writtenText == "alpha snapshot text\n");

        auto createDirBlocker = makeTempDir("yams_snapshot_create_dir_blocker_") / "blocked";
        {
            std::ofstream blocker(createDirBlocker);
            REQUIRE(blocker.good());
            blocker << "blocker";
        }
        RestoreSnapshotRequest createDirFailReq;
        createDirFailReq.snapshotId = snapshotId;
        createDirFailReq.outputDirectory = createDirBlocker.string();
        createDirFailReq.createDirs = true;
        createDirFailReq.overwrite = true;
        createDirFailReq.dryRun = false;

        auto createDirFailResp = dispatchRequest(readyDispatcher, Request{createDirFailReq});

        REQUIRE(std::holds_alternative<ErrorResponse>(createDirFailResp));
        CHECK(std::get<ErrorResponse>(createDirFailResp).code == ErrorCode::IOError);
        CHECK(std::get<ErrorResponse>(createDirFailResp)
                  .message.find("Failed to create output directory") != std::string::npos);

        auto parentFailOut = makeTempDir("yams_snapshot_parent_fail_");
        auto occupiedParent = parentFailOut / "occupied";
        {
            std::ofstream blocker(occupiedParent);
            REQUIRE(blocker.good());
            blocker << "occupied";
        }
        RestoreSnapshotRequest parentFailReq;
        parentFailReq.snapshotId = snapshotId;
        parentFailReq.outputDirectory = parentFailOut.string();
        parentFailReq.layoutTemplate = "occupied/restored.txt";
        parentFailReq.overwrite = true;
        parentFailReq.dryRun = false;

        auto parentFailResp = dispatchRequest(readyDispatcher, Request{parentFailReq});

        REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(parentFailResp));
        const auto& parentFailed = std::get<RestoreSnapshotResponse>(parentFailResp);
        REQUIRE(parentFailed.files.size() == 1);
        CHECK(parentFailed.filesRestored == 0);
        CHECK(parentFailed.files.front().skipped);
        CHECK(parentFailed.files.front().skipReason.find("Failed to create parent directory") !=
              std::string::npos);

        auto openFailOut = makeTempDir("yams_snapshot_open_fail_");
        RestoreSnapshotRequest openFailReq;
        openFailReq.snapshotId = snapshotId;
        openFailReq.outputDirectory = openFailOut.string();
        openFailReq.layoutTemplate = ".";
        openFailReq.overwrite = true;
        openFailReq.dryRun = false;

        auto openFailResp = dispatchRequest(readyDispatcher, Request{openFailReq});

        REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(openFailResp));
        const auto& openFailed = std::get<RestoreSnapshotResponse>(openFailResp);
        REQUIRE(openFailed.files.size() == 1);
        CHECK(openFailed.filesRestored == 0);
        CHECK(openFailed.files.front().skipped);
        CHECK(openFailed.files.front().skipReason == "Failed to open output file");

        auto writeFailOut = makeTempDir("yams_snapshot_write_fail_");
        RestoreSnapshotRequest writeFailReq;
        writeFailReq.snapshotId = snapshotId;
        writeFailReq.outputDirectory = writeFailOut.string();
        writeFailReq.layoutTemplate = "write-fail/{hash}{ext}";
        writeFailReq.overwrite = true;
        writeFailReq.dryRun = false;

        RequestDispatcher::__test_forceRestoreWriteFailureOnce();
        auto writeFailResp = dispatchRequest(readyDispatcher, Request{writeFailReq});

        REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(writeFailResp));
        const auto& writeFailed = std::get<RestoreSnapshotResponse>(writeFailResp);
        REQUIRE(writeFailed.files.size() == 1);
        CHECK(writeFailed.filesRestored == 0);
        CHECK(writeFailed.files.front().skipped);
        CHECK(writeFailed.files.front().skipReason == "Failed to write file content");
    }
}

TEST_CASE("DaemonMetrics: snapshot includes canonical readiness flags",
          "[daemon][metrics][readiness]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_metrics_readiness_flags_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());

    auto snap = metrics.getSnapshot();
    REQUIRE(snap != nullptr);

    const std::array<std::string_view, 12> requiredCoreReadinessKeys = {
        readiness::kIpcServer,     readiness::kContentStore,
        readiness::kDatabase,      readiness::kMetadataRepo,
        readiness::kSearchEngine,  readiness::kModelProvider,
        readiness::kVectorIndex,   readiness::kVectorDb,
        readiness::kPlugins,       readiness::kVectorDbInitAttempted,
        readiness::kVectorDbReady, readiness::kVectorDbDim};

    for (const auto key : requiredCoreReadinessKeys) {
        INFO("missing readiness key: " << key);
        REQUIRE(snap->readinessStates.count(std::string(key)) > 0);
    }
}

TEST_CASE("DaemonMetrics: snapshot reports generated build version", "[daemon][metrics][version]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_metrics_version_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());

    auto snap = metrics.getSnapshot();
    REQUIRE(snap != nullptr);
    CHECK(snap->version == yams::version::string_v);
}

TEST_CASE("DaemonMetrics: cold snapshot refreshes synchronously", "[daemon][metrics][freshness]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_metrics_freshness_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());

    auto snap = metrics.getSnapshot();
    REQUIRE(snap != nullptr);
    CHECK_FALSE(snap->statusSnapshotStale);
    CHECK(snap->statusSnapshotAgeMs == 0);
    CHECK(snap->statusDetailAgeMs == 0);
    CHECK(snap->statusDetailStale);
    CHECK_FALSE(snap->statusDetailAvailable);
}

TEST_CASE("RequestDispatcher: status refreshes cold snapshot without retry hint",
          "[daemon][status][freshness]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_busy_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
    RequestDispatcher dispatcher(nullptr, &svc, &state, &metrics);

    auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
    REQUIRE(std::holds_alternative<StatusResponse>(resp));
    const auto& status = std::get<StatusResponse>(resp);
    CHECK(status.running);
    CHECK(status.retryAfterMs == 0);
}

TEST_CASE("RequestDispatcher: cold status still reports active repair work",
          "[daemon][status][freshness]") {
    StateComponent state;
    state.stats.repairRunning.store(true, std::memory_order_relaxed);
    state.stats.repairQueueDepth.store(7, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_busy_repair_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
    RequestDispatcher dispatcher(nullptr, &svc, &state, &metrics);

    auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
    REQUIRE(std::holds_alternative<StatusResponse>(resp));
    const auto& status = std::get<StatusResponse>(resp);
    CHECK(status.retryAfterMs == 0);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairRunning)) == 1);
    REQUIRE(status.requestCounts.count(std::string(metrics::kRepairQueueDepth)) == 1);
    CHECK(status.requestCounts.at(std::string(metrics::kRepairRunning)) == 1);
    CHECK(status.requestCounts.at(std::string(metrics::kRepairQueueDepth)) == 7);
}

TEST_CASE("RequestDispatcher: status mirrors memory breakdown into requestCounts",
          "[daemon][status][memory]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_status_mem_breakdown_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
    RequestDispatcher dispatcher(nullptr, &svc, &state, &metrics);

    auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
    REQUIRE(std::holds_alternative<StatusResponse>(resp));
    const auto& status = std::get<StatusResponse>(resp);

    REQUIRE(status.requestCounts.count("status_mem_rss_bytes") == 1);
    CHECK(status.requestCounts.at("status_mem_rss_bytes") > 0);
#if defined(__APPLE__)
    REQUIRE(status.requestCounts.count("status_mem_phys_footprint_bytes") == 1);
    const auto footprintBytes = status.requestCounts.at("status_mem_phys_footprint_bytes");
    CHECK(footprintBytes > 0);
    const auto expectedMb = static_cast<double>(footprintBytes) / (1024.0 * 1024.0);
    CHECK(std::abs(status.memoryUsageMb - expectedMb) < std::max(1.0, expectedMb * 0.05));
#endif
}

TEST_CASE("DaemonMetrics: repair idle and busy counters export from state",
          "[daemon][metrics][repair]") {
    StateComponent state;
    state.stats.repairIdleTicks.store(3, std::memory_order_relaxed);
    state.stats.repairBusyTicks.store(9, std::memory_order_relaxed);
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_repair_counter_export_");
    ServiceManager svc(cfg, state, lifecycleFsm);
    DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());

    auto snap = metrics.getSnapshot();
    REQUIRE(snap != nullptr);
    CHECK(snap->repairIdleTicks == 3);
    CHECK(snap->repairBusyTicks == 9);
}

TEST_CASE("StatusResponse: post_ingest_rpc requestCounts keys round-trip",
          "[daemon][status][protocol][post_ingest]") {
    StatusResponse s{};
    s.requestCounts["post_ingest_rpc_queued"] = 3;
    s.requestCounts["post_ingest_rpc_capacity"] = 64;
    s.requestCounts["post_ingest_rpc_max_per_batch"] = 8;

    Message m{};
    m.payload = Response{std::in_place_type<StatusResponse>, s};

    auto enc = ProtoSerializer::encode_payload(m);
    REQUIRE(enc.has_value());

    auto dec = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(dec.has_value());

    const auto& resp = std::get<Response>(dec.value().payload);
    REQUIRE(std::holds_alternative<StatusResponse>(resp));

    const auto& decoded = std::get<StatusResponse>(resp);
    REQUIRE(decoded.requestCounts.at("post_ingest_rpc_queued") == 3);
    REQUIRE(decoded.requestCounts.at("post_ingest_rpc_capacity") == 64);
    REQUIRE(decoded.requestCounts.at("post_ingest_rpc_max_per_batch") == 8);
}

TEST_CASE("StatusResponse: maintenance phase fields round-trip",
          "[daemon][status][protocol][maintenance]") {
    StatusResponse s{};
    s.databasePhase = "ready";
    s.databasePhaseElapsedMs = 11;
    s.maintenancePhase = "salvaging";
    s.maintenancePhaseElapsedMs = 222;

    Message m{};
    m.payload = Response{std::in_place_type<StatusResponse>, s};

    auto enc = ProtoSerializer::encode_payload(m);
    REQUIRE(enc.has_value());

    auto dec = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(dec.has_value());

    const auto& resp = std::get<Response>(dec.value().payload);
    REQUIRE(std::holds_alternative<StatusResponse>(resp));

    const auto& decoded = std::get<StatusResponse>(resp);
    CHECK(decoded.databasePhase == "ready");
    CHECK(decoded.databasePhaseElapsedMs == 11);
    CHECK(decoded.maintenancePhase == "salvaging");
    CHECK(decoded.maintenancePhaseElapsedMs == 222);
}

TEST_CASE("StatusResponse: freshness requestCounts keys round-trip",
          "[daemon][status][protocol][freshness]") {
    StatusResponse s{};
    s.requestCounts["status_snapshot_age_ms"] = 42;
    s.requestCounts["status_snapshot_stale"] = 0;
    s.requestCounts["status_detail_age_ms"] = 17;
    s.requestCounts["status_detail_stale"] = 1;
    s.requestCounts["status_detail_available"] = 1;

    Message m{};
    m.payload = Response{std::in_place_type<StatusResponse>, s};

    auto enc = ProtoSerializer::encode_payload(m);
    REQUIRE(enc.has_value());

    auto dec = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(dec.has_value());

    const auto& resp = std::get<Response>(dec.value().payload);
    REQUIRE(std::holds_alternative<StatusResponse>(resp));

    const auto& decoded = std::get<StatusResponse>(resp);
    REQUIRE(decoded.requestCounts.at("status_snapshot_age_ms") == 42);
    REQUIRE(decoded.requestCounts.at("status_snapshot_stale") == 0);
    REQUIRE(decoded.requestCounts.at("status_detail_age_ms") == 17);
    REQUIRE(decoded.requestCounts.at("status_detail_stale") == 1);
    REQUIRE(decoded.requestCounts.at("status_detail_available") == 1);
}

TEST_CASE("StatusResponse: backpressure requestCounts keys round-trip",
          "[daemon][status][protocol][post_ingest]") {
    StatusResponse s{};
    s.requestCounts["post_ingest_backpressure_rejects"] = 123;
    s.requestCounts["kg_jobs_depth"] = 3880;
    s.requestCounts["kg_jobs_capacity"] = 4096;
    s.requestCounts["kg_jobs_fill_pct"] = 95;

    Message m{};
    m.payload = Response{std::in_place_type<StatusResponse>, s};

    auto enc = ProtoSerializer::encode_payload(m);
    REQUIRE(enc.has_value());

    auto dec = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(dec.has_value());

    const auto& resp = std::get<Response>(dec.value().payload);
    REQUIRE(std::holds_alternative<StatusResponse>(resp));

    const auto& decoded = std::get<StatusResponse>(resp);
    REQUIRE(decoded.requestCounts.at("post_ingest_backpressure_rejects") == 123);
    REQUIRE(decoded.requestCounts.at("kg_jobs_depth") == 3880);
    REQUIRE(decoded.requestCounts.at("kg_jobs_capacity") == 4096);
    REQUIRE(decoded.requestCounts.at("kg_jobs_fill_pct") == 95);
}

TEST_CASE("StatusResponse: memory breakdown requestCounts keys round-trip",
          "[daemon][status][protocol][memory]") {
    StatusResponse s{};
    s.requestCounts["status_mem_rss_bytes"] = 123456;
    s.requestCounts["status_mem_phys_footprint_bytes"] = 654321;
    s.requestCounts["status_mem_malloc_allocated_bytes"] = 777777;

    Message m{};
    m.payload = Response{std::in_place_type<StatusResponse>, s};

    auto enc = ProtoSerializer::encode_payload(m);
    REQUIRE(enc.has_value());

    auto dec = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(dec.has_value());

    const auto& resp = std::get<Response>(dec.value().payload);
    REQUIRE(std::holds_alternative<StatusResponse>(resp));

    const auto& decoded = std::get<StatusResponse>(resp);
    REQUIRE(decoded.requestCounts.at("status_mem_rss_bytes") == 123456);
    REQUIRE(decoded.requestCounts.at("status_mem_phys_footprint_bytes") == 654321);
    REQUIRE(decoded.requestCounts.at("status_mem_malloc_allocated_bytes") == 777777);
}

TEST_CASE("AbiPluginHost: failed plugin loads are retained in last scan skips",
          "[daemon][plugin][status][load-failure]") {
#if defined(__APPLE__)
    const std::filesystem::path pluginPath{"/opt/homebrew/lib/yams/plugins/yams_glint.dylib"};
    if (!std::filesystem::exists(pluginPath)) {
        SKIP("Installed yams_glint plugin not present");
    }

    AbiPluginHost host(nullptr);
    const auto trustFile = makeTempDir("yams_glint_skip_status_") / "plugins.trust";
    host.setTrustFile(trustFile);
    REQUIRE(host.trustAdd(pluginPath.parent_path()));

    const auto loadResult = host.load(pluginPath, "{}");
    if (loadResult) {
        SUCCEED("yams_glint loaded successfully on this machine");
        REQUIRE(host.unload(loadResult.value().name));
        return;
    }

    const auto skips = host.getLastScanSkips();
    REQUIRE_FALSE(skips.empty());
    const auto it = std::find_if(skips.begin(), skips.end(),
                                 [&](const auto& entry) { return entry.first == pluginPath; });
    REQUIRE(it != skips.end());
    CHECK((it->second.find("dlopen failed") != std::string::npos ||
           it->second.find("preflight failed") != std::string::npos ||
           it->second.find("Plugin init failed") != std::string::npos));
#else
    SUCCEED("macOS-specific plugin load retention test");
#endif
}

// =============================================================================
// StatusResponse Protocol Serialization Tests
// =============================================================================

TEST_CASE("StatusResponse: Protocol serialization", "[daemon][status][protocol]") {
    SECTION("Round-trip with lifecycle state and error") {
        StatusResponse s{};
        s.running = true;
        s.ready = false;
        s.uptimeSeconds = 42;
        s.requestsProcessed = 7;
        s.activeConnections = 1;
        s.memoryUsageMb = 12.5;
        s.cpuUsagePercent = 3.2;
        s.version = "test";
        s.overallStatus = "failed";
        s.lifecycleState = "starting";
        s.lastError = "boom";
        s.dataDir = "/tmp/daemon-data-dir";
        s.metadataDbPath = "/tmp/daemon-data-dir/yams.db";
        s.vectorDbPath = "/tmp/daemon-data-dir/vectors.db";
        s.requestCounts["worker_threads"] = 4;

        // FSM-exported fields
        s.requestCounts["service_fsm_state"] = 3;
        s.requestCounts["embedding_state"] = 2;
        s.requestCounts["plugin_host_state"] = 1;

        // Readiness states
        s.readinessStates["embedding_ready"] = false;
        s.readinessStates["plugins_ready"] = true;
        s.readinessStates["plugins_degraded"] = true;

        Message m{};
        m.payload = Response{std::in_place_type<StatusResponse>, s};

        // Encode
        auto enc = ProtoSerializer::encode_payload(m);
        REQUIRE(enc.has_value());

        // Decode
        auto dec = ProtoSerializer::decode_payload(enc.value());
        REQUIRE(dec.has_value());

        const Message& out = dec.value();
        REQUIRE(std::holds_alternative<Response>(out.payload));

        const auto& resp = std::get<Response>(out.payload);
        REQUIRE(std::holds_alternative<StatusResponse>(resp));

        const auto& decoded = std::get<StatusResponse>(resp);

        // Verify key fields survived round-trip
        REQUIRE(decoded.running == true);
        REQUIRE(decoded.ready == false);
        REQUIRE(decoded.uptimeSeconds == 42);
        REQUIRE(decoded.version == "test");
        REQUIRE(decoded.overallStatus == "failed");
        REQUIRE(decoded.lastError == "boom");
        REQUIRE(decoded.dataDir == "/tmp/daemon-data-dir");
        REQUIRE(decoded.metadataDbPath == "/tmp/daemon-data-dir/yams.db");
        REQUIRE(decoded.vectorDbPath == "/tmp/daemon-data-dir/vectors.db");

        // Verify FSM state counts
        REQUIRE(decoded.requestCounts.at("worker_threads") == 4);
        REQUIRE(decoded.requestCounts.at("service_fsm_state") == 3);

        // Verify readiness flags
        REQUIRE(decoded.readinessStates.at("embedding_ready") == false);
        REQUIRE(decoded.readinessStates.at("plugins_ready") == true);
        REQUIRE(decoded.readinessStates.at("plugins_degraded") == true);
    }

    SECTION("Empty status response serializes correctly") {
        StatusResponse s{};
        Message m{};
        m.payload = Response{std::in_place_type<StatusResponse>, s};

        auto enc = ProtoSerializer::encode_payload(m);
        REQUIRE(enc.has_value());

        auto dec = ProtoSerializer::decode_payload(enc.value());
        REQUIRE(dec.has_value());

        const auto& resp = std::get<Response>(dec.value().payload);
        REQUIRE(std::holds_alternative<StatusResponse>(resp));
    }
}

// =============================================================================
// Plugin Degradation Tests
// =============================================================================
// Note: Full integration tests (DISABLED in original) are skipped here
// These would test daemon start with trust failures and recovery hooks
// They belong in integration tests due to full daemon lifecycle requirements

TEST_CASE("ServiceManager: Plugin degradation tracking", "[daemon][status][plugins]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_plugin_degrade_");

    SECTION("Plugin degradation flag is tracked in FSM") {
        // This is a unit test of the FSM tracking mechanism
        // Full daemon integration tests are separate

        ServiceManager svc(cfg, state, lifecycleFsm);

        // ServiceManager should track plugin states via FSM
        // Actual degradation testing requires daemon start, which is integration level

        // Basic verification: ServiceManager exists and has FSM snapshot capability
        auto snapshot = svc.getServiceManagerFsmSnapshot();
        // ServiceManager starts uninitialized until explicitly initialized
        REQUIRE(snapshot.state == ServiceManagerState::Uninitialized);
    }
}

// =============================================================================
// FSM State Metrics
// =============================================================================

TEST_CASE("DaemonMetrics: FSM state export", "[daemon][metrics][fsm]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_fsm_metrics_");
    ServiceManager svc(cfg, state, lifecycleFsm);

    SECTION("Lifecycle FSM state is tracked") {
        lifecycleFsm.dispatch(BootstrappedEvent{});

        auto snapshot = lifecycleFsm.snapshot();
        REQUIRE(snapshot.state != LifecycleState::Unknown);
    }

    SECTION("Service manager FSM state is accessible") {
        auto snapshot = svc.getServiceManagerFsmSnapshot();
        // ServiceManager starts in Uninitialized state until initialized
        REQUIRE(snapshot.state == ServiceManagerState::Uninitialized);
    }

    SECTION("Embedding provider FSM state is accessible") {
        auto snapshot = svc.getEmbeddingProviderFsmSnapshot();
        // Embedding provider starts in Unavailable state until plugin loaded
        REQUIRE(snapshot.state == EmbeddingProviderState::Unavailable);
    }

    SECTION("Plugin host FSM state is accessible") {
        auto snapshot = svc.getPluginHostFsmSnapshot();
        // When ABI plugins are disabled, PluginManager initializes successfully with
        // an empty plugin list, transitioning to Ready state. When plugins are enabled
        // but PluginManager hasn't been created yet, the fallback is NotInitialized.
        REQUIRE((snapshot.state == PluginHostState::NotInitialized ||
                 snapshot.state == PluginHostState::Ready));
    }
}

// =============================================================================
// DatabaseManager Metrics Tests
// =============================================================================

TEST_CASE("DatabaseManager: Metrics tracking", "[daemon][metrics][database]") {
    StateComponent state;
    DatabaseManager::Dependencies deps{.state = &state};
    DatabaseManager dbm(deps);

    SECTION("Stats start at zero") {
        const auto& stats = dbm.getStats();
        REQUIRE(stats.openDurationMs.load() == 0);
        REQUIRE(stats.migrationDurationMs.load() == 0);
        REQUIRE(stats.openErrors.load() == 0);
        REQUIRE(stats.migrationErrors.load() == 0);
        REQUIRE(stats.repositoryInitErrors.load() == 0);
    }

    SECTION("Stats are returned by reference") {
        // Verify we can access stats multiple times
        const auto& stats1 = dbm.getStats();
        const auto& stats2 = dbm.getStats();
        REQUIRE(stats1.openErrors.load() == stats2.openErrors.load());
    }
}

TEST_CASE("DaemonMetrics: DatabaseManager metrics export", "[daemon][metrics][database]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_db_metrics_");
    ServiceManager svc(cfg, state, lifecycleFsm);

    SECTION("DatabaseManager metrics are included in snapshot") {
        DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
        auto snap = metrics.getSnapshot();

        REQUIRE(snap != nullptr);
        // DatabaseManager metrics should be present (may be 0 if not initialized)
        REQUIRE(snap->dbOpenErrors == 0); // Default value when no errors
    }
}

// =============================================================================
// WorkCoordinator Metrics Tests
// =============================================================================

TEST_CASE("WorkCoordinator: Metrics tracking", "[daemon][metrics][work]") {
    WorkCoordinator wc;

    SECTION("Stats before start") {
        auto stats = wc.getStats();
        REQUIRE(stats.workerCount == 0);
        REQUIRE(stats.activeWorkers == 0);
        REQUIRE(stats.isRunning == false);
    }

    SECTION("Stats accessors don't crash") {
        REQUIRE(wc.getWorkerCount() == 0);
        REQUIRE(wc.getActiveWorkerCount() == 0);
        REQUIRE(wc.isRunning() == false);
    }
}

TEST_CASE("DaemonMetrics: WorkCoordinator metrics export", "[daemon][metrics][work]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_wc_metrics_");
    ServiceManager svc(cfg, state, lifecycleFsm);

    SECTION("WorkCoordinator metrics are included in snapshot") {
        DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
        auto snap = metrics.getSnapshot();

        REQUIRE(snap != nullptr);
        // WorkCoordinator metrics should be present
        // ServiceManager creates WorkCoordinator with hardware_concurrency() workers
        REQUIRE(snap->workerThreads > 0);              // Should have workers (worker_threads)
        REQUIRE(snap->workCoordinatorRunning == true); // Should be running
    }
}

// =============================================================================
// Search Load Metrics Tests
// =============================================================================

TEST_CASE("ServiceManager: Search load metrics", "[daemon][metrics][search]") {
    StateComponent state;
    DaemonLifecycleFsm lifecycleFsm;
    DaemonConfig cfg;
    cfg.dataDir = makeTempDir("yams_search_metrics_");
    ServiceManager svc(cfg, state, lifecycleFsm);

    SECTION("Search load metrics return valid structure") {
        auto metrics = svc.getSearchLoadMetrics();
        REQUIRE(metrics.active == 0);
        REQUIRE(metrics.queued == 0);
        REQUIRE(metrics.concurrencyLimit > 0);
    }

    SECTION("Search lifecycle counters update load metrics") {
        svc.onSearchRequestQueued();
        auto queued = svc.getSearchLoadMetrics();
        REQUIRE(queued.queued == 1);
        REQUIRE(queued.active == 0);

        const auto cap = queued.concurrencyLimit;
        REQUIRE(svc.tryStartSearchRequest(cap));

        auto started = svc.getSearchLoadMetrics();
        REQUIRE(started.queued == 0);
        REQUIRE(started.active == 1);

        svc.onSearchRequestFinished();
        auto finished = svc.getSearchLoadMetrics();
        REQUIRE(finished.active == 0);
        REQUIRE(finished.queued == 0);
    }

    SECTION("Failed search start keeps queued count until rejection") {
        // Pin cap to 1 so the second start is guaranteed to fail regardless of the
        // machine-derived concurrencyLimit (varies with CPU cores).
        constexpr std::uint32_t cap = 1;
        svc.onSearchRequestQueued();
        REQUIRE(svc.tryStartSearchRequest(cap));

        svc.onSearchRequestQueued();
        REQUIRE_FALSE(svc.tryStartSearchRequest(cap));

        auto waiting = svc.getSearchLoadMetrics();
        REQUIRE(waiting.active == 1);
        REQUIRE(waiting.queued == 1);

        svc.onSearchRequestRejected();
        auto rejected = svc.getSearchLoadMetrics();
        REQUIRE(rejected.active == 1);
        REQUIRE(rejected.queued == 0);

        svc.onSearchRequestFinished();
        auto finished = svc.getSearchLoadMetrics();
        REQUIRE(finished.active == 0);
        REQUIRE(finished.queued == 0);
    }
}

// =============================================================================
// FsmMetricsRegistry Integration Tests
// =============================================================================

TEST_CASE("FsmMetricsRegistry: Metrics collection", "[daemon][metrics][fsm]") {
    // Reset registry to known state
    FsmMetricsRegistry::instance().reset();

    SECTION("Registry starts with zeros") {
        auto snap = FsmMetricsRegistry::instance().snapshot();
        REQUIRE(snap.transitions == 0);
        REQUIRE(snap.headerReads == 0);
        REQUIRE(snap.payloadReads == 0);
        REQUIRE(snap.payloadWrites == 0);
        REQUIRE(snap.bytesSent == 0);
        REQUIRE(snap.bytesReceived == 0);
    }

    SECTION("Registry increments correctly") {
        FsmMetricsRegistry::instance().incrementTransitions(5);
        FsmMetricsRegistry::instance().incrementHeaderReads(3);
        FsmMetricsRegistry::instance().incrementPayloadReads(2);
        FsmMetricsRegistry::instance().addBytesSent(100);
        FsmMetricsRegistry::instance().addBytesReceived(200);

        auto snap = FsmMetricsRegistry::instance().snapshot();
        REQUIRE(snap.transitions == 5);
        REQUIRE(snap.headerReads == 3);
        REQUIRE(snap.payloadReads == 2);
        REQUIRE(snap.bytesSent == 100);
        REQUIRE(snap.bytesReceived == 200);
    }

    SECTION("Registry reset works") {
        FsmMetricsRegistry::instance().incrementTransitions(10);
        FsmMetricsRegistry::instance().reset();

        auto snap = FsmMetricsRegistry::instance().snapshot();
        REQUIRE(snap.transitions == 0);
    }
}