// 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 // pi-lens-ignore: fatal error #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "../../common/metadata_test_db.h" #include "../../common/test_helpers_catch2.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include 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& 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 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> 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(dim_, 0.25f); } Result>> generateBatchEmbeddings(const std::vector& 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>(lastBatchSizeArg_, std::vector(dim_, 0.5f)); } Result> generateEmbeddingFor(const std::string&, const std::string& text) override { lastTextArg_ = text; return generateEmbedding(text); } Result>> generateBatchEmbeddingsFor(const std::string&, const std::vector& texts) override { lastBatchSizeArg_ = texts.size(); if (!texts.empty()) { lastTextArg_ = texts.front(); } return generateBatchEmbeddings(texts); } Result loadModel(const std::string& modelName) override { if (loadError_) { return *loadError_; } lastOptionsModel_.clear(); lastOptionsJson_.clear(); loaded_.push_back(modelName); return Result(); } Result loadModelWithOptions(const std::string& modelName, const std::string& optionsJson) override { if (loadError_) { return *loadError_; } lastOptionsModel_ = modelName; lastOptionsJson_ = optionsJson; loaded_.push_back(modelName); return Result(); } Result 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(); } 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 getLoadedModels() const override { return loaded_; } size_t getLoadedModelCount() const override { return loaded_.size(); } Result 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 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 embedding) { singleEmbeddingResult_ = std::move(embedding); } void setBatchEmbeddingResult(std::vector> 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& texts) { static const std::array 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 loaded_; std::string lastOptionsModel_; std::string lastOptionsJson_; std::optional loadError_; std::optional generateEmbeddingError_; std::optional generateBatchError_; bool throwOnMemoryUsage_{false}; bool failModelInfo_{false}; mutable int throwOnIsModelLoadedCount_{0}; std::vector singleEmbeddingResult_; std::vector> batchEmbeddingResult_; std::unordered_map 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 class ScopeExit { public: explicit ScopeExit(Fn fn) : fn_(std::move(fn)) {} ~ScopeExit() { fn_(); } private: Fn fn_; }; std::shared_ptr getPruneTestPool() { static std::shared_ptr pool = [] { metadata::ConnectionPoolConfig cfg{}; cfg.enableWAL = false; cfg.enableForeignKeys = false; auto p = std::make_shared(":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 makeReadyMetadataRepo(const std::shared_ptr& pool) { return std::make_shared( *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 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 snapshots) { snapshots_ = std::move(snapshots); } void setSnapshotError(Error error) { snapshotError_ = std::move(error); } void setSnapshotDocuments(std::vector docs) { snapshotDocs_ = std::move(docs); } void setSnapshotDocumentsError(Error error) { snapshotDocsError_ = std::move(error); } void setExactPathDocument(std::optional doc) { std::lock_guard lk(mu_); exactPathDocument_ = std::move(doc); } void setExactPathError(Error error) { std::lock_guard lk(mu_); exactPathError_ = std::move(error); } void setAllMetadata(int64_t id, std::unordered_map values) { std::lock_guard lk(mu_); metadataById_[id] = std::move(values); } void setAllMetadataError(int64_t id, Error error) { std::lock_guard lk(mu_); metadataByIdErrors_[id] = std::move(error); } void setPathTreeDocuments(std::vector docs) { std::lock_guard lk(mu_); pathTreeDocs_ = std::move(docs); } void setMetadataValueCounts( std::unordered_map> counts) { metadataValueCounts_ = std::move(counts); } void setMetadataValueCountsError(Error error) { metadataValueCountsError_ = std::move(error); } std::size_t deleteCallCount() const { std::lock_guard lk(mu_); return deleteCallCount_; } std::vector deletedIds() const { std::lock_guard lk(mu_); return deletedIds_; } Result> queryDocuments(const metadata::DocumentQueryOptions&) override { std::lock_guard lk(mu_); if (throwOnQuery_) { throw std::runtime_error(*throwOnQuery_); } if (queryError_) { return *queryError_; } return docs_; } Result> findDocumentsBySnapshot(const std::string&) override { std::lock_guard lk(mu_); if (snapshotDocsError_) { return *snapshotDocsError_; } return snapshotDocs_; } Result> getSnapshots() override { std::lock_guard lk(mu_); if (snapshotError_) { return *snapshotError_; } return snapshots_; } Result>> getMetadataValueCounts(const std::vector&, const metadata::DocumentQueryOptions&) override { std::lock_guard lk(mu_); if (metadataValueCountsError_) { return *metadataValueCountsError_; } return metadataValueCounts_; } Result> getDocumentByHash(const std::string& hash) override { std::lock_guard 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(std::nullopt); } if (auto it = docsByHash_.find(hash); it != docsByHash_.end()) { return std::optional(it->second); } return std::optional(std::nullopt); } Result> getAllMetadata(int64_t documentId) override { std::lock_guard 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{}; } Result> getDocumentSnapshots(int64_t documentId) override { auto all = getAllMetadata(documentId); if (!all) { return all.error(); } return metadata::legacySnapshotEntries(all.value()); } Result> findDocumentByExactPath(const std::string&) override { std::lock_guard lk(mu_); if (exactPathError_) { return *exactPathError_; } if (exactPathDocument_.has_value()) { return exactPathDocument_; } return std::optional(std::nullopt); } Result> findDocumentsByPathTreePrefix(std::string_view, bool = true, int limit = 0) override { std::lock_guard lk(mu_); if (limit > 0 && static_cast(limit) < pathTreeDocs_.size()) { return std::vector(pathTreeDocs_.begin(), pathTreeDocs_.begin() + limit); } return pathTreeDocs_; } Result deleteDocument(int64_t id) override { std::lock_guard 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(); } private: mutable std::mutex mu_; std::vector docs_; std::unordered_map docsByHash_; std::unordered_map docsById_; std::optional queryError_; std::optional snapshotError_; std::optional snapshotDocsError_; std::optional metadataValueCountsError_; std::optional throwOnQuery_; std::unordered_set missingHashes_; std::unordered_map throwOnLookup_; std::unordered_map deleteErrors_; std::optional exactPathDocument_; std::optional exactPathError_; std::unordered_map> metadataById_; std::unordered_map metadataByIdErrors_; std::vector pathTreeDocs_; std::vector snapshots_; std::vector snapshotDocs_; std::unordered_map> metadataValueCounts_; std::size_t deleteCallCount_{0}; std::vector deletedIds_; }; class StubContentStore : public api::IContentStore { public: void setBlob(const std::string& hash, const std::string& content) { blobs_[hash] = std::vector(reinterpret_cast(content.data()), reinterpret_cast(content.data()) + content.size()); } void setRemoveResult(const std::string& hash, Result result) { removeResults_[hash] = std::move(result); } bool hasBlob(const std::string& hash) const { return blobs_.contains(hash); } std::vector removedHashes() const { return removedHashes_; } Result storeBytes(std::span data, const api::ContentMetadata&) override { std::string hash = "stub-hash-" + std::to_string(blobs_.size() + 1); blobs_[hash] = std::vector(data.begin(), data.end()); api::StoreResult result; result.contentHash = hash; result.bytesStored = blobs_[hash].size(); return result; } Result> 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> 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 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> retrieveRawAsync(const std::string& hash) override { return std::async(std::launch::deferred, [this, hash]() { return retrieveRaw(hash); }); } Result 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 bytes; char ch = 0; while (ifs.get(ch)) { bytes.push_back(static_cast(ch)); } if (!ifs.eof()) { return Error{ErrorCode::IOError, "stub read failed"}; } return storeBytes(bytes, {}); } Result retrieve(const std::string&, const std::filesystem::path&, api::ProgressCallback) override { return ErrorCode::NotImplemented; } Result storeStream(std::istream&, const api::ContentMetadata&, api::ProgressCallback) override { return ErrorCode::NotImplemented; } Result retrieveStream(const std::string&, std::ostream&, api::ProgressCallback) override { return ErrorCode::NotImplemented; } Result exists(const std::string& hash) const override { return blobs_.contains(hash); } Result 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 getMetadata(const std::string&) const override { return ErrorCode::NotImplemented; } Result updateMetadata(const std::string&, const api::ContentMetadata&) override { return ErrorCode::NotImplemented; } std::vector> storeBatch(const std::vector&, const std::vector&) override { return {}; } std::vector> removeBatch(const std::vector&) override { return {}; } api::ContentStoreStats getStats() const override { return {}; } api::HealthStatus checkHealth() const override { return {}; } Result verify(api::ProgressCallback) override { return ErrorCode::NotImplemented; } Result compact(api::ProgressCallback) override { return ErrorCode::NotImplemented; } Result garbageCollect(api::ProgressCallback) override { return ErrorCode::NotImplemented; } private: std::unordered_map> blobs_; std::unordered_map> removeResults_; std::vector 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 query(const app::services::GraphQueryRequest&) override { return response_; } Result listSnapshots(const app::services::ListSnapshotsRequest&) override { return app::services::ListSnapshotsResponse{}; } Result getPathHistory(const app::services::PathHistoryRequest&) override { return app::services::PathHistoryResponse{}; } Result> resolveToNodeId(const std::string&) override { return std::optional(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::filesystem::file_size(path)); info.sha256Hash = hash; info.mimeType = "text/plain"; const auto now = std::chrono::floor(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; // Ingest stores resolved paths (populatePathDerivedFields); keys and labels built from // testDir must match, including where the temp dir is behind a symlink (macOS /var). testDir = std::filesystem::canonical(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(cfg, state, lifecycleFsm); dispatcher = std::make_unique(&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(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(std::move(kgResult).value()); repo->setKnowledgeGraphStore(kgStore); writeIo.restart(); writeWork = std::make_unique< boost::asio::executor_work_guard>( writeIo.get_executor()); writeThread = std::thread([this] { writeIo.run(); }); auto writeCoord = std::make_unique(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 { 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 svc; std::unique_ptr dispatcher; boost::asio::io_context writeIo; std::unique_ptr> writeWork; std::thread writeThread; std::shared_ptr pool; std::shared_ptr repo; std::shared_ptr 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, bool requireTrustedConfig = false) { 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 EXPECTED_MODE = )PY" << (requireTrustedConfig ? R"PY("trusted")PY" : "None") << R"PY( 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": if EXPECTED_MODE and req.get("params", {}).get("mode") != EXPECTED_MODE: raise ValueError("missing configured mode") 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> getStoreDocumentTaskChannel(std::size_t capacity = 0) { auto existing = InternalEventBus::instance().get_channel( "store_document_tasks"); if (existing) { return existing; } if (capacity == 0) { capacity = static_cast(TuneAdvisor::storeDocumentChannelCapacity()); } return InternalEventBus::instance().get_or_create_channel( "store_document_tasks", capacity); } void drainStoreDocumentTasks() { auto channel = InternalEventBus::instance().get_channel( "store_document_tasks"); if (!channel) { return; } InternalEventBus::StoreDocumentTask task; while (channel->try_pop(task)) { } } template 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(384, "/tmp/test-model.onnx"); svc.testing_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:" 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()); } } TEST_CASE("DaemonMetrics reports effective embedding policy before provider adoption", "[daemon][metrics][embedding][config][snapshot]") { const auto dataDir = makeTempDir("yams_metrics_embed_policy_"); const auto configPath = yams::test::write_file(dataDir / "config.toml", R"toml( [embeddings] backend = "simeon" preferred_model = "simeon-default" embedding_dim = 1024 preload_on_startup = false )toml"); yams::test::ScopedEnvVar configPathEnvironment{"YAMS_CONFIG_PATH", std::nullopt}; yams::test::ScopedEnvVar configEnvironment{"YAMS_CONFIG", std::nullopt}; yams::test::ScopedEnvVar backendEnvironment{"YAMS_EMBED_BACKEND", std::nullopt}; yams::test::ScopedEnvVar modelEnvironment{"YAMS_PREFERRED_MODEL", std::nullopt}; yams::test::ScopedEnvVar dimensionEnvironment{"YAMS_EMBED_DIM", std::nullopt}; StateComponent state; DaemonLifecycleFsm lifecycleFsm; DaemonConfig cfg; cfg.dataDir = dataDir; cfg.configFilePath = configPath; ServiceManager svc(cfg, state, lifecycleFsm); REQUIRE(svc.initialize().has_value()); DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator()); const auto snapshot = metrics.getSnapshot(); REQUIRE((snapshot != nullptr)); CHECK((snapshot->embeddingBackend == "simeon")); CHECK((snapshot->embeddingModel == "simeon-default")); CHECK((snapshot->embeddingDim == 1024U)); const auto policy = svc.getResolvedEmbeddingConfig(); REQUIRE((policy != nullptr)); CHECK((policy->policyIdentity == "simeon:simeon-default:1024")); CHECK((policy->effectiveConfigPath == configPath)); } // ============================================================================= // 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(resp)); auto stats = std::get(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(resp)); const auto& status = std::get(resp); const std::vector 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::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); } // Symbol extraction was removed in v0.20; status must not advertise it as unavailable. CHECK(status.readinessStates.count("symbol_extractors_ready") == 0); CHECK(status.requestCounts.count("symbol_extractors_loaded") == 0); const std::array requiredCapabilityMetricKeys = { metrics::kContentExtractorsLoaded, 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(resp)); return std::get(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: detailed status includes disk pressure policy", "[daemon][status][disk-pressure]") { DaemonConfig cfg; cfg.dataDir = makeTempDir("yams_status_disk_pressure_"); cfg.diskPressure.warningFreePercent = 12.5; cfg.diskPressure.minimumWriteAdmissionBytes = 512ULL * 1024ULL * 1024ULL; cfg.diskPressure.emergencyReserveBytes = 64ULL * 1024ULL * 1024ULL; 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 request = req; boost::asio::io_context io; auto future = boost::asio::co_spawn(io, dispatcher.dispatch(request), boost::asio::use_future); io.run(); const auto response = future.get(); REQUIRE(std::holds_alternative(response)); return std::get(response); }; const auto compact = fetch(false); CHECK_FALSE(compact.requestCounts.contains(std::string(metrics::kStoragePressureLevel))); const auto detailed = fetch(true); CHECK(detailed.requestCounts.at(std::string(metrics::kStoragePressureLevel)) == static_cast(storage::DiskPressureLevel::Unknown)); CHECK(detailed.requestCounts.at(std::string(metrics::kStorageWarningFreePercentBp)) == 1250); CHECK(detailed.requestCounts.at(std::string(metrics::kStorageWriteAdmissionBytesLow)) == 512ULL * 1024ULL * 1024ULL); CHECK(detailed.requestCounts.at(std::string(metrics::kStorageWriteAdmissionBytesHigh)) == 0); CHECK(detailed.requestCounts.at(std::string(metrics::kStorageEmergencyReserveBytesLow)) == 64ULL * 1024ULL * 1024ULL); CHECK(detailed.requestCounts.at(std::string(metrics::kStorageEmergencyReserveBytesHigh)) == 0); } 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(std::chrono::duration_cast( 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(resp)); const auto& status = std::get(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(resp)); const auto& status = std::get(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(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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::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(); auto store = std::make_shared(); 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(resp)); const auto& pruneResp = std::get(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(monthResp)); const auto& monthPruneResp = std::get(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(yearResp)); const auto& yearPruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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{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(resp)); const auto& pruneResp = std::get(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{targetHash}); CHECK(repo->deletedIds() == std::vector{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(resp)); const auto& pruneResp = std::get(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(resp)); const auto& pruneResp = std::get(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(384, "/tmp/model.onnx"); svc.testing_setModelProvider(provider); LoadModelRequest req; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(768, "/tmp/provider-model.onnx"); svc.testing_setModelProvider(provider); LoadModelRequest req; req.modelName = "dispatcher-model"; req.optionsJson = R"({"mode":"offline"})"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& loadResp = std::get(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(384, "/tmp/failing-model.onnx"); provider->setLoadError(Error{ErrorCode::InvalidState, "provider refused load"}); svc.testing_setModelProvider(provider); lifecycle.resetDegradedState(); LoadModelRequest req; req.modelName = "failing-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(384, "/tmp/utf8-model.onnx"); std::string invalidMessage = "prefix "; invalidMessage.push_back(static_cast(0xC2)); invalidMessage.push_back(static_cast(0xA9)); invalidMessage.push_back(static_cast(0xE2)); invalidMessage.push_back(static_cast(0x82)); invalidMessage.push_back(static_cast(0xAC)); invalidMessage.push_back(static_cast(0xF0)); invalidMessage.push_back(static_cast(0x9F)); invalidMessage.push_back(static_cast(0x92)); invalidMessage.push_back(static_cast(0xA9)); invalidMessage.push_back(static_cast(0xC0)); invalidMessage.push_back(static_cast(0xFF)); provider->setLoadError(Error{ErrorCode::InvalidState, invalidMessage}); svc.testing_setModelProvider(provider); LoadModelRequest req; req.modelName = "utf8-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(resp); std::string expected = "prefix "; expected.push_back(static_cast(0xC2)); expected.push_back(static_cast(0xA9)); expected.push_back(static_cast(0xE2)); expected.push_back(static_cast(0x82)); expected.push_back(static_cast(0xAC)); expected.push_back(static_cast(0xF0)); expected.push_back(static_cast(0x9F)); expected.push_back(static_cast(0x92)); expected.push_back(static_cast(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(384, "/tmp/throwing-model.onnx"); provider->setThrowOnMemoryUsage(true); svc.testing_setModelProvider(provider); LoadModelRequest req; req.modelName = "throwing-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(384, "/tmp/already-loaded-model.onnx"); REQUIRE(provider->loadModel("already-loaded-model").has_value()); svc.testing_setModelProvider(provider); LoadModelRequest req; req.modelName = "already-loaded-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& loadResp = std::get(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(384, "/tmp/timeout-model.onnx"); svc.testing_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(resp)); CHECK(std::get(resp).success); CHECK(provider->isModelLoaded("timeout-clamped-model")); } SECTION("load model survives invalid timeout env value") { auto provider = std::make_shared(384, "/tmp/timeout-invalid.onnx"); svc.testing_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(resp)); CHECK(std::get(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(384, "/tmp/rebuild-health.onnx"); REQUIRE(provider->loadModel("already-loaded-health-model").has_value()); svc.testing_setModelProvider(provider); metadata::ConnectionPoolConfig poolCfg{}; poolCfg.enableWAL = false; poolCfg.enableForeignKeys = false; auto pool = std::make_shared(":memory:", poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = std::make_shared(*pool); search::SearchEngineConfig config; config.vectorWeight = 1.0f; auto unhealthyEngine = std::make_shared(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(resp)); CHECK(std::get(resp).success); CHECK(provider->getLoadedModelCount() == 1); } SECTION("load model ignores isModelLoaded exceptions before loading") { auto provider = std::make_shared(4, "/tmp/isloaded-throw.onnx"); svc.testing_setModelProvider(provider); metadata::ConnectionPoolConfig poolCfg{}; poolCfg.enableWAL = false; poolCfg.enableForeignKeys = false; auto pool = std::make_shared(":memory:", poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = std::make_shared(*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(vectorCfg); REQUIRE(vectorDb->initialize()); search::SearchEngineConfig config; config.vectorWeight = 1.0f; auto healthyEngine = std::make_shared( 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(resp)); const auto& loadResp = std::get(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(384, "/tmp/lifecycle-throw.onnx"); svc.testing_setModelProvider(provider); lifecycle.setThrowOnSetSubsystemDegraded(true); LoadModelRequest req; req.modelName = "lifecycle-throw-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& loadResp = std::get(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(384, "/tmp/model.onnx"); svc.testing_setModelProvider(provider); UnloadModelRequest req; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(384, "/tmp/model.onnx"); svc.testing_setModelProvider(provider); UnloadModelRequest req; req.modelName = "missing-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(512, "/tmp/status-model.onnx"); svc.testing_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(resp)); const auto& statusResp = std::get(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(256, "/tmp/status-fallback.onnx"); REQUIRE(provider->loadModel("status-fallback").has_value()); provider->setFailModelInfo(true); svc.testing_setModelProvider(provider); auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}}); REQUIRE(std::holds_alternative(resp)); const auto& statusResp = std::get(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(384, "/tmp/model.onnx"); provider->setAvailable(false); svc.testing_setModelProvider(provider); auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}}); REQUIRE(std::holds_alternative(resp)); const auto& statusResp = std::get(resp); CHECK(statusResp.models.empty()); CHECK(statusResp.totalMemoryMb == 0); } SECTION("model status converts provider exceptions into internal errors") { auto provider = std::make_shared(384, "/tmp/model.onnx"); REQUIRE(provider->loadModel("status-throw").has_value()); provider->setThrowOnMemoryUsage(true); svc.testing_setModelProvider(provider); auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(); auto store = std::make_shared(); 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(); RequestContextRegistry::instance().register_context(4242, ctx); auto resp = dispatchRequest(dispatcher, Request{CancelRequest{4242}}); RequestContextRegistry::instance().deregister_context(4242); REQUIRE(std::holds_alternative(resp)); CHECK(std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(); svc.__test_setContentStore(store); GetInitRequest req; req.hash = "missing-bytes"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(); 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(resp)); const auto& err = std::get(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(); store->setBlob("session-flow-hash", "hello retrieval flow"); svc.__test_setContentStore(store); svc.__test_setRetrievalSessionManager(std::make_unique()); GetInitRequest initReq; initReq.hash = "session-flow-hash"; initReq.chunkSize = 5; auto initResp = dispatchRequest(dispatcher, Request{initReq}); REQUIRE(std::holds_alternative(initResp)); const auto& init = std::get(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(chunkResp)); const auto& chunk = std::get(chunkResp); CHECK(chunk.data == "hello"); CHECK(chunk.bytesRemaining == 15); auto endResp = dispatchRequest(dispatcher, Request{GetEndRequest{init.transferId}}); REQUIRE(std::holds_alternative(endResp)); CHECK(std::get(endResp).message == "OK"); auto postEndChunk = dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 0, 5}}); REQUIRE(std::holds_alternative(postEndChunk)); const auto& postEndErr = std::get(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(); store->setBlob("offset-error-hash", "hello retrieval flow"); svc.__test_setContentStore(store); svc.__test_setRetrievalSessionManager(std::make_unique()); GetInitRequest initReq; initReq.hash = "offset-error-hash"; auto initResp = dispatchRequest(dispatcher, Request{initReq}); REQUIRE(std::holds_alternative(initResp)); const auto& init = std::get(initResp); auto chunkResp = dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 99, 5}}); REQUIRE(std::holds_alternative(chunkResp)); const auto& err = std::get(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(); store->setBlob("cap-flow-hash", "hello retrieval flow"); svc.__test_setContentStore(store); svc.__test_setRetrievalSessionManager(std::make_unique()); GetInitRequest initReq; initReq.hash = "cap-flow-hash"; initReq.chunkSize = 10; initReq.maxBytes = 7; auto initResp = dispatchRequest(dispatcher, Request{initReq}); REQUIRE(std::holds_alternative(initResp)); const auto& init = std::get(initResp); auto chunkResp = dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 7, 5}}); REQUIRE(std::holds_alternative(chunkResp)); const auto& chunk = std::get(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(resp)); const auto& err = std::get(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()); auto resp = dispatchRequest(dispatcher, Request{GetChunkRequest{9999, 0, 4}}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); CHECK(std::get(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(resp)); const auto& err = std::get(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(); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(); auto store = std::make_shared(); 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(resp)); const auto& deleteResp = std::get(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 publishes durable document and content tombstones") { auto repo = std::make_shared(); auto store = std::make_shared(); const std::string hash(64, 'a'); auto doc = makeDoc(23, "/tmp/delete/replicated.txt", hash); repo->addDocument(doc); store->setBlob(hash, "replicated"); svc.__test_setMetadataRepo(repo); svc.__test_setContentStore(store); storage::BackendConfig backendConfig; backendConfig.type = "filesystem"; backendConfig.localPath = cfg.dataDir / "memory-sync"; auto backend = std::make_unique(); REQUIRE(backend->initialize(backendConfig).has_value()); auto sync = std::make_unique( std::move(backend), memory_sync::MemorySyncConfig{"delete-node", 60'000, "delete-corpus", 1}); REQUIRE(sync->syncFully().has_value()); svc.testingSetMemorySyncService(std::move(sync)); DeleteRequest req; req.hash = hash; req.force = true; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); CHECK(std::get(resp).successCount == 1); memory_sync::VersionVector empty; auto deltas = svc.testingMemorySyncService()->exportLocalDeltasAfter(empty, 4); REQUIRE(deltas.has_value()); REQUIRE(deltas.value().deltas.size() == 2); CHECK(deltas.value().deltas[0].logicalKey == "content-blob/" + hash); CHECK(deltas.value().deltas[0].record.isTombstone()); CHECK(deltas.value().deltas[1].logicalKey == "document/" + hash); CHECK(deltas.value().deltas[1].record.isTombstone()); } SECTION("delete reports missing content store") { auto repo = std::make_shared(); svc.__test_setMetadataRepo(repo); DeleteRequest req; req.hash = "no-store-hash"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); auto graphQuery = std::make_shared(); 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(resp)); const auto& getResp = std::get(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 caps related documents at graphLimit") { // #280 follow-up: GetRequest.graphLimit must reach DocumentService. auto repoPath = makeTempMetadataDbPath("yams_get_graph_limit_repo_"); metadata::ConnectionPoolConfig poolCfg{}; auto pool = std::make_shared(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); auto graphQuery = std::make_shared(); auto mainDoc = makeDoc(72, "/tmp/get-graph-limit/main.txt", std::string(64, 'e')); REQUIRE(repo->insertDocument(mainDoc).has_value()); for (int i = 0; i < 3; ++i) { auto related = makeDoc(73 + i, "/tmp/get-graph-limit/related" + std::to_string(i) + ".txt", std::string(64, static_cast('a' + i))); REQUIRE(repo->insertDocument(related).has_value()); graphQuery->addConnectedDocument(related.sha256Hash, "semantic_neighbor", 1, 0.5); } store->setBlob(mainDoc.sha256Hash, "main content"); svc.__test_setMetadataRepo(repo); svc.__test_setContentStore(store); svc.__test_setGraphQueryService(graphQuery); auto relatedCount = [&](std::uint32_t graphLimit) { GetRequest req; req.hash = mainDoc.sha256Hash; req.metadataOnly = true; req.showGraph = true; req.graphDepth = 1; req.graphLimit = graphLimit; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); return std::get(resp).related.size(); }; CHECK(relatedCount(0) == 3); CHECK(relatedCount(2) == 2); CHECK(relatedCount(10) == 3); } 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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& getResp = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& getResp = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(); 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(resp)); const auto& history = std::get(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(); svc.__test_setMetadataRepo(repo); FileHistoryRequest req; req.filepath = "ghost.txt"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& history = std::get(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(); 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(resp)); const auto& history = std::get(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(); 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(resp)); const auto& history = std::get(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(); 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(resp)); const auto& history = std::get(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(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(resp)); const auto& listResp = std::get(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(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(resp)); const auto& listResp = std::get(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(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(resp)); const auto& listResp = std::get(resp); static_cast(listResp.totalCount); } 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(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(resp)); const auto& listResp = std::get(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(resp)); const auto& err = std::get(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(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)) { return std::get(resp).message; } return std::string("not an error"); }()); REQUIRE(std::holds_alternative(resp)); const auto& addResp = std::get(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(resp)); const auto& addResp = std::get(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(resp)); const auto& addResp = std::get(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(resp)); const auto& err = std::get(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(¬ReadyLifecycle, ¬ReadySvc, ¬ReadyState); 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(resp)); const auto& addResp = std::get(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(¬ReadyLifecycle, ¬ReadySvc, ¬ReadyState); 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(resp)); const auto& err = std::get(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(¬ReadyLifecycle, ¬ReadySvc, ¬ReadyState); 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(resp)); const auto& addResp = std::get(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(¬ReadyLifecycle, ¬ReadySvc, ¬ReadyState); 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(resp)); const auto& addResp = std::get(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(resp)); const auto& addResp = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& addResp = std::get(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(resp)); const auto& addResp = std::get(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(resp)); const auto& addResp = std::get(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(¬ReadyLifecycle, ¬ReadySvc, ¬ReadyState); 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(resp)); const auto& addResp = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& addResp = std::get(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()); AddDocumentRequest req; req.name = "sync-add.txt"; req.content = "sync path content"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& addResp = std::get(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()); 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(resp)); const auto& addResp = std::get(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{}); AddDocumentRequest req; req.name = "sync-error.txt"; req.content = "sync failure"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(); 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(resp)); const auto& history = std::get(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(); 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(resp)); const auto& history = std::get(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(); 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(resp)); const auto& history = std::get(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(); 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(resp)); const auto& err = std::get(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(); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(repoPath.string(), poolCfg); REQUIRE(pool->initialize().has_value()); auto repo = makeReadyMetadataRepo(pool); auto store = std::make_shared(); 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(resp)); const auto& err = std::get(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(resp)) { latest = std::get(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 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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& jobs = std::get(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(resp)); const auto& jobs = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& dl = std::get(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(resp)); const auto& err = std::get(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(resp)) { latest = std::get(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(resp)); const auto& dl = std::get(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(statusResp)); const auto& status = std::get(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(listResp)); const auto& list = std::get(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(resp)); const auto& err = std::get(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(3, "/tmp/embed-model.onnx"); provider->setSingleEmbeddingResult({0.1f, 0.2f, 0.3f}); svc.testing_setModelProvider(provider); GenerateEmbeddingRequest req{"hello world", "embed-model", true}; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& embedResp = std::get(resp); CHECK(embedResp.embedding == std::vector{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(3, "/tmp/embed-error.onnx"); provider->setGenerateEmbeddingError( Error{ErrorCode::InvalidState, "generator unavailable"}); svc.testing_setModelProvider(provider); GenerateEmbeddingRequest req{"hello world", "embed-error", true}; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::InvalidState); CHECK(err.message == "generator unavailable"); } SECTION("generate embedding sanitizes utf8 in load failures") { auto provider = std::make_shared(3, "/tmp/embed-utf8.onnx"); std::string invalidMessage = "prefix "; invalidMessage.push_back(static_cast(0xC2)); invalidMessage.push_back(static_cast(0xA9)); invalidMessage.push_back(static_cast(0xE2)); invalidMessage.push_back(static_cast(0x82)); invalidMessage.push_back(static_cast(0xAC)); invalidMessage.push_back(static_cast(0xF0)); invalidMessage.push_back(static_cast(0x9F)); invalidMessage.push_back(static_cast(0x92)); invalidMessage.push_back(static_cast(0xA9)); invalidMessage.push_back(static_cast(0xC0)); invalidMessage.push_back(static_cast(0xFF)); provider->setLoadError(Error{ErrorCode::InvalidState, invalidMessage}); svc.testing_setModelProvider(provider); GenerateEmbeddingRequest req{"hello world", "embed-utf8", true}; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(resp); std::string expected = "prefix "; expected.push_back(static_cast(0xC2)); expected.push_back(static_cast(0xA9)); expected.push_back(static_cast(0xE2)); expected.push_back(static_cast(0x82)); expected.push_back(static_cast(0xAC)); expected.push_back(static_cast(0xF0)); expected.push_back(static_cast(0x9F)); expected.push_back(static_cast(0x92)); expected.push_back(static_cast(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(3, "/tmp/embed-timeout.onnx"); provider->setSingleEmbeddingResult({0.7f, 0.8f, 0.9f}); svc.testing_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(resp)); const auto& embedResp = std::get(resp); CHECK(embedResp.embedding == std::vector{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(3, "/tmp/embed-throw.onnx"); provider->setThrowOnIsModelLoadedCount(2); svc.testing_setModelProvider(provider); GenerateEmbeddingRequest req{"hello world", "embed-throw", true}; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(2, "/tmp/batch-fallback.onnx"); provider->setGenerateBatchError(Error{ErrorCode::NotImplemented, "batch unsupported"}); provider->setSingleEmbeddingResult({0.4f, 0.6f}); svc.testing_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(resp)); const auto& batchResp = std::get(resp); REQUIRE(batchResp.embeddings.size() == 2); CHECK(batchResp.embeddings[0] == std::vector{0.4f, 0.6f}); CHECK(batchResp.embeddings[1] == std::vector{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(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.testing_setModelProvider(provider); BatchEmbeddingRequest req; req.texts = {"good", "bad", "good-again"}; req.modelName = "batch-partial"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& batchResp = std::get(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(2, "/tmp/batch-error.onnx"); provider->setGenerateBatchError(Error{ErrorCode::InvalidState, "batch failed"}); svc.testing_setModelProvider(provider); BatchEmbeddingRequest req; req.texts = {"one", "two"}; req.modelName = "batch-error"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(2, "/tmp/batch-direct.onnx"); provider->setBatchEmbeddingResult({{0.2f, 0.8f}}); svc.testing_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(resp)); const auto& batchResp = std::get(resp); REQUIRE(batchResp.embeddings.size() == 1); CHECK(batchResp.embeddings[0] == std::vector{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(2, "/tmp/batch-load-fail.onnx"); provider->setLoadError(Error{ErrorCode::InvalidState, "batch load refused"}); svc.testing_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(resp)); const auto& err = std::get(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(2, "/tmp/batch-throw.onnx"); provider->setThrowOnIsModelLoadedCount(2); svc.testing_setModelProvider(provider); BatchEmbeddingRequest req; req.texts = {"one"}; req.modelName = "batch-throw"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::NotInitialized); CHECK(err.message == "ServiceManager not available"); } SECTION("embed documents reports unavailable model provider") { auto provider = std::make_shared(3, "/tmp/repair-provider.onnx"); provider->setAvailable(false); svc.testing_setModelProvider(provider); EmbedDocumentsRequest req; req.modelName = "repair-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::NotInitialized); CHECK(err.message == "Model provider not available"); } SECTION("embed documents refuses without a resolved data directory") { // Repair used to fall back to a relative "vectors.db" and create it in the working // directory when the data directory was not resolved. auto provider = std::make_shared(3, "/tmp/repair-empty.onnx"); svc.testing_setModelProvider(provider); svc.__test_setContentStore(std::make_shared()); svc.__test_setMetadataRepo(std::make_shared()); svc.testingSetResolvedDataDir({}); const auto strayDb = std::filesystem::current_path() / "vectors.db"; const bool existedBefore = std::filesystem::exists(strayDb); EmbedDocumentsRequest req; req.modelName = "repair-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); CHECK(std::get(resp).message.find("data directory") != std::string::npos); CHECK(std::filesystem::exists(strayDb) == existedBefore); } SECTION("embed documents returns success when repair finds no work") { svc.testingSetResolvedDataDir(cfg.dataDir); auto provider = std::make_shared(3, "/tmp/repair-empty.onnx"); svc.testing_setModelProvider(provider); svc.__test_setContentStore(std::make_shared()); svc.__test_setMetadataRepo(std::make_shared()); EmbedDocumentsRequest req; req.modelName = "repair-model"; req.batchSize = 4; req.skipExisting = true; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& embedResp = std::get(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") { svc.testingSetResolvedDataDir(cfg.dataDir); auto provider = std::make_shared(3, "/tmp/repair-missing-hashes.onnx"); svc.testing_setModelProvider(provider); svc.__test_setContentStore(std::make_shared()); svc.__test_setMetadataRepo(std::make_shared()); 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(resp)); const auto& err = std::get(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(3, "/tmp/repair-degraded.onnx"); svc.testing_setModelProvider(provider); svc.testing_setModelProviderDegraded(true, "simulated degraded state"); auto resp = dispatchRequest(dispatcher, Request{EmbedDocumentsRequest{}}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::InvalidState); CHECK(err.message == "Embedding generation disabled: provider degraded"); svc.testing_setModelProviderDegraded(false); } SECTION("embed documents reports missing content store") { auto provider = std::make_shared(3, "/tmp/repair-content.onnx"); svc.testing_setModelProvider(provider); svc.__test_setMetadataRepo(std::make_shared()); EmbedDocumentsRequest req; req.modelName = "repair-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(3, "/tmp/repair-meta.onnx"); svc.testing_setModelProvider(provider); svc.__test_setContentStore(std::make_shared()); EmbedDocumentsRequest req; req.modelName = "repair-model"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(3, "/tmp/repair-empty-model.onnx"); svc.testing_setModelProvider(provider); svc.__test_setContentStore(std::make_shared()); svc.__test_setMetadataRepo(std::make_shared()); EmbedDocumentsRequest req; req.modelName = ""; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(3, "/tmp/repair-config-model.onnx"); svc.testing_setModelProvider(provider); svc.__test_setContentStore(std::make_shared()); svc.__test_setMetadataRepo(std::make_shared()); 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.testing_setModelProvider(provider); configSvc.__test_setContentStore(std::make_shared()); configSvc.__test_setMetadataRepo(std::make_shared()); configSvc.testingSetResolvedDataDir(cfg.dataDir); 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(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(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(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::InternalError); CHECK(err.message == "Metadata repository unavailable"); } SECTION("legacy impact and affected-tests requests explain the removed symbol graph") { // No metadata repository: the answer does not depend on daemon state. GraphImpactRequest impactReq; impactReq.symbol = "callee"; auto impactResp = dispatchRequest(*fixture.dispatcher, Request{impactReq}); REQUIRE(std::holds_alternative(impactResp)); const auto& impact = std::get(impactResp); CHECK(impact.symbol == "callee"); CHECK(impact.affectedSymbols.empty()); REQUIRE(impact.warnings.size() == 1); CHECK(impact.warnings.front() == app::services::kImpactNeedsSymbolGraphWarning); GraphAffectedTestsRequest testsReq; testsReq.changedFiles = {"src/callee.cpp"}; auto testsResp = dispatchRequest(*fixture.dispatcher, Request{testsReq}); REQUIRE(std::holds_alternative(testsResp)); const auto& tests = std::get(testsResp); CHECK(tests.changedFiles == testsReq.changedFiles); CHECK(tests.affectedTests.empty()); REQUIRE(tests.warnings.size() == 1); CHECK(tests.warnings.front() == app::services::kAffectedTestsNeedSymbolGraphWarning); } SECTION("graph explore reports unavailable metadata repository") { auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphExploreRequest{.query = "entry"}}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(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(resp)); const auto& err = std::get(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(fullRespRaw)); const auto& fullResp = std::get(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(omittedRespRaw)); const auto& omittedResp = std::get(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(resp)); const auto& graphResp = std::get(resp); CHECK(graphResp.kgAvailable); CHECK(graphResp.originNode.label == "listTypes"); REQUIRE(graphResp.nodeTypeCounts.size() == 2); CHECK(graphResp.nodeTypeCounts[0] == std::pair{"function", 2}); CHECK(graphResp.nodeTypeCounts[1] == std::pair{"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(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(resp); CHECK(graphResp.originNode.label == "listRelations"); REQUIRE(graphResp.relationTypeCounts.size() == 2); CHECK(graphResp.relationTypeCounts[0] == std::pair{"calls", 1}); CHECK(graphResp.relationTypeCounts[1] == std::pair{"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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(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 the client cwd") { fixture.initMetadata(); const auto scopeRoot = fixture.testDir / "repo"; const auto sourcePath = scopeRoot / "src" / "inside.cpp"; const auto includePath = scopeRoot / "include" / "inside.hpp"; const auto testPath = fixture.testDir / "elsewhere" / "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.generic_string(), "insideSource", "function"); fixture.upsertNode("function:insideHeader@" + includePath.generic_string(), "insideHeader", "function"); fixture.upsertNode("function:outsideTest@" + testPath.generic_string(), "outsideTest", "function"); GraphQueryRequest firstPage; firstPage.listByType = true; firstPage.nodeType = "function"; firstPage.scopePathPrefix = scopeRoot.string(); firstPage.limit = 1; auto firstRaw = dispatchRequest(*fixture.dispatcher, Request{firstPage}); REQUIRE(std::holds_alternative(firstRaw)); const auto& first = std::get(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(secondRaw)); const auto& second = std::get(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 scopes document and path nodes under the client cwd") { // Issue #280: --scope-cwd returned nothing for a corpus outside src/ and include/, // because document nodes carry their path only in the label and snapshot path nodes // only in $.path. fixture.initMetadata(); const auto realRoot = fixture.testDir / "corpus"; const auto insideHeader = realRoot / "simeon" / "include" / "simeon" / "bm25.hpp"; const auto insideSource = realRoot / "simeon" / "src" / "bm25.cpp"; const auto outside = fixture.testDir / "other" / "outside.hpp"; for (const auto& path : {insideHeader, insideSource, outside}) { std::filesystem::create_directories(path.parent_path()); std::ofstream out(path); REQUIRE(out.good()); out << "int value;\n"; } const std::string snapshot = "2026-10-08T22:06:30.518048Z"; for (const auto& path : {insideHeader, insideSource, outside}) { const auto p = path.generic_string(); fixture.upsertNode("doc:hash-" + path.filename().string(), p, "document"); fixture.upsertNode("path:" + snapshot + ":" + p, p, "path", R"({"snapshot_id":")" + snapshot + R"(","path":")" + p + R"("})"); fixture.upsertNode("path:logical:" + p, p, "path", R"({"path":")" + p + R"(","logical":true})"); } auto listScoped = [&](const std::string& nodeType, const std::filesystem::path& scope) { GraphQueryRequest req; req.listByType = true; req.nodeType = nodeType; req.scopePathPrefix = scope.generic_string(); req.limit = 50; auto raw = dispatchRequest(*fixture.dispatcher, Request{req}); REQUIRE(std::holds_alternative(raw)); return std::get(raw); }; auto checkScoped = [&](const std::filesystem::path& scope) { const auto docs = listScoped("document", scope); CHECK(docs.totalNodesFound == 2); REQUIRE(docs.connectedNodes.size() == 2); for (const auto& node : docs.connectedNodes) { CHECK(node.label.starts_with(realRoot.generic_string() + "/")); } const auto paths = listScoped("path", scope); CHECK(paths.totalNodesFound == 4); REQUIRE(paths.connectedNodes.size() == 4); for (const auto& node : paths.connectedNodes) { CHECK(node.label.starts_with(realRoot.generic_string() + "/")); } }; checkScoped(realRoot); // A cwd reached through a symlink still matches the real stored paths. const auto linkRoot = fixture.testDir / "corpus-link"; std::error_code linkEc; std::filesystem::create_directory_symlink(realRoot, linkRoot, linkEc); if (!linkEc) { checkScoped(linkRoot); } // A sibling directory sharing the prefix is not in scope. const auto siblingDocs = listScoped("document", fixture.testDir / "corp"); CHECK(siblingDocs.totalNodesFound == 0); CHECK(siblingDocs.connectedNodes.empty()); } 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(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(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 defaults to documents without semantic neighbours") { fixture.initMetadata(); auto docA = fixture.upsertNode("doc:isolated-a", "a", "document"); auto docB = fixture.upsertNode("doc:isolated-b", "b", "document"); fixture.upsertNode("doc:isolated-c", "c", "document"); fixture.addEdge(docA, docB, "semantic_neighbor"); GraphQueryRequest req; req.isolatedMode = true; req.limit = 10; auto resp = dispatchRequest(*fixture.dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& graphResp = std::get(resp); CHECK(graphResp.originNode.label == "isolated:document:semantic_neighbor"); CHECK(graphResp.totalNodesFound == 2); for (const auto& node : graphResp.connectedNodes) { CHECK(node.nodeKey != "doc:isolated-b"); } } 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.nodeType = "function"; req.limit = 10; auto resp = dispatchRequest(*fixture.dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& graphResp = std::get(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(resp)); const auto& graphResp = std::get(resp); REQUIRE(graphResp.edges.size() == 7); std::unordered_set 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(resp)); const auto& err = std::get(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(resp)); const auto& historyResp = std::get(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(resp)); const auto& historyResp = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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(resp)); const auto& ingestResp = std::get(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]") { struct ReadyServiceFixture { // ServiceManager keeps non-owning references to both dependencies. Member teardown is // reverse declaration order, so the service shuts down before either dependency. std::unique_ptr state; std::unique_ptr lifecycleFsm; std::unique_ptr service; }; auto makeReadyService = [](bool configureExternalPlugin = false) { DaemonConfig cfg; cfg.dataDir = makeTempDir("yams_plugin_dispatcher_"); if (configureExternalPlugin) { cfg.pluginConfigs["dispatcher_external_plugin"] = R"({"mode":"trusted"})"; } auto state = std::make_unique(); state->readiness.contentStoreReady.store(true, std::memory_order_relaxed); state->readiness.metadataRepoReady.store(true, std::memory_order_relaxed); auto lifecycleFsm = std::make_unique(); auto svc = std::make_unique(cfg, *state, *lifecycleFsm); // The data directory is only resolved during initialize(); tests that place plugins // under it would otherwise write to the working directory. svc->testingSetResolvedDataDir(cfg.dataDir); svc->__test_pluginScanComplete(0); REQUIRE(svc->getPluginHostFsmSnapshot().state == PluginHostState::Ready); return ReadyServiceFixture{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(resp)); const auto& err = std::get(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(); 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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(scanResp)); const auto& scanErr = std::get(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(loadResp)); const auto& loadErr = std::get(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(unloadResp)); const auto& unloadErr = std::get(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(trustAddResp)); const auto& trustAddErr = std::get(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(trustRemoveResp)); const auto& trustRemoveErr = std::get(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(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::NotFound); CHECK(err.message == "Plugin not found"); } SECTION("plugin dry run does not execute an untrusted standalone script") { auto [state, lifecycleFsm, svc] = makeReadyService(); StubLifecycle lifecycle; RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get()); auto shimPath = makePythonShimPath(); ScopedEnvVar pythonGuard("PATH", shimPath.c_str()); const auto scanDir = makeTempDir("yams_untrusted_plugin_scan_"); const auto marker = scanDir / "scan-executed"; const auto script = scanDir / "untrusted.py"; { std::ofstream out(script); CHECK(out.good()); out << "from pathlib import Path\nPath(" << nlohmann::json(marker.string()).dump() << ").write_text('executed')\n"; } PluginLoadRequest request; request.pathOrName = script.string(); request.dryRun = true; auto response = dispatchRequest(dispatcher, Request{request}); CHECK(std::holds_alternative(response)); CHECK_FALSE(std::filesystem::exists(marker)); std::filesystem::remove_all(scanDir); } 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(missingResp)); const auto& missingErr = std::get(missingResp); CHECK(missingErr.code == ErrorCode::NotFound); CHECK(missingErr.message == "Plugin not found"); auto scanResp = dispatchRequest(dispatcher, Request{PluginScanRequest{}}); REQUIRE(std::holds_alternative(scanResp)); const auto& scan = std::get(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(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(resp)); const auto& err = std::get(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(externalResp)); const auto& externalLoad = std::get(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(externalUnloadResp)); CHECK((std::get(externalUnloadResp).code == ErrorCode::InvalidState)); CHECK((external->listLoaded().size() == 1)); } 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(scanResp)); const auto& scan = std::get(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(dryRunResp)); const auto& dryRun = std::get(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(loadResp)); const auto& load = std::get(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(unloadResp)); CHECK((std::get(unloadResp).code == ErrorCode::InvalidState)); CHECK((external->listLoaded().size() == 1)); } 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(resp)); const auto& load = std::get(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(unloadResp)); CHECK((std::get(unloadResp).code == ErrorCode::InvalidState)); CHECK((external->listLoaded().size() == 1)); } 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(resp)); const auto& load = std::get(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(unloadResp)); CHECK((std::get(unloadResp).code == ErrorCode::InvalidState)); CHECK((external->listLoaded().size() == 1)); #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(resp)); const auto& err = std::get(resp); CHECK(err.code == ErrorCode::InternalError); CHECK(err.message == "Plugin load failed for: " + pluginFile.string()); } SECTION("plugin trust add loads regular files with configured plugin settings") { auto [state, lifecycleFsm, svc] = makeReadyService(true); 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", true); auto pluginFile = pluginDir / "plugin.py"; auto trustAddResp = dispatchRequest(dispatcher, Request{PluginTrustAddRequest{pluginFile.string()}}); REQUIRE(std::holds_alternative(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(secondTrustResp)); CHECK(waitForCondition([&] { return external->listLoaded().size() == 1; })); auto unloadResp = dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}}); REQUIRE(std::holds_alternative(unloadResp)); CHECK((std::get(unloadResp).code == ErrorCode::InvalidState)); CHECK((external->listLoaded().size() == 1)); } } 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 pool; std::shared_ptr metadata; std::shared_ptr content; std::shared_ptr 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( makeTempMetadataDbPath(prefix + "db_").string(), poolConfig); REQUIRE(pool->initialize().has_value()); ReadyCollectionService ready{ .pool = pool, .metadata = makeReadyMetadataRepo(pool), .content = std::make_shared(), .manager = std::make_shared(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(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(text.size()); doc.sha256Hash = storeRes.value().contentHash; doc.mimeType = "text/plain"; const auto now = std::chrono::time_point_cast(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(resp)); const auto& err = std::get(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(); repo->setSnapshotError(Error{ErrorCode::DatabaseError, "snapshot lookup failed"}); svc.__test_setMetadataRepo(repo); auto resp = dispatchRequest(dispatcher, Request{ListSnapshotsRequest{}}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(); svc.__test_setMetadataRepo(repo); RestoreCollectionRequest req; req.collection = "dispatcher-collection"; req.outputDirectory = "out"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(); repo->setQueryError(Error{ErrorCode::DatabaseError, "collection query failed"}); svc.__test_setMetadataRepo(repo); svc.__test_setContentStore(std::make_shared()); RestoreCollectionRequest req; req.collection = "dispatcher-collection"; req.outputDirectory = "out"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(); svc.__test_setMetadataRepo(repo); RestoreSnapshotRequest req; req.snapshotId = "dispatcher-snapshot"; req.outputDirectory = "out"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(); repo->setSnapshotDocumentsError(Error{ErrorCode::DatabaseError, "snapshot query failed"}); svc.__test_setMetadataRepo(repo); svc.__test_setContentStore(std::make_shared()); RestoreSnapshotRequest req; req.snapshotId = "dispatcher-snapshot"; req.outputDirectory = "out"; auto resp = dispatchRequest(dispatcher, Request{req}); REQUIRE(std::holds_alternative(resp)); const auto& err = std::get(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(resp)); const auto& err = std::get(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(); 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(resp)); const auto& err = std::get(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(resp)); const auto& batchResp = std::get(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(batchResp.items[0].response)); CHECK(std::get(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(includeResp)); const auto& included = std::get(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(invalidPatternResp)); const auto& invalidPattern = std::get(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(excludeResp)); const auto& excluded = std::get(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(writeResp)); const auto& written = std::get(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(writtenFile)), std::istreambuf_iterator()); 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(createDirFailResp)); CHECK(std::get(createDirFailResp).code == ErrorCode::IOError); CHECK(std::get(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(parentFailResp)); const auto& parentFailed = std::get(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(openFailResp)); const auto& openFailed = std::get(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(writeFailResp)); const auto& writeFailed = std::get(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(defaultResp)); const auto& defaultRestore = std::get(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(invalidPatternResp)); const auto& invalidPattern = std::get(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(writeResp)); const auto& written = std::get(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(writtenFile)), std::istreambuf_iterator()); 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(createDirFailResp)); CHECK(std::get(createDirFailResp).code == ErrorCode::IOError); CHECK(std::get(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(parentFailResp)); const auto& parentFailed = std::get(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(openFailResp)); const auto& openFailed = std::get(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(writeFailResp)); const auto& writeFailed = std::get(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 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 exposes metadata integrity fast-path use", "[daemon][metrics][readiness]") { StateComponent state; state.readiness.databaseIntegrityFastPath.store(true, std::memory_order_release); DaemonLifecycleFsm lifecycleFsm; DaemonConfig cfg; cfg.dataDir = makeTempDir("yams_metrics_integrity_fast_path_"); ServiceManager svc(cfg, state, lifecycleFsm); DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator()); auto snap = metrics.getSnapshot(); REQUIRE(snap != nullptr); REQUIRE((snap->diagnosticCounters.count("database_integrity_fast_path") == 1)); CHECK((snap->diagnosticCounters.at("database_integrity_fast_path") == 1)); } 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 preserves failed lifecycle instead of reporting initialization", "[daemon][status][memory-sync-recovery]") { StateComponent state; StubLifecycle lifecycle{LifecycleState::Failed}; // Lifecycle rendering is independent of services. Avoid bootstrapping plugins, // background workers and checkpoint threads just to exercise this status seam. RequestDispatcher dispatcher(&lifecycle, nullptr, &state); const auto response = dispatchRequest(dispatcher, Request{StatusRequest{}}); REQUIRE(std::holds_alternative(response)); const auto& status = std::get(response); CHECK(status.lifecycleState == "failed"); CHECK(status.overallStatus == "failed"); CHECK_FALSE(status.ready); } 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(resp)); const auto& status = std::get(resp); CHECK(status.running); CHECK(status.retryAfterMs == 0); } TEST_CASE("RequestDispatcher: status reports effective search maintenance provenance", "[daemon][status][search][config]") { StateComponent state; DaemonLifecycleFsm lifecycleFsm; DaemonConfig config; config.dataDir = makeTempDir("yams_status_search_maintenance_"); config.searchMaintenance.automaticRebuildsEnabled = false; config.searchMaintenance.automaticRebuildsSource = "typed:test"; ServiceManager services(config, state, lifecycleFsm); DaemonMetrics metrics(nullptr, &state, &services, services.getWorkCoordinator()); RequestDispatcher dispatcher(nullptr, &services, &state, &metrics); const auto response = dispatchRequest(dispatcher, Request{StatusRequest{}}); REQUIRE(std::holds_alternative(response)); const auto& status = std::get(response); CHECK_FALSE(status.searchAutomaticRebuildsEnabled); CHECK((status.searchAutomaticRebuildsSource == "typed:test")); CHECK(status.runtimeTuning.contains("ipc.timeout_ms")); CHECK(status.runtimeTuning.contains("ipc.timeout_ms.source")); CHECK(status.runtimeTuning.contains("post_ingest.total_concurrent")); CHECK(status.runtimeTuning.contains("resource.memory_budget_bytes")); } 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(resp)); const auto& status = std::get(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(resp)); const auto& status = std::get(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(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, 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(dec.value().payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(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, 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(dec.value().payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(resp); CHECK(decoded.databasePhase == "ready"); CHECK(decoded.databasePhaseElapsedMs == 11); CHECK(decoded.maintenancePhase == "salvaging"); CHECK(decoded.maintenancePhaseElapsedMs == 222); } TEST_CASE("StatusResponse: host pressure fields round-trip", "[daemon][status][protocol][host_pressure]") { StatusResponse s{}; s.hostPressureSource = "psi"; s.hostPressureElevated = true; s.hostCpuPressurePct = 72.5; s.hostIoPressurePct = 3.25; s.hostMemoryPressurePct = 0.0; s.hostLoadPerCpu = 1.75; s.deferredBackgroundWork = "topology,repair"; Message m{}; m.payload = Response{std::in_place_type, 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(dec.value().payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(resp); CHECK(decoded.hostPressureSource == "psi"); CHECK(decoded.hostPressureElevated); CHECK(decoded.hostCpuPressurePct == Catch::Approx(72.5)); CHECK(decoded.hostIoPressurePct == Catch::Approx(3.25)); CHECK(decoded.hostMemoryPressurePct == Catch::Approx(0.0)); CHECK(decoded.hostLoadPerCpu == Catch::Approx(1.75)); CHECK(decoded.deferredBackgroundWork == "topology,repair"); // The piggyback keys must not leak into the numeric request counters. CHECK_FALSE(decoded.requestCounts.contains("host_pressure_source")); SECTION("an unmeasured host stays unmeasured") { StatusResponse none{}; Message m2{}; m2.payload = Response{std::in_place_type, none}; auto enc2 = ProtoSerializer::encode_payload(m2); REQUIRE(enc2.has_value()); auto dec2 = ProtoSerializer::decode_payload(enc2.value()); REQUIRE(dec2.has_value()); const auto& out = std::get(std::get(dec2.value().payload)); CHECK(out.hostPressureSource.empty()); CHECK_FALSE(out.hostPressureElevated); CHECK(out.hostCpuPressurePct < 0.0); CHECK(out.deferredBackgroundWork.empty()); } } TEST_CASE("StatusResponse: search maintenance provenance round-trips", "[daemon][status][protocol][search][config]") { StatusResponse status{}; status.searchAutomaticRebuildsEnabled = false; status.searchAutomaticRebuildsSource = "compatibility-environment:YAMS_DISABLE_SEARCH_REBUILDS"; status.runtimeTuning = {{"ipc.timeout_ms", "4321"}, {"ipc.timeout_ms.source", "config:tuning.ipc.timeout_ms"}}; Message message{}; message.payload = Response{std::in_place_type, status}; const auto encoded = ProtoSerializer::encode_payload(message); REQUIRE(encoded.has_value()); const auto decodedMessage = ProtoSerializer::decode_payload(encoded.value()); REQUIRE(decodedMessage.has_value()); const auto& response = std::get(decodedMessage.value().payload); REQUIRE(std::holds_alternative(response)); const auto& decoded = std::get(response); CHECK_FALSE(decoded.searchAutomaticRebuildsEnabled); CHECK((decoded.searchAutomaticRebuildsSource == "compatibility-environment:YAMS_DISABLE_SEARCH_REBUILDS")); CHECK((decoded.runtimeTuning.at("ipc.timeout_ms") == "4321")); CHECK((decoded.runtimeTuning.at("ipc.timeout_ms.source") == "config:tuning.ipc.timeout_ms")); } 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, 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(dec.value().payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(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, 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(dec.value().payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(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, 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(dec.value().payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(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; 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, 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(out.payload)); const auto& resp = std::get(out.payload); REQUIRE(std::holds_alternative(resp)); const auto& decoded = std::get(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, 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(dec.value().payload); REQUIRE(std::holds_alternative(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 REQUIRE(snap->workCoordinatorProgressProbesPosted > 0); CHECK(snap->workCoordinatorProgressProbesCompleted <= snap->workCoordinatorProgressProbesPosted); CHECK(snap->vectorCheckpointPhase == 0); // VectorCheckpointPhase::Idle } } // ============================================================================= // 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); } }