yams/tests/unit/daemon/daemon_catch2_test.cpp

// Catch2 migration of daemon_test.cpp
// Migration: yams-3s4 (daemon unit tests)
// Unit tests for YamsDaemon - creation, start/stop, signal handling, error paths

#include <catch2/catch_test_macros.hpp>

#include <atomic>
#include <chrono>
#include <filesystem>
#include <fstream>
#include <future>
#include <iomanip>
#include <sstream>
#include <string_view>
#include <thread>

#include <nlohmann/json.hpp>
using nlohmann::json;

#include <yams/compat/unistd.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/daemon_lifecycle.h>
#include <yams/daemon/ipc/ipc_protocol.h>
#include <yams/daemon/ipc/proto_serializer.h>
#include <yams/metadata/document_metadata.h>
#include <yams/metadata/metadata_repository.h>

#include "tests/common/test_helpers_catch2.h"

#ifdef _WIN32
#include <fcntl.h>
#include <io.h>
#include <share.h>
using pid_t = int;
#else
#include <signal.h>
#include <sys/wait.h>
#endif

namespace fs = std::filesystem;
using namespace std::chrono_literals;
using namespace yams;
using namespace yams::daemon;

namespace yams::daemon::test {

// Test fixture for Daemon tests
struct DaemonFixture {
    DaemonConfig config_;
    std::unique_ptr<YamsDaemon> daemon_;
    fs::path runtime_root_;
    std::thread runLoopThread_;

    DaemonFixture() {
#ifdef _WIN32
        fs::path tmp = fs::temp_directory_path() / "yams_test";
#else
        fs::path tmp{"/tmp"};
#endif
        auto unique_suffix =
            std::to_string(::getpid()) + "_" +
            std::to_string(static_cast<unsigned long>(reinterpret_cast<uintptr_t>(this) & 0xffff));

        runtime_root_ = tmp / ("ydtest_" + unique_suffix);

        config_.dataDir = runtime_root_ / "data";
        config_.socketPath = runtime_root_ / "sock";
        config_.pidFile = runtime_root_ / "daemon.pid";
        config_.logFile = runtime_root_ / "daemon.log";
        config_.maxMemoryGb = 1.0;

        // When ABI plugins are disabled, no model provider can be adopted,
        // so disable the requirement to avoid init failure.
        if (const char* v = std::getenv("YAMS_DISABLE_ABI_PLUGINS");
            v && std::string_view(v) == "1") {
            config_.enableModelProvider = false;
        } else {
            config_.enableModelProvider = true;
        }
        config_.autoLoadPlugins = true;
        config_.modelPoolConfig.lazyLoading = true;
        config_.modelPoolConfig.preloadModels.clear();

        ::setenv("YAMS_DB_OPEN_TIMEOUT_MS", "1500", 1);
        ::setenv("YAMS_DB_MIGRATE_TIMEOUT_MS", "2000", 1);
        ::setenv("YAMS_SEARCH_BUILD_TIMEOUT_MS", "1500", 1);
        ::setenv("YAMS_DISABLE_VECTORS", "1", 1);
        ::setenv("YAMS_DISABLE_SESSION_WATCHER", "1", 1);

        std::error_code se;
        fs::create_directories(config_.dataDir, se);

        ::setenv("YAMS_RUNTIME_DIR", runtime_root_.string().c_str(), 1);
        ::setenv("YAMS_SOCKET_PATH", config_.socketPath.string().c_str(), 1);
        ::setenv("YAMS_PID_FILE", config_.pidFile.string().c_str(), 1);
    }

    ~DaemonFixture() {
        stopRunLoop();
        if (daemon_) {
            daemon_->stop();
            daemon_.reset();
        }
        cleanupDaemonFiles();
    }

    void startRunLoop() {
        if (!daemon_ || runLoopThread_.joinable()) {
            return;
        }
        runLoopThread_ = std::thread([this]() { daemon_->runLoop(); });
        std::this_thread::sleep_for(50ms);
    }

