/*
  ingestion_e2e_bench.cpp - End-to-End Ingestion Pipeline Benchmark (PBI-004)

  Measures complete ingestion pipeline performance with per-stage timing,
  queue depth monitoring, and throughput characterization.

  Usage:
    YAMS_TEST_SAFE_SINGLE_INSTANCE=1 ./builddir/tests/benchmarks/ingestion_e2e_bench

  Environment variables:
    YAMS_TEST_SAFE_SINGLE_INSTANCE=1  - Required to prevent GlobalIOContext reset
    YAMS_BENCH_CORPUS_SIZE=N          - Number of documents to ingest (default: 100)
    YAMS_BENCH_DOC_SIZE=N             - Document size in bytes (default: 1000)
    YAMS_BENCH_POLL_INTERVAL_MS=N     - Queue polling interval (default: 100)
    YAMS_BENCH_INGEST_MODE=MODE       - directory, pipelined_single_file, or
                                        single_file_serial (default)
    YAMS_BENCH_INGEST_CONCURRENCY=N   - In-flight window for pipelined_single_file
                                        (default: 4)
    YAMS_BENCH_POST_INGEST_COALESCE_MS=N
                                      - Typed post-ingest coalescing window (default: 2)
    YAMS_BENCH_POST_INGEST_BATCH_SIZE=N
                                      - Existing typed batch-size axis (default: product value)
    YAMS_BENCH_OUTPUT=path            - JSON output file (default: stdout)
    YAMS_BENCH_DISABLE_KG=1           - Disable KG post-ingest enrichment for ablation
    YAMS_BENCH_SEARCH_PROBES=0        - Disable fixed post-ingest search probes
    YAMS_BENCH_PHASE=initial|replay|resume
                                      - Select lifecycle contract (default: initial)
    YAMS_BENCH_FIXTURE_ROOT=path      - Externally owned yams_e2e_bench_* temp root used across
                                        fresh processes; caller must remove it

  Metrics collected:
    - Total pipeline duration (start → all stages complete)
    - Per-stage latency: metadata, FTS5, embeddings, KG
    - Queue depths over time (store_document_tasks, embed_jobs, fts5_jobs, postIngest)
    - Throughput (docs/sec) and success/fail counts
    - WorkCoordinator thread pool utilization
    - Fixed post-ingest keyword/semantic/hybrid search probe impact
*/

#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>
#include <fmt/ranges.h>

#include "tests/common/test_helpers_catch2.h"
#include "tests/integration/daemon/test_async_helpers.h"
#include "tests/integration/daemon/test_daemon_harness.h"
#include <boost/asio/awaitable.hpp>
#include <boost/asio/io_context.hpp>
#include <yams/api/content_store.h>
#include <yams/app/services/document_ingestion_service.h>
#include <yams/app/services/services.hpp>
#include <yams/common/fs_utils.h>
#include <yams/daemon/client/daemon_client.h>
#include <yams/daemon/client/global_io_context.h>
#include <yams/daemon/components/InternalEventBus.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/VectorIndexCoordinator.h>
#include <yams/daemon/components/WriteCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/metadata/metadata_repository.h>
#include <yams/storage/reference_counter.h>
#include <yams/storage/reference_counter_writer.h>
#include <yams/vector/simeon_embedding_backend.h>
#include <yams/vector/vector_database.h>

#include <sqlite3.h>

#include <algorithm>
#include <atomic>
#include <bit>
#include <cctype>
#include <charconv>
#include <chrono>
#include <cmath>
#include <cstdio>
#include <cstdlib>
#include <ctime>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <random>
#include <sstream>
#include <string>
#include <string_view>
#include <thread>
#include <unordered_map>
#include <vector>
#include <yams/compat/unistd.h>

#if defined(__APPLE__) || defined(__linux__)
#include <sys/resource.h>
#endif

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

namespace {

constexpr const char* kSimeonModel = "simeon-default";

int64_t nonNegativeDeltaMs(int64_t endMs, int64_t startMs) noexcept {
    return endMs > startMs ? endMs - startMs : 0;
}

uint64_t counterDelta(uint64_t current, uint64_t baseline) noexcept {
    return current > baseline ? current - baseline : 0;
}

uint64_t saturatedAdd(uint64_t lhs, uint64_t rhs) noexcept {
    if (lhs > std::numeric_limits<uint64_t>::max() - rhs) {
        return std::numeric_limits<uint64_t>::max();
    }
    return lhs + rhs;
}

std::string fnvFingerprint(std::uint64_t value) {
    std::ostringstream out;
    out << "fnv1a64:" << std::hex << std::setfill('0') << std::setw(16) << value;
    return out.str();
}

class Fnv1a64 {
public:
    void addBytes(std::string_view value) {
        addInteger(value.size());
        for (const auto byte : value) {
            addByte(static_cast<unsigned char>(byte));
        }
    }

    void addInteger(std::uint64_t value) {
        for (unsigned shift = 0; shift < 64; shift += 8) {
            addByte(static_cast<unsigned char>((value >> shift) & 0xffU));
        }
    }

    void addFloat(float value) { addInteger(std::bit_cast<std::uint32_t>(value)); }

    [[nodiscard]] std::string fingerprint() const { return fnvFingerprint(value_); }

private:
    void addByte(unsigned char byte) {
        value_ ^= byte;
        value_ *= 1099511628211ULL;
    }

    std::uint64_t value_{14695981039346656037ULL};
};

std::uint64_t percentileUs(std::vector<std::uint64_t> samples, double percentile) {
    if (samples.empty()) {
        return 0;
    }
    std::sort(samples.begin(), samples.end());
    const auto rank =
        static_cast<std::size_t>(std::ceil(percentile * static_cast<double>(samples.size())));
    return samples[std::min(samples.size() - 1, rank > 0 ? rank - 1 : 0)];
}

std::uint64_t peakRssBytes() noexcept {
#if defined(__APPLE__) || defined(__linux__)
    rusage usage{};
    if (getrusage(RUSAGE_SELF, &usage) != 0) {
        return 0;
    }
#if defined(__APPLE__)
    return static_cast<std::uint64_t>(usage.ru_maxrss);
#else
    return static_cast<std::uint64_t>(usage.ru_maxrss) * 1024ULL;
#endif
#else
    return 0;
#endif
}

std::uint64_t sidecarBytes(const fs::path& dataDir, std::string_view suffix) {
    std::uint64_t total = 0;
    std::error_code ec;
    for (fs::recursive_directory_iterator it(dataDir, ec), end; !ec && it != end;
         it.increment(ec)) {
        if (!it->is_regular_file(ec)) {
            continue;
        }
        const auto name = it->path().filename().string();
        if (name.size() < suffix.size() ||
            name.compare(name.size() - suffix.size(), suffix.size(), suffix) != 0) {
            continue;
        }
        const auto size = it->file_size(ec);
        if (!ec) {
            total = saturatedAdd(total, static_cast<std::uint64_t>(size));
        }
    }
    return total;
}

bool hasSidecars(const fs::path& dataDir) {
    std::error_code ec;
    for (fs::recursive_directory_iterator it(dataDir, ec), end; !ec && it != end;
         it.increment(ec)) {
        const auto name = it->path().filename().string();
        if (name.ends_with("-wal") || name.ends_with("-shm")) {
            return true;
        }
    }
    return false;
}

bool envFlagEnabled(const char* name) {
    const char* raw = std::getenv(name);
    if (raw == nullptr || *raw == '\0') {
        return false;
    }
    std::string value(raw);
    std::transform(value.begin(), value.end(), value.begin(),
                   [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
    return value == "1" || value == "true" || value == "yes" || value == "on";
}

bool pathWithin(const fs::path& candidate, const fs::path& root) {
    const auto relative = candidate.lexically_relative(root);
    return !relative.empty() && relative.native() != "." && *relative.begin() != "..";
}

std::optional<fs::path> externalFixtureRoot() {
    const char* configured = std::getenv("YAMS_BENCH_FIXTURE_ROOT");
    if (configured == nullptr || *configured == '\0') {
        return std::nullopt;
    }

    auto candidate = fs::weakly_canonical(fs::absolute(configured));
    const auto tempRoot = fs::weakly_canonical(fs::temp_directory_path());
    bool isolated = pathWithin(candidate, tempRoot);
#ifndef _WIN32
    isolated = isolated || pathWithin(candidate, fs::weakly_canonical("/tmp"));
#endif
    const auto name = candidate.filename().string();
    if (!isolated || !name.starts_with("yams_e2e_bench_")) {
        throw std::runtime_error(
            "YAMS_BENCH_FIXTURE_ROOT must be an isolated yams_e2e_bench_* temp directory");
    }
    return candidate;
}

std::string benchPhase() {
    const char* configured = std::getenv("YAMS_BENCH_PHASE");
    const std::string phase = configured != nullptr && *configured != '\0' ? configured : "initial";
    if (phase != "initial" && phase != "replay" && phase != "resume") {
        throw std::runtime_error("YAMS_BENCH_PHASE must be initial, replay, or resume");
    }
    return phase;
}

int benchEnvInt(const char* name, int fallback, int minimum, int maximum) {
    const char* raw = std::getenv(name);
    if (raw == nullptr || *raw == '\0') {
        return fallback;
    }
    const std::string_view value(raw);
    int parsed = 0;
    const auto [end, error] = std::from_chars(value.data(), value.data() + value.size(), parsed);
    return error == std::errc{} && end == value.data() + value.size() && parsed >= minimum &&
                   parsed <= maximum
               ? parsed
               : fallback;
}

std::uint32_t benchPostIngestCoalesceMs() {
    constexpr std::uint32_t kMaxCoalesceMs = 20;
    const auto defaultValue = yams::daemon::TuningConfig{}.postIngestCoalesceMs;
    const char* raw = std::getenv("YAMS_BENCH_POST_INGEST_COALESCE_MS");
    if (!raw || !*raw) {
        return defaultValue;
    }
    try {
        const auto parsed = std::stoul(raw);
        return parsed <= kMaxCoalesceMs ? static_cast<std::uint32_t>(parsed) : defaultValue;
    } catch (const std::exception&) {
        return defaultValue;
    }
}

std::uint32_t benchPostIngestBatchSize() {
    constexpr std::uint32_t kMaxBatchSize = 256;
    const auto defaultValue = yams::daemon::TuneAdvisor::postIngestBatchSize();
    const char* raw = std::getenv("YAMS_BENCH_POST_INGEST_BATCH_SIZE");
    if (!raw || !*raw) {
        return defaultValue;
    }
    try {
        const auto parsed = std::stoul(raw);
        return parsed >= 1 && parsed <= kMaxBatchSize ? static_cast<std::uint32_t>(parsed)
                                                      : defaultValue;
    } catch (const std::exception&) {
        return defaultValue;
    }
}

} // namespace

// ============================================================================
// SimpleDaemonHarness - Minimal daemon lifecycle for benchmarking
// ============================================================================

class SimpleDaemonHarness {
public:
    SimpleDaemonHarness() {
        if (const auto configuredRoot = externalFixtureRoot()) {
            root_ = *configuredRoot;
            externallyOwnedRoot_ = true;
        } else {
            auto id = randomId();
#ifdef _WIN32
            const auto tempBase = fs::temp_directory_path();
#else
            // Keep Unix-domain socket paths well below sockaddr_un::sun_path limits while placing
            // every fixture artifact under one directory for deterministic cleanup.
            const auto tempBase = fs::path("/tmp");
#endif
            root_ = tempBase / ("yams_e2e_bench_" + id);
        }
        yams::common::ensureDirectories(root_);
        data_ = root_ / "data";
        yams::common::ensureDirectories(data_);
        sock_ = root_ / "daemon.sock";
        pid_ = root_ / "bench.pid";
        log_ = root_ / "bench.log";
    }

    ~SimpleDaemonHarness() noexcept {
        try {
            stop();
        } catch (...) {
            std::fputs("SimpleDaemonHarness stop failed during teardown\n", stderr);
        }
        try {
            cleanup();
        } catch (...) {
            std::fputs("SimpleDaemonHarness cleanup failed during teardown\n", stderr);
        }
    }

    SimpleDaemonHarness(const SimpleDaemonHarness&) = delete;
    SimpleDaemonHarness& operator=(const SimpleDaemonHarness&) = delete;

    bool start() {
        if (harness_ || !savedEnvironment_.empty()) {
            spdlog::error("SimpleDaemonHarness start called while a prior start is still active");
            return false;
        }

        // Save original working directory and change to temp dir
        // This prevents session auto-watch from indexing CWD
        originalCwd_ = fs::current_path();
        fs::current_path(root_);

        // Isolate every daemon-owned path. Plugin/model discovery must not fall through to the
        // installed daemon's config, cache, corpus, or embedding selection.
        saveAndUnsetEnvironment("YAMS_CONFIG_PATH");
        saveAndUnsetEnvironment("YAMS_CONFIG");
        saveAndUnsetEnvironment("YAMS_EMBED_BACKEND");
        saveAndUnsetEnvironment("YAMS_PREFERRED_MODEL");
        saveAndSetEnvironment("XDG_STATE_HOME", root_ / "state");
        saveAndSetEnvironment("XDG_CONFIG_HOME", root_ / "config");
        saveAndSetEnvironment("XDG_CACHE_HOME", root_ / "cache");
        saveAndSetEnvironment("YAMS_DATA_DIR", data_);
        yams::common::ensureDirectories(root_ / "state" / "yams" / "sessions");

        const auto configPath = root_ / "config" / "yams" / "config.toml";
        yams::common::ensureDirectories(configPath.parent_path());
        {
            std::ofstream configFile(configPath);
            if (!configFile) {
                spdlog::error("Failed to create isolated benchmark config: {}",
                              configPath.string());
                return false;
            }
            configFile << "config_version = 3\n"
                          "[embeddings]\n"
                          "backend = \"simeon\"\n"
                          "preferred_model = \"simeon-default\"\n";
        }

        // Benchmark embedding-mode contract:
        // - YAMS_BENCH_FORCE_MOCK_EMBEDDINGS=1 forces mock model provider
        // - YAMS_DISABLE_VECTORS=1 also implies mock/FTS-only mode
        const char* forceMockEmbeddings = std::getenv("YAMS_BENCH_FORCE_MOCK_EMBEDDINGS");
        const bool mockEmbeddingsRequested =
            forceMockEmbeddings && std::string(forceMockEmbeddings) == "1";
        const char* disableVectors = std::getenv("YAMS_DISABLE_VECTORS");
        bool vectorsDisabled = disableVectors && std::string(disableVectors) == "1";
        vectorsDisabled = vectorsDisabled || mockEmbeddingsRequested;

        yams::test::DaemonHarness::Options options;
        options.rootDir = root_;
        options.preserveRoot = true;
        options.dataDir = data_;
        options.socketPath = sock_;
        options.pidFile = pid_;
        options.logFile = log_;
        options.configPath = configPath;
        options.enableModelProvider = true;
        options.enableAutoRepair = false;
        options.requireReadyLifecycle = true;
        options.configureDaemon = [](yams::daemon::DaemonConfig& config) {
            config.tuning.postIngestCoalesceMs = benchPostIngestCoalesceMs();
        };

        glinerRequested_ = !envFlagEnabled("YAMS_DISABLE_GLINER_TITLES");
        if (const char* pluginDir = std::getenv("YAMS_BENCH_GLINT_PLUGIN_DIR");
            glinerRequested_ && pluginDir && *pluginDir && fs::is_directory(pluginDir)) {
            options.pluginDir = fs::absolute(pluginDir);
            options.pluginDirStrict = true;
            options.trustedPluginPaths = {*options.pluginDir};
            pluginAutoloadEnabled_ = true;
        }

        if (vectorsDisabled) {
            options.useMockModelProvider = true;
            options.autoLoadPlugins = false;
            spdlog::info("Using mock model provider (YAMS_BENCH_FORCE_MOCK_EMBEDDINGS={} "
                         "YAMS_DISABLE_VECTORS={})",
                         mockEmbeddingsRequested ? 1 : 0, disableVectors ? disableVectors : "0");
        } else {
            // Simeon is the benchmark's fixed production-default embedding provider. Optional
            // Glint autoload exercises the independent GLiNER title stage without changing the
            // embedding space.
            options.useMockModelProvider = false;
            options.autoLoadPlugins = pluginAutoloadEnabled_;
            options.configureModelPool = true;
            options.modelPoolLazyLoading = false;
            options.preloadModels = {kSimeonModel};
            spdlog::info("Using Simeon embedding provider for ingestion benchmark");
        }

        harness_ = std::make_unique<yams::test::DaemonHarness>(std::move(options));
        if (!harness_->startWithRetry(30s, 2, [](yams::daemon::YamsDaemon*) {})) {
            spdlog::error("Failed to start shared daemon harness");
            return false;
        }

        spdlog::info("Daemon ready at socket: {}", sock_.string());
        if (!vectorsDisabled) {
            if (auto* sm = harness_->daemon()->getServiceManager()) {
                auto ready =
                    sm->ensureEmbeddingModelReadySync(kSimeonModel, {}, 10000, false, false);
                if (!ready) {
                    spdlog::warn("Embedding model not ready: {}", ready.error().message);
                }
            }
        }
        return true;
    }

    void stop() {
        if (harness_) {
            harness_->stop();
            shutdownSucceeded_ = harness_->shutdownSucceeded();
            harness_.reset();
        }

        // Restore original working directory only after the shared harness has released daemon
        // resources and restored its own scoped process state.
        if (!originalCwd_.empty()) {
            std::error_code ec;
            fs::current_path(originalCwd_, ec);
        }
        restoreEnvironment();
    }

    [[nodiscard]] const fs::path& socketPath() const { return sock_; }
    [[nodiscard]] const fs::path& root() const { return root_; }
    [[nodiscard]] const fs::path& dataDir() const { return data_; }
    [[nodiscard]] const fs::path& pidPath() const { return pid_; }
    [[nodiscard]] bool glinerRequested() const { return glinerRequested_; }
    [[nodiscard]] bool pluginAutoloadEnabled() const { return pluginAutoloadEnabled_; }
    [[nodiscard]] bool shutdownSucceeded() const { return shutdownSucceeded_; }
    [[nodiscard]] bool stopRequested() const {
        return harness_ && harness_->daemon() && harness_->daemon()->isStopRequested();
    }
    [[nodiscard]] yams::daemon::YamsDaemon* daemon() const {
        return harness_ ? harness_->daemon() : nullptr;
    }

private:
    static std::string randomId() {
        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 < 8; ++i)
            out.push_back(cs[dist(rng)]);
        return out;
    }

    void saveAndSetEnvironment(const std::string& key, const fs::path& value) {
        savedEnvironment_.emplace_back(key, value.string());
    }

    void saveAndUnsetEnvironment(const std::string& key) {
        savedEnvironment_.emplace_back(key, std::nullopt);
    }

    void restoreEnvironment() {
        while (!savedEnvironment_.empty()) {
            savedEnvironment_.pop_back();
        }
    }

    void cleanup() {
        std::error_code ec;
        if (!root_.empty() && !externallyOwnedRoot_)
            fs::remove_all(root_, ec);
    }

    std::unique_ptr<yams::test::DaemonHarness> harness_;
    fs::path root_, data_, sock_, pid_, log_;
    fs::path originalCwd_;
    std::vector<yams::test::ScopedEnvVar> savedEnvironment_;
    bool glinerRequested_{false};
    bool pluginAutoloadEnabled_{false};
    bool shutdownSucceeded_{false};
    bool externallyOwnedRoot_{false};
};

// ============================================================================
// CorpusGenerator - Generate synthetic documents for benchmarking
// ============================================================================

struct CorpusGenerator {
    static constexpr std::uint32_t kSeed = 42;
    static constexpr std::uint64_t kFnvOffsetBasis = 14695981039346656037ULL;
    static constexpr std::uint64_t kFnvPrime = 1099511628211ULL;

