/**
 * @file retrieval_and_ingestion_catch2_test.cpp
 * @brief Catch2 integration tests for RetrievalService and DocumentIngestionService
 *
 * Tests the daemon-based services layer for document ingestion and retrieval,
 * using DaemonHarness for proper lifecycle management.
 */

#include <algorithm>
#include <chrono>
#include <filesystem>
#include <fstream>
#include <iostream>
#include <string_view>
#include <thread>
#include <vector>

#include <catch2/catch_test_macros.hpp>

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

#include "../../common/test_helpers_catch2.h"
#include "../daemon/test_daemon_harness.h"
#include "common/capability.h"
#include "common/fixture_manager.h"

#include <yams/app/services/document_ingestion_service.h>
#include <yams/app/services/retrieval_service.h>
#include <yams/app/services/services.hpp>
#include <yams/compat/unistd.h>
#include <yams/daemon/client/asio_connection_pool.h>
#include <yams/daemon/client/daemon_client.h>
#include <yams/daemon/client/global_io_context.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/ipc/ipc_protocol.h>

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

namespace {
constexpr int kRequestTimeoutMs = 5000;
constexpr int kBodyTimeoutMs = 15000;

using yams::test::ScopedEnvVar;

struct BarrierStats {
    std::uint64_t lastPostIngestQueued{0};
    std::uint64_t lastPostIngestInflight{0};
    std::uint64_t lastPostIngestDrained{0};
    std::uint64_t lastIndexVisible{0};
    std::uint32_t statusPolls{0};
    std::uint32_t statusErrors{0};
    std::uint32_t snapshotPolls{0};
    std::size_t lastSnapshotCount{0};
    std::size_t lastDocsByLabelCount{0};
};

// Helper: wait until post-ingest queue drains (queued==0 && inflight==0) or timeout.
// Returns true when condition is observed; false on timeout.
bool waitForPostIngestQuiescent(const fs::path& socket, const fs::path& dataDir,
                                std::chrono::milliseconds timeout,
                                std::chrono::milliseconds poll = 50ms,
                                BarrierStats* stats = nullptr) {
    yams::daemon::ClientConfig cfg;
    cfg.socketPath = socket;
    cfg.dataDir = dataDir;
    cfg.requestTimeout = 1500ms;

    auto deadline = std::chrono::steady_clock::now() + timeout;
    while (std::chrono::steady_clock::now() < deadline) {
        auto client = std::make_shared<yams::daemon::DaemonClient>(cfg);
        std::promise<yams::Result<yams::daemon::StatusResponse>> prom;
        auto fut = prom.get_future();
        boost::asio::co_spawn(
            yams::daemon::GlobalIOContext::global_executor(),
            [client, p = std::move(prom)]() mutable -> boost::asio::awaitable<void> {
                auto r = co_await client->status();
                p.set_value(std::move(r));
                co_return;
            },
            boost::asio::detached);

        if (stats) {
            stats->statusPolls++;
        }

        if (fut.wait_for(1500ms) == std::future_status::ready) {
            auto res = fut.get();
            if (res) {
                const auto& s = res.value();
                auto getU64 = [&](const char* key) -> std::uint64_t {
                    auto it = s.requestCounts.find(key);
                    return (it == s.requestCounts.end()) ? 0ull
                                                         : static_cast<std::uint64_t>(it->second);
                };

                const auto queued = getU64("post_ingest_queued");
                const auto inflight = getU64("post_ingest_inflight");
                const auto drained = getU64("post_ingest_drained");
                const auto indexVisible = getU64("index_visible");
                if (stats) {
                    stats->lastPostIngestQueued = queued;
                    stats->lastPostIngestInflight = inflight;
                    stats->lastPostIngestDrained = drained;
                    stats->lastIndexVisible = indexVisible;
                }

                if (queued == 0 && inflight == 0)
                    return true;
            } else if (stats) {
                stats->statusErrors++;
            }
        } else if (stats) {
            stats->statusErrors++;
        }
        std::this_thread::sleep_for(poll);
    }

    return false;
}

template <typename MetadataRepoT>
bool waitForSnapshotVisible(MetadataRepoT* repo, const std::string& snapshotId,
                            const std::string& snapshotLabel, std::size_t minDocsByLabel,
                            std::chrono::milliseconds timeout,
                            std::chrono::milliseconds poll = 25ms, BarrierStats* stats = nullptr) {
    if (!repo || snapshotId.empty()) {
        return false;
    }

    const auto deadline = std::chrono::steady_clock::now() + timeout;
    while (std::chrono::steady_clock::now() < deadline) {
        if (stats) {
            stats->snapshotPolls++;
        }

        auto snaps = repo->listTreeSnapshots(256);
        if (snaps) {
            auto records = snaps.value();
            std::sort(records.begin(), records.end(), [](const auto& lhs, const auto& rhs) {
                if (lhs.createdTime != rhs.createdTime)
                    return lhs.createdTime > rhs.createdTime;
                return lhs.snapshotId > rhs.snapshotId;
            });

            if (stats) {
                stats->lastSnapshotCount = records.size();
            }

            auto it = std::find_if(records.begin(), records.end(),
                                   [&](const auto& rec) { return rec.snapshotId == snapshotId; });
            if (it != records.end()) {
                bool labelOk = true;
                if (!snapshotLabel.empty()) {
                    auto labelIt = it->metadata.find("snapshot_label");
                    labelOk = (labelIt != it->metadata.end() && labelIt->second == snapshotLabel);
                }

                std::size_t docsByLabelCount = 0;
                if (minDocsByLabel > 0 && !snapshotLabel.empty()) {
                    auto docsByLabel = repo->findDocumentsBySnapshotLabel(snapshotLabel);
                    if (docsByLabel) {
                        docsByLabelCount = docsByLabel.value().size();
                    }
                    if (stats) {
                        stats->lastDocsByLabelCount = docsByLabelCount;
                    }
                }

                if (labelOk && docsByLabelCount >= minDocsByLabel) {
                    return true;
                }
            }
        }

        std::this_thread::sleep_for(poll);
    }

    return false;
}

} // namespace