    void stopRunLoop() {
        if (daemon_ && daemon_->isRunning()) {
            daemon_->requestStop();
        }
        if (runLoopThread_.joinable()) {
            runLoopThread_.join();
        }
    }

    void cleanupDaemonFiles() {
        std::error_code ec;
        if (!runtime_root_.empty()) {
            fs::remove_all(runtime_root_, ec);
        }
    }

    static bool isSocketPermissionDenied(const yams::Error& error) {
        const std::string_view message{error.message};
        return message.find("Operation not permitted") != std::string_view::npos ||
               message.find("Permission denied") != std::string_view::npos;
    }

    bool waitForDaemonReady(std::chrono::milliseconds timeout = 5000ms) {
        auto deadline = std::chrono::steady_clock::now() + timeout;
        while (std::chrono::steady_clock::now() < deadline) {
            const auto& state = daemon_->getState();
            if (state.readiness.contentStoreReady.load()) {
                return true;
            }
            std::this_thread::sleep_for(50ms);
        }
        return false;
    }

    static bool waitForCondition(std::chrono::milliseconds timeout,
                                 const std::function<bool()>& predicate) {
        auto deadline = std::chrono::steady_clock::now() + timeout;
        while (std::chrono::steady_clock::now() < deadline) {
            if (predicate()) {
                return true;
            }
            std::this_thread::sleep_for(10ms);
        }
        return predicate();
    }
};

#ifdef _WIN32
#define SKIP_ON_WINDOWS() SKIP("Daemon IPC tests unstable on Windows")
#else
#define SKIP_ON_WINDOWS() ((void)0)
#endif

TEST_CASE_METHOD(DaemonFixture, "Daemon creation and destruction", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    REQUIRE(daemon_ != nullptr);
    REQUIRE_FALSE(daemon_->isRunning());
}

TEST_CASE_METHOD(DaemonFixture, "Daemon tuning reload preserves unspecified values",
                 "[daemon][tuning][reload]") {
    SKIP_ON_WINDOWS();

    const auto configPath = runtime_root_ / "config.toml";
    {
        std::ofstream out(configPath);
        REQUIRE(out.is_open());
        out << "[tuning]\n";
        out << "target_cpu_percent = 321\n";
        out << "post_ingest_capacity = 111\n";
        out << "control_interval_ms = 222\n";
    }

    config_.configFilePath = configPath;
    config_.tuning.postIngestThreadsMin = 5;
    config_.tuning.holdMs = 777;

    daemon_ = std::make_unique<YamsDaemon>(config_);
    REQUIRE(daemon_ != nullptr);

    daemon_->reloadTuningConfig();

    CHECK(daemon_->config_.tuning.targetCpuPercent == 321u);
    CHECK(daemon_->config_.tuning.postIngestCapacity == 111u);
    CHECK(daemon_->config_.tuning.controlIntervalMs == 222u);
    CHECK(daemon_->config_.tuning.postIngestThreadsMin == 5u);
    CHECK(daemon_->config_.tuning.holdMs == 777u);
}

TEST_CASE_METHOD(DaemonFixture, "Lifecycle shutdown waits for owner-thread stop",
                 "[daemon][lifecycle][shutdown-request]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);

    startRunLoop();
    REQUIRE(waitForCondition(
        2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));

    DaemonLifecycleAdapter lifecycle(daemon_.get());
    lifecycle.requestShutdown(true, true);

    REQUIRE(waitForCondition(2s, [&] {
        return daemon_->stopRequested_.load(std::memory_order_acquire) &&
               daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
    }));

    std::this_thread::sleep_for(100ms);
    CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));

    stopRunLoop();
    auto stopResult = daemon_->stop();
    REQUIRE(stopResult);

    REQUIRE(waitForCondition(
        2s, [&] { return !daemon_->shutdownThreadActive_.load(std::memory_order_acquire); }));
    daemon_->reapCompletedShutdownThread();
}