    fs::path corpusDir;
    std::vector<std::string> createdFiles;
    std::mt19937 rng{kSeed};
    size_t docSize;
    std::uint64_t generatedBytes{0};
    std::uint64_t fingerprintValue{kFnvOffsetBasis};

    CorpusGenerator(const fs::path& dir, size_t size) : corpusDir(dir), docSize(size) {
        yams::common::ensureDirectories(corpusDir);
    }

    void hashBytes(std::string_view bytes) {
        for (const unsigned char byte : bytes) {
            fingerprintValue ^= byte;
            fingerprintValue *= kFnvPrime;
        }
    }

    std::string fingerprint() const {
        std::ostringstream out;
        out << "fnv1a64:" << std::hex << std::setfill('0') << std::setw(16) << fingerprintValue;
        return out.str();
    }

    void generateDocuments(int count) {
        spdlog::info("Generating {} documents of ~{} bytes each", count, docSize);
        std::uniform_int_distribution<int> wordDist(0, 999);

        for (int i = 0; i < count; ++i) {
            std::ostringstream content;
            content << "Document " << i << " - Synthetic Test Corpus\n\n";

            // Generate content to approximately match docSize
            size_t written = content.str().size();
            while (written < docSize) {
                content << "word" << wordDist(rng) << " ";
                written += 10; // Approximate word length
                if (written % 100 == 0) {
                    content << "\n";
                }
            }

            std::string filename = "doc_" + std::to_string(i) + ".txt";
            const std::string document = content.str();
            std::ofstream(corpusDir / filename) << document;
            generatedBytes = saturatedAdd(generatedBytes, document.size());
            hashBytes(filename);
            hashBytes(document);
            createdFiles.push_back(filename);

            if ((i + 1) % 10 == 0 || (i + 1) == count) {
                spdlog::debug("Generated {}/{} documents", i + 1, count);
            }
        }
        spdlog::info("Corpus generation complete: {} files", createdFiles.size());
    }
};

struct VectorOracleSnapshot {
    bool complete{false};
    std::string error;
    std::uint64_t storedVectors{0};
    std::uint64_t fetchedVectors{0};
    std::uint64_t embeddedDocuments{0};
    std::uint64_t chunkVectors{0};
    std::uint64_t documentVectors{0};
    std::uint64_t invalidDimensions{0};
    std::uint64_t nonFiniteValues{0};
    std::uint64_t searchResultCount{0};
    std::uint64_t nonFiniteSearchScores{0};
    std::string fingerprint;
    std::string searchFingerprint;
};

VectorOracleSnapshot captureVectorOracle(const yams::vector::VectorDatabase& vectorDb,
                                         std::size_t expectedDim) {
    VectorOracleSnapshot snapshot;
    snapshot.storedVectors = vectorDb.getVectorCount();
    auto hashesResult = vectorDb.getEmbeddedDocumentHashesChecked();
    if (!hashesResult) {
        snapshot.error = hashesResult.error().message;
        return snapshot;
    }

    std::vector<std::string> hashes(hashesResult.value().begin(), hashesResult.value().end());
    std::sort(hashes.begin(), hashes.end());
    snapshot.embeddedDocuments = hashes.size();

    Fnv1a64 fingerprint;
    fingerprint.addInteger(expectedDim);
    std::vector<float> canonicalQuery;
    for (const auto& hash : hashes) {
        auto records = vectorDb.getVectorsByDocument(hash);
        std::sort(records.begin(), records.end(), [](const auto& lhs, const auto& rhs) {
            if (lhs.chunk_id != rhs.chunk_id) {
                return lhs.chunk_id < rhs.chunk_id;
            }
            if (lhs.document_hash != rhs.document_hash) {
                return lhs.document_hash < rhs.document_hash;
            }
            if (lhs.level != rhs.level) {
                return static_cast<int>(lhs.level) < static_cast<int>(rhs.level);
            }
            if (lhs.start_offset != rhs.start_offset) {
                return lhs.start_offset < rhs.start_offset;
            }
            if (lhs.end_offset != rhs.end_offset) {
                return lhs.end_offset < rhs.end_offset;
            }
            if (lhs.model_id != rhs.model_id) {
                return lhs.model_id < rhs.model_id;
            }
            if (lhs.model_version != rhs.model_version) {
                return lhs.model_version < rhs.model_version;
            }
            const auto sharedSize = std::min(lhs.embedding.size(), rhs.embedding.size());
            for (std::size_t index = 0; index < sharedSize; ++index) {
                const auto lhsBits = std::bit_cast<std::uint32_t>(lhs.embedding[index]);
                const auto rhsBits = std::bit_cast<std::uint32_t>(rhs.embedding[index]);
                if (lhsBits != rhsBits) {
                    return lhsBits < rhsBits;
                }
            }
            return lhs.embedding.size() < rhs.embedding.size();
        });
        fingerprint.addBytes(hash);
        fingerprint.addInteger(records.size());
        for (const auto& record : records) {
            fingerprint.addBytes(record.chunk_id);
            fingerprint.addBytes(record.document_hash);
            fingerprint.addInteger(static_cast<std::uint64_t>(record.level));
            fingerprint.addInteger(record.start_offset);
            fingerprint.addInteger(record.end_offset);
            fingerprint.addBytes(record.model_id);
            fingerprint.addBytes(record.model_version);
            fingerprint.addInteger(record.embedding.size());
            if (record.embedding.size() != expectedDim) {
                ++snapshot.invalidDimensions;
            }
            for (const auto value : record.embedding) {
                if (!std::isfinite(value)) {
                    ++snapshot.nonFiniteValues;
                }
                fingerprint.addFloat(value);
            }
            if (canonicalQuery.empty() && record.embedding.size() == expectedDim) {
                canonicalQuery = record.embedding;
            }
            ++snapshot.fetchedVectors;
            if (record.level == yams::vector::EmbeddingLevel::DOCUMENT) {
                ++snapshot.documentVectors;
            } else {
                ++snapshot.chunkVectors;
            }
        }
    }
    snapshot.fingerprint = fingerprint.fingerprint();
    if (!canonicalQuery.empty()) {
        yams::vector::VectorSearchParams params;
        params.k = 10;
        params.similarity_threshold = -1.0F;
        const auto results = vectorDb.search(canonicalQuery, params);
        Fnv1a64 searchFingerprint;
        searchFingerprint.addInteger(results.size());
        for (const auto& record : results) {
            searchFingerprint.addBytes(record.chunk_id);
            searchFingerprint.addBytes(record.document_hash);
            if (!std::isfinite(record.relevance_score)) {
                ++snapshot.nonFiniteSearchScores;
            }
            searchFingerprint.addFloat(record.relevance_score);
        }
        snapshot.searchResultCount = results.size();
        snapshot.searchFingerprint = searchFingerprint.fingerprint();
    }
    snapshot.complete = snapshot.fetchedVectors == snapshot.storedVectors &&
                        snapshot.invalidDimensions == 0 && snapshot.nonFiniteValues == 0 &&
                        snapshot.searchResultCount > 0 && snapshot.nonFiniteSearchScores == 0 &&
                        snapshot.error.empty();
    return snapshot;
}

struct VectorPersistenceSnapshot {
    bool readable{false};
    std::string error;
    std::int64_t authoritativeGeneration{-1};
    std::int64_t sourceGeneration{-1};
    std::uint64_t persistedVectorCount{0};
    std::uint64_t codeCount{0};
    bool trained{false};
};

VectorPersistenceSnapshot captureVectorPersistence(const fs::path& databasePath,
                                                   std::size_t dimension) {
    VectorPersistenceSnapshot snapshot;
    sqlite3* db = nullptr;
    const int openRc = sqlite3_open_v2(databasePath.string().c_str(), &db,
                                       SQLITE_OPEN_READONLY | SQLITE_OPEN_NOMUTEX, nullptr);
    if (openRc != SQLITE_OK || db == nullptr) {
        snapshot.error = db ? sqlite3_errmsg(db) : "sqlite open failed";
        if (db) {
            sqlite3_close(db);
        }
        return snapshot;
    }

    auto queryOne = [&](const char* sql, std::int64_t bindValue,
                        std::vector<std::int64_t>& values) -> bool {
        sqlite3_stmt* stmt = nullptr;
        if (sqlite3_prepare_v2(db, sql, -1, &stmt, nullptr) != SQLITE_OK) {
            snapshot.error = sqlite3_errmsg(db);
            return false;
        }
        if (bindValue >= 0) {
            sqlite3_bind_int64(stmt, 1, bindValue);
        }
        const int stepRc = sqlite3_step(stmt);
        if (stepRc == SQLITE_ROW) {
            const int columns = sqlite3_column_count(stmt);
            values.reserve(columns);
            for (int column = 0; column < columns; ++column) {
                values.push_back(sqlite3_column_int64(stmt, column));
            }
        } else if (stepRc != SQLITE_DONE) {
            snapshot.error = sqlite3_errmsg(db);
        }
        sqlite3_finalize(stmt);
        return stepRc == SQLITE_ROW;
    };

    std::vector<std::int64_t> generation;
    const bool generationOk =
        queryOne("SELECT generation FROM vector_index_generation WHERE id = 1", -1, generation);
    std::vector<std::int64_t> pqMeta;
    const bool pqOk = queryOne(
        "SELECT source_generation, trained, vector_count FROM simeon_pq_meta WHERE dim = ?",
        static_cast<std::int64_t>(dimension), pqMeta);
    std::vector<std::int64_t> codes;
    const bool codesOk = queryOne("SELECT COUNT(*) FROM simeon_pq_codes WHERE dim = ?",
                                  static_cast<std::int64_t>(dimension), codes);
    sqlite3_close(db);

    if (generationOk && !generation.empty()) {
        snapshot.authoritativeGeneration = generation[0];
    }
    if (pqOk && pqMeta.size() == 3) {
        snapshot.sourceGeneration = pqMeta[0];
        snapshot.trained = pqMeta[1] != 0;
        snapshot.persistedVectorCount =
            static_cast<std::uint64_t>(std::max<std::int64_t>(0, pqMeta[2]));
    }
    if (codesOk && !codes.empty()) {
        snapshot.codeCount = static_cast<std::uint64_t>(std::max<std::int64_t>(0, codes[0]));
    }
    snapshot.readable = generationOk && pqOk && codesOk;
    return snapshot;
}

// ============================================================================
// QueueSnapshot - Capture queue depths and counters at a point in time
// Uses InternalEventBus::instance() to access daemon-internal state
// ============================================================================

struct QueueSnapshot {
    int64_t timestamp_ms = 0;
    // Queue depths (actual items in queues)
    uint64_t store_document_tasks = 0;
    uint64_t embed_jobs = 0;
    uint64_t fts5_jobs = 0;
    uint64_t post_ingest = 0;
    // Cumulative counters (for stage completion detection)
    uint64_t embed_queued = 0;
    uint64_t embed_consumed = 0;
    uint64_t embed_dropped = 0;
    uint64_t embed_prepared_docs_queued = 0;
    uint64_t embed_prepared_chunks_queued = 0;
    uint64_t embed_hash_only_docs_queued = 0;
    uint64_t kg_jobs = 0;
    uint64_t symbol_jobs = 0;
    uint64_t entity_jobs = 0;
    uint64_t title_jobs = 0;
    uint64_t fts5_queued = 0;
    uint64_t fts5_consumed = 0;
    uint64_t fts5_dropped = 0;
    uint64_t post_queued = 0;
    uint64_t post_consumed = 0;
    uint64_t post_dropped = 0;
    uint64_t kg_queued = 0;
    uint64_t kg_consumed = 0;
    uint64_t kg_dropped = 0;
    uint64_t symbol_queued = 0;
    uint64_t symbol_consumed = 0;
    uint64_t symbol_dropped = 0;
    uint64_t entity_queued = 0;
    uint64_t entity_consumed = 0;
    uint64_t entity_dropped = 0;
    uint64_t title_queued = 0;
    uint64_t title_consumed = 0;
    uint64_t title_dropped = 0;