// Windows daemon shutdown hangs during WorkCoordinator thread join.
// The tests themselves work - IPC communication succeeds - but cleanup blocks indefinitely.
// Windows daemon IPC does not yet support this path.
// TODO: Remove this skip once Windows daemon shutdown is fixed.
#ifdef _WIN32
#define SKIP_ON_WINDOWS_DAEMON_SHUTDOWN()                                                          \
    SKIP("Windows daemon shutdown hangs - see windows-daemon-ipc-plan.md")
#else
#define SKIP_ON_WINDOWS_DAEMON_SHUTDOWN() ((void)0)
#endif

/**
 * @brief Test fixture using DaemonHarness for proper lifecycle management
 */
class ServicesRetrievalIngestionFixture {
public:
    ServicesRetrievalIngestionFixture() {
        auto unique = std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
        testRoot_ = fs::temp_directory_path() / ("yams_services_it_" + unique);
        fs::create_directories(testRoot_);
        fixtures_ = std::make_unique<yams::test::FixtureManager>(testRoot_ / "fixtures");

        // Ensure test daemon never kills any existing user/system daemon
        killOthersEnv_.emplace("YAMS_DAEMON_KILL_OTHERS", "0");

        harness_ = std::make_unique<yams::test::DaemonHarness>();
    }

    ~ServicesRetrievalIngestionFixture() {
        bool keepArtifacts = std::getenv("YAMS_KEEP_IT_DIR") != nullptr;

        if (harness_) {
            harness_->stop();
            harness_.reset();
        }

        fixtures_.reset();

        if (keepArtifacts) {
            std::cout << "Keeping integration test artifacts under " << testRoot_.string()
                      << " (YAMS_KEEP_IT_DIR set)." << std::endl;
        } else {
            std::error_code ec;
            fs::remove_all(testRoot_, ec);
        }
    }

    bool startDaemon(std::chrono::seconds timeout = 30s) {
        if (!harness_->start(timeout))
            return false;
        socketPath_ = harness_->socketPath();
        storageDir_ = harness_->dataDir();
        return true;
    }

    yams::daemon::YamsDaemon* daemon() const { return harness_ ? harness_->daemon() : nullptr; }

