#include <algorithm>
#include <chrono>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <map>
#include <memory>
#include <optional>
#include <set>
#include <stdexcept>
#include <string>
#include <string_view>
#include <tuple>
#include <utility>
#include <vector>
#include <yams/compat/unistd.h>

#include "../common/test_helpers_catch2.h"

#include <nlohmann/json.hpp>
#include <numeric>
#include <unordered_map>
#include <unordered_set>

#include "../integration/daemon/test_daemon_harness.h"

#include <yams/app/services/document_ingestion_service.h>
#include <yams/cli/cli_sync.h>
#include <yams/daemon/client/daemon_client.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/TuneAdvisor.h>
#include <yams/daemon/metric_keys.h>
#include <yams/daemon/resource/model_provider.h>
#include <yams/vector/simeon_embedding_backend.h>

#include <spdlog/spdlog.h>

namespace fs = std::filesystem;
using json = nlohmann::json;
using yams::app::services::AddOptions;
using yams::app::services::DocumentIngestionService;

namespace {

class BenchmarkMockModelProvider final : public yams::daemon::IModelProvider {
public:
    explicit BenchmarkMockModelProvider(std::size_t dim) : dim_(dim) {}

    void setProgressCallback(std::function<void(const yams::daemon::ModelLoadEvent&)> cb) override {
        progress_ = std::move(cb);
    }

    yams::Result<std::vector<float>> generateEmbedding(const std::string& text) override {
        return generateEmbeddingFor(defaultModelName_, text);
    }

    yams::Result<std::vector<std::vector<float>>>
    generateBatchEmbeddings(const std::vector<std::string>& texts) override {
        return generateBatchEmbeddingsFor(defaultModelName_, texts);
    }

    yams::Result<std::vector<float>> generateEmbeddingFor(const std::string& modelName,
                                                          const std::string& text) override {
        if (!isModelLoaded(modelName)) {
            return yams::Error{yams::ErrorCode::NotFound, "model not loaded"};
        }
        std::vector<float> embedding(dim_, 0.0f);
        std::hash<std::string> hasher;
        const auto seed = hasher(text);
        for (std::size_t i = 0; i < embedding.size(); ++i) {
            embedding[i] = static_cast<float>((seed ^ (i * 131)) % 1000) / 1000.0f;
        }
        return embedding;
    }

    yams::Result<std::vector<std::vector<float>>>
    generateBatchEmbeddingsFor(const std::string& modelName,
                               const std::vector<std::string>& texts) override {
        if (!isModelLoaded(modelName)) {
            return yams::Error{yams::ErrorCode::NotFound, "model not loaded"};
        }
        std::vector<std::vector<float>> batch;
        batch.reserve(texts.size());
        for (const auto& text : texts) {
            auto embedding = generateEmbeddingFor(modelName, text);
            if (!embedding) {
                return embedding.error();
            }
            batch.push_back(std::move(embedding.value()));
        }
        return batch;
    }

    yams::Result<void> loadModel(const std::string& modelName) override {
        if (progress_) {
            yams::daemon::ModelLoadEvent ev;
            ev.modelName = modelName;
            ev.phase = "loading";
            ev.message = "benchmark mock loading";
            progress_(ev);
        }
        loadedModels_.insert(modelName);
        defaultModelName_ = modelName;
        if (progress_) {
            yams::daemon::ModelLoadEvent ev;
            ev.modelName = modelName;
            ev.phase = "completed";
            ev.message = "benchmark mock ready";
            progress_(ev);
        }
        return yams::Result<void>();
    }

    yams::Result<void> unloadModel(const std::string& modelName) override {
        loadedModels_.erase(modelName);
        return yams::Result<void>();
    }

    bool isModelLoaded(const std::string& modelName) const override {
        return loadedModels_.contains(modelName);
    }

    std::vector<std::string> getLoadedModels() const override {
        return std::vector<std::string>(loadedModels_.begin(), loadedModels_.end());
    }

    size_t getLoadedModelCount() const override { return loadedModels_.size(); }

    yams::Result<yams::daemon::ModelInfo>
    getModelInfo(const std::string& modelName) const override {
        if (!isModelLoaded(modelName)) {
            return yams::Error{yams::ErrorCode::NotFound, "model not loaded"};
        }
        yams::daemon::ModelInfo info;
        info.name = modelName;
        info.embeddingDim = dim_;
        return info;
    }