    json toJson() const {
        return json{
            {"timestamp_ms", timestamp_ms},
            {"queue_depths",
             {{"store_document_tasks", store_document_tasks},
              {"embed_jobs", embed_jobs},
              {"fts5_jobs", fts5_jobs},
              {"post_ingest", post_ingest}}},
            {"enrichment_depths",
             {{"kg_jobs", kg_jobs},
              {"symbol_jobs", symbol_jobs},
              {"entity_jobs", entity_jobs},
              {"title_jobs", title_jobs}}},
            {"counters",
             {{"embed",
               {{"queued", embed_queued},
                {"consumed", embed_consumed},
                {"dropped", embed_dropped},
                {"prepared_docs_queued", embed_prepared_docs_queued},
                {"prepared_chunks_queued", embed_prepared_chunks_queued},
                {"hash_only_docs_queued", embed_hash_only_docs_queued}}},
              {"fts5",
               {{"queued", fts5_queued}, {"consumed", fts5_consumed}, {"dropped", fts5_dropped}}},
              {"post",
               {{"queued", post_queued}, {"consumed", post_consumed}, {"dropped", post_dropped}}},
              {"kg", {{"queued", kg_queued}, {"consumed", kg_consumed}, {"dropped", kg_dropped}}},
              {"symbol",
               {{"queued", symbol_queued},
                {"consumed", symbol_consumed},
                {"dropped", symbol_dropped}}},
              {"entity",
               {{"queued", entity_queued},
                {"consumed", entity_consumed},
                {"dropped", entity_dropped}}},
              {"title",
               {{"queued", title_queued},
                {"consumed", title_consumed},
                {"dropped", title_dropped}}}}}};
    }
};

// ============================================================================
// StageMetrics - Per-stage completion tracking
// ============================================================================

struct StageMetrics {
    std::string name;
    int64_t start_ms = 0;
    int64_t end_ms = 0;
    uint64_t count = 0;
    uint64_t failures = 0;

    int64_t duration_ms() const { return nonNegativeDeltaMs(end_ms, start_ms); }

    json toJson() const {
        return json{{"name", name},
                    {"duration_ms", duration_ms()},
                    {"count", count},
                    {"failures", failures}};
    }
};

struct BenchmarkEmbeddingPhaseTiming {
    static constexpr std::size_t kMaxSamples = 4096;

    std::uint64_t calls{0};
    std::uint64_t totalUs{0};
    std::uint64_t maxUs{0};
    std::vector<std::uint64_t> samplesUs;
    bool samplesTruncated{false};
};

class BenchmarkEmbeddingTimingSink final : public yams::daemon::EmbeddingPhaseTimingSink {
public:
    void record(std::string_view phase, std::uint64_t elapsedUs) override {
        std::lock_guard lock(mutex_);
        auto& timing = timings_[std::string(phase)];
        ++timing.calls;
        timing.totalUs += elapsedUs;
        timing.maxUs = std::max(timing.maxUs, elapsedUs);
        if (timing.samplesUs.size() < BenchmarkEmbeddingPhaseTiming::kMaxSamples) {
            timing.samplesUs.push_back(elapsedUs);
        } else {
            timing.samplesTruncated = true;
        }
    }

    std::unordered_map<std::string, BenchmarkEmbeddingPhaseTiming> snapshot() const {
        std::lock_guard lock(mutex_);
        return timings_;
    }

private:
    mutable std::mutex mutex_;
    std::unordered_map<std::string, BenchmarkEmbeddingPhaseTiming> timings_;
};

struct SearchImpactProbe {
    std::string name;
    std::string query;
    std::string search_type;
    bool required_by_contract = false;
    bool skipped = false;
    std::string skip_reason{};
    bool ok = false;
    std::string error{};
    int64_t wall_ms = 0;
    int64_t completed_ms = 0;
    int64_t daemon_elapsed_ms = 0;
    uint64_t total_count = 0;
    uint64_t returned_count = 0;
    std::vector<std::string> top_ids{};
    std::vector<std::string> top_paths{};
    std::vector<double> top_scores{};
    bool scores_finite = true;

    json toJson() const {
        return json{{"name", name},
                    {"query", query},
                    {"search_type", search_type},
                    {"required_by_contract", required_by_contract},
                    {"skipped", skipped},
                    {"skip_reason", skip_reason},
                    {"ok", ok},
                    {"error", error},
                    {"wall_ms", wall_ms},
                    {"completed_ms", completed_ms},
                    {"daemon_elapsed_ms", daemon_elapsed_ms},
                    {"total_count", total_count},
                    {"returned_count", returned_count},
                    {"top_ids", top_ids},
                    {"top_paths", top_paths},
                    {"top_scores", top_scores},
                    {"scores_finite", scores_finite}};
    }
};

struct AddDispatchMetrics {
    uint64_t samples = 0;
    uint64_t total_us = 0;
    uint64_t max_us = 0;
    uint64_t fingerprint_total_us = 0;
    uint64_t fingerprint_max_us = 0;
    uint64_t enqueue_total_us = 0;
    uint64_t enqueue_max_us = 0;

    json toJson() const {
        const auto average = [this](uint64_t total) {
            return samples == 0 ? 0.0 : static_cast<double>(total) / static_cast<double>(samples);
        };
        return json{{"samples", samples},
                    {"total_us", total_us},
                    {"max_us", max_us},
                    {"avg_us", average(total_us)},
                    {"fingerprint_total_us", fingerprint_total_us},
                    {"fingerprint_max_us", fingerprint_max_us},
                    {"fingerprint_avg_us", average(fingerprint_total_us)},
                    {"enqueue_total_us", enqueue_total_us},
                    {"enqueue_max_us", enqueue_max_us},
                    {"enqueue_avg_us", average(enqueue_total_us)}};
    }
};

// ============================================================================
// BenchmarkResult - Complete benchmark output
// ============================================================================

struct BenchmarkResult {
    // Test configuration
    int corpus_size = 0;
    int doc_size = 0;
    int poll_interval_ms = 0;
    std::uint32_t corpus_seed = CorpusGenerator::kSeed;
    std::string corpus_fingerprint;
    std::string ingest_mode = "single_file_serial";
    std::size_t ingest_concurrency = 1;
    std::string fixture_phase = "initial";
    std::uint32_t post_ingest_coalesce_ms = 0;
    std::uint32_t post_ingest_batch_size = 0;
    std::uint64_t enrichment_timeout_ms = 30000;
    std::string timestamp;
    std::string embedding_model = "simeon-default";
    std::string embedding_backend = "simeon";
    std::string embedding_provider;
    std::string embedding_provider_version;
    std::string embedding_model_uri;
    std::string embedding_space_identity;
    std::size_t embedding_dimension = 0;
    std::string simeon_recipe;
    std::uint64_t generated_bytes = 0;
    bool kg_enabled = true;
    bool gliner_requested = false;
    bool plugin_autoload_enabled = false;
    bool gliner_stage_active = false;
    std::string gliner_model_path;

    // Overall metrics
    int64_t total_duration_ms = 0;
    double throughput_docs_per_sec = 0.0;
    double throughput_mib_per_sec = 0.0;
    bool pipeline_complete = false;
    uint64_t expected_post = 0;
    uint64_t expected_embed = 0;
    uint64_t expected_kg = 0;
    uint64_t expected_title = 0;
    uint64_t observed_post = 0;
    uint64_t observed_embed = 0;
    uint64_t observed_kg = 0;
    uint64_t observed_title = 0;
    bool stages_settled = false;
    bool enrichment_drained = false;
    bool write_coordinator_flushed = false;
    int64_t admission_ms = 0;
    int64_t storage_ready_ms = 0;
    int64_t pipeline_drain_ms = 0;
    int64_t enrichment_ready_ms = 0;
    int64_t vector_index_finalize_ms = 0;
    int64_t searchability_ready_ms = 0;
    int64_t daemon_startup_ms = 0;
    int64_t shutdown_ms = 0;
    int64_t lifecycle_ms = 0;
    std::uint64_t cpu_ms = 0;
    std::uint64_t peak_rss_before_bytes = 0;
    std::uint64_t peak_rss_after_bytes = 0;
    std::uint64_t wal_bytes_before = 0;
    std::uint64_t wal_bytes_after = 0;
    std::uint64_t wal_growth_bytes = 0;
    std::uint64_t db_lock_errors = 0;
    bool shutdown_succeeded = false;
    bool socket_removed = false;
    bool pid_removed = false;
    bool sidecars_removed = false;
    AddDispatchMetrics add_dispatch_metrics;

    std::uint64_t stored_documents = 0;
    bool stored_document_count_readable = false;
    std::string stored_document_count_error;
    VectorOracleSnapshot vector_oracle;
    bool vector_rebuild_ok = false;
    std::string vector_rebuild_error;
    bool vector_index_reusable = false;
    yams::daemon::VectorIndexTelemetry vector_index_telemetry;
    VectorPersistenceSnapshot vector_persistence;
    std::string search_result_fingerprint;
    std::string search_score_fingerprint;
    bool startup_oracle_required = false;
    bool startup_oracle_complete = false;
    std::uint64_t startup_stored_documents = 0;
    bool startup_stored_document_count_readable = false;
    bool startup_vector_reusable = false;
    VectorOracleSnapshot startup_vector_oracle;
    yams::daemon::VectorIndexTelemetry startup_vector_index_telemetry;
    std::vector<SearchImpactProbe> startup_search_impact;
    std::string startup_search_result_fingerprint;
    std::string startup_search_score_fingerprint;