    fs::path testRoot_;
    fs::path socketPath_;
    fs::path storageDir_;
    std::optional<ScopedEnvVar> killOthersEnv_;
    std::unique_ptr<yams::test::DaemonHarness> harness_;
    std::unique_ptr<yams::test::FixtureManager> fixtures_;
};

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "AddViaDaemonAndListGetGrep",
                 "[integration][services][retrieval][ingestion]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    // 1) Create a small file to add
    auto doc = fixtures_->createTextFixture("hello/hello.txt", "hello yams services",
                                            {"services", "smoke"});
    fs::path filePath = doc.path;

    // 2) Add via DocumentIngestionService (daemon-first)
    DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.path = filePath.string();
    opts.recursive = false;
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    opts.explicitDataDir = storageDir_;
    auto addRes = ing.addViaDaemon(opts);
    REQUIRE(addRes);
    REQUIRE_FALSE(addRes.value().hash.empty());

    BarrierStats barrierStats;
    const bool quiescent =
        waitForPostIngestQuiescent(socketPath_, storageDir_, 5000ms, 50ms, &barrierStats);
    CAPTURE(barrierStats.statusPolls, barrierStats.statusErrors, barrierStats.lastPostIngestQueued,
            barrierStats.lastPostIngestInflight, barrierStats.lastPostIngestDrained,
            barrierStats.lastIndexVisible);
    REQUIRE(quiescent);

    // 3) List via RetrievalService
    RetrievalService rsvc;
    RetrievalOptions ropts;
    ropts.socketPath = socketPath_;
    ropts.explicitDataDir = storageDir_;
    ropts.headerTimeoutMs = kRequestTimeoutMs;
    ropts.bodyTimeoutMs = kBodyTimeoutMs;
    ropts.requestTimeoutMs = kRequestTimeoutMs;

    yams::app::services::ListOptions lreq;
    lreq.limit = 25;
    bool found = false;
    for (int attempt = 0; attempt < 40 && !found; ++attempt) {
        auto lres = rsvc.list(lreq, ropts);
        REQUIRE(lres);
        for (const auto& e : lres.value().items) {
            if (e.name == "hello.txt") {
                found = true;
                break;
            }
        }
        if (!found)
            std::this_thread::sleep_for(50ms);
    }
    if (!found) {
        SKIP("List output did not surface hello.txt; eventual consistency issue.");
    }

    // 4) Get by hash
    yams::app::services::GetOptions greq;
    greq.hash = addRes.value().hash;
    greq.metadataOnly = false;
    auto gres = rsvc.get(greq, ropts);
    REQUIRE(gres);
    CHECK((gres.value().name == "hello.txt"));

    // 5) Grep for pattern
    yams::app::services::GrepOptions gpreq;
    gpreq.pattern = "hello";
    gpreq.pathsOnly = true;
    bool grepReady = false;
    for (int attempt = 0; attempt < 40 && !grepReady; ++attempt) {
        auto gpres = rsvc.grep(gpreq, ropts);
        REQUIRE(gpres);
        grepReady = !gpres.value().matches.empty();
        if (!grepReady)
            std::this_thread::sleep_for(50ms);
    }
    if (!grepReady) {
        SKIP("Grep matches not available; skipping on this runner.");
    }

    // 6) Name-smart get
    auto gname = rsvc.getByNameSmart("hello.txt", false, true, false, std::string{}, ropts, {});
    REQUIRE(gname);
    CHECK((gname.value().name == "hello.txt"));

    // 7) Chunked buffer get
    yams::app::services::GetInitOptions ginit;
    ginit.hash = addRes.value().hash;
    ropts.requestTimeoutMs = 30000;
    auto chunked = rsvc.getChunkedBuffer(ginit, 8, ropts);
    REQUIRE(chunked);
    CHECK((chunked.value().content.size() <= 8u));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "AddBatchPreservesOrderAndMixedResults",
                 "[integration][services][retrieval][ingestion][batch]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    auto docA = fixtures_->createTextFixture("batch/a.txt", "alpha", {"batch"});
    auto docB = fixtures_->createTextFixture("batch/b.txt", "bravo", {"batch"});

    DocumentIngestionService ing;
    std::vector<yams::app::services::AddOptions> batch;

    yams::app::services::AddOptions a;
    a.socketPath = socketPath_;
    a.explicitDataDir = storageDir_;
    a.path = docA.path.string();
    a.noEmbeddings = true;
    a.timeoutMs = kRequestTimeoutMs;
    batch.push_back(a);

    yams::app::services::AddOptions bad = a;
    bad.content = "invalid inline content";
    batch.push_back(bad);

    yams::app::services::AddOptions b = a;
    b.path = docB.path.string();
    batch.push_back(b);

    auto batchRes = boost::asio::co_spawn(yams::daemon::GlobalIOContext::global_executor(),
                                          ing.addBatchAsync(batch, 2), boost::asio::use_future)
                        .get();
    REQUIRE((batchRes.results.size() == batch.size()));
    CHECK((batchRes.succeeded == 2));
    CHECK((batchRes.failed == 1));
    REQUIRE(batchRes.results[0]);
    REQUIRE_FALSE(batchRes.results[1]);
    REQUIRE(batchRes.results[2]);

    BarrierStats barrierStats;
    const bool quiescent =
        waitForPostIngestQuiescent(socketPath_, storageDir_, 5000ms, 50ms, &barrierStats);
    CAPTURE(barrierStats.statusPolls, barrierStats.statusErrors, barrierStats.lastPostIngestQueued,
            barrierStats.lastPostIngestInflight, barrierStats.lastPostIngestDrained,
            barrierStats.lastIndexVisible);
    REQUIRE(quiescent);

    RetrievalService rsvc;
    RetrievalOptions ropts;
    ropts.socketPath = socketPath_;
    ropts.explicitDataDir = storageDir_;
    ropts.headerTimeoutMs = kRequestTimeoutMs;
    ropts.bodyTimeoutMs = kBodyTimeoutMs;
    ropts.requestTimeoutMs = kRequestTimeoutMs;

    yams::app::services::GetOptions getA;
    getA.hash = batchRes.results[0].value().hash;
    getA.metadataOnly = true;
    yams::Result<yams::daemon::GetResponse> getARes;
    for (int attempt = 0; attempt < 40; ++attempt) {
        getARes = rsvc.get(getA, ropts);
        if (getARes) {
            break;
        }
        std::this_thread::sleep_for(50ms);
    }
    REQUIRE(getARes);
    CHECK((getARes.value().name == "a.txt"));

    yams::app::services::GetOptions getB;
    getB.hash = batchRes.results[2].value().hash;
    getB.metadataOnly = true;
    yams::Result<yams::daemon::GetResponse> getBRes;
    for (int attempt = 0; attempt < 40; ++attempt) {
        getBRes = rsvc.get(getB, ropts);
        if (getBRes) {
            break;
        }
        std::this_thread::sleep_for(50ms);
    }
    REQUIRE(getBRes);
    CHECK((getBRes.value().name == "b.txt"));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture,
                 "SearchDoesNotPersistRetrievalServedFeedbackEvent",
                 "[integration][services][retrieval][feedback]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    REQUIRE((ctx.metadataRepo != nullptr));

    using yams::app::services::DocumentIngestionService;

    auto doc = fixtures_->createTextFixture("feedback/search_trace.txt", "trace feedback body",
                                            {"feedback", "trace"});

    DocumentIngestionService ing;
    yams::app::services::AddOptions addOpts;
    addOpts.socketPath = socketPath_;
    addOpts.explicitDataDir = storageDir_;
    addOpts.path = doc.path.string();
    addOpts.recursive = false;
    addOpts.noEmbeddings = true;
    addOpts.timeoutMs = kRequestTimeoutMs;
    auto addRes = ing.addViaDaemon(addOpts);
    REQUIRE(addRes);

    BarrierStats barrierStats;
    const bool quiescent =
        waitForPostIngestQuiescent(socketPath_, storageDir_, 5000ms, 50ms, &barrierStats);
    CAPTURE(barrierStats.statusPolls, barrierStats.statusErrors, barrierStats.lastPostIngestQueued,
            barrierStats.lastPostIngestInflight, barrierStats.lastPostIngestDrained,
            barrierStats.lastIndexVisible);
    REQUIRE(quiescent);

    yams::daemon::ClientConfig cfg;
    cfg.socketPath = socketPath_;
    cfg.dataDir = storageDir_;
    cfg.requestTimeout = 5000ms;
    auto client = std::make_shared<yams::daemon::DaemonClient>(cfg);

    auto beforeEvents = ctx.metadataRepo->getRecentFeedbackEvents(50);
    REQUIRE(beforeEvents);
    const auto beforeCount = std::count_if(
        beforeEvents.value().begin(), beforeEvents.value().end(), [](const auto& event) {
            return event.eventType == "retrieval_served" && event.source == "daemon";
        });

    yams::daemon::SearchRequest req;
    req.query = "trace feedback body";
    req.limit = 10;
    req.searchType = "keyword";

    std::promise<yams::Result<yams::daemon::SearchResponse>> prom;
    auto fut = prom.get_future();
    boost::asio::co_spawn(
        yams::daemon::GlobalIOContext::global_executor(),
        [client, req, p = std::move(prom)]() mutable -> boost::asio::awaitable<void> {
            auto r = co_await client->call(req);
            p.set_value(std::move(r));
            co_return;
        },
        boost::asio::detached);

    REQUIRE((fut.wait_for(5s) == std::future_status::ready));
    auto searchRes = fut.get();
    REQUIRE(searchRes);

    auto afterEvents = ctx.metadataRepo->getRecentFeedbackEvents(50);
    REQUIRE(afterEvents);
    const auto afterCount = std::count_if(
        afterEvents.value().begin(), afterEvents.value().end(), [](const auto& event) {
            return event.eventType == "retrieval_served" && event.source == "daemon";
        });
    CHECK((afterCount == beforeCount));
    CHECK((searchRes.value().searchStats["phase_dispatch_feedback_us"] == "0"));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "StatsZeroThenGrowthAndUnreachableHints",
                 "[integration][services][retrieval]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::makeStatsService;
    using yams::app::services::StatsRequest;

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    auto statsSvc = makeStatsService(ctx);

    // Capture baseline stats
    StatsRequest rq{};
    auto s0 = statsSvc->getStats(rq);
    REQUIRE(s0);
    auto baseObjects = s0.value().totalObjects;
    auto baseBytes = s0.value().totalBytes;

    // Ingest one small doc
    auto doc = fixtures_->createTextFixture("stats/one.txt", "alpha beta", {"stats", "it"});
    DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = doc.path.string();
    opts.recursive = false;
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    auto addRes = ing.addViaDaemon(opts);
    REQUIRE(addRes);

    waitForPostIngestQuiescent(socketPath_, storageDir_, 5000ms);

    auto s2 = statsSvc->getStats(StatsRequest{});
    REQUIRE(s2);
    CHECK((s2.value().totalObjects >= baseObjects));
    CHECK((s2.value().totalBytes >= baseBytes));

    // Unreachable daemon should fail cleanly
    {
        DocumentIngestionService ing2;
        yams::app::services::AddOptions bad;
        bad.socketPath =
            fs::path("/tmp") / ("yams-does-not-exist-" + std::to_string(::getpid()) + ".sock");
        bad.explicitDataDir = storageDir_;
        bad.path = doc.path.string();
        bad.recursive = false;
        bad.noEmbeddings = true;
        bad.timeoutMs = kRequestTimeoutMs;
        auto r = ing2.addViaDaemon(bad);
        CHECK_FALSE(r);
    }
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "AddDirectoryWithPatternsAndTags",
                 "[integration][services][directory][patterns]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    // Create fixtures
    auto keepCpp = fixtures_->createTextFixture("ingest/dirA/dirB/keep.cpp",
                                                "int main() { return 0; }\n", {"code", "cpp"});
    auto keepMd = fixtures_->createTextFixture("ingest/dirA/dirB/keep.md",
                                               "# Title\nhello pattern tags\n", {"docs", "md"});
    std::vector<std::byte> skipData = {std::byte{0x00}, std::byte{0x01}, std::byte{0x02}};
    auto skipBin = fixtures_->createBinaryFixture("ingest/dirA/dirB/skip.bin", skipData, {"bin"});
    (void)keepCpp;
    (void)keepMd;
    (void)skipBin;

    DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.path = (fixtures_->root() / "ingest").string();
    opts.recursive = true;
    opts.includePatterns = {"*.cpp", "*.md"};
    opts.excludePatterns = {"*.bin"};
    opts.tags = {"code", "working", "kg"};
    opts.metadata = {{"pbi", "002"}, {"system", "yams"}};
    opts.collection = "workspace";
    opts.noEmbeddings = true;
    opts.explicitDataDir = storageDir_;
    opts.timeoutMs = kRequestTimeoutMs;

    auto addRes = ing.addViaDaemon(opts);
    REQUIRE(addRes);
    const auto& firstResp = addRes.value();
    // Directory ingestion is async: immediate response has documentsAdded=0
    // Actual indexing happens asynchronously; we verify via list operations after quiescence
    CHECK((firstResp.documentsUpdated == 0));

    waitForPostIngestQuiescent(socketPath_, storageDir_, 7000ms);

    // Verify via list
    RetrievalService rsvc;
    RetrievalOptions ropts;
    ropts.socketPath = socketPath_;
    yams::app::services::ListOptions lreq;
    lreq.limit = 10;
    lreq.namePattern = (fixtures_->root() / "ingest" / "**").string();
    lreq.showTags = true;
    lreq.showMetadata = true;

    bool haveCpp = false, haveMd = false, skipPresent = false;
    yams::daemon::ListEntry keepCppEntry, keepMdEntry;

    for (int attempt = 0; attempt < 40; ++attempt) {
        auto lres = rsvc.list(lreq, ropts);
        REQUIRE(lres);
        const auto& items = lres.value().items;
        for (const auto& e : items) {
            if (e.name == "keep.cpp") {
                haveCpp = true;
                keepCppEntry = e;
            }
            if (e.name == "keep.md") {
                haveMd = true;
                keepMdEntry = e;
            }
            if (e.name == "skip.bin")
                skipPresent = true;
        }
        if (haveCpp && haveMd && !skipPresent)
            break;
        std::this_thread::sleep_for(50ms);
    }

    REQUIRE(haveCpp);
    REQUIRE(haveMd);
    CHECK_FALSE(skipPresent);

    // Check tags/metadata were persisted
    CHECK(std::find(keepCppEntry.tags.begin(), keepCppEntry.tags.end(), "code") !=
          keepCppEntry.tags.end());
    auto cppMeta = keepCppEntry.metadata.find("pbi");
    REQUIRE((cppMeta != keepCppEntry.metadata.end()));
    CHECK((cppMeta->second == "002"));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "LightIndexInvalidHashTolerantError",
                 "[integration][services][search][error]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    auto searchSvc = yams::app::services::makeSearchService(ctx);

    std::string bogus = "notahashvalue";
    auto r = searchSvc->lightIndexForHash(bogus);
    REQUIRE_FALSE(r);

    std::string msg = r.error().message;
    if (msg.find("Worker executor not available") != std::string::npos) {
        SKIP("SearchService worker executor unavailable on this runner.");
    }
    bool ok = (msg.find("not found") != std::string::npos) ||
              (msg.find("Invalid") != std::string::npos) ||
              (msg.find("invalid") != std::string::npos);
    CHECK(ok);
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "IndexingAddDirectoryVerifyAndDefer",
                 "[integration][services][indexing]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::AddDirectoryRequest;
    using yams::app::services::makeIndexingService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    auto dir = fixtures_->root() / "idx";
    fs::create_directories(dir);
    std::ofstream(dir / "a.md") << "alpha";
    std::ofstream(dir / "b.bin") << std::string(4, '\0');

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    auto idxSvc = makeIndexingService(ctx);

    if (!yams::test::capability::indexing_available(ctx)) {
        SKIP("IndexingService capability not available on this runner.");
    }

    AddDirectoryRequest req;
    req.directoryPath = dir.string();
    req.includePatterns = {"*.md", "*.bin"};
    req.excludePatterns = {"*.bin"};
    req.tags = {"idx", "phase2"};
    req.metadata = {{"pbi", "028-45"}};
    req.recursive = true;
    req.verify = true;

    auto ar = idxSvc->addDirectory(req);
    if (!ar) {
        SKIP("IndexingService.addDirectory unavailable: " + ar.error().message);
    }
    const auto& resp = ar.value();
    if (resp.filesProcessed == 0) {
        SKIP("IndexingService processed=0 (capability gap)");
    }
    CHECK((resp.filesIndexed + resp.filesFailed + resp.filesSkipped >= 1));
    if (resp.filesIndexed == 0 && resp.filesFailed > 0) {
        SKIP("IndexingService reports failure for all files.");
    }
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "RetrievalGetToFileWritesContent",
                 "[integration][services][retrieval]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    auto doc = fixtures_->createTextFixture("rtv/one.txt", "retrieval content", {"retrieval"});

    DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = doc.path.string();
    opts.recursive = false;
    opts.noEmbeddings = true;
    auto addRes = ing.addViaDaemon(opts);
    REQUIRE(addRes);

    RetrievalService rsvc;
    RetrievalOptions ro;
    ro.socketPath = socketPath_;
    ro.explicitDataDir = storageDir_;

    yams::app::services::GetInitOptions gi;
    gi.hash = addRes.value().hash;
    auto outPath = testRoot_ / "out_one.txt";

    if (auto chunked = rsvc.getChunkedBuffer(gi, 8, ro); chunked) {
        CHECK((chunked.value().content.size() > 0u));
        return;
    }
    auto wr = rsvc.getToFile(gi, outPath, ro);
    if (!wr) {
        SKIP("RetrievalService getToFile/chunked unavailable: " + wr.error().message);
    }

    std::error_code ec;
    if (!fs::exists(outPath, ec)) {
        SKIP("Output path not created.");
    }
    auto sz = fs::file_size(outPath, ec);
    if (ec) {
        SKIP("Cannot stat output path: " + ec.message());
    }
    CHECK((sz > 0u));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "BinaryChunkedGetBytesRoundTrip",
                 "[integration][services][retrieval]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    fs::create_directories(fixtures_->root() / "bin");
    auto p = fixtures_->root() / "bin" / "b.dat";
    std::string bytes;
    bytes.push_back('\0');
    bytes.push_back((char)0xFF);
    bytes.push_back((char)0x10);
    bytes.push_back('A');
    bytes.push_back('B');
    bytes.push_back('C');
    bytes.push_back('\n');
    bytes.push_back((char)0x7F);
    {
        std::ofstream out(p, std::ios::binary);
        out.write(bytes.data(), static_cast<std::streamsize>(bytes.size()));
    }

    yams::app::services::DocumentIngestionService ing;
    yams::app::services::AddOptions a;
    a.socketPath = socketPath_;
    a.explicitDataDir = storageDir_;
    a.path = p.string();
    a.recursive = false;
    a.noEmbeddings = true;
    a.timeoutMs = kRequestTimeoutMs;
    auto addRes = ing.addViaDaemon(a);
    REQUIRE(addRes);

    yams::app::services::RetrievalService rsvc;
    yams::app::services::RetrievalOptions ropts;
    ropts.socketPath = socketPath_;
    ropts.explicitDataDir = storageDir_;
    ropts.headerTimeoutMs = kRequestTimeoutMs;
    ropts.bodyTimeoutMs = kBodyTimeoutMs;
    ropts.requestTimeoutMs = kRequestTimeoutMs;

    yams::app::services::GetInitOptions gi{};
    gi.name = p.filename().string();
    gi.maxBytes = 0;
    gi.chunkSize = 4;
    auto cr = rsvc.getChunkedBuffer(gi, 16, ropts);
    if (!cr) {
        SKIP("Chunked retrieval unavailable: " + cr.error().message);
    }
    auto out = cr.value().content;
    if (out.empty()) {
        SKIP("Chunked retrieval returned empty content.");
    }
    REQUIRE((bytes.size() >= out.size()));
    CHECK((out == bytes.substr(0, out.size())));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "RetrievalListRespectsLimitAndPattern",
                 "[integration][services][retrieval]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    fs::create_directories(fixtures_->root() / "list" / "d");
    fixtures_->createTextFixture("list/d/a.md", "alpha", {"list", "md"});
    fixtures_->createTextFixture("list/d/b.md", "bravo", {"list", "md"});
    fixtures_->createTextFixture("list/d/c.txt", "charlie", {"list", "txt"});

    DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = (fixtures_->root() / "list").string();
    opts.recursive = true;
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    auto addRes = ing.addViaDaemon(opts);
    REQUIRE(addRes);

    RetrievalService rsvc;
    RetrievalOptions ro;
    ro.socketPath = socketPath_;
    ro.explicitDataDir = storageDir_;
    ro.headerTimeoutMs = kRequestTimeoutMs;
    ro.bodyTimeoutMs = kBodyTimeoutMs;
    ro.requestTimeoutMs = kRequestTimeoutMs;

    yams::app::services::ListOptions lreq;
    lreq.limit = 1;
    lreq.namePattern = (fixtures_->root() / "list" / "**" / "*.md").string();

    auto lres = rsvc.list(lreq, ro);
    REQUIRE(lres);
    const auto& items = lres.value().items;
    REQUIRE((items.size() <= 1));
    for (const auto& e : items) {
        CHECK((e.name.rfind(".md") != std::string::npos));
    }
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "RetrievalListEchoesTotalCountAndStructure",
                 "[integration][services][retrieval][list][structure]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;
    using yams::app::services::RetrievalOptions;
    using yams::app::services::RetrievalService;

    fs::create_directories(fixtures_->root() / "list2");
    fixtures_->createTextFixture("list2/one.txt", "one", {"l2"});
    fixtures_->createTextFixture("list2/two.txt", "two", {"l2"});
    fixtures_->createTextFixture("list2/three.md", "three", {"l2"});

    DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = (fixtures_->root() / "list2").string();
    opts.recursive = true;
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    REQUIRE(ing.addViaDaemon(opts));

    RetrievalService rsvc;
    RetrievalOptions ro;
    ro.socketPath = socketPath_;
    ro.explicitDataDir = storageDir_;
    ro.headerTimeoutMs = kRequestTimeoutMs;
    ro.bodyTimeoutMs = kBodyTimeoutMs;
    ro.requestTimeoutMs = kRequestTimeoutMs;

    yams::app::services::ListOptions lq;
    lq.limit = 2;
    lq.showTags = true;
    lq.showMetadata = true;

    auto lr = rsvc.list(lq, ro);
    REQUIRE(lr);
    const auto& resp = lr.value();
    CHECK((resp.items.size() <= static_cast<size_t>(lq.limit)));
    CHECK((resp.totalCount >= static_cast<uint64_t>(resp.items.size())));
    for (const auto& e : resp.items) {
        CHECK_FALSE(e.name.empty());
        CHECK_FALSE(e.path.empty());
    }
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture,
                 "AppDocumentList_HotOnlyStillHydratesMetadataWhenRequested",
                 "[integration][services][document][list][hot_only]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    ScopedEnvVar listMode("YAMS_LIST_MODE", "hot_only");
    REQUIRE(startDaemon());

    using yams::app::services::ListDocumentsRequest;
    using yams::app::services::makeDocumentService;

    fs::create_directories(fixtures_->root() / "hot_list");
    fixtures_->createTextFixture("hot_list/one.txt", "one", {"it-list-hot"});

    yams::app::services::DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = (fixtures_->root() / "hot_list").string();
    opts.recursive = true;
    // Apply a tag via ingestion so we can assert metadata hydration under hot_only list mode.
    opts.tags = {"it-list-hot"};
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    REQUIRE(ing.addViaDaemon(opts));

    BarrierStats barrierStats;
    const bool quiescent =
        waitForPostIngestQuiescent(socketPath_, storageDir_, 8000ms, 50ms, &barrierStats);
    CAPTURE(quiescent, barrierStats.statusPolls, barrierStats.statusErrors,
            barrierStats.lastPostIngestQueued, barrierStats.lastPostIngestInflight,
            barrierStats.lastPostIngestDrained, barrierStats.lastIndexVisible);

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    auto docSvc = makeDocumentService(ctx);

    ListDocumentsRequest rq;
    rq.limit = 10;
    rq.pattern = (fixtures_->root() / "hot_list" / "**").string();
    rq.showSnippets = false;
    rq.showTags = true;
    rq.showMetadata = true;

    // In hot_only mode the list path may depend on post-ingest indexing visibility.
    // Retry briefly to reduce flakiness.
    decltype(docSvc->list(rq)) r1;
    std::optional<std::reference_wrapper<const yams::app::services::ListDocumentsResponse>> resp;
    bool gotDocs = false;
    for (int i = 0; i < 40; ++i) {
        r1 = docSvc->list(rq);
        REQUIRE(r1);
        if (!r1.value().documents.empty()) {
            resp = std::cref(r1.value());
            gotDocs = true;
            break;
        }
        std::this_thread::sleep_for(50ms);
    }
    REQUIRE(gotDocs);

    bool foundTagMetadata = false;
    for (const auto& e : resp->get().documents) {
        if (e.metadata.contains("tag:it-list-hot")) {
            foundTagMetadata = true;
            break;
        }
    }
    CHECK(foundTagMetadata);
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "AppDocumentListEchoDetailsAndBurst",
                 "[integration][services][document][list][burst]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::ListDocumentsRequest;
    using yams::app::services::makeDocumentService;

    fs::create_directories(fixtures_->root() / "app_list");
    fixtures_->createTextFixture("app_list/a.txt", "alpha", {"app-list"});
    fixtures_->createTextFixture("app_list/b.md", "bravo", {"app-list"});

    yams::app::services::DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = (fixtures_->root() / "app_list").string();
    opts.recursive = true;
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    REQUIRE(ing.addViaDaemon(opts));

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    auto docSvc = makeDocumentService(ctx);

    ListDocumentsRequest rq;
    rq.limit = 2;
    rq.pattern = (fixtures_->root() / "app_list" / "**").string();
    rq.showSnippets = false;
    rq.showMetadata = false;
    rq.showTags = true;

    auto r1 = docSvc->list(rq);
    REQUIRE(r1);
    const auto& resp = r1.value();
    REQUIRE((resp.documents.size() <= 2));
    CHECK((resp.totalFound >= resp.documents.size()));

    // Burst: call list repeatedly
    for (int i = 0; i < 25; ++i) {
        auto ri = docSvc->list(rq);
        REQUIRE(ri);
        CHECK((ri.value().documents.size() <= 2));
    }
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture,
                 "SnapshotContract_NonDirectoryAddsPersistTreeSnapshot",
                 "[integration][services][snapshot][add]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    REQUIRE((ctx.metadataRepo != nullptr));

    const std::string snapshotId =
        "it-snap-nondir-" +
        std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
    const std::string snapshotLabel = "it-snapshot-nondir";

    auto doc1 = fixtures_->createTextFixture("snapshot/non_dir/one.txt", "snapshot one", {"snap"});
    auto doc2 = fixtures_->createTextFixture("snapshot/non_dir/two.txt", "snapshot two", {"snap"});

    DocumentIngestionService ing;

    yams::app::services::AddOptions addFileOne;
    addFileOne.socketPath = socketPath_;
    addFileOne.explicitDataDir = storageDir_;
    addFileOne.path = doc1.path.string();
    addFileOne.recursive = false;
    addFileOne.noEmbeddings = true;
    addFileOne.timeoutMs = kRequestTimeoutMs;
    addFileOne.snapshotId = snapshotId;
    addFileOne.snapshotLabel = snapshotLabel;
    REQUIRE(ing.addViaDaemon(addFileOne));

    yams::app::services::AddOptions addFileTwo = addFileOne;
    addFileTwo.path = doc2.path.string();
    REQUIRE(ing.addViaDaemon(addFileTwo));

    yams::app::services::AddOptions addStdinLike;
    addStdinLike.socketPath = socketPath_;
    addStdinLike.explicitDataDir = storageDir_;
    addStdinLike.content = "snapshot stdin-like content";
    addStdinLike.name = "stdin_like.txt";
    addStdinLike.recursive = false;
    addStdinLike.noEmbeddings = true;
    addStdinLike.timeoutMs = kRequestTimeoutMs;
    addStdinLike.snapshotId = snapshotId;
    addStdinLike.snapshotLabel = snapshotLabel;
    REQUIRE(ing.addViaDaemon(addStdinLike));

    BarrierStats barrierStats;
    waitForPostIngestQuiescent(socketPath_, storageDir_, 8000ms, 50ms, &barrierStats);
    const bool snapshotVisible = waitForSnapshotVisible(
        ctx.metadataRepo.get(), snapshotId, snapshotLabel, 3, 8000ms, 25ms, &barrierStats);
    CAPTURE(snapshotVisible, barrierStats.snapshotPolls, barrierStats.lastSnapshotCount,
            barrierStats.lastDocsByLabelCount);
    REQUIRE(snapshotVisible);

    auto snaps = ctx.metadataRepo->listTreeSnapshots(200);
    REQUIRE(snaps);
    auto it = std::find_if(snaps.value().begin(), snaps.value().end(),
                           [&](const auto& rec) { return rec.snapshotId == snapshotId; });
    REQUIRE((it != snaps.value().end()));
    CHECK((it->metadata.count("snapshot_label") > 0));
    CHECK((it->metadata.at("snapshot_label") == snapshotLabel));

    auto docsByLabel = ctx.metadataRepo->findDocumentsBySnapshotLabel(snapshotLabel);
    REQUIRE(docsByLabel);
    CHECK((docsByLabel.value().size() >= 3));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture,
                 "SnapshotContract_DirectoryAddPreservesProvidedSnapshotFields",
                 "[integration][services][snapshot][add][directory]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    using yams::app::services::DocumentIngestionService;

    auto* sm = daemon()->getServiceManager();
    REQUIRE((sm != nullptr));
    auto ctx = sm->getAppContext();
    REQUIRE((ctx.metadataRepo != nullptr));

    const std::string snapshotId =
        "it-snap-dir-" +
        std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
    const std::string snapshotLabel = "it-snapshot-directory";

    fs::create_directories(fixtures_->root() / "snapshot" / "dir_add");
    fixtures_->createTextFixture("snapshot/dir_add/a.txt", "a", {"snapdir"});
    fixtures_->createTextFixture("snapshot/dir_add/b.txt", "b", {"snapdir"});

    DocumentIngestionService ing;
    yams::app::services::AddOptions dirOpts;
    dirOpts.socketPath = socketPath_;
    dirOpts.explicitDataDir = storageDir_;
    dirOpts.path = (fixtures_->root() / "snapshot" / "dir_add").string();
    dirOpts.recursive = true;
    dirOpts.noEmbeddings = true;
    dirOpts.timeoutMs = kRequestTimeoutMs;
    dirOpts.snapshotId = snapshotId;
    dirOpts.snapshotLabel = snapshotLabel;
    dirOpts.waitForProcessing = true;
    dirOpts.waitTimeoutSeconds = 10;
    REQUIRE(ing.addViaDaemon(dirOpts));

    BarrierStats barrierStats;
    const bool quiescent =
        waitForPostIngestQuiescent(socketPath_, storageDir_, 5000ms, 50ms, &barrierStats);
    CAPTURE(barrierStats.statusPolls, barrierStats.statusErrors, barrierStats.lastPostIngestQueued,
            barrierStats.lastPostIngestInflight, barrierStats.lastPostIngestDrained,
            barrierStats.lastIndexVisible);
    REQUIRE(quiescent);

    const bool snapshotVisible = waitForSnapshotVisible(
        ctx.metadataRepo.get(), snapshotId, snapshotLabel, 0, 5000ms, 25ms, &barrierStats);
    CAPTURE(barrierStats.snapshotPolls, barrierStats.lastSnapshotCount,
            barrierStats.lastDocsByLabelCount);
    REQUIRE(snapshotVisible);

    auto snaps = ctx.metadataRepo->listTreeSnapshots(200);
    REQUIRE(snaps);
    auto it = std::find_if(snaps.value().begin(), snaps.value().end(),
                           [&](const auto& rec) { return rec.snapshotId == snapshotId; });
    REQUIRE((it != snaps.value().end()));
    CHECK((it->metadata.count("snapshot_label") > 0));
    CHECK((it->metadata.at("snapshot_label") == snapshotLabel));
}

