// Comprehensive daemon socket client integration test // Covers connection lifecycle, reconnection, timeouts, multiplexing, and error handling #define CATCH_CONFIG_MAIN #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "../../common/test_helpers_catch2.h" #include "test_async_helpers.h" #include "test_daemon_harness.h" #include #include #include #include #include #include #include #include #include using namespace yams::daemon; using namespace yams::test; using namespace std::chrono_literals; namespace { class EnvGuard { public: EnvGuard(const std::string& envName, const std::string& newValue) : name_(envName) { if (const char* orig = std::getenv(name_.c_str())) { originalValue_ = orig; } #ifdef _WIN32 _putenv_s(name_.c_str(), newValue.c_str()); #else setenv(name_.c_str(), newValue.c_str(), 1); #endif } ~EnvGuard() { if (originalValue_) { #ifdef _WIN32 _putenv_s(name_.c_str(), originalValue_->c_str()); #else setenv(name_.c_str(), originalValue_->c_str(), 1); #endif } else { #ifdef _WIN32 _putenv_s(name_.c_str(), ""); #else unsetenv(name_.c_str()); #endif } } private: std::string name_; std::optional originalValue_; }; class StubModelProvider : public yams::daemon::IModelProvider { public: yams::Result> generateEmbedding(const std::string& text) override { return std::vector(getEmbeddingDim(""), static_cast(text.size())); } yams::Result>> generateBatchEmbeddings(const std::vector& texts) override { std::vector> out; out.reserve(texts.size()); for (const auto& text : texts) { auto one = generateEmbedding(text); if (!one) { return one.error(); } out.push_back(std::move(one.value())); } return out; } yams::Result> generateEmbeddingFor(const std::string& modelName, const std::string& text) override { if (!isModelLoaded(modelName)) { return yams::Error{yams::ErrorCode::NotFound, "Requested model not loaded: " + modelName}; } return generateEmbedding(text); } yams::Result>> generateBatchEmbeddingsFor(const std::string& modelName, const std::vector& texts) override { if (!isModelLoaded(modelName)) { return yams::Error{yams::ErrorCode::NotFound, "Requested model not loaded: " + modelName}; } return generateBatchEmbeddings(texts); } yams::Result loadModel(const std::string& modelName) override { std::lock_guard lock(mutex_); lastLoadWithOptionsModel_.clear(); lastLoadWithOptionsJson_.clear(); if (!available_) { return yams::Error{yams::ErrorCode::InvalidState, "Provider unavailable"}; } if (std::find(loadedModels_.begin(), loadedModels_.end(), modelName) == loadedModels_.end()) { loadedModels_.push_back(modelName); } return yams::Result(); } yams::Result loadModelWithOptions(const std::string& modelName, const std::string& optionsJson) override { std::lock_guard lock(mutex_); if (!available_) { return yams::Error{yams::ErrorCode::InvalidState, "Provider unavailable"}; } lastLoadWithOptionsModel_ = modelName; lastLoadWithOptionsJson_ = optionsJson; if (std::find(loadedModels_.begin(), loadedModels_.end(), modelName) == loadedModels_.end()) { loadedModels_.push_back(modelName); } return yams::Result(); } yams::Result unloadModel(const std::string& modelName) override { std::lock_guard lock(mutex_); auto it = std::find(loadedModels_.begin(), loadedModels_.end(), modelName); if (it == loadedModels_.end()) { return yams::Error{yams::ErrorCode::NotFound, "Model not loaded: " + modelName}; } loadedModels_.erase(it); return yams::Result(); } bool isModelLoaded(const std::string& modelName) const override { std::lock_guard lock(mutex_); return std::find(loadedModels_.begin(), loadedModels_.end(), modelName) != loadedModels_.end(); } std::vector getLoadedModels() const override { std::lock_guard lock(mutex_); return loadedModels_; } size_t getLoadedModelCount() const override { std::lock_guard lock(mutex_); return loadedModels_.size(); } yams::Result getModelInfo(const std::string& modelName) const override { std::lock_guard lock(mutex_); if (std::find(loadedModels_.begin(), loadedModels_.end(), modelName) == loadedModels_.end()) { return yams::Error{yams::ErrorCode::NotFound, "Model not loaded: " + modelName}; } yams::daemon::ModelInfo info; info.name = modelName; info.embeddingDim = 384; info.maxSequenceLength = 512; info.memoryUsageBytes = 32 * 1024 * 1024; return info; } size_t getEmbeddingDim(const std::string&) const override { return 384; } std::shared_ptr getEmbeddingGenerator(const std::string& = "") override { return nullptr; } std::string getProviderName() const override { return "StubModelProvider"; } std::string getProviderVersion() const override { return "1.0-test"; } bool isAvailable() const override { std::lock_guard lock(mutex_); return available_; } size_t getMemoryUsage() const override { std::lock_guard lock(mutex_); return loadedModels_.size() * 32 * 1024 * 1024; } void releaseUnusedResources() override {} void shutdown() override { std::lock_guard lock(mutex_); available_ = false; } void setAvailable(bool available) { std::lock_guard lock(mutex_); available_ = available; } std::string lastLoadWithOptionsModel() const { std::lock_guard lock(mutex_); return lastLoadWithOptionsModel_; } std::string lastLoadWithOptionsJson() const { std::lock_guard lock(mutex_); return lastLoadWithOptionsJson_; } private: mutable std::mutex mutex_; bool available_{true}; std::vector loadedModels_; std::string lastLoadWithOptionsModel_; std::string lastLoadWithOptionsJson_; }; // Helper to create a client with custom config // Increased default timeouts to handle thread resource pressure from repeated daemon restarts DaemonClient createClient(const std::filesystem::path& socketPath, std::chrono::milliseconds connectTimeout = 5s, bool autoStart = false) { ClientConfig config; config.socketPath = socketPath; config.connectTimeout = connectTimeout; config.autoStart = autoStart; config.requestTimeout = 10s; return DaemonClient(config); } void startHarnessWithRetry(DaemonHarness& harness, int maxRetries = 3, std::chrono::milliseconds retryDelay = 250ms) { for (int attempt = 0; attempt < maxRetries; ++attempt) { if (harness.start(30s)) { return; } harness.stop(); std::this_thread::sleep_for(retryDelay); } SKIP("Skipping socket integration section due to daemon startup instability"); } // Helper to retry get request with delay - handles async post-ingest processing // streamingAddDocument returns immediately but document may not be available until // post-ingest completes (FTS5 indexing, metadata extraction, etc.) template yams::Result getWithRetry(Client& client, const std::string& hash, int maxRetries = 10, std::chrono::milliseconds retryDelay = 100ms) { GetRequest getReq; getReq.hash = hash; for (int i = 0; i < maxRetries; ++i) { auto getResult = yams::cli::run_sync(client.get(getReq), 5s); if (getResult.has_value()) { return getResult; } std::this_thread::sleep_for(retryDelay); } // Return last attempt's result return yams::cli::run_sync(client.get(getReq), 5s); } DownloadResponse waitForDownloadJobStatus(DaemonClient& client, const std::string& jobId, std::chrono::milliseconds timeout = 5s, std::chrono::milliseconds pollInterval = 25ms) { const auto deadline = std::chrono::steady_clock::now() + timeout; DownloadResponse latest; latest.jobId = jobId; latest.state = "unknown"; while (std::chrono::steady_clock::now() < deadline) { auto statusResult = yams::cli::run_sync(client.downloadStatus(DownloadStatusRequest{jobId}), 5s); if (statusResult.has_value()) { latest = statusResult.value(); if (latest.state != "queued" && latest.state != "running") { return latest; } } std::this_thread::sleep_for(pollInterval); } return latest; } std::filesystem::path createMockExternalPlugin(const std::filesystem::path& baseDir) { auto pluginDir = baseDir / "socket_mock_plugin"; std::filesystem::create_directories(pluginDir); { std::ofstream pluginFile(pluginDir / "plugin.py"); pluginFile << R"PY(#!/usr/bin/env python3 import json import sys def handle_request(req): method = req.get("method", "") if method == "handshake.manifest": return { "name": "socket_mock_plugin", "version": "1.0.0", "interfaces": ["content_extractor_v1"], "capabilities": { "content_extraction": { "formats": ["text/plain"], "extensions": [".txt"] } } } if method == "plugin.init": return {"status": "initialized"} if method == "plugin.health": return {"status": "ok"} if method == "plugin.shutdown": return {"status": "ok"} raise ValueError(f"Method not found: {method}") for line in sys.stdin: line = line.strip() if not line: continue req_id = None try: req = json.loads(line) req_id = req.get("id") response = {"jsonrpc": "2.0", "id": req_id, "result": handle_request(req)} except Exception as exc: response = { "jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": str(exc)} } print(json.dumps(response), flush=True) )PY"; } { std::ofstream manifestFile(pluginDir / "yams-plugin.json"); manifestFile << R"JSON({ "name": "socket_mock_plugin", "version": "1.0.0", "interfaces": ["content_extractor_v1"], "capabilities": { "content_extraction": { "formats": ["text/plain"], "extensions": [".txt"] } }, "entry": { "fallback_cmd": ["/usr/bin/env", "python3", "-u", "${plugin_dir}/plugin.py"], "env": { "PYTHONUNBUFFERED": "1" } } })JSON"; } return pluginDir; } bool containsCanonicalPath(const std::vector& paths, const std::filesystem::path& target) { std::error_code ec; auto targetCanonical = std::filesystem::weakly_canonical(target, ec); if (ec) { targetCanonical = target.lexically_normal(); } for (const auto& path : paths) { std::error_code itemEc; auto itemCanonical = std::filesystem::weakly_canonical(path, itemEc); if (itemEc) { itemCanonical = std::filesystem::path(path).lexically_normal(); } if (itemCanonical == targetCanonical) { return true; } } return false; } class SimpleHttpRedirectServer { public: explicit SimpleHttpRedirectServer(std::string location) : io_(), acceptor_(io_), location_(std::move(location)) { using boost::asio::ip::tcp; tcp::endpoint endpoint(boost::asio::ip::make_address("127.0.0.1"), 0); acceptor_.open(endpoint.protocol()); acceptor_.set_option(tcp::acceptor::reuse_address(true)); acceptor_.bind(endpoint); acceptor_.listen(); port_ = acceptor_.local_endpoint().port(); ready_.set_value(); thread_ = std::thread([this]() { serveOnce(); }); ready_.get_future().wait(); } ~SimpleHttpRedirectServer() { boost::system::error_code ec; acceptor_.close(ec); io_.stop(); if (thread_.joinable()) { thread_.join(); } } uint16_t port() const { return port_; } private: void serveOnce() { using boost::asio::ip::tcp; try { tcp::socket socket(io_); acceptor_.accept(socket); std::array buffer{}; boost::system::error_code readEc; (void)socket.read_some(boost::asio::buffer(buffer), readEc); std::ostringstream response; response << "HTTP/1.1 302 Found\r\n"; response << "Location: " << location_ << "\r\n"; response << "Content-Length: 0\r\n"; response << "Connection: close\r\n\r\n"; boost::system::error_code writeEc; boost::asio::write(socket, boost::asio::buffer(response.str()), writeEc); socket.shutdown(tcp::socket::shutdown_both, writeEc); socket.close(writeEc); } catch (...) { } } boost::asio::io_context io_; boost::asio::ip::tcp::acceptor acceptor_; std::string location_; uint16_t port_{0}; std::promise ready_; std::thread thread_; }; } // namespace TEST_CASE("Daemon socket connection lifecycle", "[daemon][socket][integration]") { SKIP_DAEMON_TEST_ON_WINDOWS(); SECTION("connects to running daemon") { DaemonHarness harness; startHarnessWithRetry(harness); auto client = createClient(harness.socketPath()); auto result = yams::cli::run_sync(client.connect(), 5s); REQUIRE(result.has_value()); REQUIRE(client.isConnected()); } SECTION("fails gracefully when daemon not running") { // Use a separate harness that won't start DaemonHarness harness; auto client = createClient(harness.socketPath(), 500ms); auto result = yams::cli::run_sync(client.connect(), 1s); REQUIRE(!result.has_value()); REQUIRE(!client.isConnected()); } SECTION("connects after daemon restart") { // This test needs its own daemon to restart DaemonHarness harness; startHarnessWithRetry(harness); auto client = createClient(harness.socketPath()); auto connectResult = yams::cli::run_sync(client.connect(), 5s); REQUIRE(connectResult.has_value()); REQUIRE(client.isConnected()); // Stop daemon - this resets GlobalIOContext, invalidating existing clients harness.stop(); std::this_thread::sleep_for(500ms); // Restart daemon startHarnessWithRetry(harness); // Create NEW client after restart (old client's io_context was reset) // Allow extra time for GlobalIOContext threads to fully stabilize after restart std::this_thread::sleep_for(200ms); // Use retry logic to handle potential timing issues with GlobalIOContext restart auto client2 = createClient(harness.socketPath()); yams::Result reconnectResult; int maxRetries = 3; for (int attempt = 0; attempt < maxRetries; ++attempt) { reconnectResult = yams::cli::run_sync(client2.connect(), 5s); if (reconnectResult.has_value()) { break; } spdlog::warn("[TEST] Reconnect attempt {} failed: {}, retrying...", attempt + 1, reconnectResult.error().message); std::this_thread::sleep_for(200ms); } if (!reconnectResult.has_value()) { spdlog::error("[TEST] All reconnect attempts failed: {}", reconnectResult.error().message); } REQUIRE(reconnectResult.has_value()); REQUIRE(client2.isConnected()); } SECTION("handles explicit disconnect") { DaemonHarness harness; startHarnessWithRetry(harness); auto client = createClient(harness.socketPath()); auto connectResult = yams::cli::run_sync(client.connect(), 5s); REQUIRE(connectResult.has_value()); REQUIRE(client.isConnected()); client.disconnect(); REQUIRE(!client.isConnected()); } SECTION("respects connection timeout") { // Use a separate harness that won't start DaemonHarness harness; auto client = createClient(harness.socketPath(), 100ms); auto start = std::chrono::steady_clock::now(); auto result = yams::cli::run_sync(client.connect(), 500ms); auto elapsed = std::chrono::steady_clock::now() - start; REQUIRE(!result.has_value()); REQUIRE(elapsed < 200ms); // Should fail fast with 100ms timeout } SECTION("survives repeated restart cycles") { DaemonHarness harness; constexpr int cycles = 2; // Reduced from 3 to avoid macOS thread limit for (int i = 0; i < cycles; ++i) { INFO("restart cycle " << i); startHarnessWithRetry(harness); // Allow time for GlobalIOContext to stabilize after restart std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); // Retry connection to handle timing issues yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); auto statusResult = yams::cli::run_sync(client.status(), 5s); REQUIRE(statusResult.has_value()); harness.stop(); // Allow GlobalIOContext to fully restart before next cycle std::this_thread::sleep_for(500ms); } } } TEST_CASE("Daemon client request execution", "[daemon][socket][requests]") { SKIP_DAEMON_TEST_ON_WINDOWS(); // Use isolated daemon for tests that modify database state DaemonHarness harness; startHarnessWithRetry(harness); // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); SECTION("status request succeeds") { auto statusResult = yams::cli::run_sync(client.status(), 5s); REQUIRE(statusResult.has_value()); auto& status = statusResult.value(); REQUIRE(status.readinessStates.at("ipc_server")); REQUIRE(status.readinessStates.at("content_store")); } SECTION("list request succeeds") { ListRequest req; req.limit = 10; auto listResult = yams::cli::run_sync(client.list(req), 5s); REQUIRE(listResult.has_value()); // Database may have documents from other tests - just verify response is valid } SECTION("add and retrieve document") { // Add a document with unique content AddDocumentRequest addReq; addReq.name = "test_request_execution.txt"; addReq.content = "Hello from request execution test!"; // Retry document add in case of transient connection issues yams::Result addResult; for (int attempt = 0; attempt < 3; ++attempt) { addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); if (addResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(addResult.has_value()); std::string docHash = addResult.value().hash; REQUIRE(!docHash.empty()); // Retrieve it (with retry since streamingAddDocument is async) auto getResult = getWithRetry(client, docHash); REQUIRE(getResult.has_value()); REQUIRE(getResult.value().content == "Hello from request execution test!"); REQUIRE(getResult.value().name == "test_request_execution.txt"); } SECTION("search for unique content") { SearchRequest searchReq; searchReq.query = "xyzzy_nonexistent_unique_string_12345"; // Retry search request in case of transient connection issues yams::Result searchResult; for (int attempt = 0; attempt < 3; ++attempt) { searchResult = yams::cli::run_sync(client.search(searchReq), 10s); if (searchResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(searchResult.has_value()); // Should return empty results for unique non-existent query REQUIRE(searchResult.value().results.empty()); } SECTION("grep with unique pattern") { GrepRequest grepReq; grepReq.pattern = "xyzzy_nonexistent_pattern_67890"; // Retry grep request in case of transient connection issues yams::Result grepResult; for (int attempt = 0; attempt < 3; ++attempt) { grepResult = yams::cli::run_sync(client.grep(grepReq), 10s); if (grepResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(grepResult.has_value()); // Should return empty matches for unique non-existent pattern REQUIRE(grepResult.value().matches.empty()); } SECTION("prepare session request succeeds") { PrepareSessionRequest req; req.sessionName = "default"; req.limit = 16; req.snippetLen = 48; auto result = yams::cli::run_sync(client.prepareSession(req), 10s); REQUIRE(result.has_value()); CHECK(result.value().warmedCount == 0); CHECK(result.value().message == "OK"); } SECTION("chunked get protocol succeeds by name") { const std::string docName = "test_chunk_protocol.txt"; const std::string content = "chunked retrieval content for request execution"; AddDocumentRequest addReq; addReq.name = docName; addReq.content = content; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); auto getResult = getWithRetry(client, addResult.value().hash); REQUIRE(getResult.has_value()); GetInitRequest initReq; initReq.name = docName; initReq.byName = true; initReq.chunkSize = 7; auto initResult = yams::cli::run_sync(client.getInit(initReq), 10s); REQUIRE(initResult.has_value()); REQUIRE(initResult.value().transferId != 0); CHECK(initResult.value().totalSize == content.size()); CHECK(initResult.value().chunkSize == 7); CHECK(initResult.value().metadata.contains("hash")); CHECK(initResult.value().metadata.at("hash") == addResult.value().hash); std::string assembled; uint64_t offset = 0; uint64_t remaining = initResult.value().totalSize; while (remaining > 0) { GetChunkRequest chunkReq{initResult.value().transferId, offset, 7}; auto chunkResult = yams::cli::run_sync(client.getChunk(chunkReq), 10s); REQUIRE(chunkResult.has_value()); assembled += chunkResult.value().data; offset += chunkResult.value().data.size(); remaining = chunkResult.value().bytesRemaining; } CHECK(assembled == content); auto endResult = yams::cli::run_sync(client.getEnd(GetEndRequest{initResult.value().transferId}), 10s); REQUIRE(endResult.has_value()); CHECK(endResult.value().message == "OK"); } SECTION("cat request returns stored content") { AddDocumentRequest addReq; addReq.name = "test_cat_request.txt"; addReq.content = "cat request body"; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE(getWithRetry(client, addResult.value().hash).has_value()); CatRequest catReq; catReq.hash = addResult.value().hash; auto catResult = yams::cli::run_sync(client.cat(catReq), 10s); REQUIRE(catResult.has_value()); CHECK(catResult.value().hash == addResult.value().hash); CHECK(catResult.value().name == "test_cat_request.txt"); CHECK(catResult.value().content.find("cat request body") != std::string::npos); } SECTION("update document request succeeds") { AddDocumentRequest addReq; addReq.name = "test_update_request.txt"; addReq.content = "update request body"; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE(getWithRetry(client, addResult.value().hash).has_value()); UpdateDocumentRequest updateReq; updateReq.hash = addResult.value().hash; updateReq.metadata["owner"] = "socket-test"; updateReq.addTags.push_back("request-execution"); auto updateResult = yams::cli::run_sync(client.updateDocument(updateReq), 10s); REQUIRE(updateResult.has_value()); CHECK((updateResult.value().contentUpdated || !updateResult.value().hash.empty() || updateResult.value().metadataUpdated || updateResult.value().tagsUpdated)); } SECTION("delete request succeeds") { AddDocumentRequest addReq; addReq.name = "test_delete_request.txt"; addReq.content = "delete request body"; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE(getWithRetry(client, addResult.value().hash).has_value()); DeleteRequest deleteReq; deleteReq.hash = addResult.value().hash; auto deleteResult = yams::cli::run_sync(client.call(deleteReq), 10s); REQUIRE(deleteResult.has_value()); CHECK_FALSE(deleteResult.value().dryRun); } SECTION("download request returns policy reminder") { DownloadRequest downloadReq; downloadReq.url = "https://example.com/archive.txt"; auto downloadResult = yams::cli::run_sync(client.call(downloadReq), 10s); REQUIRE(downloadResult.has_value()); CHECK(downloadResult.value().url == downloadReq.url); CHECK_FALSE(downloadResult.value().success); CHECK(downloadResult.value().error.find("Daemon download is disabled") != std::string::npos); } SECTION("cancel request reports missing request id") { auto cancelResult = yams::cli::run_sync(client.executeRequest(Request{CancelRequest{999999999ULL}}), 10s); REQUIRE_FALSE(cancelResult.has_value()); } SECTION("file history request reports document outside snapshots") { const std::string docName = "test_file_history_request.txt"; AddDocumentRequest addReq; addReq.name = docName; addReq.content = "file history request body"; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE(getWithRetry(client, addResult.value().hash).has_value()); FileHistoryRequest historyReq; historyReq.filepath = (std::filesystem::current_path() / docName).string(); auto historyResult = yams::cli::run_sync(client.fileHistory(historyReq), 10s); REQUIRE(historyResult.has_value()); CHECK(historyResult.value().filepath == std::filesystem::absolute(std::filesystem::current_path() / docName) .lexically_normal() .string()); CHECK(historyResult.value().totalVersions >= 1); CHECK_FALSE(historyResult.value().versions.empty()); } } TEST_CASE("Daemon client download job request execution", "[daemon][socket][download-jobs]") { SKIP_DAEMON_TEST_ON_WINDOWS(); ScopedEnvVar daemonDownloadEnabled("YAMS_ENABLE_DAEMON_DOWNLOAD", std::string("1")); DaemonHarnessOptions options; options.configureDaemon = [](yams::daemon::DaemonConfig& cfg) { cfg.downloadPolicy.enable = true; cfg.downloadPolicy.allowedHosts = {"127.0.0.1", "example.com"}; cfg.downloadPolicy.allowedSchemes = {"http", "https"}; cfg.downloadPolicy.requireChecksum = false; cfg.downloadPolicy.storeOnly = true; cfg.downloadPolicy.timeout = 5s; }; DaemonHarness harness(options); startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) { break; } std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); SECTION("download request creates tracked job and reaches completed status") { RequestDispatcher::__test_resetDownloadJobs(); RequestDispatcher::__test_forceDownloadServiceSuccessOnce(); DownloadRequest downloadReq; downloadReq.url = "https://example.com/archive.txt"; downloadReq.checksum = "sha256:abcd1234"; auto startResult = yams::cli::run_sync(client.call(downloadReq), 10s); REQUIRE(startResult.has_value()); CHECK(startResult.value().url == downloadReq.url); CHECK_FALSE(startResult.value().jobId.empty()); CHECK(startResult.value().state == "queued"); CHECK_FALSE(startResult.value().success); const auto finalStatus = waitForDownloadJobStatus(client, startResult.value().jobId); CHECK(finalStatus.jobId == startResult.value().jobId); CHECK(finalStatus.state == "completed"); CHECK(finalStatus.success); CHECK(finalStatus.hash == "sha256:forced-download-success"); CHECK(finalStatus.error.empty()); auto listResult = yams::cli::run_sync(client.listDownloadJobs(ListDownloadJobsRequest{10}), 10s); REQUIRE(listResult.has_value()); REQUIRE_FALSE(listResult.value().jobs.empty()); CHECK(std::any_of( listResult.value().jobs.begin(), listResult.value().jobs.end(), [&](const auto& job) { return job.jobId == startResult.value().jobId && job.state == "completed"; })); } SECTION("cancel download job transitions seeded running job to canceled") { RequestDispatcher::__test_resetDownloadJobs(); DownloadResponse seeded; seeded.url = "https://example.com/seeded-archive.txt"; seeded.jobId = "download-seeded-running"; seeded.state = "running"; seeded.success = false; seeded.createdAtMs = 111; seeded.updatedAtMs = 222; RequestDispatcher::__test_seedDownloadJob(seeded); yams::Result cancelResult; for (int attempt = 0; attempt < 3; ++attempt) { cancelResult = yams::cli::run_sync( client.cancelDownloadJob(CancelDownloadJobRequest{seeded.jobId}), 10s); if (cancelResult.has_value()) { break; } std::this_thread::sleep_for(200ms); } INFO("cancel download job error: " << (cancelResult.has_value() ? "" : cancelResult.error().message)); REQUIRE(cancelResult.has_value()); CHECK(cancelResult.value().jobId == seeded.jobId); CHECK(cancelResult.value().state == "canceled"); CHECK_FALSE(cancelResult.value().success); auto statusResult = yams::cli::run_sync(client.downloadStatus(DownloadStatusRequest{seeded.jobId}), 10s); REQUIRE(statusResult.has_value()); CHECK(statusResult.value().jobId == seeded.jobId); CHECK(statusResult.value().state == "canceled"); auto listResult = yams::cli::run_sync(client.listDownloadJobs(ListDownloadJobsRequest{10}), 10s); REQUIRE(listResult.has_value()); CHECK(std::any_of( listResult.value().jobs.begin(), listResult.value().jobs.end(), [&](const auto& job) { return job.jobId == seeded.jobId && job.state == "canceled"; })); } SECTION("redirecting download is blocked and surfaced to the client") { RequestDispatcher::__test_resetDownloadJobs(); SimpleHttpRedirectServer redirectServer("http://example.com/escaped.bin"); DownloadRequest downloadReq; downloadReq.url = "http://127.0.0.1:" + std::to_string(redirectServer.port()) + "/redirect.bin"; auto startResult = yams::cli::run_sync(client.call(downloadReq), 10s); REQUIRE(startResult.has_value()); CHECK(startResult.value().url == downloadReq.url); CHECK_FALSE(startResult.value().jobId.empty()); CHECK(startResult.value().state == "queued"); CHECK_FALSE(startResult.value().success); const auto finalStatus = waitForDownloadJobStatus(client, startResult.value().jobId, 10s); CHECK(finalStatus.jobId == startResult.value().jobId); CHECK(finalStatus.state == "failed"); CHECK_FALSE(finalStatus.success); CHECK(finalStatus.error == "Redirect blocked by policy during probe (follow_redirects=false)"); } } TEST_CASE("Daemon client model request execution", "[daemon][socket][requests][model]") { SKIP_DAEMON_TEST_ON_WINDOWS(); EnvGuard backendGuard("YAMS_EMBED_BACKEND", "daemon"); DaemonHarnessOptions options; options.useMockModelProvider = false; DaemonHarness harness(options); startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); const auto* daemon = harness.daemon(); REQUIRE(daemon != nullptr); auto* serviceManager = const_cast(daemon->getServiceManager()); REQUIRE(serviceManager != nullptr); ModelStatusRequest statusReq; auto statusResult = yams::cli::run_sync(client.executeRequest(Request{statusReq}), 5s); REQUIRE(statusResult.has_value()); REQUIRE(std::holds_alternative(statusResult.value())); CHECK(std::get(statusResult.value()).models.empty()); LoadModelRequest emptyLoadReq; auto emptyLoadResult = yams::cli::run_sync(client.executeRequest(Request{emptyLoadReq}), 10s); REQUIRE(emptyLoadResult.has_value()); REQUIRE(std::holds_alternative(emptyLoadResult.value())); CHECK(std::get(emptyLoadResult.value()).code == yams::ErrorCode::InvalidState); LoadModelRequest loadReq; loadReq.modelName = "test-model-that-does-not-exist"; auto loadResult = yams::cli::run_sync(client.executeRequest(Request{loadReq}), 10s); REQUIRE(loadResult.has_value()); REQUIRE(std::holds_alternative(loadResult.value())); CHECK(std::get(loadResult.value()).code == yams::ErrorCode::InvalidState); auto injectedProvider = std::make_shared(); REQUIRE(injectedProvider != nullptr); serviceManager->__test_setModelProvider(injectedProvider); LoadModelRequest emptyLoadWithProviderReq; auto emptyLoadWithProviderResult = yams::cli::run_sync(client.executeRequest(Request{emptyLoadWithProviderReq}), 10s); REQUIRE(emptyLoadWithProviderResult.has_value()); REQUIRE(std::holds_alternative(emptyLoadWithProviderResult.value())); CHECK(std::get(emptyLoadWithProviderResult.value()).code == yams::ErrorCode::InvalidData); LoadModelRequest optionsLoadReq; optionsLoadReq.modelName = "dispatcher-model-with-options"; optionsLoadReq.optionsJson = R"({"revision":"test","offline":true})"; auto optionsLoadResult = yams::cli::run_sync(client.loadModel(optionsLoadReq), 10s); REQUIRE(optionsLoadResult.has_value()); CHECK(optionsLoadResult.value().success); CHECK(optionsLoadResult.value().modelName == "dispatcher-model-with-options"); CHECK(injectedProvider->isModelLoaded("dispatcher-model-with-options")); CHECK(injectedProvider->lastLoadWithOptionsModel() == "dispatcher-model-with-options"); CHECK(injectedProvider->lastLoadWithOptionsJson() == optionsLoadReq.optionsJson); LoadModelRequest successLoadReq; successLoadReq.modelName = "dispatcher-model"; auto successLoadResult = yams::cli::run_sync(client.loadModel(successLoadReq), 10s); REQUIRE(successLoadResult.has_value()); CHECK(successLoadResult.value().success); CHECK(successLoadResult.value().modelName == "dispatcher-model"); CHECK(injectedProvider->isModelLoaded("dispatcher-model")); ModelStatusRequest loadedStatusReq; loadedStatusReq.modelName = "dispatcher-model"; auto loadedStatusResult = yams::cli::run_sync(client.getModelStatus(loadedStatusReq), 5s); REQUIRE(loadedStatusResult.has_value()); REQUIRE(loadedStatusResult.value().models.size() == 1); CHECK(loadedStatusResult.value().models.front().name == "dispatcher-model"); CHECK(loadedStatusResult.value().models.front().loaded); ModelStatusRequest missingStatusReq; missingStatusReq.modelName = "missing-dispatcher-model"; auto missingStatusResult = yams::cli::run_sync(client.getModelStatus(missingStatusReq), 5s); REQUIRE(missingStatusResult.has_value()); CHECK(missingStatusResult.value().models.empty()); UnloadModelRequest unloadReq; unloadReq.modelName = "test-model-that-does-not-exist"; auto unloadResult = yams::cli::run_sync(client.executeRequest(Request{unloadReq}), 5s); REQUIRE(unloadResult.has_value()); REQUIRE(std::holds_alternative(unloadResult.value())); CHECK(std::get(unloadResult.value()).code == yams::ErrorCode::NotFound); UnloadModelRequest successUnloadReq; successUnloadReq.modelName = "dispatcher-model"; auto successUnloadResult = yams::cli::run_sync(client.unloadModel(successUnloadReq), 5s); REQUIRE(successUnloadResult.has_value()); CHECK(successUnloadResult.value().message == "Model unloaded"); CHECK_FALSE(injectedProvider->isModelLoaded("dispatcher-model")); UnloadModelRequest successOptionsUnloadReq; successOptionsUnloadReq.modelName = "dispatcher-model-with-options"; auto successOptionsUnloadResult = yams::cli::run_sync(client.unloadModel(successOptionsUnloadReq), 5s); REQUIRE(successOptionsUnloadResult.has_value()); CHECK(successOptionsUnloadResult.value().message == "Model unloaded"); CHECK_FALSE(injectedProvider->isModelLoaded("dispatcher-model-with-options")); injectedProvider->setAvailable(false); ModelStatusRequest unavailableStatusReq; auto unavailableStatusResult = yams::cli::run_sync(client.getModelStatus(unavailableStatusReq), 5s); REQUIRE(unavailableStatusResult.has_value()); CHECK(unavailableStatusResult.value().models.empty()); CHECK(unavailableStatusResult.value().totalMemoryMb == 0); } TEST_CASE("Daemon client plugin request execution", "[daemon][socket][requests][plugin]") { SKIP_DAEMON_TEST_ON_WINDOWS(); const auto uniqueId = std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()); const auto pluginRoot = std::filesystem::temp_directory_path() / ("yams_socket_plugin_" + uniqueId); const auto emptyTrustedDir = pluginRoot / "trusted_empty"; std::filesystem::create_directories(emptyTrustedDir); const auto pluginDir = createMockExternalPlugin(pluginRoot); DaemonHarnessOptions options; options.isolateState = true; options.trustedPluginPaths = {pluginDir}; DaemonHarness harness(options); startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); PluginScanRequest scanReq; scanReq.target = pluginDir.string(); auto scanResult = yams::cli::run_sync(client.call(scanReq), 10s); REQUIRE(scanResult.has_value()); REQUIRE(scanResult.value().plugins.size() == 1); CHECK(scanResult.value().plugins.front().name == "socket_mock_plugin"); CHECK(scanResult.value().plugins.front().path == pluginDir.string()); PluginScanRequest dirScanReq; dirScanReq.dir = pluginRoot.string(); auto dirScanResult = yams::cli::run_sync(client.call(dirScanReq), 10s); REQUIRE(dirScanResult.has_value()); REQUIRE(dirScanResult.value().plugins.size() == 1); CHECK(dirScanResult.value().plugins.front().name == "socket_mock_plugin"); auto initialTrustList = yams::cli::run_sync(client.call(PluginTrustListRequest{}), 5s); REQUIRE(initialTrustList.has_value()); CHECK(containsCanonicalPath(initialTrustList.value().paths, pluginDir)); PluginTrustAddRequest trustAddReq; trustAddReq.path = emptyTrustedDir.string(); auto trustAddResult = yams::cli::run_sync(client.executeRequest(Request{trustAddReq}), 5s); REQUIRE(trustAddResult.has_value()); REQUIRE(std::holds_alternative(trustAddResult.value())); CHECK(std::get(trustAddResult.value()).message == "ok"); auto trustListAfterAdd = yams::cli::run_sync(client.call(PluginTrustListRequest{}), 5s); REQUIRE(trustListAfterAdd.has_value()); CHECK(containsCanonicalPath(trustListAfterAdd.value().paths, emptyTrustedDir)); PluginLoadRequest dryRunReq; dryRunReq.pathOrName = pluginDir.string(); dryRunReq.dryRun = true; auto dryRunResult = yams::cli::run_sync(client.call(dryRunReq), 10s); REQUIRE(dryRunResult.has_value()); CHECK_FALSE(dryRunResult.value().loaded); CHECK(dryRunResult.value().message == "dry-run"); CHECK(dryRunResult.value().record.name == "socket_mock_plugin"); PluginLoadRequest loadReq; loadReq.pathOrName = pluginDir.string(); auto loadResult = yams::cli::run_sync(client.call(loadReq), 10s); REQUIRE(loadResult.has_value()); CHECK(loadResult.value().loaded); CHECK(loadResult.value().message == "loaded"); CHECK(loadResult.value().record.name == "socket_mock_plugin"); PluginUnloadRequest unloadLoadedReq; unloadLoadedReq.name = loadResult.value().record.name; auto unloadLoadedResult = yams::cli::run_sync(client.executeRequest(Request{unloadLoadedReq}), 5s); REQUIRE(unloadLoadedResult.has_value()); REQUIRE(std::holds_alternative(unloadLoadedResult.value())); CHECK(std::get(unloadLoadedResult.value()).message == "unloaded"); PluginUnloadRequest unloadReq; unloadReq.name = "definitely_missing_plugin_for_dispatcher_coverage"; auto unloadResult = yams::cli::run_sync(client.executeRequest(Request{unloadReq}), 5s); REQUIRE(unloadResult.has_value()); REQUIRE(std::holds_alternative(unloadResult.value())); CHECK(std::get(unloadResult.value()).message == "unloaded"); PluginScanRequest missingScanReq; missingScanReq.target = (pluginRoot / "missing_plugin_dir").string(); auto missingScanResult = yams::cli::run_sync(client.executeRequest(Request{missingScanReq}), 10s); REQUIRE(missingScanResult.has_value()); REQUIRE(std::holds_alternative(missingScanResult.value())); CHECK(std::get(missingScanResult.value()).code == yams::ErrorCode::NotFound); PluginTrustRemoveRequest trustRemoveReq; trustRemoveReq.path = emptyTrustedDir.string(); auto trustRemoveResult = yams::cli::run_sync(client.executeRequest(Request{trustRemoveReq}), 5s); REQUIRE(trustRemoveResult.has_value()); REQUIRE(std::holds_alternative(trustRemoveResult.value())); CHECK(std::get(trustRemoveResult.value()).message == "ok"); auto trustListAfterRemove = yams::cli::run_sync(client.call(PluginTrustListRequest{}), 5s); REQUIRE(trustListAfterRemove.has_value()); CHECK_FALSE(containsCanonicalPath(trustListAfterRemove.value().paths, emptyTrustedDir)); } TEST_CASE("Daemon client prune request execution", "[daemon][socket][requests][prune]") { SKIP_DAEMON_TEST_ON_WINDOWS(); DaemonHarness harness; startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); const auto* daemon = harness.daemon(); REQUIRE(daemon != nullptr); auto* serviceManager = daemon->getServiceManager(); REQUIRE(serviceManager != nullptr); auto metadataRepo = serviceManager->getMetadataRepo(); REQUIRE(metadataRepo != nullptr); AddDocumentRequest addReq; addReq.name = "dispatcher_coverage.log"; addReq.content = "prune coverage content\n"; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE_FALSE(addResult.value().hash.empty()); auto getResult = getWithRetry(client, addResult.value().hash); REQUIRE(getResult.has_value()); AddDocumentRequest largeReq; largeReq.name = "dispatcher_coverage_large.log"; largeReq.content = std::string(4096, 'L'); auto largeResult = yams::cli::run_sync(client.streamingAddDocument(largeReq), 10s); REQUIRE(largeResult.has_value()); REQUIRE_FALSE(largeResult.value().hash.empty()); REQUIRE(getWithRetry(client, largeResult.value().hash, 50, 200ms).has_value()); auto smallDocResult = metadataRepo->getDocumentByHash(addResult.value().hash); REQUIRE(smallDocResult.has_value()); REQUIRE(smallDocResult.value().has_value()); const auto smallDoc = *smallDocResult.value(); auto largeDocResult = metadataRepo->getDocumentByHash(largeResult.value().hash); REQUIRE(largeDocResult.has_value()); REQUIRE(largeDocResult.value().has_value()); auto largeDoc = *largeDocResult.value(); largeDoc.modifiedTime = std::chrono::time_point_cast( std::chrono::system_clock::now() - std::chrono::hours(48)); REQUIRE(metadataRepo->updateDocument(largeDoc).has_value()); PruneRequest pruneReq; pruneReq.extensions = {"log"}; pruneReq.smallerThan = "1MB"; pruneReq.dryRun = true; pruneReq.verbose = true; auto pruneResult = yams::cli::run_sync(client.call(pruneReq), 30s); REQUIRE(pruneResult.has_value()); CHECK(pruneResult.value().filesDeleted == 0); CHECK(pruneResult.value().filesFailed == 0); CHECK(pruneResult.value().totalBytesFreed >= static_cast(smallDoc.fileSize + largeDoc.fileSize)); CHECK_FALSE(pruneResult.value().categoryCounts.empty()); PruneRequest olderThanReq; olderThanReq.extensions = {"log"}; olderThanReq.olderThan = "1d"; olderThanReq.dryRun = true; auto olderThanResult = yams::cli::run_sync(client.call(olderThanReq), 30s); REQUIRE(olderThanResult.has_value()); CHECK(olderThanResult.value().filesDeleted == 0); CHECK(olderThanResult.value().filesFailed == 0); CHECK(olderThanResult.value().totalBytesFreed == static_cast(largeDoc.fileSize)); PruneRequest unknownAgeUnitReq; unknownAgeUnitReq.extensions = {"log"}; unknownAgeUnitReq.olderThan = "1q"; unknownAgeUnitReq.dryRun = true; auto unknownAgeUnitResult = yams::cli::run_sync(client.call(unknownAgeUnitReq), 30s); REQUIRE(unknownAgeUnitResult.has_value()); CHECK(unknownAgeUnitResult.value().filesDeleted == 0); CHECK(unknownAgeUnitResult.value().filesFailed == 0); CHECK(unknownAgeUnitResult.value().totalBytesFreed >= static_cast(smallDoc.fileSize + largeDoc.fileSize)); PruneRequest largerThanReq; largerThanReq.extensions = {"log"}; largerThanReq.largerThan = "1KB"; largerThanReq.dryRun = true; auto largerThanResult = yams::cli::run_sync(client.call(largerThanReq), 30s); REQUIRE(largerThanResult.has_value()); CHECK(largerThanResult.value().filesDeleted == 0); CHECK(largerThanResult.value().filesFailed == 0); CHECK(largerThanResult.value().totalBytesFreed == static_cast(largeDoc.fileSize)); PruneRequest applyReq; applyReq.extensions = {"log"}; applyReq.olderThan = "1d"; applyReq.largerThan = "1KB"; applyReq.dryRun = false; auto applyResult = yams::cli::run_sync(client.call(applyReq), 30s); REQUIRE(applyResult.has_value()); CHECK(applyResult.value().filesDeleted == 1); CHECK(applyResult.value().filesFailed == 0); CHECK(applyResult.value().totalBytesFreed == static_cast(largeDoc.fileSize)); auto largeDocAfterPrune = metadataRepo->getDocumentByHash(largeResult.value().hash); REQUIRE(largeDocAfterPrune.has_value()); CHECK_FALSE(largeDocAfterPrune.value().has_value()); auto smallDocAfterPrune = metadataRepo->getDocumentByHash(addResult.value().hash); REQUIRE(smallDocAfterPrune.has_value()); CHECK(smallDocAfterPrune.value().has_value()); } TEST_CASE("Daemon client graph maintenance request execution", "[daemon][socket][requests][graph]") { SKIP_DAEMON_TEST_ON_WINDOWS(); DaemonHarness harness; startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); GraphRepairRequest repairReq; repairReq.dryRun = true; auto graphRepairResult = yams::cli::run_sync(client.graphRepair(repairReq), 30s); REQUIRE(graphRepairResult.has_value()); CHECK(graphRepairResult.value().dryRun); CHECK(graphRepairResult.value().errors == 0); CHECK(std::any_of( graphRepairResult.value().issues.begin(), graphRepairResult.value().issues.end(), [](const auto& issue) { return issue.find("dry-run") != std::string::npos; })); GraphValidateRequest validateReq; auto graphValidateResult = yams::cli::run_sync(client.graphValidate(validateReq), 30s); REQUIRE(graphValidateResult.has_value()); CHECK(graphValidateResult.value().totalNodes >= 0); CHECK(graphValidateResult.value().totalEdges >= 0); CHECK(graphValidateResult.value().issues.empty()); AddDocumentRequest addReq; addReq.name = "graph_dispatcher_coverage.txt"; addReq.content = "graph maintenance coverage content\n"; addReq.tags = {"graph-coverage", "dispatcher"}; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE_FALSE(addResult.value().hash.empty()); REQUIRE(getWithRetry(client, addResult.value().hash).has_value()); GraphRepairRequest applyRepairReq; applyRepairReq.dryRun = false; auto applyRepairResult = yams::cli::run_sync(client.graphRepair(applyRepairReq), 30s); REQUIRE(applyRepairResult.has_value()); CHECK_FALSE(applyRepairResult.value().dryRun); CHECK(applyRepairResult.value().errors == 0); CHECK(applyRepairResult.value().nodesCreated >= 1); CHECK(applyRepairResult.value().edgesCreated >= 1); auto postRepairValidateResult = yams::cli::run_sync(client.graphValidate(validateReq), 30s); REQUIRE(postRepairValidateResult.has_value()); CHECK(postRepairValidateResult.value().totalNodes >= 1); CHECK(postRepairValidateResult.value().totalEdges >= 1); } TEST_CASE("Daemon client collection request execution", "[daemon][socket][requests][collections]") { SKIP_DAEMON_TEST_ON_WINDOWS(); DaemonHarness harness; startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); const auto* daemon = harness.daemon(); REQUIRE(daemon != nullptr); auto* serviceManager = daemon->getServiceManager(); REQUIRE(serviceManager != nullptr); const std::string collectionName = "dispatcher-coverage-collection"; const std::string snapshotId = "dispatcher-coverage-snapshot"; AddDocumentRequest addReq; addReq.name = "dispatcher_collection_coverage.txt"; addReq.content = "collection coverage content\n"; addReq.collection = collectionName; addReq.snapshotId = snapshotId; addReq.snapshotLabel = "Dispatcher coverage snapshot"; auto addResult = yams::cli::run_sync(client.streamingAddDocument(addReq), 10s); REQUIRE(addResult.has_value()); REQUIRE_FALSE(addResult.value().hash.empty()); auto getResult = getWithRetry(client, addResult.value().hash); REQUIRE(getResult.has_value()); ListSnapshotsRequest listReq; auto listResult = yams::cli::run_sync(client.call(listReq), 10s); REQUIRE(listResult.has_value()); CHECK(listResult.value().totalCount >= 1); CHECK(std::any_of(listResult.value().snapshots.begin(), listResult.value().snapshots.end(), [&](const SnapshotInfo& info) { return info.id == snapshotId; })); MetadataValueCountsRequest countsReq; countsReq.keys = {"collection", "snapshot_id"}; auto countsResult = yams::cli::run_sync(client.call(countsReq), 10s); REQUIRE(countsResult.has_value()); REQUIRE(countsResult.value().valueCounts.count("collection") == 1); REQUIRE(countsResult.value().valueCounts.count("snapshot_id") == 1); CHECK(std::any_of( countsResult.value().valueCounts.at("collection").begin(), countsResult.value().valueCounts.at("collection").end(), [&](const auto& entry) { return entry.first == collectionName && entry.second >= 1; })); CHECK(std::any_of( countsResult.value().valueCounts.at("snapshot_id").begin(), countsResult.value().valueCounts.at("snapshot_id").end(), [&](const auto& entry) { return entry.first == snapshotId && entry.second >= 1; })); MetadataValueCountsRequest emptyCountsReq; auto emptyCountsResult = yams::cli::run_sync(client.call(emptyCountsReq), 10s); REQUIRE(emptyCountsResult.has_value()); CHECK(emptyCountsResult.value().valueCounts.empty()); RestoreCollectionRequest missingCollectionReq; missingCollectionReq.collection = "dispatcher-missing-collection"; missingCollectionReq.outputDirectory = "collection-restore-missing"; missingCollectionReq.dryRun = true; auto missingCollectionResult = yams::cli::run_sync(client.call(missingCollectionReq), 30s); REQUIRE(missingCollectionResult.has_value()); CHECK(missingCollectionResult.value().dryRun); CHECK(missingCollectionResult.value().filesRestored == 0); CHECK(missingCollectionResult.value().files.empty()); RestoreCollectionRequest restoreCollectionReq; restoreCollectionReq.collection = collectionName; restoreCollectionReq.outputDirectory = "collection-restore-out"; restoreCollectionReq.layoutTemplate = "{name}{ext}"; restoreCollectionReq.dryRun = true; auto restoreCollectionResult = yams::cli::run_sync(client.call(restoreCollectionReq), 30s); REQUIRE(restoreCollectionResult.has_value()); CHECK(restoreCollectionResult.value().dryRun); CHECK(restoreCollectionResult.value().filesRestored == 1); REQUIRE(restoreCollectionResult.value().files.size() == 1); CHECK(restoreCollectionResult.value().files.front().hash == addResult.value().hash); CHECK_FALSE(restoreCollectionResult.value().files.front().skipped); CHECK(restoreCollectionResult.value().files.front().path.find( "dispatcher_collection_coverage") != std::string::npos); const auto existingCollectionOut = std::filesystem::temp_directory_path() / "yams_dispatcher_collection_existing.txt"; { std::ofstream existing(existingCollectionOut); REQUIRE(existing.good()); existing << "existing collection content"; } RestoreCollectionRequest existingCollectionReq; existingCollectionReq.collection = collectionName; existingCollectionReq.outputDirectory = existingCollectionOut.parent_path().string(); existingCollectionReq.layoutTemplate = existingCollectionOut.filename().string(); existingCollectionReq.dryRun = false; auto existingCollectionResult = yams::cli::run_sync(client.call(existingCollectionReq), 30s); REQUIRE(existingCollectionResult.has_value()); REQUIRE(existingCollectionResult.value().files.size() == 1); CHECK(existingCollectionResult.value().filesRestored == 0); CHECK(existingCollectionResult.value().files.front().skipped); CHECK(existingCollectionResult.value().files.front().skipReason == "File exists (overwrite=false)"); auto metadataRepo = serviceManager->getMetadataRepo(); REQUIRE(metadataRepo != nullptr); auto originalCollectionDocResult = metadataRepo->getDocumentByHash(addResult.value().hash); REQUIRE(originalCollectionDocResult.has_value()); REQUIRE(originalCollectionDocResult.value().has_value()); const auto originalCollectionDoc = *originalCollectionDocResult.value(); auto missingCollectionDoc = originalCollectionDoc; missingCollectionDoc.sha256Hash = std::string(64, 'a'); REQUIRE(metadataRepo->updateDocument(missingCollectionDoc).has_value()); const auto missingCollectionOut = std::filesystem::temp_directory_path() / "yams_dispatcher_missing_collection_root" / "missing_collection" / "restored.txt"; RestoreCollectionRequest missingContentCollectionReq; missingContentCollectionReq.collection = collectionName; missingContentCollectionReq.outputDirectory = (missingCollectionOut.parent_path().parent_path()).string(); missingContentCollectionReq.layoutTemplate = "missing_collection/restored.txt"; missingContentCollectionReq.overwrite = true; missingContentCollectionReq.dryRun = false; auto missingContentCollectionResult = yams::cli::run_sync( client.call(missingContentCollectionReq), 30s); REQUIRE(missingContentCollectionResult.has_value()); REQUIRE(missingContentCollectionResult.value().files.size() == 1); CHECK(missingContentCollectionResult.value().filesRestored == 0); CHECK(missingContentCollectionResult.value().files.front().skipped); CHECK(missingContentCollectionResult.value().files.front().skipReason.find( "Failed to retrieve content") != std::string::npos); REQUIRE(metadataRepo->updateDocument(originalCollectionDoc).has_value()); RestoreCollectionRequest filteredCollectionReq; filteredCollectionReq.collection = collectionName; filteredCollectionReq.outputDirectory = "collection-restore-filtered"; filteredCollectionReq.includePatterns = {"*.md"}; filteredCollectionReq.dryRun = true; auto filteredCollectionResult = yams::cli::run_sync(client.call(filteredCollectionReq), 30s); REQUIRE(filteredCollectionResult.has_value()); CHECK(filteredCollectionResult.value().filesRestored == 0); CHECK(filteredCollectionResult.value().files.empty()); RestoreCollectionRequest invalidCollectionReq; invalidCollectionReq.outputDirectory = "collection-restore-invalid"; invalidCollectionReq.dryRun = true; auto invalidCollectionResult = yams::cli::run_sync(client.executeRequest(Request{invalidCollectionReq}), 30s); REQUIRE(invalidCollectionResult.has_value()); REQUIRE(std::holds_alternative(invalidCollectionResult.value())); CHECK(std::get(invalidCollectionResult.value()).code == yams::ErrorCode::InvalidArgument); RestoreSnapshotRequest restoreSnapshotReq; restoreSnapshotReq.snapshotId = snapshotId; restoreSnapshotReq.outputDirectory = "snapshot-restore-out"; restoreSnapshotReq.layoutTemplate = "{hash}{ext}"; restoreSnapshotReq.dryRun = true; auto restoreSnapshotResult = yams::cli::run_sync(client.call(restoreSnapshotReq), 30s); REQUIRE(restoreSnapshotResult.has_value()); CHECK(restoreSnapshotResult.value().dryRun); CHECK(restoreSnapshotResult.value().filesRestored == 1); REQUIRE(restoreSnapshotResult.value().files.size() == 1); CHECK(restoreSnapshotResult.value().files.front().hash == addResult.value().hash); CHECK_FALSE(restoreSnapshotResult.value().files.front().skipped); CHECK(restoreSnapshotResult.value().files.front().path.find(addResult.value().hash) != std::string::npos); const auto existingSnapshotOut = std::filesystem::temp_directory_path() / "yams_dispatcher_snapshot_existing.txt"; { std::ofstream existing(existingSnapshotOut); REQUIRE(existing.good()); existing << "existing snapshot content"; } RestoreSnapshotRequest existingSnapshotReq; existingSnapshotReq.snapshotId = snapshotId; existingSnapshotReq.outputDirectory = existingSnapshotOut.parent_path().string(); existingSnapshotReq.layoutTemplate = existingSnapshotOut.filename().string(); existingSnapshotReq.dryRun = false; auto existingSnapshotResult = yams::cli::run_sync(client.call(existingSnapshotReq), 30s); REQUIRE(existingSnapshotResult.has_value()); REQUIRE(existingSnapshotResult.value().files.size() == 1); CHECK(existingSnapshotResult.value().filesRestored == 0); CHECK(existingSnapshotResult.value().files.front().skipped); CHECK(existingSnapshotResult.value().files.front().skipReason == "File exists (overwrite=false)"); auto originalSnapshotDocResult = metadataRepo->getDocumentByHash(addResult.value().hash); REQUIRE(originalSnapshotDocResult.has_value()); REQUIRE(originalSnapshotDocResult.value().has_value()); const auto originalSnapshotDoc = *originalSnapshotDocResult.value(); auto missingSnapshotDoc = originalSnapshotDoc; missingSnapshotDoc.sha256Hash = std::string(64, 'b'); REQUIRE(metadataRepo->updateDocument(missingSnapshotDoc).has_value()); const auto missingSnapshotOut = std::filesystem::temp_directory_path() / "yams_dispatcher_missing_snapshot_root" / "missing_snapshot" / "restored.txt"; RestoreSnapshotRequest missingContentSnapshotReq; missingContentSnapshotReq.snapshotId = snapshotId; missingContentSnapshotReq.outputDirectory = (missingSnapshotOut.parent_path().parent_path()).string(); missingContentSnapshotReq.layoutTemplate = "missing_snapshot/restored.txt"; missingContentSnapshotReq.overwrite = true; missingContentSnapshotReq.dryRun = false; auto missingContentSnapshotResult = yams::cli::run_sync(client.call(missingContentSnapshotReq), 30s); REQUIRE(missingContentSnapshotResult.has_value()); REQUIRE(missingContentSnapshotResult.value().files.size() == 1); CHECK(missingContentSnapshotResult.value().filesRestored == 0); CHECK(missingContentSnapshotResult.value().files.front().skipped); CHECK(missingContentSnapshotResult.value().files.front().skipReason.find( "Failed to retrieve content") != std::string::npos); REQUIRE(metadataRepo->updateDocument(originalSnapshotDoc).has_value()); RestoreSnapshotRequest filteredSnapshotReq; filteredSnapshotReq.snapshotId = snapshotId; filteredSnapshotReq.outputDirectory = "snapshot-restore-filtered"; filteredSnapshotReq.excludePatterns = {"*coverage.txt"}; filteredSnapshotReq.dryRun = true; auto filteredSnapshotResult = yams::cli::run_sync(client.call(filteredSnapshotReq), 30s); REQUIRE(filteredSnapshotResult.has_value()); CHECK(filteredSnapshotResult.value().filesRestored == 0); CHECK(filteredSnapshotResult.value().files.empty()); RestoreSnapshotRequest missingSnapshotReq; missingSnapshotReq.snapshotId = "dispatcher-missing-snapshot"; missingSnapshotReq.outputDirectory = "snapshot-restore-missing"; missingSnapshotReq.dryRun = true; auto missingSnapshotResult = yams::cli::run_sync(client.call(missingSnapshotReq), 30s); REQUIRE(missingSnapshotResult.has_value()); CHECK(missingSnapshotResult.value().filesRestored == 0); CHECK(missingSnapshotResult.value().files.empty()); RestoreSnapshotRequest invalidSnapshotReq; invalidSnapshotReq.outputDirectory = "snapshot-restore-invalid"; invalidSnapshotReq.dryRun = true; auto invalidSnapshotResult = yams::cli::run_sync(client.executeRequest(Request{invalidSnapshotReq}), 30s); REQUIRE(invalidSnapshotResult.has_value()); REQUIRE(std::holds_alternative(invalidSnapshotResult.value())); CHECK(std::get(invalidSnapshotResult.value()).code == yams::ErrorCode::InvalidArgument); } TEST_CASE("Daemon client session request execution", "[daemon][socket][requests][session]") { SKIP_DAEMON_TEST_ON_WINDOWS(); DaemonHarness harness({.isolateState = true}); startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); const auto* daemon = harness.daemon(); REQUIRE(daemon != nullptr); auto* serviceManager = daemon->getServiceManager(); REQUIRE(serviceManager != nullptr); auto appContext = serviceManager->getAppContext(); auto sessionService = yams::app::services::makeSessionService(&appContext); REQUIRE(sessionService); const std::string sessionA = "dispatcher-session-a"; const std::string sessionB = "dispatcher-session-b"; const std::string selectorPath = "src/**/*.cpp"; sessionService->init(sessionA, "dispatcher coverage session A"); sessionService->create(sessionB, "dispatcher coverage session B"); auto listResult = yams::cli::run_sync(client.executeRequest(Request{ListSessionsRequest{}}), 10s); REQUIRE(listResult.has_value()); REQUIRE(std::holds_alternative(listResult.value())); const auto& listResp = std::get(listResult.value()); std::set sessionNames(listResp.session_names.begin(), listResp.session_names.end()); CHECK(sessionNames.contains(sessionA)); CHECK(sessionNames.contains(sessionB)); CHECK(listResp.current_session == sessionA); UseSessionRequest missingReq; missingReq.session_name = "dispatcher-session-missing"; auto missingResult = yams::cli::run_sync(client.executeRequest(Request{missingReq}), 10s); REQUIRE(missingResult.has_value()); REQUIRE(std::holds_alternative(missingResult.value())); CHECK(std::get(missingResult.value()).code == yams::ErrorCode::NotFound); sessionService->close(); auto noCurrentListResult = yams::cli::run_sync(client.executeRequest(Request{ListSessionsRequest{}}), 10s); REQUIRE(noCurrentListResult.has_value()); REQUIRE(std::holds_alternative(noCurrentListResult.value())); CHECK(std::get(noCurrentListResult.value()).current_session.empty()); AddPathSelectorRequest noCurrentReq; noCurrentReq.path = selectorPath; auto noCurrentResult = yams::cli::run_sync(client.executeRequest(Request{noCurrentReq}), 10s); REQUIRE(noCurrentResult.has_value()); REQUIRE(std::holds_alternative(noCurrentResult.value())); CHECK(std::get(noCurrentResult.value()).code == yams::ErrorCode::InvalidState); RemovePathSelectorRequest noCurrentRemoveReq; noCurrentRemoveReq.path = selectorPath; auto noCurrentRemoveResult = yams::cli::run_sync(client.executeRequest(Request{noCurrentRemoveReq}), 10s); REQUIRE(noCurrentRemoveResult.has_value()); REQUIRE(std::holds_alternative(noCurrentRemoveResult.value())); CHECK(std::get(noCurrentRemoveResult.value()).code == yams::ErrorCode::InvalidState); UseSessionRequest useReq; useReq.session_name = sessionB; auto useResult = yams::cli::run_sync(client.executeRequest(Request{useReq}), 10s); REQUIRE(useResult.has_value()); REQUIRE(std::holds_alternative(useResult.value())); CHECK(std::get(useResult.value()).message == "OK"); REQUIRE(sessionService->current().has_value()); CHECK(sessionService->current().value() == sessionB); AddPathSelectorRequest missingSessionAddReq; missingSessionAddReq.session_name = "dispatcher-session-missing"; missingSessionAddReq.path = "missing/**/*.txt"; auto missingSessionAddResult = yams::cli::run_sync(client.executeRequest(Request{missingSessionAddReq}), 10s); REQUIRE(missingSessionAddResult.has_value()); REQUIRE(std::holds_alternative(missingSessionAddResult.value())); CHECK(std::get(missingSessionAddResult.value()).code == yams::ErrorCode::NotFound); AddPathSelectorRequest addReq; addReq.path = selectorPath; addReq.tags = {"pinned", "coverage"}; addReq.metadata = {{"team", "daemon"}, {"task", "dispatcher-session"}}; auto addResult = yams::cli::run_sync(client.executeRequest(Request{addReq}), 10s); REQUIRE(addResult.has_value()); REQUIRE(std::holds_alternative(addResult.value())); CHECK(std::get(addResult.value()).message == "OK"); auto selectors = sessionService->listPathSelectors(sessionB); REQUIRE(selectors.size() == 1); CHECK(selectors.front() == selectorPath); AddPathSelectorRequest explicitSessionAddReq; explicitSessionAddReq.session_name = sessionB; explicitSessionAddReq.path = "include/**/*.hpp"; explicitSessionAddReq.tags = {"explicit"}; auto explicitSessionAddResult = yams::cli::run_sync(client.executeRequest(Request{explicitSessionAddReq}), 10s); REQUIRE(explicitSessionAddResult.has_value()); REQUIRE(std::holds_alternative(explicitSessionAddResult.value())); selectors = sessionService->listPathSelectors(sessionB); REQUIRE(selectors.size() == 2); CHECK(std::find(selectors.begin(), selectors.end(), "include/**/*.hpp") != selectors.end()); RemovePathSelectorRequest missingSessionRemoveReq; missingSessionRemoveReq.session_name = "dispatcher-session-missing"; missingSessionRemoveReq.path = selectorPath; auto missingSessionRemoveResult = yams::cli::run_sync(client.executeRequest(Request{missingSessionRemoveReq}), 10s); REQUIRE(missingSessionRemoveResult.has_value()); REQUIRE(std::holds_alternative(missingSessionRemoveResult.value())); CHECK(std::get(missingSessionRemoveResult.value()).code == yams::ErrorCode::NotFound); RemovePathSelectorRequest explicitRemoveReq; explicitRemoveReq.session_name = sessionB; explicitRemoveReq.path = "include/**/*.hpp"; auto explicitRemoveResult = yams::cli::run_sync(client.executeRequest(Request{explicitRemoveReq}), 10s); REQUIRE(explicitRemoveResult.has_value()); REQUIRE(std::holds_alternative(explicitRemoveResult.value())); RemovePathSelectorRequest removeReq; removeReq.path = selectorPath; auto removeResult = yams::cli::run_sync(client.executeRequest(Request{removeReq}), 10s); REQUIRE(removeResult.has_value()); REQUIRE(std::holds_alternative(removeResult.value())); CHECK(std::get(removeResult.value()).message == "OK"); CHECK(sessionService->listPathSelectors(sessionB).empty()); } TEST_CASE("Daemon client repair request execution", "[daemon][socket][requests][repair]") { SKIP_DAEMON_TEST_ON_WINDOWS(); DaemonHarness harness({.enableModelProvider = false, .useMockModelProvider = false}); startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); const auto* daemon = harness.daemon(); REQUIRE(daemon != nullptr); auto* serviceManager = const_cast(daemon->getServiceManager()); REQUIRE(serviceManager != nullptr); std::shared_ptr repairService; for (int attempt = 0; attempt < 30; ++attempt) { repairService = serviceManager->getRepairServiceShared(); if (repairService) { break; } std::this_thread::sleep_for(100ms); } REQUIRE(repairService != nullptr); auto injectedProvider = std::make_shared(); REQUIRE(injectedProvider != nullptr); serviceManager->__test_setModelProvider(injectedProvider); REQUIRE(injectedProvider->loadModel("repair-embed-model").has_value()); EmbedDocumentsRequest embedReq; embedReq.documentHashes = {std::string(64, 'a')}; embedReq.modelName = "repair-embed-model"; embedReq.normalize = true; embedReq.batchSize = 8; embedReq.skipExisting = false; auto embedResult = yams::cli::run_sync(client.call(embedReq), 10s); REQUIRE_FALSE(embedResult.has_value()); CHECK(embedResult.error().message.find( "No requested documents were found in the daemon metadata repository") != std::string::npos); RepairRequest repairReq; repairReq.repairOrphans = true; repairReq.dryRun = true; repairReq.verbose = true; std::vector events; auto repairResult = yams::cli::run_sync( client.callRepair(repairReq, [&events](const RepairEvent& event) { events.push_back(event); }), 30s); REQUIRE(repairResult.has_value()); CHECK(repairResult.value().success); CHECK(repairResult.value().totalOperations == 1); CHECK(repairResult.value().operationResults.size() == 1); CHECK(repairResult.value().operationResults.front().operation == "orphans"); CHECK_FALSE(events.empty()); CHECK(std::any_of(events.begin(), events.end(), [](const RepairEvent& event) { return event.phase == "repairing" && event.operation == "orphans"; })); CHECK(std::any_of(events.begin(), events.end(), [](const RepairEvent& event) { return event.phase == "completed" && event.operation == "orphans"; })); } TEST_CASE("Daemon client tree diff request execution", "[daemon][socket][requests][tree-diff]") { SKIP_DAEMON_TEST_ON_WINDOWS(); DaemonHarness harness; startHarnessWithRetry(harness); std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); const auto* daemon = harness.daemon(); REQUIRE(daemon != nullptr); const auto* serviceManager = daemon->getServiceManager(); REQUIRE(serviceManager != nullptr); auto metadataRepo = serviceManager->getMetadataRepo(); REQUIRE(metadataRepo != nullptr); const std::string baseSnapshotId = "dispatcher-coverage-base"; const std::string targetSnapshotId = "dispatcher-coverage-target"; const auto nowMicros = std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch()) .count(); yams::metadata::TreeSnapshotRecord baseSnapshot; baseSnapshot.snapshotId = baseSnapshotId; baseSnapshot.createdTime = nowMicros; baseSnapshot.metadata["directory_path"] = "/tmp/dispatcher-base"; REQUIRE(metadataRepo->upsertTreeSnapshot(baseSnapshot).has_value()); yams::metadata::TreeSnapshotRecord targetSnapshot; targetSnapshot.snapshotId = targetSnapshotId; targetSnapshot.createdTime = nowMicros + 1; targetSnapshot.metadata["directory_path"] = "/tmp/dispatcher-target"; REQUIRE(metadataRepo->upsertTreeSnapshot(targetSnapshot).has_value()); yams::metadata::TreeDiffDescriptor descriptor; descriptor.baseSnapshotId = baseSnapshotId; descriptor.targetSnapshotId = targetSnapshotId; descriptor.computedAt = nowMicros + 2; descriptor.status = "pending"; auto diffIdResult = metadataRepo->beginTreeDiff(descriptor); REQUIRE(diffIdResult.has_value()); std::vector changes; yams::metadata::TreeChangeRecord added; added.type = yams::metadata::TreeChangeType::Added; added.newPath = "src/new_file.cpp"; added.newHash = "hash-added"; added.mode = 0644; changes.push_back(added); yams::metadata::TreeChangeRecord modified; modified.type = yams::metadata::TreeChangeType::Modified; modified.oldPath = "src/existing.cpp"; modified.newPath = "src/existing.cpp"; modified.oldHash = "hash-old"; modified.newHash = "hash-new"; modified.mode = 0644; modified.contentDeltaHash = "delta-existing"; changes.push_back(modified); yams::metadata::TreeChangeRecord renamed; renamed.type = yams::metadata::TreeChangeType::Renamed; renamed.oldPath = "src/old_name.cpp"; renamed.newPath = "src/new_name.cpp"; renamed.oldHash = "hash-rename"; renamed.newHash = "hash-rename"; renamed.mode = 0644; changes.push_back(renamed); yams::metadata::TreeChangeRecord deleted; deleted.type = yams::metadata::TreeChangeType::Deleted; deleted.oldPath = "src/removed.cpp"; deleted.oldHash = "hash-deleted"; deleted.mode = 0644; changes.push_back(deleted); yams::metadata::TreeChangeRecord moved; moved.type = yams::metadata::TreeChangeType::Moved; moved.oldPath = "include/old_header.hpp"; moved.newPath = "include/new_header.hpp"; moved.oldHash = "hash-moved"; moved.newHash = "hash-moved"; moved.mode = 0644; changes.push_back(moved); auto appendResult = metadataRepo->appendTreeChanges(diffIdResult.value(), changes); REQUIRE(appendResult.has_value()); auto finalizeResult = metadataRepo->finalizeTreeDiff(diffIdResult.value(), changes.size(), "complete"); REQUIRE(finalizeResult.has_value()); yams::metadata::TreeDiffQuery directQuery; directQuery.baseSnapshotId = baseSnapshotId; directQuery.targetSnapshotId = targetSnapshotId; directQuery.limit = 20000; auto directListResult = metadataRepo->listTreeChanges(directQuery); if (!directListResult.has_value()) { INFO("direct listTreeChanges error: " << directListResult.error().message); } REQUIRE(directListResult.has_value()); CHECK(directListResult.value().size() == 5); ListTreeDiffRequest listReq; listReq.baseSnapshotId = baseSnapshotId; listReq.targetSnapshotId = targetSnapshotId; listReq.limit = 20000; auto listResult = yams::cli::run_sync(client.executeRequest(Request{listReq}), 10s); INFO("listTreeDiff has_value=" << listResult.has_value()); if (!listResult.has_value()) { INFO("listTreeDiff transport error: " << listResult.error().message); } REQUIRE(listResult.has_value()); if (std::holds_alternative(listResult.value())) { INFO("listTreeDiff daemon error: " << std::get(listResult.value()).message); } REQUIRE(std::holds_alternative(listResult.value())); const auto& listResp = std::get(listResult.value()); CHECK(listResp.totalCount == 5); REQUIRE(listResp.changes.size() == 5); CHECK(std::any_of(listResp.changes.begin(), listResp.changes.end(), [](const auto& entry) { return entry.changeType == "added" && entry.path == "src/new_file.cpp" && entry.oldPath.empty(); })); CHECK(std::any_of(listResp.changes.begin(), listResp.changes.end(), [](const auto& entry) { return entry.changeType == "modified" && entry.path == "src/existing.cpp" && entry.oldPath.empty() && entry.contentDeltaHash == "delta-existing"; })); CHECK(std::any_of(listResp.changes.begin(), listResp.changes.end(), [](const auto& entry) { return entry.changeType == "renamed" && entry.path == "src/new_name.cpp" && entry.oldPath == "src/old_name.cpp"; })); CHECK(std::any_of(listResp.changes.begin(), listResp.changes.end(), [](const auto& entry) { return entry.changeType == "deleted" && entry.path.empty() && entry.oldPath == "src/removed.cpp"; })); CHECK(std::any_of(listResp.changes.begin(), listResp.changes.end(), [](const auto& entry) { return entry.changeType == "unknown" && entry.path == "include/new_header.hpp" && entry.oldPath == "include/old_header.hpp"; })); ListTreeDiffRequest filteredReq; filteredReq.baseSnapshotId = baseSnapshotId; filteredReq.targetSnapshotId = targetSnapshotId; filteredReq.pathPrefix = "src/new"; filteredReq.typeFilter = "renamed"; filteredReq.limit = 10; auto filteredResult = yams::cli::run_sync(client.executeRequest(Request{filteredReq}), 10s); INFO("filteredTreeDiff has_value=" << filteredResult.has_value()); if (!filteredResult.has_value()) { INFO("filteredTreeDiff transport error: " << filteredResult.error().message); } REQUIRE(filteredResult.has_value()); if (std::holds_alternative(filteredResult.value())) { INFO("filteredTreeDiff daemon error: " << std::get(filteredResult.value()).message); } REQUIRE(std::holds_alternative(filteredResult.value())); const auto& filteredResp = std::get(filteredResult.value()); CHECK(filteredResp.totalCount == 1); REQUIRE(filteredResp.changes.size() == 1); CHECK(filteredResp.changes.front().changeType == "renamed"); CHECK(filteredResp.changes.front().path == "src/new_name.cpp"); CHECK(filteredResp.changes.front().oldPath == "src/old_name.cpp"); ListTreeDiffRequest unknownFilterReq; unknownFilterReq.baseSnapshotId = baseSnapshotId; unknownFilterReq.targetSnapshotId = targetSnapshotId; unknownFilterReq.pathPrefix = "include/new"; unknownFilterReq.typeFilter = "moved"; unknownFilterReq.limit = 10; auto unknownFilterResult = yams::cli::run_sync(client.executeRequest(Request{unknownFilterReq}), 10s); REQUIRE(unknownFilterResult.has_value()); REQUIRE(std::holds_alternative(unknownFilterResult.value())); const auto& unknownFilterResp = std::get(unknownFilterResult.value()); CHECK(unknownFilterResp.totalCount == 1); REQUIRE(unknownFilterResp.changes.size() == 1); CHECK(unknownFilterResp.changes.front().changeType == "unknown"); CHECK(unknownFilterResp.changes.front().path == "include/new_header.hpp"); CHECK(unknownFilterResp.changes.front().oldPath == "include/old_header.hpp"); ListTreeDiffRequest pagedReq; pagedReq.baseSnapshotId = baseSnapshotId; pagedReq.targetSnapshotId = targetSnapshotId; pagedReq.limit = 1; pagedReq.offset = 1; auto pagedResult = yams::cli::run_sync(client.executeRequest(Request{pagedReq}), 10s); REQUIRE(pagedResult.has_value()); REQUIRE(std::holds_alternative(pagedResult.value())); const auto& pagedResp = std::get(pagedResult.value()); CHECK(pagedResp.totalCount == 1); REQUIRE(pagedResp.changes.size() == 1); CHECK(pagedResp.changes.front().changeType == "modified"); CHECK(pagedResp.changes.front().path == "src/existing.cpp"); CHECK(pagedResp.changes.front().contentDeltaHash == "delta-existing"); ListTreeDiffRequest invalidReq; invalidReq.targetSnapshotId = targetSnapshotId; auto invalidResult = yams::cli::run_sync(client.executeRequest(Request{invalidReq}), 10s); REQUIRE(invalidResult.has_value()); REQUIRE(std::holds_alternative(invalidResult.value())); CHECK(std::get(invalidResult.value()).code == yams::ErrorCode::InvalidArgument); } TEST_CASE("Daemon client error handling", "[daemon][socket][errors]") { SKIP_DAEMON_TEST_ON_WINDOWS(); // Use isolated daemon for tests that modify database state DaemonHarness harness; startHarnessWithRetry(harness); // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); SECTION("get nonexistent document fails gracefully") { GetRequest req; req.hash = "0000000000000000000000000000000000000000000000000000000000000000"; auto result = yams::cli::run_sync(client.get(req), 5s); REQUIRE(!result.has_value()); // Should return error, not crash } SECTION("invalid search query handled") { SearchRequest req; req.query = ""; // Empty query auto result = yams::cli::run_sync(client.search(req), 5s); // Either succeeds with empty results or returns error // Both are acceptable error handling REQUIRE((!result.has_value() || result.value().results.empty())); } SECTION("malformed grep pattern handled") { GrepRequest req; req.pattern = "[[[invalid regex"; // Invalid regex auto result = yams::cli::run_sync(client.grep(req), 5s); // Should return error, not crash REQUIRE(!result.has_value()); } } TEST_CASE("Daemon client concurrent requests", "[daemon][socket][concurrency]") { SKIP_DAEMON_TEST_ON_WINDOWS(); // Use isolated daemon for tests that modify database state DaemonHarness harness; startHarnessWithRetry(harness); // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); SECTION("sequential status requests") { for (int i = 0; i < 5; ++i) { auto result = yams::cli::run_sync(client.status(), 5s); REQUIRE(result.has_value()); } } SECTION("add multiple documents") { std::vector hashes; for (int i = 0; i < 3; ++i) { AddDocumentRequest req; req.name = "doc" + std::to_string(i) + ".txt"; req.content = "Content " + std::to_string(i); auto result = yams::cli::run_sync(client.streamingAddDocument(req), 5s); REQUIRE(result.has_value()); hashes.push_back(result.value().hash); } REQUIRE(hashes.size() == 3); // Verify all hashes are unique std::set uniqueHashes(hashes.begin(), hashes.end()); REQUIRE(uniqueHashes.size() == 3); } SECTION("list after adding documents") { // Add documents with delay to avoid database lock contention from async post-ingest std::vector hashes; for (int i = 0; i < 3; ++i) { AddDocumentRequest req; req.name = "multi_doc" + std::to_string(i) + ".txt"; req.content = "Content " + std::to_string(i); // Retry document add in case of transient connection issues yams::Result result; for (int attempt = 0; attempt < 3; ++attempt) { result = yams::cli::run_sync(client.streamingAddDocument(req), 10s); if (result.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(result.has_value()); hashes.push_back(result.value().hash); // Small delay to let post-ingest process and release database lock std::this_thread::sleep_for(50ms); } // Wait for all documents to be available (async post-ingest processing) for (const auto& hash : hashes) { auto getResult = getWithRetry(client, hash); REQUIRE(getResult.has_value()); } // List should show all documents ListRequest listReq; listReq.limit = 100; auto listResult = yams::cli::run_sync(client.list(listReq), 5s); REQUIRE(listResult.has_value()); REQUIRE(listResult.value().items.size() >= 3); } } TEST_CASE("Daemon client timeout behavior", "[daemon][socket][timeout]") { SKIP_DAEMON_TEST_ON_WINDOWS(); // Use isolated daemon DaemonHarness harness; startHarnessWithRetry(harness); SECTION("short header timeout") { ClientConfig config; config.socketPath = harness.socketPath(); config.connectTimeout = 2s; config.headerTimeout = 500ms; // Reasonable short timeout (increased from 100ms for stability) config.autoStart = false; // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); DaemonClient client(config); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); // Fast operation should still succeed even with short timeout auto statusResult = yams::cli::run_sync(client.status(), 5s); REQUIRE(statusResult.has_value()); } SECTION("request timeout handling") { // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); // Use a short timeout for the operation (50ms is short but avoids race conditions) // Note: 1ms caused SIGABRT due to mutex race conditions during timeout handling auto statusResult = yams::cli::run_sync(client.status(), 50ms); // May timeout or succeed depending on timing // Just verify it doesn't crash REQUIRE(true); } } TEST_CASE("Daemon client move semantics", "[daemon][socket][move]") { SKIP_DAEMON_TEST_ON_WINDOWS(); // Use isolated daemon DaemonHarness harness; startHarnessWithRetry(harness); SECTION("move construction") { // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client1 = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client1.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); REQUIRE(client1.isConnected()); // Move construct DaemonClient client2(std::move(client1)); REQUIRE(client2.isConnected()); // Original should be in valid but unspecified state // Just verify we can destroy it } SECTION("move assignment") { // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client1 = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client1.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); REQUIRE(client1.isConnected()); // Move assign DaemonClient client2(std::move(client1)); REQUIRE(client2.isConnected()); // Can make request with moved-to client auto statusResult = yams::cli::run_sync(client2.status(), 5s); REQUIRE(statusResult.has_value()); } } TEST_CASE("Daemon socket file lifecycle", "[daemon][socket][filesystem]") { SKIP_DAEMON_TEST_ON_WINDOWS(); SECTION("socket file created on start") { // Use its own daemon DaemonHarness harness; startHarnessWithRetry(harness); REQUIRE(std::filesystem::exists(harness.socketPath())); harness.stop(); } SECTION("daemon stops on shutdown request") { // This test needs its own daemon to shutdown DaemonHarness harness; startHarnessWithRetry(harness); REQUIRE(std::filesystem::exists(harness.socketPath())); // Allow time for GlobalIOContext to stabilize std::this_thread::sleep_for(200ms); auto client = createClient(harness.socketPath()); // Retry connection to handle timing issues with GlobalIOContext yams::Result connectResult; for (int attempt = 0; attempt < 3; ++attempt) { connectResult = yams::cli::run_sync(client.connect(), 5s); if (connectResult.has_value()) break; std::this_thread::sleep_for(200ms); } REQUIRE(connectResult.has_value()); // Send shutdown request - the daemon may close the connection before responding, // so we don't require the IPC response to succeed. The important thing is that // the daemon eventually stops. (void)yams::cli::run_sync(client.shutdown(true), 10s); REQUIRE(harness.daemon() != nullptr); REQUIRE(harness.daemon()->waitForStopCompletion(10s)); } SECTION("socket file removed on stop") { // This test needs its own daemon to stop DaemonHarness harness; startHarnessWithRetry(harness); REQUIRE(std::filesystem::exists(harness.socketPath())); harness.stop(); std::this_thread::sleep_for(500ms); // Socket may be cleaned up // Just verify we don't crash REQUIRE(true); } SECTION("multiple clients can connect to same socket") { DaemonHarness harness; startHarnessWithRetry(harness); auto client1 = createClient(harness.socketPath()); auto client2 = createClient(harness.socketPath()); auto connect1 = yams::cli::run_sync(client1.connect(), 5s); auto connect2 = yams::cli::run_sync(client2.connect(), 5s); REQUIRE(connect1.has_value()); REQUIRE(connect2.has_value()); REQUIRE(client1.isConnected()); REQUIRE(client2.isConnected()); } }