    // Per-stage metrics
    StageMetrics metadata_storage;
    StageMetrics fts5_extraction;
    StageMetrics embedding_generation;
    StageMetrics kg_extraction;
    std::unordered_map<std::string, BenchmarkEmbeddingPhaseTiming> embedding_phase_timings;
    std::unordered_map<std::string, yams::app::services::DocumentStorePhaseTiming>
        document_store_phase_timings;
    std::unordered_map<std::string, yams::api::ContentStorePhaseTiming> content_store_phase_timings;
    std::unordered_map<std::string, yams::metadata::MetadataInsertPhaseTiming>
        metadata_insert_phase_timings;
    std::unordered_map<std::string, yams::storage::RefCounterCommitPhaseTiming>
        ref_counter_commit_phase_timings;
    std::unordered_map<std::string, yams::storage::RefCounterWriterTiming>
        ref_counter_writer_timings;
    std::unordered_map<std::string, yams::storage::RefCounterWriterValueMetric>
        ref_counter_writer_value_metrics;
    yams::metadata::MetadataInsertWriter::MetricsSnapshot metadata_insert_writer_metrics;
    yams::daemon::PostIngestQueue::MetricsSnapshot post_ingest_metrics;
    std::vector<SearchImpactProbe> search_impact;

    // Queue monitoring
    std::vector<QueueSnapshot> queue_samples;
    uint64_t max_store_document_tasks = 0;
    uint64_t max_embed_jobs = 0;
    uint64_t max_fts5_jobs = 0;
    uint64_t max_post_ingest = 0;
    uint64_t max_kg_jobs = 0;
    uint64_t max_symbol_jobs = 0;
    uint64_t max_entity_jobs = 0;
    uint64_t max_title_jobs = 0;
    uint64_t final_kg_queued = 0;
    uint64_t final_kg_consumed = 0;
    uint64_t final_kg_dropped = 0;
    uint64_t final_symbol_queued = 0;
    uint64_t final_symbol_consumed = 0;
    uint64_t final_symbol_dropped = 0;
    uint64_t final_entity_queued = 0;
    uint64_t final_entity_consumed = 0;
    uint64_t final_entity_dropped = 0;
    uint64_t final_title_queued = 0;
    uint64_t final_title_consumed = 0;
    uint64_t final_title_dropped = 0;
    uint64_t dropped_batches = 0;

    // WorkCoordinator metrics
    size_t thread_count = 0;
    bool has_write_coordinator_stats = false;
    yams::daemon::WriteCoordinator::Stats write_coordinator_stats;

    [[nodiscard]] json toJson() const {
        json j;
        j["test_config"] = {{"corpus_size", corpus_size},
                            {"doc_size", doc_size},
                            {"poll_interval_ms", poll_interval_ms},
                            {"corpus_seed", corpus_seed},
                            {"corpus_fingerprint", corpus_fingerprint},
                            {"ingest_mode", ingest_mode},
                            {"ingest_concurrency", ingest_concurrency},
                            {"fixture_phase", fixture_phase},
                            {"post_ingest_coalesce_ms", post_ingest_coalesce_ms},
                            {"post_ingest_batch_size", post_ingest_batch_size},
                            {"enrichment_timeout_ms", enrichment_timeout_ms},
                            {"timestamp", timestamp},
                            {"embedding_model", embedding_model},
                            {"embedding_backend", embedding_backend},
                            {"embedding_provider", embedding_provider},
                            {"embedding_provider_version", embedding_provider_version},
                            {"embedding_model_uri", embedding_model_uri},
                            {"embedding_space_identity", embedding_space_identity},
                            {"embedding_dimension", embedding_dimension},
                            {"simeon_recipe", simeon_recipe},
                            {"generated_bytes", generated_bytes},
                            {"kg_enabled", kg_enabled},
                            {"gliner_requested", gliner_requested},
                            {"plugin_autoload_enabled", plugin_autoload_enabled},
                            {"gliner_stage_active", gliner_stage_active},
                            {"gliner_model_path", gliner_model_path}};

        j["total_duration_ms"] = total_duration_ms;
        j["throughput_docs_per_sec"] = throughput_docs_per_sec;
        j["throughput_mib_per_sec"] = throughput_mib_per_sec;
        j["pipeline_status"] = {{"complete", pipeline_complete},
                                {"stages_settled", stages_settled},
                                {"enrichment_drained", enrichment_drained},
                                {"write_coordinator_flushed", write_coordinator_flushed},
                                {"expected_post", expected_post},
                                {"observed_post", observed_post},
                                {"expected_embed", expected_embed},
                                {"observed_embed", observed_embed},
                                {"expected_kg", expected_kg},
                                {"observed_kg", observed_kg},
                                {"expected_title", expected_title},
                                {"observed_title", observed_title}};
        j["phase_timings"] = {{"daemon_startup_ms", daemon_startup_ms},
                              {"admission_ms", admission_ms},
                              {"storage_ready_ms", storage_ready_ms},
                              {"pipeline_drain_ms", pipeline_drain_ms},
                              {"enrichment_ready_ms", enrichment_ready_ms},
                              {"vector_index_finalize_ms", vector_index_finalize_ms},
                              {"searchability_ready_ms", searchability_ready_ms},
                              {"shutdown_ms", shutdown_ms},
                              {"lifecycle_ms", lifecycle_ms}};
        j["storage_oracle"] = {
            {"stored_documents", stored_documents},
            {"stored_document_count_readable", stored_document_count_readable},
            {"stored_document_count_error", stored_document_count_error},
            {"stored_vectors", vector_oracle.storedVectors},
            {"fetched_vectors", vector_oracle.fetchedVectors},
            {"embedded_documents", vector_oracle.embeddedDocuments},
            {"chunk_vectors", vector_oracle.chunkVectors},
            {"document_vectors", vector_oracle.documentVectors},
            {"invalid_vector_dimensions", vector_oracle.invalidDimensions},
            {"non_finite_vector_values", vector_oracle.nonFiniteValues},
            {"vector_search_result_count", vector_oracle.searchResultCount},
            {"non_finite_vector_search_scores", vector_oracle.nonFiniteSearchScores},
            {"vector_search_fingerprint", vector_oracle.searchFingerprint},
            {"vector_oracle_complete", vector_oracle.complete},
            {"vector_oracle_error", vector_oracle.error},
            {"embedding_fingerprint", vector_oracle.fingerprint},
            {"search_result_fingerprint", search_result_fingerprint},
            {"search_score_fingerprint", search_score_fingerprint}};
        j["restart_oracle"] = {
            {"required", startup_oracle_required},
            {"complete", startup_oracle_complete},
            {"stored_documents", startup_stored_documents},
            {"stored_document_count_readable", startup_stored_document_count_readable},
            {"vector_reusable", startup_vector_reusable},
            {"stored_vectors", startup_vector_oracle.storedVectors},
            {"embedded_documents", startup_vector_oracle.embeddedDocuments},
            {"invalid_vector_dimensions", startup_vector_oracle.invalidDimensions},
            {"non_finite_vector_values", startup_vector_oracle.nonFiniteValues},
            {"vector_search_result_count", startup_vector_oracle.searchResultCount},
            {"non_finite_vector_search_scores", startup_vector_oracle.nonFiniteSearchScores},
            {"vector_search_fingerprint", startup_vector_oracle.searchFingerprint},
            {"vector_oracle_complete", startup_vector_oracle.complete},
            {"embedding_fingerprint", startup_vector_oracle.fingerprint},
            {"search_result_fingerprint", startup_search_result_fingerprint},
            {"search_score_fingerprint", startup_search_score_fingerprint},
            {"coordinator_ready", startup_vector_index_telemetry.ready},
            {"coordinator_rebuilding", startup_vector_index_telemetry.rebuilding},
            {"coordinator_epoch", startup_vector_index_telemetry.rebuildEpoch}};
        j["vector_index"] = {
            {"rebuild_ok", vector_rebuild_ok},
            {"rebuild_error", vector_rebuild_error},
            {"reusable_persisted_index", vector_index_reusable},
            {"coordinator_epoch", vector_index_telemetry.rebuildEpoch},
            {"coordinator_pending_reasons", vector_index_telemetry.pendingReasons},
            {"coordinator_active_bulk_scopes", vector_index_telemetry.activeBulkScopes},
            {"coordinator_rebuilding", vector_index_telemetry.rebuilding},
            {"coordinator_ready", vector_index_telemetry.ready},
            {"coordinator_progress_pct", vector_index_telemetry.progressPct},
            {"persistence_readable", vector_persistence.readable},
            {"persistence_error", vector_persistence.error},
            {"authoritative_generation", vector_persistence.authoritativeGeneration},
            {"source_generation", vector_persistence.sourceGeneration},
            {"trained", vector_persistence.trained},
            {"persisted_vector_count", vector_persistence.persistedVectorCount},
            {"code_count", vector_persistence.codeCount}};
        j["resources"] = {{"cpu_ms", cpu_ms},
                          {"peak_rss_before_bytes", peak_rss_before_bytes},
                          {"peak_rss_after_bytes", peak_rss_after_bytes},
                          {"wal_bytes_before", wal_bytes_before},
                          {"wal_bytes_after", wal_bytes_after},
                          {"wal_growth_bytes", wal_growth_bytes},
                          {"db_lock_errors", db_lock_errors},
                          {"shutdown_succeeded", shutdown_succeeded},
                          {"socket_removed", socket_removed},
                          {"pid_removed", pid_removed},
                          {"sidecars_removed", sidecars_removed}};
        j["add_dispatch_metrics"] = add_dispatch_metrics.toJson();

        j["stages"] = {{"metadata_storage", metadata_storage.toJson()},
                       {"fts5_extraction", fts5_extraction.toJson()},
                       {"embedding_generation", embedding_generation.toJson()},
                       {"kg_extraction", kg_extraction.toJson()}};

        json phaseTimings = json::object();
        for (const auto& [phase, timing] : embedding_phase_timings) {
            phaseTimings[phase] = {
                {"calls", timing.calls},
                {"total_ms", timing.totalUs / 1000},
                {"max_ms", timing.maxUs / 1000},
                {"avg_ms", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   (1000.0 * static_cast<double>(timing.calls))},
                {"total_us", timing.totalUs},
                {"max_us", timing.maxUs},
                {"avg_us", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   static_cast<double>(timing.calls)},
                {"p50_us", percentileUs(timing.samplesUs, 0.50)},
                {"p95_us", percentileUs(timing.samplesUs, 0.95)},
                {"p99_us", percentileUs(timing.samplesUs, 0.99)},
                {"samples", timing.samplesUs.size()},
                {"samples_truncated", timing.samplesTruncated}};
        }
        j["embedding_phase_timings"] = std::move(phaseTimings);

        json docStoreTimings = json::object();
        for (const auto& [phase, timing] : document_store_phase_timings) {
            docStoreTimings[phase] = {{"calls", timing.calls},
                                      {"total_ms", timing.totalMs},
                                      {"max_ms", timing.maxMs},
                                      {"avg_ms", timing.calls == 0
                                                     ? 0.0
                                                     : static_cast<double>(timing.totalMs) /
                                                           static_cast<double>(timing.calls)}};
        }
        j["document_store_phase_timings"] = std::move(docStoreTimings);

        json contentStoreTimings = json::object();
        for (const auto& [phase, timing] : content_store_phase_timings) {
            contentStoreTimings[phase] = {
                {"calls", timing.calls},
                {"total_ms", timing.totalMs},
                {"max_ms", timing.maxMs},
                {"avg_ms", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalMs) /
                                                   static_cast<double>(timing.calls)},
                {"total_us", timing.totalUs},
                {"max_us", timing.maxUs},
                {"avg_us", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   static_cast<double>(timing.calls)}};
        }
        j["content_store_phase_timings"] = std::move(contentStoreTimings);

        json metadataInsertTimings = json::object();
        for (const auto& [phase, timing] : metadata_insert_phase_timings) {
            metadataInsertTimings[phase] = {
                {"calls", timing.calls},
                {"total_us", timing.totalUs},
                {"max_us", timing.maxUs},
                {"avg_us", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   static_cast<double>(timing.calls)}};
        }
        j["metadata_insert_phase_timings"] = std::move(metadataInsertTimings);

        j["metadata_insert_writer_metrics"] = {
            {"submitted_items", metadata_insert_writer_metrics.submittedItems},
            {"completed_items", metadata_insert_writer_metrics.completedItems},
            {"rejected_items", metadata_insert_writer_metrics.rejectedItems},
            {"batches", metadata_insert_writer_metrics.batches},
            {"batch_items", metadata_insert_writer_metrics.batchItems},
            {"max_batch_size", metadata_insert_writer_metrics.maxBatchSize},
            {"avg_batch_size",
             metadata_insert_writer_metrics.batches == 0
                 ? 0.0
                 : static_cast<double>(metadata_insert_writer_metrics.batchItems) /
                       static_cast<double>(metadata_insert_writer_metrics.batches)},
            {"queue_wait_samples", metadata_insert_writer_metrics.queueWaitSamples},
            {"queue_wait_total_us", metadata_insert_writer_metrics.queueWaitTotalUs},
            {"queue_wait_max_us", metadata_insert_writer_metrics.queueWaitMaxUs},
            {"queue_wait_avg_us",
             metadata_insert_writer_metrics.queueWaitSamples == 0
                 ? 0.0
                 : static_cast<double>(metadata_insert_writer_metrics.queueWaitTotalUs) /
                       static_cast<double>(metadata_insert_writer_metrics.queueWaitSamples)},
            {"batch_apply_samples", metadata_insert_writer_metrics.batchApplySamples},
            {"batch_apply_total_us", metadata_insert_writer_metrics.batchApplyTotalUs},
            {"batch_apply_max_us", metadata_insert_writer_metrics.batchApplyMaxUs},
            {"batch_apply_avg_us",
             metadata_insert_writer_metrics.batchApplySamples == 0
                 ? 0.0
                 : static_cast<double>(metadata_insert_writer_metrics.batchApplyTotalUs) /
                       static_cast<double>(metadata_insert_writer_metrics.batchApplySamples)},
            {"failed_batches", metadata_insert_writer_metrics.failedBatches},
            {"fallback_items", metadata_insert_writer_metrics.fallbackItems}};