    size_t getEmbeddingDim(const std::string&) const override { return dim_; }
    std::shared_ptr<yams::vector::EmbeddingGenerator>
    getEmbeddingGenerator(const std::string& = "") override {
        return nullptr;
    }
    std::string getProviderName() const override { return "BenchmarkMockProvider"; }
    std::string getProviderVersion() const override { return "1.0"; }
    bool isAvailable() const override { return true; }
    size_t getMemoryUsage() const override { return 0; }
    void releaseUnusedResources() override {}
    void shutdown() override { loadedModels_.clear(); }

private:
    std::size_t dim_;
    std::string defaultModelName_;
    std::unordered_set<std::string> loadedModels_;
    std::function<void(const yams::daemon::ModelLoadEvent&)> progress_;
};

struct RunConfig {
    std::string label;
    int workers{1};
    int repeat{1};
    bool enableEmbeddings{false};
    bool requireTopologyFresh{false};
    bool requireSearchProbe{false};
    std::string embeddingModel{"simeon-default"};
    int searchProbeTimeoutSecs{120};
    std::vector<std::string> args;
};

struct BenchConfig {
    fs::path datasetPath;
    fs::path manifestPath;
    fs::path metricsPath;
    bool postRunCleanup{true};
    std::vector<RunConfig> runs;
};

json loadJsonFile(const fs::path& path) {
    std::ifstream in(path);
    if (!in) {
        throw std::runtime_error("Failed to open config file: " + path.string());
    }
    json data;
    in >> data;
    return data;
}

fs::path normalizePath(const fs::path& base, const fs::path& value) {
    fs::path resolved = value;
    if (resolved.is_relative()) {
        resolved = base / resolved;
    }
    return fs::absolute(resolved).lexically_normal();
}

BenchConfig loadConfig(const fs::path& configPath) {
    const json root = loadJsonFile(configPath);
    BenchConfig cfg;

    const fs::path base =
        configPath.parent_path().empty() ? fs::current_path() : configPath.parent_path();

    cfg.datasetPath =
        normalizePath(base, root.value("dataset_path", std::string("tests/data/ingestion")));
    cfg.manifestPath = normalizePath(
        base, root.value("fixture_manifest", std::string("data/benchmarks/fixtures.json")));
    cfg.metricsPath =
        normalizePath(base, root.value("output_metrics",
                                       std::string("data/benchmarks/ingestion_baseline.jsonl")));
    cfg.postRunCleanup = root.value("post_run_cleanup", true);

    for (const auto& entry : root.at("runs")) {
        RunConfig run;
        run.label = entry.value("label", std::string("baseline"));
        run.workers = entry.value("workers", 1);
        run.repeat = std::max(1, entry.value("repeat", 1));
        run.enableEmbeddings = entry.value("enable_embeddings", false);
        run.requireTopologyFresh = entry.value("require_topology_fresh", run.enableEmbeddings);
        run.embeddingModel = entry.value("embedding_model", std::string("simeon-default"));
        if (entry.contains("args")) {
            for (const auto& arg : entry["args"]) {
                run.args.push_back(arg.get<std::string>());
            }
        }
        cfg.runs.push_back(std::move(run));
    }

    return cfg;
}

std::string isoTimestamp() {
    const auto now = std::chrono::system_clock::now();
    const auto tt = std::chrono::system_clock::to_time_t(now);
    std::tm tm{};
#if defined(_WIN32)
    gmtime_s(&tm, &tt);
#else
    gmtime_r(&tt, &tm);
#endif
    std::ostringstream oss;
    oss << std::put_time(&tm, "%Y-%m-%dT%H:%M:%SZ");
    return oss.str();
}

std::vector<std::string> splitCsv(const std::string& csv) {
    std::vector<std::string> tokens;
    std::string current;
    std::istringstream iss(csv);
    while (std::getline(iss, current, ',')) {
        if (!current.empty()) {
            tokens.push_back(current);
        }
    }
    return tokens;
}

struct IngestOptions {
    bool recursive{true};
    bool verify{false};
    std::vector<std::string> includePatterns;
    std::vector<std::string> excludePatterns;
    std::vector<std::string> tags;
    std::map<std::string, std::string> metadata;
    std::string collection;
};

IngestOptions parseArgs(const std::vector<std::string>& args) {
    IngestOptions opts;
    for (const auto& arg : args) {
        if (arg == "--recursive") {
            opts.recursive = true;
        } else if (arg == "--no-recursive") {
            opts.recursive = false;
        } else if (arg.rfind("--include=", 0) == 0) {
            auto val = arg.substr(std::string("--include=").size());
            auto patterns = splitCsv(val);
            opts.includePatterns.insert(opts.includePatterns.end(), patterns.begin(),
                                        patterns.end());
        } else if (arg.rfind("--exclude=", 0) == 0) {
            auto val = arg.substr(std::string("--exclude=").size());
            auto patterns = splitCsv(val);
            opts.excludePatterns.insert(opts.excludePatterns.end(), patterns.begin(),
                                        patterns.end());
        } else if (arg.rfind("--tags=", 0) == 0) {
            auto val = arg.substr(std::string("--tags=").size());
            auto tags = splitCsv(val);
            opts.tags.insert(opts.tags.end(), tags.begin(), tags.end());
        } else if (arg.rfind("--collection=", 0) == 0) {
            opts.collection = arg.substr(std::string("--collection=").size());
        } else if (arg == "--verify") {
            opts.verify = true;
        } else if (arg == "--no-verify") {
            opts.verify = false;
        } else if (arg.rfind("--metadata=", 0) == 0) {
            auto kv = arg.substr(std::string("--metadata=").size());
            auto pos = kv.find('=');
            if (pos != std::string::npos) {
                auto key = kv.substr(0, pos);
                auto value = kv.substr(pos + 1);
                if (!key.empty()) {
                    opts.metadata.emplace(std::move(key), std::move(value));
                }
            }
        }
    }
    return opts;
}

size_t countFiles(const fs::path& root) {
    if (!fs::exists(root)) {
        throw std::runtime_error("Dataset directory does not exist: " + root.string());
    }
    size_t total = 0;
    for (const auto& entry : fs::recursive_directory_iterator(root)) {
        if (entry.is_regular_file()) {
            ++total;
        }
    }
    return total;
}

void ensureMetricsPath(const fs::path& path) {
    if (!path.parent_path().empty()) {
        fs::create_directories(path.parent_path());
    }
    std::ofstream touch(path, std::ios::app);
    if (!touch) {
        throw std::runtime_error("Failed to open metrics output: " + path.string());
    }
}

void deleteTree(const fs::path& root) {
    std::error_code ec;
    fs::remove_all(root, ec);
    if (ec) {
        std::cerr << "[bench] Warning: failed to remove " << root << ": " << ec.message() << "\n";
    }
}

using ScopedEnv = yams::test::ScopedEnvVar;

struct StatusSnapshot {
    uint64_t documentsTotal{0};
    uint64_t documentsIndexed{0};
    uint64_t workerQueued{0};
    uint64_t workerActive{0};
    uint64_t postIngestQueued{0};
    uint64_t postIngestInflight{0};
    uint64_t postIngestProcessed{0};
    uint64_t postIngestFailed{0};
    uint64_t embedQueued{0};
    uint64_t embedInflight{0};
    uint64_t topologyDirtyDocuments{0};
    uint64_t topologyLastSuccessAgeMs{0};
    uint64_t topologyRebuildLagMs{0};
    uint64_t topologyRebuildRunningAgeMs{0};
    uint64_t topologyLastDurationMs{0};
    uint64_t topologyRebuildsTotal{0};
    uint64_t topologyRebuildFailuresTotal{0};
    uint64_t topologyLastDocumentsRequested{0};
    uint64_t topologyLastDocumentsProcessed{0};
    uint64_t topologyLastClustersBuilt{0};
    uint64_t topologyLastMembershipsBuilt{0};
    uint64_t topologyLastDirtySeedCount{0};
    uint64_t topologyLastDirtyRegionDocs{0};
    uint64_t topologyLastCoalescedDirtySets{0};
    uint64_t topologyLastFallbackFullRebuilds{0};
    bool topologyArtifactsFresh{false};
    bool topologyRebuildRunning{false};
    bool simeonLexicalReady{false};
    bool simeonLexicalConfigured{false};