TEST_CASE_METHOD(DaemonFixture,
                 "Lifecycle shutdown request does not cancel an in-flight worker task",
                 "[daemon][lifecycle][shutdown-request][cancellation]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);

    startRunLoop();
    REQUIRE(waitForCondition(
        2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));

    auto* serviceManager = daemon_->getServiceManager();
    REQUIRE((serviceManager != nullptr));

    std::mutex gateMutex;
    std::condition_variable gateCv;
    bool allowExit = false;
    std::atomic<bool> workerEntered{false};
    std::atomic<bool> workerFinished{false};

    boost::asio::post(serviceManager->getWorkerExecutor(), [&]() {
        {
            std::lock_guard<std::mutex> lk(gateMutex);
            workerEntered.store(true, std::memory_order_release);
        }
        gateCv.notify_all();

        std::unique_lock<std::mutex> lk(gateMutex);
        gateCv.wait(lk, [&] { return allowExit; });
        workerFinished.store(true, std::memory_order_release);
    });

    REQUIRE(waitForCondition(2s, [&] { return workerEntered.load(std::memory_order_acquire); }));

    DaemonLifecycleAdapter lifecycle(daemon_.get());
    lifecycle.requestShutdown(true, true);

    REQUIRE(waitForCondition(2s, [&] {
        return daemon_->stopRequested_.load(std::memory_order_acquire) &&
               daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
    }));

    std::this_thread::sleep_for(150ms);
    CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
    CHECK_FALSE(workerFinished.load(std::memory_order_acquire));

    {
        std::lock_guard<std::mutex> lk(gateMutex);
        allowExit = true;
    }
    gateCv.notify_all();

    REQUIRE(waitForCondition(2s, [&] { return workerFinished.load(std::memory_order_acquire); }));

    stopRunLoop();
    auto stopResult = daemon_->stop();
    REQUIRE(stopResult);

    REQUIRE(waitForCondition(
        2s, [&] { return !daemon_->shutdownThreadActive_.load(std::memory_order_acquire); }));
    daemon_->reapCompletedShutdownThread();
}

TEST_CASE_METHOD(DaemonFixture,
                 "Lifecycle shutdown completion stays pending while worker teardown is blocked",
                 "[daemon][lifecycle][shutdown-request][budget]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);

    startRunLoop();
    REQUIRE(waitForCondition(
        2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));

    auto* serviceManager = daemon_->getServiceManager();
    REQUIRE((serviceManager != nullptr));

    std::mutex gateMutex;
    std::condition_variable gateCv;
    bool allowExit = false;
    std::atomic<bool> workerEntered{false};

    boost::asio::post(serviceManager->getWorkerExecutor(), [&]() {
        {
            std::lock_guard<std::mutex> lk(gateMutex);
            workerEntered.store(true, std::memory_order_release);
        }
        gateCv.notify_all();

        std::unique_lock<std::mutex> lk(gateMutex);
        gateCv.wait(lk, [&] { return allowExit; });
    });

    REQUIRE(waitForCondition(2s, [&] { return workerEntered.load(std::memory_order_acquire); }));

    DaemonLifecycleAdapter lifecycle(daemon_.get());
    lifecycle.requestShutdown(true, true);

    REQUIRE(waitForCondition(2s, [&] {
        return daemon_->stopRequested_.load(std::memory_order_acquire) &&
               daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
    }));

    CHECK_FALSE(daemon_->waitForStopCompletion(100ms));
    std::this_thread::sleep_for(150ms);
    CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
    CHECK_FALSE(daemon_->waitForStopCompletion(100ms));

    {
        std::lock_guard<std::mutex> lk(gateMutex);
        allowExit = true;
    }
    gateCv.notify_all();

    stopRunLoop();
    auto stopResult = daemon_->stop();
    REQUIRE(stopResult);
    REQUIRE(daemon_->waitForStopCompletion(2s));
    daemon_->reapCompletedShutdownThread();
}