        json refCounterCommitTimings = json::object();
        for (const auto& [phase, timing] : ref_counter_commit_phase_timings) {
            refCounterCommitTimings[phase] = {
                {"calls", timing.calls},
                {"total_us", timing.totalUs},
                {"max_us", timing.maxUs},
                {"avg_us", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   static_cast<double>(timing.calls)}};
        }
        j["ref_counter_commit_phase_timings"] = std::move(refCounterCommitTimings);

        json refCounterWriterTimings = json::object();
        for (const auto& [phase, timing] : ref_counter_writer_timings) {
            refCounterWriterTimings[phase] = {
                {"calls", timing.calls},
                {"total_us", timing.totalUs},
                {"max_us", timing.maxUs},
                {"avg_us", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   static_cast<double>(timing.calls)}};
        }
        j["ref_counter_writer_timings"] = std::move(refCounterWriterTimings);

        json refCounterWriterValueMetrics = json::object();
        for (const auto& [metric, value] : ref_counter_writer_value_metrics) {
            refCounterWriterValueMetrics[metric] = {
                {"calls", value.calls},
                {"total", value.total},
                {"max", value.max},
                {"avg", value.calls == 0
                            ? 0.0
                            : static_cast<double>(value.total) / static_cast<double>(value.calls)}};
        }
        j["ref_counter_writer_value_metrics"] = std::move(refCounterWriterValueMetrics);

        json postTimings = json::object();
        for (const auto& [phase, timing] : post_ingest_metrics.timings) {
            postTimings[phase] = {
                {"calls", timing.calls},
                {"total_ms", timing.totalMs},
                {"max_ms", timing.maxMs},
                {"avg_ms", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalMs) /
                                                   static_cast<double>(timing.calls)},
                {"total_us", timing.totalUs},
                {"max_us", timing.maxUs},
                {"avg_us", timing.calls == 0 ? 0.0
                                             : static_cast<double>(timing.totalUs) /
                                                   static_cast<double>(timing.calls)},
                {"first_start_ms", timing.firstStartMs},
                {"last_end_ms", timing.lastEndMs},
                {"span_ms", nonNegativeDeltaMs(timing.lastEndMs, timing.firstStartMs)}};
        }
        j["post_ingest_phase_timings"] = std::move(postTimings);
        json searchImpact = json::array();
        for (const auto& probe : search_impact) {
            searchImpact.push_back(probe.toJson());
        }
        j["search_impact"] = std::move(searchImpact);
        json startupSearchImpact = json::array();
        for (const auto& probe : startup_search_impact) {
            startupSearchImpact.push_back(probe.toJson());
        }
        j["startup_search_impact"] = std::move(startupSearchImpact);

        j["post_ingest_batch_metrics"] = {
            {"extraction_batches", post_ingest_metrics.batches.extractionBatches},
            {"extraction_tasks", post_ingest_metrics.batches.extractionTasks},
            {"extraction_successes", post_ingest_metrics.batches.extractionSuccesses},
            {"extraction_failures", post_ingest_metrics.batches.extractionFailures},
            {"embed_jobs_emitted", post_ingest_metrics.batches.embedJobsEmitted},
            {"embed_docs_emitted", post_ingest_metrics.batches.embedDocsEmitted},
            {"embed_prepared_docs_emitted", post_ingest_metrics.batches.embedPreparedDocsEmitted},
            {"embed_hash_only_docs_emitted", post_ingest_metrics.batches.embedHashOnlyDocsEmitted},
            {"content_index_calls", post_ingest_metrics.batches.contentIndexCalls},
            {"content_index_entries", post_ingest_metrics.batches.contentIndexEntries},
            {"content_index_chunks", post_ingest_metrics.batches.contentIndexChunks},
            {"content_index_max_entries", post_ingest_metrics.batches.contentIndexMaxEntries},
            {"content_index_max_chunk_entries",
             post_ingest_metrics.batches.contentIndexMaxChunkEntries},
            {"content_index_avg_entries_per_call",
             post_ingest_metrics.batches.contentIndexCalls == 0
                 ? 0.0
                 : static_cast<double>(post_ingest_metrics.batches.contentIndexEntries) /
                       static_cast<double>(post_ingest_metrics.batches.contentIndexCalls)},
            {"content_index_avg_entries_per_chunk",
             post_ingest_metrics.batches.contentIndexChunks == 0
                 ? 0.0
                 : static_cast<double>(post_ingest_metrics.batches.contentIndexEntries) /
                       static_cast<double>(post_ingest_metrics.batches.contentIndexChunks)}};

        j["queues"] = {{"max_store_document_tasks", max_store_document_tasks},
                       {"max_embed_jobs", max_embed_jobs},
                       {"max_fts5_jobs", max_fts5_jobs},
                       {"max_post_ingest", max_post_ingest},
                       {"max_kg_jobs", max_kg_jobs},
                       {"max_symbol_jobs", max_symbol_jobs},
                       {"max_entity_jobs", max_entity_jobs},
                       {"max_title_jobs", max_title_jobs},
                       {"dropped_batches", dropped_batches},
                       {"sample_count", queue_samples.size()}};
        j["enrichment_status"] = {{"kg",
                                   {{"queued", final_kg_queued},
                                    {"consumed", final_kg_consumed},
                                    {"dropped", final_kg_dropped}}},
                                  {"symbol",
                                   {{"queued", final_symbol_queued},
                                    {"consumed", final_symbol_consumed},
                                    {"dropped", final_symbol_dropped}}},
                                  {"entity",
                                   {{"queued", final_entity_queued},
                                    {"consumed", final_entity_consumed},
                                    {"dropped", final_entity_dropped}}},
                                  {"title",
                                   {{"queued", final_title_queued},
                                    {"consumed", final_title_consumed},
                                    {"dropped", final_title_dropped}}}};

        // Include first/last 5 samples for debugging
        if (!queue_samples.empty()) {
            json samples_head = json::array();
            json samples_tail = json::array();
            for (size_t i = 0; i < std::min<size_t>(5, queue_samples.size()); ++i) {
                samples_head.push_back(queue_samples[i].toJson());
            }
            size_t tail_start = queue_samples.size() > 5 ? queue_samples.size() - 5 : 0;
            for (size_t i = tail_start; i < queue_samples.size(); ++i) {
                samples_tail.push_back(queue_samples[i].toJson());
            }
            j["queues"]["samples_head"] = samples_head;
            j["queues"]["samples_tail"] = samples_tail;
        }

        j["work_coordinator"] = {{"thread_count", thread_count}};

        if (has_write_coordinator_stats) {
            const auto& stats = write_coordinator_stats;
            j["write_coordinator"] = {
                {"batches_enqueued", stats.batchesEnqueued},
                {"batches_committed", stats.batchesCommitted},
                {"ops_applied", stats.opsApplied},
                {"commit_errors", stats.commitErrors},
                {"nodes_upserted", stats.nodesUpserted},
                {"edges_added", stats.edgesAdded},
                {"edges_coalesced", stats.edgesCoalesced},
                {"aliases_added", stats.aliasesAdded},
                {"doc_entities_added", stats.docEntitiesAdded},
                {"symbols_upserted", stats.symbolsUpserted},
                {"node_key_lookups_batched", stats.nodeKeyLookupsBatched},
                {"max_queue_depth", stats.maxQueueDepth},
                {"capacity_rejections", stats.capacityRejections},
                {"forced_enqueues_over_capacity", stats.forcedEnqueuesOverCapacity},
                {"max_batch_apply_ms", stats.maxBatchApplyMs},
                {"max_batch_queue_wait_ms", stats.maxBatchQueueWaitMs},
                {"max_batch_excess_queue_wait_ms", stats.maxBatchExcessQueueWaitMs}};

            json hotSources = json::array();
            for (const auto& source : stats.hotSources) {
                hotSources.push_back({{"source", source.source},
                                      {"batches", source.batches},
                                      {"ops", source.ops},
                                      {"errors", source.errors},
                                      {"total_queue_wait_ms", source.totalQueueWaitMs},
                                      {"max_queue_wait_ms", source.maxQueueWaitMs},
                                      {"max_excess_queue_wait_ms", source.maxExcessQueueWaitMs},
                                      {"total_apply_ms", source.totalApplyMs},
                                      {"max_apply_ms", source.maxApplyMs}});
            }
            j["write_coordinator"]["hot_sources"] = std::move(hotSources);
        }

        return j;
    }
};

// ============================================================================
// Helper: Get current timestamp in milliseconds
// ============================================================================

int64_t nowMs() {
    return std::chrono::duration_cast<std::chrono::milliseconds>(
               std::chrono::steady_clock::now().time_since_epoch())
        .count();
}

std::string nowIso8601() {
    auto now = std::chrono::system_clock::now();
    auto time_t = std::chrono::system_clock::to_time_t(now);
    std::ostringstream oss;
    oss << std::put_time(std::gmtime(&time_t), "%Y-%m-%dT%H:%M:%SZ");
    return oss.str();
}

// ============================================================================
// captureQueueSnapshot - Read queue depths directly from InternalEventBus
// ============================================================================

QueueSnapshot captureQueueSnapshot() {
    using namespace yams::daemon;
    auto& bus = InternalEventBus::instance();

    QueueSnapshot snap;
    snap.timestamp_ms = nowMs();

    // Get actual queue depths by accessing channels
    if (auto q = bus.get_or_create_channel<InternalEventBus::StoreDocumentTask>(
            "store_document_tasks", 4096)) {
        snap.store_document_tasks = q->size_approx();
    }

    if (auto q = bus.get_or_create_channel<InternalEventBus::EmbedJob>("embed_jobs", 4096)) {
        snap.embed_jobs = q->size_approx();
    }

    if (auto q = bus.get_or_create_channel<InternalEventBus::Fts5Job>("fts5_jobs", 4096)) {
        snap.fts5_jobs = q->size_approx();
    }

    if (auto q = bus.get_or_create_channel<InternalEventBus::PostIngestTask>("post_ingest", 4096)) {
        snap.post_ingest = q->size_approx();
    }
    if (auto q = bus.get_channel<InternalEventBus::KgJob>("kg_jobs")) {
        snap.kg_jobs = q->size_approx();
    }
    if (auto q = bus.get_channel<InternalEventBus::SymbolExtractionJob>("symbol_extraction")) {
        snap.symbol_jobs = q->size_approx();
    }
    if (auto q = bus.get_channel<InternalEventBus::EntityExtractionJob>("entity_extraction")) {
        snap.entity_jobs = q->size_approx();
    }
    if (auto q = bus.get_channel<InternalEventBus::TitleExtractionJob>("title_extraction")) {
        snap.title_jobs = q->size_approx();
    }

    // Get cumulative counters (these work for completion detection)
    snap.embed_queued = bus.embedQueued();
    snap.embed_consumed = bus.embedConsumed();
    snap.embed_dropped = bus.embedDropped();
    snap.embed_prepared_docs_queued = bus.embedPreparedDocsQueued();
    snap.embed_prepared_chunks_queued = bus.embedPreparedChunksQueued();
    snap.embed_hash_only_docs_queued = bus.embedHashOnlyDocsQueued();

    snap.fts5_queued = bus.fts5Queued();
    snap.fts5_consumed = bus.fts5Consumed();
    snap.fts5_dropped = bus.fts5Dropped();

    snap.post_queued = bus.postQueued();
    snap.post_consumed = bus.postConsumed();
    snap.post_dropped = bus.postDropped();
    snap.kg_queued = bus.kgQueued();
    snap.kg_consumed = bus.kgConsumed();
    snap.kg_dropped = bus.kgDropped();
    snap.symbol_queued = bus.symbolQueued();
    snap.symbol_consumed = bus.symbolConsumed();
    snap.symbol_dropped = bus.symbolDropped();
    snap.entity_queued = bus.entityQueued();
    snap.entity_consumed = bus.entityConsumed();
    snap.entity_dropped = bus.entityDropped();
    snap.title_queued = bus.titleQueued();
    snap.title_consumed = bus.titleConsumed();
    snap.title_dropped = bus.titleDropped();

    return snap;
}