TEST_CASE_METHOD(ServicesRetrievalIngestionFixture, "StressTail",
                 "[integration][services][stress]") {
    SKIP_ON_WINDOWS_DAEMON_SHUTDOWN();
    REQUIRE(startDaemon());

    auto p = fixtures_->createTextFixture("stress/tail.txt", "hello", {"stress"});
    yams::app::services::DocumentIngestionService ing;
    yams::app::services::AddOptions opts;
    opts.socketPath = socketPath_;
    opts.explicitDataDir = storageDir_;
    opts.path = p.path.string();
    opts.recursive = false;
    opts.noEmbeddings = true;
    opts.timeoutMs = kRequestTimeoutMs;
    REQUIRE(ing.addViaDaemon(opts));

    yams::app::services::RetrievalService rsvc;
    yams::app::services::RetrievalOptions ro;
    ro.socketPath = socketPath_;
    ro.explicitDataDir = storageDir_;
    ro.headerTimeoutMs = kRequestTimeoutMs;
    ro.bodyTimeoutMs = kBodyTimeoutMs;
    ro.requestTimeoutMs = kRequestTimeoutMs;

    yams::app::services::ListOptions lq;
    lq.limit = 5;
    lq.namePattern = (fixtures_->root() / "stress" / "**").string();

    int stress_iters = 100;
    if (const char* s = std::getenv("YAMS_STRESS_ITERS")) {
        int v = std::atoi(s);
        if (v > 0 && v < 100000)
            stress_iters = v;
    }

    int ok = 0;
    for (int i = 0; i < stress_iters; ++i) {
        auto lr = rsvc.list(lq, ro);
        if (lr)
            ++ok;
        std::this_thread::sleep_for(5ms);
    }
    CHECK((ok > 0));
}