TEST_CASE_METHOD(DaemonFixture,
                 "Lifecycle shutdown timeout floor covers work coordinator join budget",
                 "[daemon][lifecycle][shutdown-request][budget]") {
    SKIP_ON_WINDOWS();

    yams::test::ScopedEnvVar shutdownBudget("YAMS_SHUTDOWN_FORCE_EXIT_MS", std::string("1000"));

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);

    startRunLoop();
    REQUIRE(waitForCondition(
        2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));

    auto* serviceManager = daemon_->getServiceManager();
    REQUIRE((serviceManager != nullptr));

    std::mutex gateMutex;
    std::condition_variable gateCv;
    bool allowExit = false;
    std::atomic<bool> workerEntered{false};

    boost::asio::post(serviceManager->getWorkerExecutor(), [&]() {
        {
            std::lock_guard<std::mutex> lk(gateMutex);
            workerEntered.store(true, std::memory_order_release);
        }
        gateCv.notify_all();

        std::unique_lock<std::mutex> lk(gateMutex);
        gateCv.wait(lk, [&] { return allowExit; });
    });

    REQUIRE(waitForCondition(2s, [&] { return workerEntered.load(std::memory_order_acquire); }));

    DaemonLifecycleAdapter lifecycle(daemon_.get());
    lifecycle.requestShutdown(true, true);

    REQUIRE(waitForCondition(2s, [&] {
        return daemon_->stopRequested_.load(std::memory_order_acquire) &&
               daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
    }));

    std::this_thread::sleep_for(1200ms);
    CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
    CHECK_FALSE(daemon_->waitForStopCompletion(100ms));

    {
        std::lock_guard<std::mutex> lk(gateMutex);
        allowExit = true;
    }
    gateCv.notify_all();

    stopRunLoop();
    auto stopResult = daemon_->stop();
    REQUIRE(stopResult);
    REQUIRE(daemon_->waitForStopCompletion(2s));
    daemon_->reapCompletedShutdownThread();
}

TEST_CASE_METHOD(DaemonFixture, "Lifecycle shutdown falls back to direct stop without run loop",
                 "[daemon][lifecycle][shutdown-request][fallback]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);
    REQUIRE_FALSE(daemon_->runLoopStarted_.load(std::memory_order_acquire));

    DaemonLifecycleAdapter lifecycle(daemon_.get());
    lifecycle.requestShutdown(true, true);

    REQUIRE(waitForCondition(2s, [&] {
        return !daemon_->isRunning() &&
               !daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
    }));
    CHECK(daemon_->getLifecycle().snapshot().state == LifecycleState::Stopped);
    if (const auto* serviceManager = daemon_->getServiceManager()) {
        CHECK(serviceManager->getServiceManagerFsmSnapshot().state == ServiceManagerState::Stopped);
    }
    daemon_->reapCompletedShutdownThread();
}

TEST_CASE_METHOD(DaemonFixture,
                 "Lifecycle shutdown avoids forced exit for foreground-managed daemon",
                 "[daemon][lifecycle][shutdown-request][foreground-managed]") {
    SKIP_ON_WINDOWS();

    yams::test::ScopedEnvVar foregroundManaged("YAMS_DAEMON_FOREGROUND", std::string("1"));

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);

    startRunLoop();
    REQUIRE(waitForCondition(
        2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));

    DaemonLifecycleAdapter lifecycle(daemon_.get());
    lifecycle.requestShutdown(true, false);

    REQUIRE(waitForCondition(2s, [&] {
        return daemon_->stopRequested_.load(std::memory_order_acquire) &&
               daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
    }));

    stopRunLoop();
    auto stopResult = daemon_->stop();
    REQUIRE(stopResult);

    REQUIRE(waitForCondition(
        2s, [&] { return !daemon_->shutdownThreadActive_.load(std::memory_order_acquire); }));
    daemon_->reapCompletedShutdownThread();
    REQUIRE_FALSE(daemon_->isRunning());
}