std::vector<SearchImpactProbe>
runSearchImpactProbes(const std::shared_ptr<yams::daemon::DaemonClient>& client,
                      bool vectorsDisabled, bool kgEnabled, bool pipelineComplete) {
    if (const char* searchProbes = std::getenv("YAMS_BENCH_SEARCH_PROBES");
        searchProbes != nullptr && std::string(searchProbes) == "0") {
        return {};
    }

    std::vector<SearchImpactProbe> probes;
    probes.push_back({.name = "keyword",
                      .query = "Synthetic Test Corpus",
                      .search_type = "keyword",
                      .required_by_contract = pipelineComplete});
    probes.push_back({.name = "semantic",
                      .query = "synthetic document corpus",
                      .search_type = "semantic",
                      .required_by_contract = pipelineComplete && !vectorsDisabled});
    probes.push_back({.name = "graph_rerank_hybrid",
                      .query = "synthetic document corpus",
                      .search_type = "hybrid",
                      .required_by_contract = pipelineComplete && !vectorsDisabled && kgEnabled});

    for (auto& probe : probes) {
        if (probe.name == "semantic" && vectorsDisabled) {
            probe.skipped = true;
            probe.skip_reason = "vectors_disabled";
            continue;
        }
        if (probe.name == "graph_rerank_hybrid" && !kgEnabled) {
            probe.skipped = true;
            probe.skip_reason = "kg_disabled";
            continue;
        }
        if (probe.name == "graph_rerank_hybrid" && vectorsDisabled) {
            probe.skipped = true;
            probe.skip_reason = "vectors_disabled";
            continue;
        }
        if (!pipelineComplete) {
            probe.skipped = true;
            probe.skip_reason = "pipeline_incomplete";
            continue;
        }

        yams::daemon::SearchRequest req;
        req.query = probe.query;
        req.searchType = probe.search_type;
        req.limit = 10;
        req.useSession = false;
        req.globalSearch = true;
        req.showHash = true;
        req.timeout = std::chrono::milliseconds(10000);

        const auto start = nowMs();
        auto response = yams::cli::run_sync(client->search(req), 15s);
        const auto end = nowMs();
        probe.wall_ms = nonNegativeDeltaMs(end, start);
        probe.completed_ms = end;

        if (!response) {
            probe.ok = false;
            probe.error = response.error().message;
            continue;
        }

        const auto& value = response.value();
        probe.daemon_elapsed_ms = value.elapsed.count();
        probe.total_count = static_cast<uint64_t>(value.totalCount);
        probe.returned_count = static_cast<uint64_t>(value.results.size());
        for (const auto& item : value.results) {
            probe.top_ids.push_back(item.id);
            probe.top_paths.push_back(item.path.empty() ? std::string{}
                                                        : fs::path(item.path).filename().string());
            probe.top_scores.push_back(item.score);
            probe.scores_finite = probe.scores_finite && std::isfinite(item.score);
            if (probe.top_paths.size() >= 5) {
                break;
            }
        }
        probe.ok =
            probe.scores_finite && !probe.top_paths.empty() &&
            std::ranges::all_of(probe.top_paths, [](const auto& path) { return !path.empty(); });
        if (!probe.ok) {
            probe.error = "search returned no stable document paths";
        }
    }

    return probes;
}

std::string fingerprintSearchImpact(const std::vector<SearchImpactProbe>& probes) {
    Fnv1a64 fingerprint;
    for (const auto& probe : probes) {
        fingerprint.addBytes(probe.name);
        fingerprint.addInteger(probe.ok ? 1 : 0);
        fingerprint.addInteger(probe.total_count);
        fingerprint.addInteger(probe.top_paths.size());
        for (const auto& path : probe.top_paths) {
            fingerprint.addBytes(path);
        }
    }
    return fingerprint.fingerprint();
}

std::string fingerprintSearchScores(const std::vector<SearchImpactProbe>& probes) {
    Fnv1a64 fingerprint;
    for (const auto& probe : probes) {
        fingerprint.addBytes(probe.name);
        fingerprint.addInteger(probe.top_scores.size());
        for (const double score : probe.top_scores) {
            fingerprint.addInteger(std::bit_cast<std::uint64_t>(score));
        }
    }
    return fingerprint.fingerprint();
}

boost::asio::awaitable<yams::Result<yams::app::services::BatchAddResult>>
submitBatchAsync(yams::app::services::DocumentIngestionService& service,
                 const std::vector<yams::app::services::AddOptions>& batch, int requestWindow) {
    co_return co_await service.addBatchAsync(batch, requestWindow);
}

// ============================================================================
// runBenchmark - Main benchmark logic
// ============================================================================