    [[nodiscard]] bool pipelineIdle(bool requireTopologyFresh) const {
        const bool topologyIdle =
            !requireTopologyFresh ||
            (topologyArtifactsFresh && !topologyRebuildRunning && topologyDirtyDocuments == 0);
        return workerQueued == 0 && workerActive == 0 && postIngestQueued == 0 &&
               postIngestInflight == 0 && topologyIdle;
    }

    [[nodiscard]] bool searchable(bool requireTopologyFresh) const {
        return pipelineIdle(requireTopologyFresh) && embedQueued == 0 && embedInflight == 0 &&
               (!requireTopologyFresh || (topologyArtifactsFresh && !topologyRebuildRunning &&
                                          topologyDirtyDocuments == 0)) &&
               (!simeonLexicalConfigured || simeonLexicalReady);
    }

    static StatusSnapshot capture(yams::daemon::DaemonClient& client) {
        StatusSnapshot snapshot;
        auto statusResult = yams::cli::run_sync(client.status(), std::chrono::seconds(5));
        if (!statusResult) {
            return snapshot;
        }

        const auto& status = statusResult.value();
        auto getCount = [&](std::string_view key) -> uint64_t {
            auto it = status.requestCounts.find(std::string(key));
            return it != status.requestCounts.end() ? it->second : 0;
        };
        auto getReady = [&](std::string_view key) -> bool {
            auto it = status.readinessStates.find(std::string(key));
            return it != status.readinessStates.end() ? it->second : false;
        };

        snapshot.documentsTotal = getCount(yams::daemon::metrics::kDocumentsTotal);
        snapshot.documentsIndexed = getCount(yams::daemon::metrics::kDocumentsIndexed);
        snapshot.workerQueued = getCount(yams::daemon::metrics::kWorkerQueued);
        snapshot.workerActive = getCount(yams::daemon::metrics::kWorkerActive);
        snapshot.postIngestQueued = getCount(yams::daemon::metrics::kPostIngestQueued);
        snapshot.postIngestInflight = getCount(yams::daemon::metrics::kPostIngestInflight);
        snapshot.postIngestProcessed = getCount(yams::daemon::metrics::kPostIngestProcessed);
        snapshot.postIngestFailed = getCount(yams::daemon::metrics::kPostIngestFailed);
        snapshot.embedQueued = getCount(yams::daemon::metrics::kEmbedQueued);
        snapshot.embedInflight = getCount(yams::daemon::metrics::kEmbedInflight);
        snapshot.topologyDirtyDocuments = getCount(yams::daemon::metrics::kTopologyDirtyDocuments);
        snapshot.topologyLastSuccessAgeMs =
            getCount(yams::daemon::metrics::kTopologyLastSuccessAgeMs);
        snapshot.topologyRebuildLagMs = getCount(yams::daemon::metrics::kTopologyRebuildLagMs);
        snapshot.topologyRebuildRunningAgeMs =
            getCount(yams::daemon::metrics::kTopologyRebuildRunningAgeMs);
        snapshot.topologyLastDurationMs = getCount(yams::daemon::metrics::kTopologyLastDurationMs);
        snapshot.topologyRebuildsTotal = getCount(yams::daemon::metrics::kTopologyRebuildsTotal);
        snapshot.topologyRebuildFailuresTotal =
            getCount(yams::daemon::metrics::kTopologyRebuildFailuresTotal);
        snapshot.topologyLastDocumentsRequested =
            getCount(yams::daemon::metrics::kTopologyLastDocumentsRequested);
        snapshot.topologyLastDocumentsProcessed =
            getCount(yams::daemon::metrics::kTopologyLastDocumentsProcessed);
        snapshot.topologyLastClustersBuilt =
            getCount(yams::daemon::metrics::kTopologyLastClustersBuilt);
        snapshot.topologyLastMembershipsBuilt =
            getCount(yams::daemon::metrics::kTopologyLastMembershipsBuilt);
        snapshot.topologyLastDirtySeedCount =
            getCount(yams::daemon::metrics::kTopologyLastDirtySeedCount);
        snapshot.topologyLastDirtyRegionDocs =
            getCount(yams::daemon::metrics::kTopologyLastDirtyRegionDocs);
        snapshot.topologyLastCoalescedDirtySets =
            getCount(yams::daemon::metrics::kTopologyLastCoalescedDirtySets);
        snapshot.topologyLastFallbackFullRebuilds =
            getCount(yams::daemon::metrics::kTopologyLastFallbackFullRebuilds);
        snapshot.topologyArtifactsFresh =
            getReady(yams::daemon::readiness::kTopologyArtifactsFresh);
        snapshot.topologyRebuildRunning =
            getReady(yams::daemon::readiness::kTopologyRebuildRunning);
        snapshot.simeonLexicalConfigured =
            getReady(yams::daemon::readiness::kSearchEngineLexicalEnhancementConfigured);
        snapshot.simeonLexicalReady =
            getReady(yams::daemon::readiness::kSearchEngineLexicalEnhancementReady);
        return snapshot;
    }
};

bool probeDaemonSearch(yams::daemon::DaemonClient& client, const std::string& query,
                       const std::string& searchType, const std::string& expectedPath) {
    using namespace yams::daemon;
    SearchRequest req;
    req.query = query;
    req.searchType = searchType;
    req.limit = 5;
    req.pathsOnly = true;
    req.timeout = std::chrono::milliseconds(5000);
    auto result = yams::cli::run_sync(client.search(req), std::chrono::seconds(6));
    if (!result)
        return false;
    for (const auto& r : result.value().results) {
        if (expectedPath.empty())
            return true;
        if (!r.path.empty() && r.path.find(expectedPath) != std::string::npos)
            return true;
    }
    return false;
}

static std::string randomToken() {
    static const char* cs = "abcdefghijklmnopqrstuvwxyz0123456789";
    thread_local std::mt19937_64 rng{std::random_device{}()};
    std::uniform_int_distribution<size_t> dist(0, 35);
    std::string out;
    for (int i = 0; i < 16; ++i)
        out.push_back(cs[dist(rng)]);
    return out;
}

std::string createSentinelFile(const fs::path& rootDir, std::string& tokenOut) {
    tokenOut = "yams_bench_probe_" + randomToken();
    fs::path sentinelPath = rootDir / ".yams_bench_probe.txt";
    std::error_code ec;
    fs::create_directories(rootDir, ec);
    std::ofstream out(sentinelPath);
    if (!out) {
        throw std::runtime_error("Failed to create sentinel file: " + sentinelPath.string());
    }
    out << "YAMS_BENCH_SENTINEL_TOKEN: " << tokenOut << "\n";
    out << "This file exists solely for the ingestion throughput benchmark search probe.\n";
    return sentinelPath.string();
}

struct SearchableWaitResult {
    bool completed{false};
    bool keywordHit{false};
    bool hybridHit{false};
    StatusSnapshot lastSnapshot{};
    std::chrono::nanoseconds keywordHitAt{};
    std::chrono::nanoseconds hybridHitAt{};
    std::chrono::nanoseconds drainedAt{};
    std::chrono::nanoseconds searchableAt{};
    std::string probeQuery;
    std::string probeExpectedPath;
};

SearchableWaitResult waitForSearchable(yams::daemon::DaemonClient& client,
                                       std::chrono::seconds timeout, bool requireTopologyFresh,
                                       bool requireSearchProbe, const std::string& probeToken,
                                       const std::string& probeExpectedPath) {
    auto deadline = std::chrono::steady_clock::now() + timeout;
    SearchableWaitResult result;
    result.probeQuery = probeToken;
    result.probeExpectedPath = probeExpectedPath;
    StatusSnapshot previous;
    StatusSnapshot last;
    bool havePrevious = false;
    int stableCount = 0;
    constexpr int kStableRequired = 3;
    bool keywordFound = !requireSearchProbe;
    bool hybridFound = !requireSearchProbe;
    const auto start = std::chrono::steady_clock::now();

    while (std::chrono::steady_clock::now() < deadline) {
        StatusSnapshot current = StatusSnapshot::capture(client);
        last = current;

        if (!keywordFound && requireSearchProbe && !probeToken.empty()) {
            if (probeDaemonSearch(client, probeToken, "keyword", probeExpectedPath)) {
                keywordFound = true;
                if (result.keywordHitAt == std::chrono::nanoseconds{}) {
                    result.keywordHitAt = std::chrono::steady_clock::now() - start;
                }
            }
        }
        if (!hybridFound && requireSearchProbe && !probeToken.empty() && keywordFound) {
            if (probeDaemonSearch(client, probeToken, "hybrid", probeExpectedPath)) {
                hybridFound = true;
                if (result.hybridHitAt == std::chrono::nanoseconds{}) {
                    result.hybridHitAt = std::chrono::steady_clock::now() - start;
                }
            }
        }

        bool searchable = current.documentsTotal > 0 && current.searchable(requireTopologyFresh);
        if (searchable && havePrevious) {
            searchable = current.documentsTotal == previous.documentsTotal &&
                         current.documentsIndexed == previous.documentsIndexed &&
                         current.postIngestProcessed == previous.postIngestProcessed &&
                         current.postIngestFailed == previous.postIngestFailed &&
                         current.topologyRebuildsTotal == previous.topologyRebuildsTotal &&
                         current.topologyDirtyDocuments == previous.topologyDirtyDocuments;
        }

        if (searchable) {
            ++stableCount;
            if (stableCount >= kStableRequired && keywordFound && hybridFound) {
                result.completed = true;
                result.lastSnapshot = current;
                result.keywordHit = keywordFound;
                result.hybridHit = hybridFound;
                result.searchableAt = std::chrono::steady_clock::now() - start;
                return result;
            }
        } else {
            stableCount = 0;
        }

        previous = current;
        havePrevious = true;
        std::this_thread::sleep_for(std::chrono::milliseconds(25));
    }

    result.lastSnapshot = last;
    result.keywordHit = keywordFound;
    result.hybridHit = hybridFound;
    return result;
}

bool useMockEmbeddingsForBench() {
    if (const char* env = std::getenv("YAMS_BENCH_FORCE_MOCK_EMBEDDINGS")) {
        return std::string(env) == "1";
    }
    return false;
}

std::optional<fs::path> writeMockEmbeddingConfig(const fs::path& root) {
    std::error_code ec;
    fs::create_directories(root, ec);
    const fs::path cfgPath = root / "mock_embeddings.toml";
    std::ofstream out(cfgPath);
    if (!out) {
        return std::nullopt;
    }
    out << "[embeddings]\n";
    out << "embedding_dim = 384\n";
    out << "preferred_model = \"all-MiniLM-L6-v2\"\n";
    return cfgPath;
}

void ensureBenchmarkEmbeddingsReady(yams::daemon::YamsDaemon* daemon, bool enableEmbeddings,
                                    bool useMockEmbeddings, const std::string& embeddingModel) {
    if (!enableEmbeddings || !daemon) {
        return;
    }

    auto serviceManager = daemon->getServiceManager();
    if (!serviceManager) {
        return;
    }

    if (useMockEmbeddings) {
        std::size_t dim = 384;
        if (auto vectorDb = serviceManager->getVectorDatabase()) {
            dim = vectorDb->getConfig().embedding_dim;
        }
        auto sharedProvider =
            std::make_shared<BenchmarkMockModelProvider>(std::max<std::size_t>(dim, 1));
        serviceManager->__test_setModelProvider(sharedProvider);
    }

    const std::string model = embeddingModel.empty() ? "simeon-default" : embeddingModel;
    auto ready = serviceManager->ensureEmbeddingModelReadySync(model, {}, 10000, false, false);
    if (!ready) {
        spdlog::warn("[bench] Embedding model not ready for throughput run: {}",
                     ready.error().message);
    }
}

struct RunResult {
    int exitCode{0};
    double durationSeconds{0.0};
    double throughputFilesPerSecond{0.0};
    double timeToKeywordHitMs{0.0};
    double timeToHybridHitMs{0.0};
    double timeToDrainedMs{0.0};
    double timeToSearchableMs{0.0};
    std::string command;
    fs::path dataDir;
    StatusSnapshot finalSnapshot{};
    bool drained{false};
    bool timedOut{false};
    bool keywordHit{false};
    bool hybridHit{false};
    bool topologyValidated{false};
    std::string probeToken;
    std::string probeExpectedPath;
    std::uint64_t addDocumentsAdded{0};
    std::string addMessage;
    std::uint64_t embedQueued{0};
    std::uint64_t embedConsumed{0};
    std::uint64_t embedDropped{0};
    std::uint64_t embedPreparedDocsQueued{0};
    std::uint64_t embedPreparedChunksQueued{0};
    std::uint64_t embedHashOnlyDocsQueued{0};
    std::uint64_t postQueued{0};
    std::uint64_t postConsumed{0};
    std::uint64_t postDropped{0};
    std::uint64_t kgQueued{0};
    std::uint64_t kgConsumed{0};
    std::uint64_t kgDropped{0};
    std::uint64_t symbolQueued{0};
    std::uint64_t symbolConsumed{0};
    std::uint64_t symbolDropped{0};
    std::uint64_t entityQueued{0};
    std::uint64_t entityConsumed{0};
    std::uint64_t entityDropped{0};
    std::uint64_t titleQueued{0};
    std::uint64_t titleConsumed{0};
    std::uint64_t titleDropped{0};
};

RunResult executeRun(const BenchConfig& cfg, const RunConfig& run, size_t datasetCount,
                     int iteration) {
    (void)datasetCount;

    fs::path runRoot = yams::test::make_temp_dir("yams_ingest_bench_") /
                       (run.label + "_" + std::to_string(iteration));
    fs::path runDataDir = runRoot / "data";
    fs::create_directories(runDataDir);

    std::string workerStr = std::to_string(std::max(1, run.workers));
    ScopedEnv envWorkers("YAMS_INDEXING_WORKERS", workerStr);
    ScopedEnv envLegacy("YAMS_BENCH_INDEX_WORKERS", workerStr);
    ScopedEnv envParallel("YAMS_ENABLE_PARALLEL_INGEST", "1");
    ScopedEnv envPoolSize("YAMS_STORAGE_POOL_SIZE", "8");
    ScopedEnv envSafeSingle("YAMS_TEST_SAFE_SINGLE_INSTANCE", "1");
    const bool useMockEmbeddings = useMockEmbeddingsForBench();
    std::optional<ScopedEnv> envDisableVectors;
    if (!run.enableEmbeddings) {
        envDisableVectors.emplace("YAMS_DISABLE_VECTORS", "1");
    }
    ScopedEnv envDisableWatcher("YAMS_DISABLE_SESSION_WATCHER", "1");
    std::optional<ScopedEnv> envSkipModelLoading;
    if (!run.enableEmbeddings) {
        envSkipModelLoading.emplace("YAMS_SKIP_MODEL_LOADING", "1");
    }
    std::optional<ScopedEnv> envUseMockProvider;
    std::optional<ScopedEnv> envEmbedBackend;
    const bool useSimeon = run.embeddingModel == "simeon-default" || run.embeddingModel == "simeon";
    if (run.enableEmbeddings && !useMockEmbeddings && useSimeon) {
        envEmbedBackend.emplace("YAMS_EMBED_BACKEND", "simeon");
    }
    if (run.enableEmbeddings && useMockEmbeddings) {
        envUseMockProvider.emplace("YAMS_USE_MOCK_PROVIDER", "1");
    }
    std::optional<ScopedEnv> envMockEmbedDim;
    if (run.enableEmbeddings && useMockEmbeddings) {
        envMockEmbedDim.emplace("YAMS_EMBED_DIM", "384");
    }
    std::optional<fs::path> benchConfigPath;
    if (run.enableEmbeddings && useMockEmbeddings) {
        benchConfigPath = writeMockEmbeddingConfig(runRoot);
    }

    yams::daemon::TuneAdvisor::setEnableParallelIngest(true);
    yams::daemon::TuneAdvisor::setMaxIngestWorkers(static_cast<uint32_t>(std::max(1, run.workers)));
    yams::daemon::TuneAdvisor::setStoragePoolSize(8);

    const auto opts = parseArgs(run.args);

    yams::test::DaemonHarness::Options harnessOptions;
    harnessOptions.enableModelProvider = true;
    harnessOptions.useMockModelProvider = !run.enableEmbeddings || useMockEmbeddings;
    harnessOptions.autoLoadPlugins = run.enableEmbeddings && !useMockEmbeddings;
    harnessOptions.configureModelPool = run.enableEmbeddings && !useMockEmbeddings;
    harnessOptions.modelPoolLazyLoading = false;
    if (run.enableEmbeddings && !useMockEmbeddings) {
        harnessOptions.preloadModels = {useSimeon ? std::string{"simeon-default"}
                                                  : run.embeddingModel};
        if (const char* envPluginDir = std::getenv("YAMS_PLUGIN_DIR")) {
            harnessOptions.pluginDir = fs::path(envPluginDir);
        } else {
            harnessOptions.pluginDir = fs::current_path() / "builddir" / "plugins";
        }
    }
    harnessOptions.enableAutoRepair = false;
    harnessOptions.isolateState = true;
    harnessOptions.dataDir = runDataDir;
    if (benchConfigPath.has_value()) {
        harnessOptions.configPath = *benchConfigPath;
    }

    yams::test::DaemonHarness harness(harnessOptions);
    if (!harness.start(std::chrono::seconds(30), [](yams::daemon::YamsDaemon*) {})) {
        throw std::runtime_error("Failed to start daemon harness for ingestion benchmark");
    }
    ensureBenchmarkEmbeddingsReady(harness.daemon(), run.enableEmbeddings, useMockEmbeddings,
                                   useSimeon ? std::string{"simeon-default"} : run.embeddingModel);

    yams::daemon::ClientConfig clientCfg;
    clientCfg.socketPath = harness.socketPath();
    clientCfg.autoStart = false;
    yams::daemon::DaemonClient client(clientCfg);

    std::string probeToken;
    std::string probeExpectedPath;
    if (run.requireSearchProbe) {
        probeExpectedPath = createSentinelFile(runRoot, probeToken);
        spdlog::info("[bench] Search probe: token={} path={}", probeToken, probeExpectedPath);
    }

    DocumentIngestionService ingestion;
    AddOptions addOptions;
    addOptions.socketPath = harness.socketPath();
    addOptions.explicitDataDir = harness.dataDir();
    addOptions.path = cfg.datasetPath.string();
    addOptions.recursive = opts.recursive;
    addOptions.includePatterns = opts.includePatterns;
    addOptions.excludePatterns = opts.excludePatterns;
    addOptions.tags = opts.tags;
    addOptions.metadata = opts.metadata;
    addOptions.collection = opts.collection;
    addOptions.verify = opts.verify;
    addOptions.noEmbeddings = !run.enableEmbeddings;
    addOptions.timeoutMs = 30000;
    addOptions.retries = 2;

    int searchableTimeoutSecs = run.searchProbeTimeoutSecs;
    if (const char* envDrain = std::getenv("YAMS_BENCH_DRAIN_TIMEOUT_S"); envDrain && *envDrain) {
        try {
            searchableTimeoutSecs = std::max(10, std::stoi(envDrain));
        } catch (...) {
        }
    }

    auto& bus = yams::daemon::InternalEventBus::instance();
    const auto baseEmbedQueued = bus.embedQueued();
    const auto baseEmbedConsumed = bus.embedConsumed();
    const auto baseEmbedDropped = bus.embedDropped();
    const auto baseEmbedPreparedDocsQueued = bus.embedPreparedDocsQueued();
    const auto baseEmbedPreparedChunksQueued = bus.embedPreparedChunksQueued();
    const auto baseEmbedHashOnlyDocsQueued = bus.embedHashOnlyDocsQueued();
    const auto basePostQueued = bus.postQueued();
    const auto basePostConsumed = bus.postConsumed();
    const auto basePostDropped = bus.postDropped();
    const auto baseKgQueued = bus.kgQueued();
    const auto baseKgConsumed = bus.kgConsumed();
    const auto baseKgDropped = bus.kgDropped();
    const auto baseSymbolQueued = bus.symbolQueued();
    const auto baseSymbolConsumed = bus.symbolConsumed();
    const auto baseSymbolDropped = bus.symbolDropped();
    const auto baseEntityQueued = bus.entityQueued();
    const auto baseEntityConsumed = bus.entityConsumed();
    const auto baseEntityDropped = bus.entityDropped();
    const auto baseTitleQueued = bus.titleQueued();
    const auto baseTitleConsumed = bus.titleConsumed();
    const auto baseTitleDropped = bus.titleDropped();

    if (run.requireSearchProbe && !probeToken.empty()) {
        AddOptions sentinelOpts = addOptions;
        sentinelOpts.path = probeExpectedPath;
        sentinelOpts.recursive = false;
        auto sentinelResult = ingestion.addViaDaemon(sentinelOpts);
        if (!sentinelResult) {
            throw std::runtime_error("Sentinel ingest failed: " + sentinelResult.error().message);
        }
        spdlog::info("[bench] Sentinel document ingested: {}", probeToken);
    }

    const auto start = std::chrono::steady_clock::now();
    auto addResult = ingestion.addViaDaemon(addOptions);
    if (!addResult) {
        throw std::runtime_error("Daemon ingestion failed: " + addResult.error().message);
    }
    auto waitResult = waitForSearchable(client, std::chrono::seconds(searchableTimeoutSecs),
                                        run.requireTopologyFresh, run.requireSearchProbe,
                                        probeToken, probeExpectedPath);
    const auto end = std::chrono::steady_clock::now();

    harness.stop();

    const double elapsed = std::chrono::duration<double>(end - start).count();
    const uint64_t documentsTotal = waitResult.lastSnapshot.documentsTotal;
    const double throughput =
        (documentsTotal > 0 && elapsed > 0.0) ? static_cast<double>(documentsTotal) / elapsed : 0.0;

    auto nsToMs = [](std::chrono::nanoseconds ns) -> double {
        return ns.count() > 0 ? std::chrono::duration<double, std::milli>(ns).count() : 0.0;
    };

    std::ostringstream cmd;
    cmd << "daemon_ingest_full_pipeline label=" << run.label << " workers=" << run.workers
        << " embeddings=" << (run.enableEmbeddings ? 1 : 0);
    if (!opts.includePatterns.empty()) {
        cmd << " include=";
        for (size_t i = 0; i < opts.includePatterns.size(); ++i) {
            if (i)
                cmd << ',';
            cmd << opts.includePatterns[i];
        }
    }

    RunResult result;
    result.exitCode = 0;
    result.durationSeconds = elapsed;
    result.throughputFilesPerSecond = throughput;
    result.command = cmd.str();
    result.dataDir = runRoot;
    result.finalSnapshot = waitResult.lastSnapshot;
    result.drained = waitResult.completed;
    result.timedOut = !waitResult.completed;
    result.keywordHit = waitResult.keywordHit;
    result.hybridHit = waitResult.hybridHit;
    result.topologyValidated = run.requireTopologyFresh;
    result.probeToken = probeToken;
    result.probeExpectedPath = probeExpectedPath;
    result.timeToKeywordHitMs = nsToMs(waitResult.keywordHitAt);
    result.timeToHybridHitMs = nsToMs(waitResult.hybridHitAt);
    result.timeToDrainedMs = nsToMs(waitResult.drainedAt);
    result.timeToSearchableMs = nsToMs(waitResult.searchableAt);
    result.addDocumentsAdded = addResult.value().documentsAdded;
    result.addMessage = addResult.value().message;
    result.embedQueued = bus.embedQueued() - baseEmbedQueued;
    result.embedConsumed = bus.embedConsumed() - baseEmbedConsumed;
    result.embedDropped = bus.embedDropped() - baseEmbedDropped;
    result.embedPreparedDocsQueued = bus.embedPreparedDocsQueued() - baseEmbedPreparedDocsQueued;
    result.embedPreparedChunksQueued =
        bus.embedPreparedChunksQueued() - baseEmbedPreparedChunksQueued;
    result.embedHashOnlyDocsQueued = bus.embedHashOnlyDocsQueued() - baseEmbedHashOnlyDocsQueued;
    result.postQueued = bus.postQueued() - basePostQueued;
    result.postConsumed = bus.postConsumed() - basePostConsumed;
    result.postDropped = bus.postDropped() - basePostDropped;
    result.kgQueued = bus.kgQueued() - baseKgQueued;
    result.kgConsumed = bus.kgConsumed() - baseKgConsumed;
    result.kgDropped = bus.kgDropped() - baseKgDropped;
    result.symbolQueued = bus.symbolQueued() - baseSymbolQueued;
    result.symbolConsumed = bus.symbolConsumed() - baseSymbolConsumed;
    result.symbolDropped = bus.symbolDropped() - baseSymbolDropped;
    result.entityQueued = bus.entityQueued() - baseEntityQueued;
    result.entityConsumed = bus.entityConsumed() - baseEntityConsumed;
    result.entityDropped = bus.entityDropped() - baseEntityDropped;
    result.titleQueued = bus.titleQueued() - baseTitleQueued;
    result.titleConsumed = bus.titleConsumed() - baseTitleConsumed;
    result.titleDropped = bus.titleDropped() - baseTitleDropped;

    const bool isSimeonFullPipeline = run.enableEmbeddings && !useMockEmbeddings && useSimeon;
    if (isSimeonFullPipeline && result.titleQueued == 0 && result.titleConsumed == 0) {
        spdlog::error(
            "[bench] Simeon full-pipeline run '{}' has titleQueued==0 and titleConsumed==0 — "
            "title extractor was not wired. This benchmark is misconfigured.",
            run.label);
        result.exitCode = 4;
    }
    return result;
}

void appendMetrics(const fs::path& metricsPath, const RunConfig& run, const RunResult& result,
                   size_t datasetCount, int iteration) {
    std::ofstream out(metricsPath, std::ios::app);
    if (!out) {
        throw std::runtime_error("Failed to open metrics output: " + metricsPath.string());
    }

    const auto indexed = static_cast<std::uint64_t>(result.finalSnapshot.documentsTotal);
    const auto processed = static_cast<std::uint64_t>(datasetCount);
    const auto skipped =
        datasetCount > indexed ? static_cast<std::uint64_t>(datasetCount - indexed) : 0;
    const auto failed = static_cast<std::uint64_t>(0);

    const double throughput = result.throughputFilesPerSecond;

    json record{
        {"timestamp", isoTimestamp()},
        {"label", run.label},
        {"iteration", iteration},
        {"workers", run.workers},
        {"embedding_model", run.enableEmbeddings ? run.embeddingModel : std::string{}},
        {"require_topology_fresh", run.requireTopologyFresh},
        {"embedding_backend", run.enableEmbeddings ? ((run.embeddingModel == "simeon-default" ||
                                                       run.embeddingModel == "simeon")
                                                          ? "simeon"
                                                          : "onnxruntime")
                                                   : "disabled"},
        {"simeon_recipe", run.enableEmbeddings && (run.embeddingModel == "simeon-default" ||
                                                   run.embeddingModel == "simeon")
                              ? yams::vector::simeonRecipeLabel()
                              : std::string{}},
        {"exit_code", result.exitCode},
        {"duration_seconds", result.durationSeconds},
        {"dataset_files", datasetCount},
        {"files_indexed", indexed},
        {"files_processed", processed},
        {"files_skipped", skipped},
        {"files_failed", failed},
        {"throughput_files_per_second", throughput},
        {"time_to_keyword_hit_ms", result.timeToKeywordHitMs},
        {"time_to_hybrid_hit_ms", result.timeToHybridHitMs},
        {"time_to_drained_ms", result.timeToDrainedMs},
        {"time_to_searchable_ms", result.timeToSearchableMs},
        {"keyword_hit", result.keywordHit},
        {"hybrid_hit", result.hybridHit},
        {"probe_token", result.probeToken},
        {"probe_expected_path", result.probeExpectedPath},
        {"timed_out", result.timedOut},
        {"add_documents_added", result.addDocumentsAdded},
        {"add_message", result.addMessage},
        {"documents_total", result.finalSnapshot.documentsTotal},
        {"documents_indexed", result.finalSnapshot.documentsIndexed},
        {"worker_queued", result.finalSnapshot.workerQueued},
        {"worker_active", result.finalSnapshot.workerActive},
        {"post_ingest_queued", result.finalSnapshot.postIngestQueued},
        {"post_ingest_inflight", result.finalSnapshot.postIngestInflight},
        {"post_ingest_processed", result.finalSnapshot.postIngestProcessed},
        {"post_ingest_failed", result.finalSnapshot.postIngestFailed},
        {"pipeline_counters",
         {{"embed",
           {{"queued", result.embedQueued},
            {"consumed", result.embedConsumed},
            {"dropped", result.embedDropped},
            {"prepared_docs_queued", result.embedPreparedDocsQueued},
            {"prepared_chunks_queued", result.embedPreparedChunksQueued},
            {"hash_only_docs_queued", result.embedHashOnlyDocsQueued}}},
          {"post",
           {{"queued", result.postQueued},
            {"consumed", result.postConsumed},
            {"dropped", result.postDropped}}},
          {"kg",
           {{"queued", result.kgQueued},
            {"consumed", result.kgConsumed},
            {"dropped", result.kgDropped}}},
          {"symbol",
           {{"queued", result.symbolQueued},
            {"consumed", result.symbolConsumed},
            {"dropped", result.symbolDropped}}},
          {"entity",
           {{"queued", result.entityQueued},
            {"consumed", result.entityConsumed},
            {"dropped", result.entityDropped}}},
          {"title",
           {{"queued", result.titleQueued},
            {"consumed", result.titleConsumed},
            {"dropped", result.titleDropped}}}}},
        {"topology_validated", result.topologyValidated},
        {"topology_artifacts_fresh", result.finalSnapshot.topologyArtifactsFresh},
        {"topology_rebuild_running", result.finalSnapshot.topologyRebuildRunning},
        {"topology_dirty_documents", result.finalSnapshot.topologyDirtyDocuments},
        {"simeon_lexical_configured", result.finalSnapshot.simeonLexicalConfigured},
        {"simeon_lexical_ready", result.finalSnapshot.simeonLexicalReady},
        {"embed_queued", result.finalSnapshot.embedQueued},
        {"embed_inflight", result.finalSnapshot.embedInflight},
        {"drained", result.drained},
        {"command", result.command},
        {"data_dir", result.dataDir.string()},
    };

    out << record.dump() << '\n';
}

} // namespace