TEST_CASE_METHOD(DaemonFixture, "Daemon start and stop", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto startResult = daemon_->start();
    if (!startResult && isSocketPermissionDenied(startResult.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(startResult);
    REQUIRE(daemon_->isRunning());
    REQUIRE(fs::exists(config_.pidFile));

    startRunLoop();
    REQUIRE(waitForDaemonReady());

    stopRunLoop();

    auto result = daemon_->stop();
    REQUIRE(result);
    REQUIRE_FALSE(daemon_->isRunning());
    REQUIRE(daemon_->getLifecycle().snapshot().state == LifecycleState::Stopped);
    if (const auto* serviceManager = daemon_->getServiceManager()) {
        REQUIRE(serviceManager->getServiceManagerFsmSnapshot().state ==
                ServiceManagerState::Stopped);
    }
    REQUIRE_FALSE(fs::exists(config_.pidFile));
}

TEST_CASE_METHOD(DaemonFixture, "Daemon single instance enforcement", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    auto daemon2 = std::make_unique<YamsDaemon>(config_);
    auto result2 = daemon2->start();
    REQUIRE_FALSE(result2);
    REQUIRE(result2.error().code == ErrorCode::InvalidState);

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon rejects pid file pointing to current process",
                 "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    json payload = json::object();
    payload["pid"] = static_cast<std::int64_t>(getpid());
    payload["start_ns"] = 1;
    payload["token"] = "test";

    std::ofstream pidFile(config_.pidFile);
    pidFile << payload.dump();
    pidFile.close();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    REQUIRE_FALSE(result);
    REQUIRE(result.error().code == ErrorCode::InvalidState);
}

TEST_CASE_METHOD(DaemonFixture, "Daemon refuses pid file pointing to unrelated running process",
                 "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

#ifndef _WIN32

    pid_t child = fork();
    if (child < 0) {
        SKIP("fork not available");
    }
    if (child == 0) {
        for (;;) {
            ::pause();
        }
    }

    json payload = json::object();
    payload["pid"] = static_cast<std::int64_t>(child);
    payload["start_ns"] = 1;
    payload["token"] = "test";

    std::ofstream pidFile(config_.pidFile);
    pidFile << payload.dump();
    pidFile.close();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    REQUIRE_FALSE(result);
    REQUIRE(result.error().code == ErrorCode::InvalidState);

    ::kill(child, SIGKILL);
    int status = 0;
    (void)::waitpid(child, &status, 0);

#else
    SKIP("POSIX process APIs not available on Windows");
#endif
}

TEST_CASE_METHOD(DaemonFixture, "Daemon restart", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    result = daemon_->stop();
    REQUIRE(result);

    result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted on restart");
    }
    REQUIRE(result);
    REQUIRE(daemon_->isRunning());

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon path resolution with empty config", "[daemon][config]") {
    SKIP_ON_WINDOWS();

    DaemonConfig autoConfig;
    daemon_ = std::make_unique<YamsDaemon>(autoConfig);
    REQUIRE(daemon_ != nullptr);
}

TEST_CASE_METHOD(DaemonFixture, "Daemon signal handling and PID file", "[daemon][signals]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

#ifndef _WIN32
    std::ifstream pidFile(config_.pidFile);
    std::string content;
    std::getline(pidFile, content, '\0');
    pidFile.close();
    auto parsed = json::parse(content, nullptr, false);
    REQUIRE_FALSE(parsed.is_discarded());
    REQUIRE(parsed.value("pid", 0) == getpid());
#endif

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon stale PID file cleanup", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    std::ofstream pidFile(config_.pidFile);
    pidFile << "99999999\n";
    pidFile.close();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);
    REQUIRE(daemon_->isRunning());

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon invalid socket path configuration", "[daemon][config]") {
    SKIP_ON_WINDOWS();

    config_.socketPath = "/root/cannot_write_here.sock";
    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    REQUIRE_FALSE(result);
}

TEST_CASE_METHOD(DaemonFixture,
                 "Daemon socket preflight does not clean unrelated stale PID artifacts",
                 "[daemon][config][lifecycle]") {
    SKIP_ON_WINDOWS();

    // Seed a stale PID artifact that this process does not own.
    {
        std::ofstream pidFile(config_.pidFile);
        pidFile << "99999999\n";
    }
    REQUIRE(fs::exists(config_.pidFile));

    // Force an AF_UNIX-overlong socket path so start fails in preflight, before lifecycle init.
    config_.socketPath = runtime_root_ / std::string(140, 'x') / "daemon.sock";
    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    REQUIRE_FALSE(result);
    CHECK(result.error().message.find("Socket path too long") != std::string::npos);

    // Preflight failure must not mutate stale PID artifacts we did not create.
    CHECK(fs::exists(config_.pidFile));
}