BenchmarkResult runBenchmark(int corpusSize, int docSize, int pollIntervalMs,
                             std::string ingestMode, std::size_t ingestConcurrency) {
    const auto _lifecycleStart = std::chrono::steady_clock::now();
    const auto _cpuStart = std::clock();
    yams::daemon::TuneAdvisor::setPostIngestBatchSize(benchPostIngestBatchSize());

    BenchmarkResult result;
    auto embeddingTimingSink = std::make_shared<BenchmarkEmbeddingTimingSink>();
    result.peak_rss_before_bytes = peakRssBytes();
    result.corpus_size = corpusSize;
    result.doc_size = docSize;
    result.poll_interval_ms = pollIntervalMs;
    result.ingest_mode = std::move(ingestMode);
    result.ingest_concurrency = std::max<std::size_t>(1, ingestConcurrency);
    result.fixture_phase = benchPhase();
    result.post_ingest_coalesce_ms = benchPostIngestCoalesceMs();
    result.post_ingest_batch_size = yams::daemon::TuneAdvisor::postIngestBatchSize();
    result.timestamp = nowIso8601();
    const char* forceMockEmbeddings = std::getenv("YAMS_BENCH_FORCE_MOCK_EMBEDDINGS");
    const bool mockEmbeddingsRequested =
        forceMockEmbeddings != nullptr && std::string(forceMockEmbeddings) == "1";
    const char* disableVectors = std::getenv("YAMS_DISABLE_VECTORS");
    const bool vectorsDisabled = mockEmbeddingsRequested ||
                                 (disableVectors != nullptr && std::string(disableVectors) == "1");
    if (vectorsDisabled) {
        result.embedding_backend = "mock";
    }
    result.simeon_recipe = !vectorsDisabled ? yams::vector::simeonRecipeLabel() : "";
    result.kg_enabled = !envFlagEnabled("YAMS_BENCH_DISABLE_KG");

    spdlog::info("=== Ingestion E2E Benchmark ===");
    spdlog::info("Corpus size: {}", corpusSize);
    spdlog::info("Doc size: {} bytes", docSize);
    spdlog::info("Poll interval: {} ms", pollIntervalMs);
    spdlog::info("Ingest mode: {} (concurrency={})", result.ingest_mode, result.ingest_concurrency);
    spdlog::info("Fixture phase: {}", result.fixture_phase);
    spdlog::info("Post-ingest coalesce window: {} ms", result.post_ingest_coalesce_ms);
    spdlog::info("Post-ingest batch size: {}", result.post_ingest_batch_size);
    spdlog::info("KG enrichment: {}", result.kg_enabled ? "enabled" : "disabled");

    // Phase 1: Start daemon
    spdlog::info("Phase 1: Starting daemon...");
    SimpleDaemonHarness harness;
    const auto startupStart = std::chrono::steady_clock::now();
    if (!harness.start()) {
        spdlog::error("Failed to start daemon");
        return result;
    }
    result.daemon_startup_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                                   std::chrono::steady_clock::now() - startupStart)
                                   .count();
    result.gliner_requested = harness.glinerRequested();
    result.plugin_autoload_enabled = harness.pluginAutoloadEnabled();
    if (const char* modelPath = std::getenv("YAMS_GLINT_MODEL_PATH"); modelPath != nullptr) {
        result.gliner_model_path = modelPath;
    }
    result.wal_bytes_before = sidecarBytes(harness.dataDir(), "-wal");

    const auto stopHarness = [&](bool capturePersistence) {
        const auto shutdownStart = std::chrono::steady_clock::now();
        harness.stop();
        result.shutdown_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                                 std::chrono::steady_clock::now() - shutdownStart)
                                 .count();
        result.shutdown_succeeded = harness.shutdownSucceeded();
        result.socket_removed = !fs::exists(harness.socketPath());
        result.pid_removed = !fs::exists(harness.pidPath());
        result.sidecars_removed = !hasSidecars(harness.dataDir());
        if (capturePersistence && !vectorsDisabled) {
            result.vector_persistence = captureVectorPersistence(harness.dataDir() / "vectors.db",
                                                                 result.embedding_dimension);
        }
    };

    // Phase 2: Generate corpus
    spdlog::info("Phase 2: Generating test corpus...");
    fs::path corpusDir = harness.root() / "corpus";
    CorpusGenerator corpus(corpusDir, docSize);
    corpus.generateDocuments(corpusSize);
    result.corpus_fingerprint = corpus.fingerprint();
    result.generated_bytes = corpus.generatedBytes;

    // Phase 3: Connect to daemon
    spdlog::info("Phase 3: Connecting to daemon...");
    yams::daemon::ClientConfig clientConfig;
    clientConfig.socketPath = harness.socketPath();
    clientConfig.autoStart = false; // Daemon already started by harness
    auto client = std::make_shared<yams::daemon::DaemonClient>(clientConfig);

    // Actually connect to the daemon socket
    auto connectResult = yams::cli::run_sync(client->connect(), std::chrono::seconds(5));
    if (!connectResult) {
        spdlog::error("Failed to connect to daemon: {}", connectResult.error().message);
        return result; // Return empty result on connection failure
    }
    spdlog::info("Connected to daemon at {}", harness.socketPath().string());

    // Log model provider status
    auto* serviceManager = harness.daemon()->getServiceManager();
    if (serviceManager != nullptr) {
        auto modelProvider = serviceManager->getModelProvider();
        if (modelProvider && modelProvider->isAvailable()) {
            auto loadedModels = modelProvider->getLoadedModels();
            spdlog::info("Model provider: {} model(s) loaded: {}", loadedModels.size(),
                         loadedModels.empty() ? "none" : std::to_string(loadedModels.size()));
            result.embedding_provider = modelProvider->getProviderName();
            result.embedding_provider_version = modelProvider->getProviderVersion();
            result.embedding_dimension = modelProvider->getEmbeddingDim(result.embedding_model);
            result.embedding_space_identity =
                modelProvider->getEmbeddingSpaceIdentity(result.embedding_model);
            if (auto info = modelProvider->getModelInfo(result.embedding_model); info) {
                result.embedding_model_uri = info.value().path;
            } else {
                result.embedding_model_uri = "simeon://" + result.embedding_model;
            }
        } else {
            spdlog::warn("Model provider not available");
        }

        if (result.fixture_phase != "initial") {
            result.startup_oracle_required = result.fixture_phase == "replay";
            if (auto metadataRepo = serviceManager->getMetadataRepo()) {
                if (auto countResult = metadataRepo->getDocumentCount(); countResult) {
                    result.startup_stored_documents =
                        static_cast<std::uint64_t>(std::max<std::int64_t>(0, countResult.value()));
                    result.startup_stored_document_count_readable = true;
                }
            }

            if (!vectorsDisabled) {
                if (auto vectorDb = serviceManager->getVectorDatabase()) {
                    if (result.startup_oracle_required) {
                        const auto reuseDeadline = std::chrono::steady_clock::now() + 30s;
                        while (!vectorDb->hasReusablePersistedSearchIndex() &&
                               std::chrono::steady_clock::now() < reuseDeadline) {
                            std::this_thread::sleep_for(20ms);
                        }
                    }
                    result.startup_vector_reusable = vectorDb->hasReusablePersistedSearchIndex();
                    result.startup_vector_oracle =
                        captureVectorOracle(*vectorDb, result.embedding_dimension);
                }
                if (auto coordinator = serviceManager->getVectorIndexCoordinator()) {
                    result.startup_vector_index_telemetry = coordinator->snapshot();
                }
            }

            const bool startupStorageComplete =
                result.startup_stored_document_count_readable &&
                result.startup_stored_documents == static_cast<std::uint64_t>(corpusSize);
            const bool startupVectorComplete =
                vectorsDisabled ||
                (result.startup_vector_reusable && result.startup_vector_oracle.complete &&
                 result.startup_vector_oracle.embeddedDocuments ==
                     static_cast<std::uint64_t>(corpusSize));
            if (result.startup_oracle_required) {
                result.startup_search_impact =
                    runSearchImpactProbes(client, vectorsDisabled, result.kg_enabled,
                                          startupStorageComplete && startupVectorComplete);
                result.startup_search_result_fingerprint =
                    fingerprintSearchImpact(result.startup_search_impact);
                result.startup_search_score_fingerprint =
                    fingerprintSearchScores(result.startup_search_impact);
            }
            bool startupSearchComplete = true;
            for (const auto& probe : result.startup_search_impact) {
                startupSearchComplete =
                    startupSearchComplete && (!probe.required_by_contract || probe.ok);
            }
            result.startup_oracle_complete =
                startupStorageComplete && startupVectorComplete && startupSearchComplete;
        }

        serviceManager->setEmbeddingPhaseTimingSink(embeddingTimingSink);
        yams::app::services::resetDocumentStorePhaseTimings();
        yams::api::resetContentStorePhaseTimings();
        yams::metadata::resetMetadataInsertPhaseTimings();
        yams::storage::resetRefCounterCommitPhaseTimings();
        yams::storage::resetRefCounterWriterMetrics();
        auto appContext = serviceManager->getAppContext();
        if (appContext.metadataInsertWriter) {
            appContext.metadataInsertWriter->resetMetrics();
        }
        if (auto postIngest = serviceManager->getPostIngestQueue()) {
            postIngest->resetMetrics();
            result.gliner_stage_active =
                postIngest->hasTitleExtractor() && !envFlagEnabled("YAMS_DISABLE_GLINER_TITLES");
            if (!result.kg_enabled) {
                postIngest->setKnowledgeGraphEnabled(false);
            }
        }
    }

    // Capture baseline counters (start from zero)
    auto& bus = yams::daemon::InternalEventBus::instance();
    uint64_t baselineEmbedConsumed = bus.embedConsumed();
    uint64_t baselineEmbedDropped = bus.embedDropped();
    uint64_t baselinePostConsumed = bus.postConsumed();
    uint64_t baselinePostDropped = bus.postDropped();
    uint64_t baselineKgConsumed = bus.kgConsumed();
    uint64_t baselineKgDropped = bus.kgDropped();
    uint64_t baselineTitleQueued = bus.titleQueued();
    uint64_t baselineTitleConsumed = bus.titleConsumed();
    uint64_t baselineTitleDropped = bus.titleDropped();
    spdlog::info("Baseline counters: embed={}, post={}, kg={}, title={}", baselineEmbedConsumed,
                 baselinePostConsumed, baselineKgConsumed, baselineTitleConsumed);

    // Phase 4: Ingest documents
    spdlog::info("Phase 4: Ingesting {} documents...", corpusSize);
    int64_t ingestStartMs = nowMs();
    result.metadata_storage.name = "metadata_storage";
    result.metadata_storage.start_ms = ingestStartMs;

    std::atomic<int> successCount{0};
    std::atomic<int> failCount{0};
    const auto addSingleFile = [&](yams::daemon::DaemonClient& addClient,
                                   const std::string& filename) {
        yams::daemon::AddDocumentRequest addReq;
        addReq.path = (corpusDir / filename).string();
        addReq.noEmbeddings = vectorsDisabled;

        auto addResult = yams::cli::run_sync(addClient.streamingAddDocument(addReq), 30s);
        if (!addResult) {
            spdlog::warn("Failed to ingest {}: {}", filename, addResult.error().message);
            failCount.fetch_add(1, std::memory_order_relaxed);
            return;
        }
        const int completed = successCount.fetch_add(1, std::memory_order_relaxed) + 1;
        if (completed % 10 == 0 || completed == corpusSize) {
            spdlog::info("Ingested {}/{} documents", completed, corpusSize);
        }
    };

    if (result.ingest_mode == "directory") {
        // Directory admission is one recursive request regardless of the caller's
        // generic concurrency default. Keep the experiment identity truthful.
        result.ingest_concurrency = 1;
        yams::daemon::AddDocumentRequest addReq;
        addReq.path = corpusDir.string();
        addReq.recursive = true;
        addReq.noEmbeddings = vectorsDisabled;
        addReq.includePatterns = {"*.txt"};
        auto addResult = yams::cli::run_sync(client->streamingAddDocument(addReq), 120s);
        if (!addResult) {
            spdlog::warn("Directory ingestion failed: {}", addResult.error().message);
            failCount.store(corpusSize, std::memory_order_relaxed);
        } else {
            successCount.store(corpusSize, std::memory_order_relaxed);
            spdlog::info("Directory ingestion accepted for {} documents: {}", corpusSize,
                         addResult.value().message);
        }
    } else if (result.ingest_mode == "pipelined_single_file") {
        const std::size_t requestWindow =
            std::min<std::size_t>(result.ingest_concurrency, corpus.createdFiles.size());
        std::vector<yams::app::services::AddOptions> batch;
        batch.reserve(corpus.createdFiles.size());
        for (const auto& filename : corpus.createdFiles) {
            yams::app::services::AddOptions options;
            options.path = (corpusDir / filename).string();
            options.noEmbeddings = vectorsDisabled;
            batch.push_back(std::move(options));
        }

        yams::app::services::DocumentIngestionService ingestionService(client);
        auto batchResult = yams::cli::run_sync(
            submitBatchAsync(ingestionService, batch, static_cast<int>(requestWindow)), 120s);
        if (!batchResult) {
            spdlog::warn("Pipelined ingestion failed: {}", batchResult.error().message);
            failCount.store(corpusSize, std::memory_order_relaxed);
        } else {
            successCount.store(static_cast<int>(batchResult.value().succeeded),
                               std::memory_order_relaxed);
            failCount.store(static_cast<int>(batchResult.value().failed),
                            std::memory_order_relaxed);
        }
    } else {
        result.ingest_mode = "single_file_serial";
        result.ingest_concurrency = 1;
        for (const auto& filename : corpus.createdFiles) {
            if (harness.stopRequested()) {
                break;
            }
            addSingleFile(*client, filename);
        }
    }

    int64_t ingestEndMs = nowMs();
    result.metadata_storage.end_ms = ingestEndMs;
    result.admission_ms = nonNegativeDeltaMs(ingestEndMs, ingestStartMs);
    result.metadata_storage.count = static_cast<std::uint64_t>(successCount.load());
    result.metadata_storage.failures = static_cast<std::uint64_t>(failCount.load());
    const auto& addStats = harness.daemon()->getState().stats;
    result.add_dispatch_metrics = {
        .samples = addStats.addDispatchSamples.load(std::memory_order_relaxed),
        .total_us = addStats.addDispatchTotalUs.load(std::memory_order_relaxed),
        .max_us = addStats.addDispatchMaxUs.load(std::memory_order_relaxed),
        .fingerprint_total_us = addStats.addFingerprintTotalUs.load(std::memory_order_relaxed),
        .fingerprint_max_us = addStats.addFingerprintMaxUs.load(std::memory_order_relaxed),
        .enqueue_total_us = addStats.addEnqueueTotalUs.load(std::memory_order_relaxed),
        .enqueue_max_us = addStats.addEnqueueMaxUs.load(std::memory_order_relaxed)};

    spdlog::info("Ingestion admission complete: {} succeeded, {} failed in {} ms",
                 successCount.load(), failCount.load(),
                 nonNegativeDeltaMs(ingestEndMs, ingestStartMs));

    // Phase 5: Monitor pipeline completion via InternalEventBus
    spdlog::info("Phase 5: Monitoring pipeline completion...");

    // Capture initial state
    QueueSnapshot initialSnap = captureQueueSnapshot();
    spdlog::info("Initial state: embed_queued={} fts5_queued={} post_queued={}",
                 initialSnap.embed_queued, initialSnap.fts5_queued, initialSnap.post_queued);

    // Expected counts
    // Note: FTS5 indexing happens synchronously in post-ingest metadata stage (not via jobs)
    // So we only track post-ingest (metadata+FTS5) and embeddings completion
    uint64_t expectedPost = static_cast<std::uint64_t>(successCount.load());
    uint64_t expectedEmbed = vectorsDisabled ? 0 : static_cast<std::uint64_t>(successCount.load());
    uint64_t expectedKg = result.kg_enabled ? static_cast<std::uint64_t>(successCount.load()) : 0;
    uint64_t expectedTitle =
        result.gliner_stage_active ? static_cast<std::uint64_t>(successCount.load()) : 0;
    result.expected_post = expectedPost;
    result.expected_embed = expectedEmbed;
    result.expected_kg = expectedKg;
    result.expected_title = expectedTitle;
    constexpr std::uint64_t kBaseEnrichmentTimeoutMs = 30000;
    constexpr std::uint64_t kPerTitleTimeoutMs = 60000;
    constexpr std::uint64_t kMaxEnrichmentTimeoutMs = 2ULL * 60ULL * 60ULL * 1000ULL;
    result.enrichment_timeout_ms =
        std::min(kMaxEnrichmentTimeoutMs,
                 saturatedAdd(kBaseEnrichmentTimeoutMs, expectedTitle * kPerTitleTimeoutMs));

    // Poll until every expected item is consumed or explicitly dropped. Drops settle the
    // workload but never satisfy the lossless completion contract.
    auto pipelineDeadline = std::chrono::steady_clock::now() + 60s;
    bool allStagesComplete = false;
    bool allStagesSettled = false;
    QueueSnapshot lastSnap = initialSnap;
    uint64_t lastEmbedConsumedDelta = 0;
    uint64_t lastPostConsumedDelta = 0;
    uint64_t lastEmbedDroppedDelta = 0;
    uint64_t lastKgConsumedDelta = 0;
    uint64_t lastKgDroppedDelta = 0;

    while (std::chrono::steady_clock::now() < pipelineDeadline && !allStagesSettled &&
           !harness.stopRequested()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(pollIntervalMs));

        QueueSnapshot snap = captureQueueSnapshot();
        result.queue_samples.push_back(snap);
        lastSnap = snap;

        // Update max queue depths (approximated by queued - consumed)
        // Use signed arithmetic to avoid underflow, then clamp to 0
        auto fts5Depth =
            snap.fts5_queued > snap.fts5_consumed ? snap.fts5_queued - snap.fts5_consumed : 0;
        auto embedDepth =
            snap.embed_queued > snap.embed_consumed ? snap.embed_queued - snap.embed_consumed : 0;
        auto postDepth =
            snap.post_queued > snap.post_consumed ? snap.post_queued - snap.post_consumed : 0;

        result.max_fts5_jobs = std::max(result.max_fts5_jobs, fts5Depth);
        result.max_embed_jobs = std::max(result.max_embed_jobs, embedDepth);
        result.max_post_ingest = std::max(result.max_post_ingest, postDepth);
        result.max_kg_jobs = std::max(result.max_kg_jobs, snap.kg_jobs);
        result.max_symbol_jobs = std::max(result.max_symbol_jobs, snap.symbol_jobs);
        result.max_entity_jobs = std::max(result.max_entity_jobs, snap.entity_jobs);
        result.max_title_jobs = std::max(result.max_title_jobs, snap.title_jobs);

        // Track dropped batches (excluding warmup baseline)
        uint64_t embedDroppedDelta = counterDelta(snap.embed_dropped, baselineEmbedDropped);
        uint64_t postDroppedDelta = counterDelta(snap.post_dropped, baselinePostDropped);
        uint64_t kgDroppedDelta = counterDelta(snap.kg_dropped, baselineKgDropped);
        const uint64_t countedKgDroppedDelta = result.kg_enabled ? kgDroppedDelta : 0;
        result.dropped_batches = saturatedAdd(
            saturatedAdd(saturatedAdd(snap.fts5_dropped, embedDroppedDelta), postDroppedDelta),
            countedKgDroppedDelta);

        const uint64_t embedConsumedDelta =
            counterDelta(snap.embed_consumed, baselineEmbedConsumed);
        const uint64_t postConsumedDelta = counterDelta(snap.post_consumed, baselinePostConsumed);
        const uint64_t kgConsumedDelta =
            result.kg_enabled ? counterDelta(snap.kg_consumed, baselineKgConsumed) : 0;
        lastEmbedConsumedDelta = embedConsumedDelta;
        lastPostConsumedDelta = postConsumedDelta;
        lastKgConsumedDelta = kgConsumedDelta;
        lastEmbedDroppedDelta = embedDroppedDelta;
        lastKgDroppedDelta = kgDroppedDelta;

        // Completion is exact: consuming more work than was admitted indicates a replay or
        // duplicate enqueue. Keep the settled checks below permissive so the benchmark can stop
        // and report the over-count instead of waiting for a timeout.
        const bool embedComplete = embedConsumedDelta == expectedEmbed;
        const bool postComplete = postConsumedDelta == expectedPost;
        const bool kgComplete = kgConsumedDelta == expectedKg;
        const bool embedSettled =
            saturatedAdd(embedConsumedDelta, embedDroppedDelta) >= expectedEmbed;
        const bool postSettled = saturatedAdd(postConsumedDelta, postDroppedDelta) >= expectedPost;
        const bool kgSettled =
            saturatedAdd(kgConsumedDelta, result.kg_enabled ? kgDroppedDelta : 0) >= expectedKg;

        allStagesComplete =
            embedComplete && postComplete && kgComplete && result.dropped_batches == 0;
        allStagesSettled = embedSettled && postSettled && kgSettled;

        if (nonNegativeDeltaMs(snap.timestamp_ms, initialSnap.timestamp_ms) % 1000 <
            pollIntervalMs) {
            spdlog::info("Pipeline status: post={}/{} (dropped={}) embed={}/{} (dropped={}) "
                         "kg={}/{} (dropped={})",
                         postConsumedDelta, expectedPost, postDroppedDelta, embedConsumedDelta,
                         expectedEmbed, embedDroppedDelta, kgConsumedDelta, expectedKg,
                         kgDroppedDelta);
        }

        if (allStagesSettled) {
            // Update stage metrics (use deltas to exclude warmup)
            int64_t now = nowMs();

            result.fts5_extraction.name = "fts5_extraction";
            result.fts5_extraction.start_ms = ingestStartMs;
            result.fts5_extraction.end_ms = now;
            result.fts5_extraction.count =
                snap.fts5_consumed; // FTS5 counter not affected by warmup
            result.fts5_extraction.failures = snap.fts5_dropped;

            result.embedding_generation.name = "embedding_generation";
            result.embedding_generation.start_ms = ingestStartMs;
            result.embedding_generation.end_ms = now;
            result.embedding_generation.count =
                counterDelta(snap.embed_consumed, baselineEmbedConsumed);
            result.embedding_generation.failures = embedDroppedDelta;

            result.kg_extraction.name = "kg_extraction";
            result.kg_extraction.start_ms = ingestStartMs;
            result.kg_extraction.end_ms = now;
            result.kg_extraction.count = counterDelta(snap.kg_consumed, baselineKgConsumed);
            result.kg_extraction.failures = kgDroppedDelta;

            spdlog::info("All pipeline stages settled (lossless={})", allStagesComplete);
            break;
        }
    }

    result.stages_settled = allStagesSettled;
    const int64_t storageReadyEndMs = nowMs();
    result.storage_ready_ms = nonNegativeDeltaMs(storageReadyEndMs, ingestStartMs);
    result.observed_post = lastPostConsumedDelta;
    result.observed_embed = lastEmbedConsumedDelta;
    result.observed_kg = lastKgConsumedDelta;

    if (!allStagesSettled) {
        spdlog::warn("Pipeline settlement timeout after 60s (post={}/{} embed={}/{} kg={}/{})",
                     lastPostConsumedDelta, expectedPost, lastEmbedConsumedDelta, expectedEmbed,
                     lastKgConsumedDelta, expectedKg);

        // Preserve partial pipeline evidence instead of leaving stage metrics at zero. A timeout is
        // still a failure, but partial counters identify which PostIngestQueue branch stalled.
        const int64_t now = nowMs();
        result.fts5_extraction.name = "fts5_extraction";
        result.fts5_extraction.start_ms = ingestStartMs;
        result.fts5_extraction.end_ms = now;
        result.fts5_extraction.count = lastSnap.fts5_consumed;
        result.fts5_extraction.failures = lastSnap.fts5_dropped;

        result.embedding_generation.name = "embedding_generation";
        result.embedding_generation.start_ms = ingestStartMs;
        result.embedding_generation.end_ms = now;
        result.embedding_generation.count =
            counterDelta(lastSnap.embed_consumed, baselineEmbedConsumed);
        result.embedding_generation.failures = lastEmbedDroppedDelta;

        result.kg_extraction.name = "kg_extraction";
        result.kg_extraction.start_ms = ingestStartMs;
        result.kg_extraction.end_ms = now;
        result.kg_extraction.count = counterDelta(lastSnap.kg_consumed, baselineKgConsumed);
        result.kg_extraction.failures = lastKgDroppedDelta;
    }

    // The core counters can settle before entity/title work and deferred KG writes commit.
    // Wait for every PostIngestQueue stage to drain, then flush the WriteCoordinator.
    result.enrichment_drained = false;
    const auto enrichmentDeadline =
        std::chrono::steady_clock::now() + std::chrono::milliseconds(result.enrichment_timeout_ms);
    while (std::chrono::steady_clock::now() < enrichmentDeadline && !harness.stopRequested()) {
        lastSnap = captureQueueSnapshot();
        bool postIngestIdle = true;
        if (serviceManager != nullptr) {
            if (auto postIngest = serviceManager->getPostIngestQueue()) {
                postIngestIdle =
                    postIngest->size() == 0 && postIngest->kgQueueDepth() == 0 &&
                    postIngest->symbolQueueDepth() == 0 && postIngest->entityQueueDepth() == 0 &&
                    postIngest->titleQueueDepth() == 0 && postIngest->totalInFlight() == 0;
            }
        }
        const bool embeddingIdle =
            lastSnap.embed_jobs == 0 &&
            (serviceManager == nullptr || serviceManager->getEmbeddingInFlightJobs() == 0);
        if (postIngestIdle && embeddingIdle) {
            result.enrichment_drained = true;
            break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(pollIntervalMs));
    }

    if (harness.stopRequested()) {
        lastSnap = captureQueueSnapshot();
        result.observed_post = counterDelta(lastSnap.post_consumed, baselinePostConsumed);
        result.observed_embed = counterDelta(lastSnap.embed_consumed, baselineEmbedConsumed);
        result.observed_kg = counterDelta(lastSnap.kg_consumed, baselineKgConsumed);
        result.observed_title = counterDelta(lastSnap.title_consumed, baselineTitleConsumed);
        spdlog::warn("Termination requested; stopping isolated benchmark fixture");
        stopHarness(false);
        return result;
    }

    result.write_coordinator_flushed = true;
    if (serviceManager != nullptr) {
        if (auto* writeCoordinator = serviceManager->getWriteCoordinator()) {
            auto flushResult = writeCoordinator->flush(30s);
            result.write_coordinator_flushed = static_cast<bool>(flushResult);
            result.write_coordinator_stats = writeCoordinator->getStats();
            result.has_write_coordinator_stats = true;
            if (!flushResult) {
                spdlog::warn("WriteCoordinator flush failed: {}", flushResult.error().message);
            }
        }
    }

    lastSnap = captureQueueSnapshot();
    result.final_kg_queued = lastSnap.kg_queued;
    result.final_kg_consumed = lastSnap.kg_consumed;
    result.final_kg_dropped = lastSnap.kg_dropped;
    result.final_symbol_queued = lastSnap.symbol_queued;
    result.final_symbol_consumed = lastSnap.symbol_consumed;
    result.final_symbol_dropped = lastSnap.symbol_dropped;
    result.final_entity_queued = lastSnap.entity_queued;
    result.final_entity_consumed = lastSnap.entity_consumed;
    result.final_entity_dropped = lastSnap.entity_dropped;
    result.final_title_queued = counterDelta(lastSnap.title_queued, baselineTitleQueued);
    result.final_title_consumed = counterDelta(lastSnap.title_consumed, baselineTitleConsumed);
    result.final_title_dropped = counterDelta(lastSnap.title_dropped, baselineTitleDropped);
    result.observed_title = result.final_title_consumed;
    result.dropped_batches = saturatedAdd(
        result.dropped_batches,
        saturatedAdd(result.final_symbol_dropped,
                     saturatedAdd(result.final_entity_dropped, result.final_title_dropped)));

    const bool titleComplete = result.observed_title == result.expected_title;
    result.pipeline_complete = allStagesComplete && titleComplete && result.enrichment_drained &&
                               result.write_coordinator_flushed && result.dropped_batches == 0 &&
                               failCount.load(std::memory_order_relaxed) == 0;
    const int64_t enrichmentEndMs = nowMs();
    result.pipeline_drain_ms = nonNegativeDeltaMs(enrichmentEndMs, storageReadyEndMs);
    result.enrichment_ready_ms = nonNegativeDeltaMs(enrichmentEndMs, ingestStartMs);

    if (serviceManager != nullptr) {
        if (auto metadataRepo = serviceManager->getMetadataRepo()) {
            auto countResult = metadataRepo->getDocumentCount();
            if (countResult) {
                result.stored_documents =
                    static_cast<std::uint64_t>(std::max<std::int64_t>(0, countResult.value()));
                result.stored_document_count_readable = true;
            } else {
                result.stored_document_count_error = countResult.error().message;
            }
        }

        if (!vectorsDisabled) {
            const auto vectorFinalizeStart = std::chrono::steady_clock::now();
            if (auto coordinator = serviceManager->getVectorIndexCoordinator()) {
                auto rebuildResult = yams::cli::run_sync(
                    coordinator->requestRebuild(yams::daemon::RebuildReason::EmbeddingBatch), 300s);
                result.vector_rebuild_ok = static_cast<bool>(rebuildResult);
                if (!rebuildResult) {
                    result.vector_rebuild_error = rebuildResult.error().message;
                }
                result.vector_index_telemetry = coordinator->snapshot();
            } else {
                result.vector_rebuild_error = "vector index coordinator unavailable";
            }
            result.vector_index_finalize_ms =
                std::chrono::duration_cast<std::chrono::milliseconds>(
                    std::chrono::steady_clock::now() - vectorFinalizeStart)
                    .count();

            if (auto vectorDb = serviceManager->getVectorDatabase()) {
                result.vector_index_reusable = vectorDb->hasReusablePersistedSearchIndex();
                result.vector_oracle = captureVectorOracle(*vectorDb, result.embedding_dimension);
            } else {
                result.vector_oracle.error = "vector database unavailable";
            }
        } else {
            result.vector_rebuild_ok = true;
            result.vector_oracle.complete = true;
        }
    }

    const bool corePipelineComplete = result.pipeline_complete;
    spdlog::info("Phase 6: Running fixed search-impact probes...");
    result.search_impact =
        runSearchImpactProbes(client, vectorsDisabled, result.kg_enabled, corePipelineComplete);
    result.search_result_fingerprint = fingerprintSearchImpact(result.search_impact);
    result.search_score_fingerprint = fingerprintSearchScores(result.search_impact);
    bool searchOracleComplete = true;
    for (const auto& probe : result.search_impact) {
        if (!probe.required_by_contract) {
            continue;
        }
        searchOracleComplete = searchOracleComplete && probe.ok;
        if (probe.ok) {
            result.searchability_ready_ms =
                std::max(result.searchability_ready_ms,
                         nonNegativeDeltaMs(probe.completed_ms, ingestStartMs));
        }
    }

    const auto expectedStoredDocuments = static_cast<std::uint64_t>(corpusSize);
    const bool documentOracleComplete =
        result.stored_document_count_readable && result.stored_documents == expectedStoredDocuments;
    const bool vectorOracleComplete =
        vectorsDisabled || (result.vector_rebuild_ok && result.vector_oracle.complete &&
                            result.vector_oracle.embeddedDocuments == expectedStoredDocuments);
    const bool glinerOracleComplete = !result.gliner_requested || result.gliner_stage_active;
    result.pipeline_complete = result.pipeline_complete && documentOracleComplete &&
                               vectorOracleComplete && glinerOracleComplete && searchOracleComplete;
    if (result.startup_oracle_required) {
        const bool replayFingerprintsStable =
            result.startup_vector_oracle.fingerprint == result.vector_oracle.fingerprint &&
            result.startup_vector_oracle.searchFingerprint ==
                result.vector_oracle.searchFingerprint &&
            result.startup_search_result_fingerprint == result.search_result_fingerprint &&
            result.startup_search_score_fingerprint == result.search_score_fingerprint;
        result.pipeline_complete =
            result.pipeline_complete && result.startup_oracle_complete && replayFingerprintsStable;
    }

    result.total_duration_ms = std::max(result.enrichment_ready_ms, result.searchability_ready_ms);
    const double elapsedSeconds = static_cast<double>(result.total_duration_ms) / 1000.0;
    result.throughput_docs_per_sec =
        elapsedSeconds > 0.0 ? static_cast<double>(successCount.load()) / elapsedSeconds : 0.0;
    result.throughput_mib_per_sec =
        elapsedSeconds > 0.0
            ? (static_cast<double>(result.generated_bytes) / (1024.0 * 1024.0)) / elapsedSeconds
            : 0.0;

    if (serviceManager != nullptr) {
        result.embedding_phase_timings = embeddingTimingSink->snapshot();
        result.document_store_phase_timings =
            yams::app::services::getDocumentStorePhaseTimingsSnapshot();
        result.content_store_phase_timings = yams::api::getContentStorePhaseTimingsSnapshot();
        result.metadata_insert_phase_timings =
            yams::metadata::getMetadataInsertPhaseTimingsSnapshot();
        result.ref_counter_commit_phase_timings =
            yams::storage::getRefCounterCommitPhaseTimingsSnapshot();
        result.ref_counter_writer_timings = yams::storage::getRefCounterWriterTimingsSnapshot();
        result.ref_counter_writer_value_metrics =
            yams::storage::getRefCounterWriterValueMetricsSnapshot();
        auto appContext = serviceManager->getAppContext();
        if (appContext.metadataInsertWriter) {
            result.metadata_insert_writer_metrics =
                appContext.metadataInsertWriter->metricsSnapshot();
        }
        if (auto postIngest = serviceManager->getPostIngestQueue()) {
            result.post_ingest_metrics = postIngest->metricsSnapshot();
        }
        result.db_lock_errors =
            harness.daemon()->getState().stats.dbLockErrors.load(std::memory_order_relaxed);
    }

    result.wal_bytes_after = sidecarBytes(harness.dataDir(), "-wal");
    result.wal_growth_bytes = result.wal_bytes_after > result.wal_bytes_before
                                  ? result.wal_bytes_after - result.wal_bytes_before
                                  : 0;

    stopHarness(true);

    result.peak_rss_after_bytes = peakRssBytes();
    const auto cpuEnd = std::clock();
    if (_cpuStart != static_cast<std::clock_t>(-1) && cpuEnd != static_cast<std::clock_t>(-1) &&
        cpuEnd >= _cpuStart) {
        result.cpu_ms = static_cast<std::uint64_t>(
            (static_cast<double>(cpuEnd - _cpuStart) * 1000.0) / CLOCKS_PER_SEC);
    }
    result.lifecycle_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                              std::chrono::steady_clock::now() - _lifecycleStart)
                              .count();

    const bool cleanShutdown = result.shutdown_succeeded && result.socket_removed &&
                               result.pid_removed && result.sidecars_removed;
    result.pipeline_complete = result.pipeline_complete && cleanShutdown;

    spdlog::info("=== Benchmark Complete ===");
    spdlog::info("Total duration: {} ms", result.total_duration_ms);
    spdlog::info("Throughput: {:.2f} docs/sec ({:.2f} MiB/sec)", result.throughput_docs_per_sec,
                 result.throughput_mib_per_sec);
    spdlog::info("Dropped batches: {}", result.dropped_batches);

    return result;
}