std::unordered_map<std::string, double> loadBaselineThroughput(const fs::path& metricsPath) {
    std::unordered_map<std::string, std::pair<double, int>> accum;
    if (!fs::exists(metricsPath))
        return {};

    std::ifstream in(metricsPath);
    std::string line;
    while (std::getline(in, line)) {
        if (line.empty())
            continue;
        json record = json::parse(line, nullptr, false);
        if (!record.is_object())
            continue;
        if (!record.contains("label") || !record.contains("throughput_files_per_second"))
            continue;
        double throughput = record.value("throughput_files_per_second", 0.0);
        if (throughput <= 0.0)
            continue;
        std::string label = record.value("label", std::string{});
        if (label.empty())
            continue;
        auto& slot = accum[label];
        slot.first += throughput;
        slot.second += 1;
    }

    std::unordered_map<std::string, double> averages;
    for (auto& [label, data] : accum) {
        if (data.second > 0)
            averages[label] = data.first / static_cast<double>(data.second);
    }
    return averages;
}

int main(int argc, char** argv) {
    spdlog::set_level(spdlog::level::warn);
    fs::path configPath;
    for (int i = 1; i < argc; ++i) {
        std::string arg(argv[i]);
        if ((arg == "--config" || arg == "-c") && i + 1 < argc) {
            configPath = fs::path(argv[++i]);
        } else if (arg == "--help" || arg == "-h") {
            std::cout << "Usage: " << argv[0] << " --config <path>\n";
            return 0;
        }
    }

    if (configPath.empty()) {
        std::cerr << "error: --config <path> is required\n";
        return 2;
    }

    try {
        BenchConfig cfg = loadConfig(configPath);
        const size_t datasetCount = countFiles(cfg.datasetPath);
        std::cout << "[bench] Dataset: " << cfg.datasetPath << " files=" << datasetCount
                  << std::endl;

        ensureMetricsPath(cfg.metricsPath);
        auto baseline = loadBaselineThroughput(cfg.metricsPath);
        std::unordered_map<std::string, std::vector<double>> newThroughputs;

        for (const auto& run : cfg.runs) {
            for (int iteration = 0; iteration < run.repeat; ++iteration) {
                RunResult result = executeRun(cfg, run, datasetCount, iteration);
                appendMetrics(cfg.metricsPath, run, result, datasetCount, iteration);
                newThroughputs[run.label].push_back(result.throughputFilesPerSecond);
                if (cfg.postRunCleanup) {
                    deleteTree(result.dataDir);
                }
                if (result.exitCode != 0) {
                    return result.exitCode;
                }
            }
        }

        bool regressionDetected = false;
        for (const auto& [label, samples] : newThroughputs) {
            if (samples.empty())
                continue;
            double newAvg = std::accumulate(samples.begin(), samples.end(), 0.0) /
                            static_cast<double>(samples.size());
            auto it = baseline.find(label);
            if (it != baseline.end() && it->second > 0.0) {
                double threshold = it->second * 0.9;
                if (newAvg < threshold) {
                    std::cerr << "[bench] throughput regression detected for '" << label
                              << "': baseline=" << it->second << " new=" << newAvg
                              << " (threshold=" << threshold << ")" << std::endl;
                    regressionDetected = true;
                }
            }
        }
        if (regressionDetected)
            return 3;
    } catch (const std::exception& ex) {
        std::cerr << "error: " << ex.what() << "\n";
        return 1;
    }

    return 0;
}