TEST_CASE_METHOD(DaemonFixture, "Daemon concurrent start attempts", "[daemon][concurrency]") {
    SKIP_ON_WINDOWS();

    {
        YamsDaemon probe(config_);
        auto result = probe.start();
        if (!result && isSocketPermissionDenied(result.error())) {
            SKIP("UNIX domain sockets not permitted");
        }
        REQUIRE(result);
        probe.stop();
    }

    std::atomic<int> successCount{0};
    std::atomic<int> failCount{0};

    std::once_flag startedFlag;
    std::promise<void> startedPromise;
    auto startedFuture = startedPromise.get_future().share();
    std::promise<void> releasePromise;
    auto releaseFuture = releasePromise.get_future().share();
    std::promise<void> attemptGatePromise;
    auto attemptGateFuture = attemptGatePromise.get_future().share();
    std::mutex readyMutex;
    std::condition_variable readyCv;
    int readyCount = 0;

    std::vector<std::thread> threads;
    constexpr int kAttempts = 5;
    threads.reserve(kAttempts);

    for (int i = 0; i < kAttempts; ++i) {
        threads.emplace_back([this, &successCount, &failCount, &startedFlag, startedFuture,
                              &startedPromise, releaseFuture, attemptGateFuture, &readyMutex,
                              &readyCv, &readyCount]() {
            {
                std::lock_guard<std::mutex> lock(readyMutex);
                ++readyCount;
            }
            readyCv.notify_one();
            attemptGateFuture.wait();

            YamsDaemon local(config_);
            auto result = local.start();
            if (result) {
                successCount.fetch_add(1, std::memory_order_relaxed);
                std::call_once(startedFlag, [&]() { startedPromise.set_value(); });
                releaseFuture.wait();
                local.stop();
            } else {
                startedFuture.wait();
                failCount.fetch_add(1, std::memory_order_relaxed);
            }
        });
    }

    {
        std::unique_lock<std::mutex> lock(readyMutex);
        REQUIRE(readyCv.wait_for(lock, std::chrono::seconds(1),
                                 [&] { return readyCount == kAttempts; }));
    }
    attemptGatePromise.set_value();

    if (startedFuture.wait_for(std::chrono::seconds(1)) != std::future_status::ready) {
        std::call_once(startedFlag, [&]() { startedPromise.set_value(); });
    }
    const auto failDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    while (failCount.load(std::memory_order_relaxed) < kAttempts - 1 &&
           std::chrono::steady_clock::now() < failDeadline) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
    releasePromise.set_value();

    for (auto& t : threads) {
        t.join();
    }

    REQUIRE(successCount.load() == 1);
    REQUIRE(failCount.load() == kAttempts - 1);
}