// ============================================================================
// main - Entry point
// ============================================================================

int main(int /*argc*/, char* /*argv*/[]) noexcept {
    try {
        // Set log level
        const char* logLevel = std::getenv("YAMS_LOG_LEVEL");
        if (logLevel != nullptr && std::string(logLevel) == "debug") {
            spdlog::set_level(spdlog::level::debug);
        } else {
            spdlog::set_level(spdlog::level::info);
        }

        // Read configuration from environment
        const int corpusSize = benchEnvInt("YAMS_BENCH_CORPUS_SIZE", 100, 1, 1000000);
        const int docSize = benchEnvInt("YAMS_BENCH_DOC_SIZE", 1000, 1, 100 * 1024 * 1024);
        const int pollInterval = benchEnvInt("YAMS_BENCH_POLL_INTERVAL_MS", 100, 1, 60000);

        std::string ingestMode = "single_file_serial";
        if (const char* env = std::getenv("YAMS_BENCH_INGEST_MODE")) {
            ingestMode = env;
        }

        const auto ingestConcurrency =
            static_cast<std::size_t>(benchEnvInt("YAMS_BENCH_INGEST_CONCURRENCY", 4, 1, 1024));

        const char* outputPath = std::getenv("YAMS_BENCH_OUTPUT");

        BenchmarkResult result =
            runBenchmark(corpusSize, docSize, pollInterval, ingestMode, ingestConcurrency);
        const std::string jsonStr = result.toJson().dump(2);

        if (outputPath != nullptr) {
            std::ofstream outFile(outputPath);
            outFile << jsonStr << '\n';
            spdlog::info("Results written to: {}", outputPath);
        } else {
            std::cout << jsonStr << '\n';
        }

        return result.metadata_storage.failures > 0 || !result.pipeline_complete ||
                       result.dropped_batches > 0
                   ? 1
                   : 0;
    } catch (const std::exception& error) {
        spdlog::error("Ingestion E2E benchmark failed: {}", error.what());
        return 2;
    } catch (...) {
        spdlog::error("Ingestion E2E benchmark failed with an unknown exception");
        return 2;
    }
}