TEST_CASE_METHOD(DaemonFixture, "Daemon stats tracking", "[daemon][stats]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    [[maybe_unused]] const auto& state = daemon_->getState();

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon hybrid search smoke test", "[daemon][search]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    SearchRequest req;
    req.query = "seed";
    req.limit = 5;
    req.fuzzy = false;
    req.literalText = false;
    req.similarity = 0.7;

    Message msg;
    msg.version = 1;
    msg.requestId = 42;
    msg.payload = Request{req};

    auto enc = ProtoSerializer::encode_payload(msg);
    REQUIRE(enc);
    auto parsed = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(parsed);

    REQUIRE(daemon_->isRunning());

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon fallback search when hybrid unavailable",
                 "[daemon][search]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    SearchRequest req;
    req.query = "fallback";
    req.limit = 3;
    req.fuzzy = true;
    req.similarity = 0.6;

    Message msg;
    msg.version = 1;
    msg.requestId = 7;
    msg.payload = Request{req};

    auto enc = ProtoSerializer::encode_payload(msg);
    REQUIRE(enc);
    auto parsed = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(parsed);

    REQUIRE(daemon_->isRunning());

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "StatusResponse serialization includes ready field",
                 "[daemon][serialization]") {
    StatusResponse sr;
    sr.running = true;
    sr.ready = true;
    sr.uptimeSeconds = 0;
    sr.requestsProcessed = 0;
    sr.activeConnections = 0;
    sr.memoryUsageMb = 0.0;
    sr.cpuUsagePercent = 0.0;
    sr.version = "test";

    Message msg;
    msg.version = 1;
    msg.requestId = 0;
    msg.payload = Response{sr};

    auto enc = ProtoSerializer::encode_payload(msg);
    REQUIRE(enc);
    auto des = ProtoSerializer::decode_payload(enc.value());
    REQUIRE(des);

    Message out = des.value();
    auto respVariant = std::get<Response>(out.payload);
    REQUIRE(std::holds_alternative<StatusResponse>(respVariant));
    auto outSr = std::get<StatusResponse>(respVariant);

    REQUIRE(outSr.ready == true);
    REQUIRE(outSr.running == true);
    REQUIRE(outSr.version == "test");
}

TEST_CASE_METHOD(DaemonFixture, "Daemon start fails with invalid data directory",
                 "[daemon][config]") {
    SKIP_ON_WINDOWS();

    DaemonConfig bad = config_;
    bad.dataDir = "/root/forbidden_yams_test";
    daemon_ = std::make_unique<YamsDaemon>(bad);

    auto result = daemon_->start();
    REQUIRE_FALSE(result);
}

TEST_CASE_METHOD(DaemonFixture, "Daemon PID file cannot be created in read-only directory",
                 "[daemon][errors]") {
    SKIP_ON_WINDOWS();

    fs::path readonly_dir = runtime_root_ / "readonly";
    fs::create_directories(readonly_dir);
    fs::permissions(readonly_dir, fs::perms::owner_read | fs::perms::owner_exec,
                    fs::perm_options::replace);

    DaemonConfig bad_config = config_;
    bad_config.pidFile = readonly_dir / "yams.pid";
    daemon_ = std::make_unique<YamsDaemon>(bad_config);

    auto result = daemon_->start();
    REQUIRE_FALSE(result);
    if (!result) {
        REQUIRE_FALSE(result.error().message.empty());
    }

    fs::permissions(readonly_dir, fs::perms::owner_all, fs::perm_options::replace);
}

TEST_CASE_METHOD(DaemonFixture, "Daemon PID file cleaned up on start failure", "[daemon][errors]") {
    SKIP_ON_WINDOWS();

    DaemonConfig bad_config = config_;
    bad_config.dataDir = "/nonexistent/forbidden_path";

    daemon_ = std::make_unique<YamsDaemon>(bad_config);
    auto result = daemon_->start();
    REQUIRE_FALSE(result);

    REQUIRE_FALSE(fs::exists(bad_config.pidFile));
}

TEST_CASE_METHOD(DaemonFixture, "Daemon stop when not started is safe", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    REQUIRE_NOTHROW(daemon_->stop());
    REQUIRE_NOTHROW(daemon_->stop());
}

TEST_CASE_METHOD(DaemonFixture, "Daemon double stop is idempotent", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    daemon_->stop();
    REQUIRE_NOTHROW(daemon_->stop());
}

TEST_CASE_METHOD(DaemonFixture, "Daemon fails with missing socket directory", "[daemon][config]") {
    SKIP_ON_WINDOWS();

    DaemonConfig bad_config = config_;
    bad_config.socketPath = "/nonexistent_dir/subdir/yams.sock";

    daemon_ = std::make_unique<YamsDaemon>(bad_config);
    auto result = daemon_->start();
    REQUIRE_FALSE(result);
}

TEST_CASE_METHOD(DaemonFixture, "Daemon socket cleanup after stop", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    REQUIRE(fs::exists(config_.socketPath));

    daemon_->stop();

    REQUIRE_FALSE(fs::exists(config_.socketPath));
}

TEST_CASE_METHOD(DaemonFixture, "Daemon restart with existing socket", "[daemon][lifecycle]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result1 = daemon_->start();
    if (!result1 && isSocketPermissionDenied(result1.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result1);
    daemon_->stop();

    auto result2 = daemon_->start();
    REQUIRE(result2);
    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon empty data directory handling", "[daemon][config]") {
    SKIP_ON_WINDOWS();

    DaemonConfig bad_config = config_;
    bad_config.dataDir = "";

    daemon_ = std::make_unique<YamsDaemon>(bad_config);
    auto result = daemon_->start();

    if (result) {
        daemon_->stop();
    } else {
        REQUIRE_FALSE(result.error().message.empty());
    }
}

TEST_CASE_METHOD(DaemonFixture, "Daemon status query on stopped daemon", "[daemon][state]") {
    SKIP_ON_WINDOWS();

    daemon_ = std::make_unique<YamsDaemon>(config_);
    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);

    const auto& state_running = daemon_->getState();
    REQUIRE(state_running.readiness.ipcServerReady.load());

    daemon_->stop();

    const auto& state_stopped = daemon_->getState();
    REQUIRE_FALSE(state_stopped.readiness.ipcServerReady.load());
}

TEST_CASE_METHOD(DaemonFixture, "Daemon async init barrier prevents race condition",
                 "[daemon][concurrency]") {
    SKIP_ON_WINDOWS();

    if (const char* disable_vectors = std::getenv("YAMS_DISABLE_VECTORS")) {
        if (std::string_view(disable_vectors) == "1") {
            SKIP("Vector DB disabled via YAMS_DISABLE_VECTORS=1");
        }
    }

    daemon_ = std::make_unique<YamsDaemon>(config_);

    auto result = daemon_->start();
    if (!result && isSocketPermissionDenied(result.error())) {
        SKIP("UNIX domain sockets not permitted");
    }
    REQUIRE(result);
    REQUIRE(daemon_->isRunning());

    std::atomic<bool> runLoopEntered{false};
    std::atomic<bool> runLoopExited{false};

    std::thread runLoopThread([this, &runLoopEntered, &runLoopExited]() {
        runLoopEntered.store(true);
        daemon_->runLoop();
        runLoopExited.store(true);
    });

    std::this_thread::sleep_for(100ms);

    REQUIRE(runLoopEntered.load());

    daemon_->requestStop();

    if (runLoopThread.joinable()) {
        runLoopThread.join();
    }

    REQUIRE(runLoopExited.load());

    daemon_->stop();
}

TEST_CASE_METHOD(DaemonFixture, "Daemon async init barrier stress test",
                 "[daemon][concurrency][.slow]") {
    SKIP_ON_WINDOWS();

    constexpr int kIterations = 5;

    for (int i = 0; i < kIterations; ++i) {
        INFO("Iteration " << i);

        daemon_ = std::make_unique<YamsDaemon>(config_);

        auto result = daemon_->start();
        if (!result && isSocketPermissionDenied(result.error())) {
            SKIP("UNIX domain sockets not permitted");
        }
        if (!result) {
            std::this_thread::sleep_for(100ms);
            result = daemon_->start();
            if (!result && isSocketPermissionDenied(result.error())) {
                SKIP("UNIX domain sockets not permitted");
            }
            REQUIRE(result);
        }

        REQUIRE(daemon_->isRunning());

        std::atomic<bool> crashed{false};
        std::thread runLoopThread([this, &crashed]() {
            try {
                daemon_->runLoop();
            } catch (...) {
                crashed.store(true);
            }
        });

        std::this_thread::sleep_for(50ms);

        daemon_->requestStop();
        if (runLoopThread.joinable()) {
            runLoopThread.join();
        }

        REQUIRE_FALSE(crashed.load());

        daemon_->stop();
        daemon_.reset();

        std::this_thread::sleep_for(50ms);
    }
}

} // namespace yams::daemon::test