yams/tests/unit/daemon/daemon_metrics_status_test.cpp
Trevon 5bda4d77a2
perf+assert(storage): eliminate zero-init, temp-path overhead, add DCHECK invariants (#42)
* fix(core): add assertions and harden daemon lifecycle

* fix(daemon): tighten shutdown lifecycle invariants

* fix(test): add PressureLimitedPoller capability-sleep shutdown test

* refactor(vector): harden backend contracts

Split the vector backend surface into narrower contracts and
move entity read paths to Result<> so callers can distinguish
backend errors from empty results.

Also fixes the TurboQuant lazy-init race, makes VectorDatabase
error state thread-safe, wires eager index preparation, and
adds regression plus memory-scaling benchmark coverage.

* fix(vector): address audit findings plus hardening

Fix the five confirmed bugs: coroutine lifetime hazard in
embedding_generator (await_with_timeout outliving captured state),
VectorSystemManager init retry latch poisoning, incremental HNSW
persistence race, vec0 registry key collisions, and simeon PMI
output_dim/config() drift.

Add invariant hardening: PostIngestQueue stage validation +
numChunks assertion, VectorSystemManager success postcondition,
VectorDatabase TurboQuant non-null assertions on all quantize/
dequantize paths, and sqlite_vec_backend Simeon PQ range and
divisibility preconditions.

Also adds regression coverage for init-retry and lock-busy paths
and a locked node snapshot helper for safe incremental HNSW
persistence.

* fix(test): add PressureLimitedPoller wake-timer shutdown tests

* chore(vector): resolve clang-tidy hygiene across queue, vdb, and sqlite-vec

Replace unprotected std::getenv calls with locked getenvCopy helpers,
add debug-level logging to previously-silent catch blocks in
PostIngestQueue, VectorDatabase, and SqliteVecBackend, wrap
PostIngestQueue timer-detach allocations in try-catch, and fix
unsigned-widening / optional-deref / unused-include hygiene.

Verified with LSP diagnostics and rebuilds of catch2_vector_submodule
and catch2_daemon_component_submodule.

* fix(daemon): harden PressureLimitedPoller inFlightCounter null check

* fix(daemon): harden PostIngestQueue wait_until deadline checks

* fix(daemon): clean up recovery sentinels regardless of salvage count

Previously the .recovered-* sentinel and .corrupt-* file cleanup
was gated on documentsSalvaged > 0, leaving quarantine
artifacts to accumulate on repeated crash/recovery cycles where
the corrupt DB had 0 salvageable docs.

Move cleanup outside the salvage-count guard so stale
quarantine artifacts are always removed after examination.

* feat(vec0): restore persisted ANN index on daemon restart

After vec0 rebuild, probe for vectors_<dim>_hnsw_meta shadow tables
and load a previously-persisted HNSW index.  When found, the ANN
index is set directly on the vec0 table, skipping the O(n log n)
rebuild that was previously paid on every daemon restart.

Removes the stub restoreVec0AnnIndexIfPresentUnlocked() in favor
of inline restoration in rebuildVec0DimUnlocked(), which has
the vec0 table already created and ready for ann_index assignment.

* fix: enable DCHECK by default and assert metadata counter consistency

* fix(services): assert document deletion ordering content-before-metadata

* fix(test): add fork-based assertion validation tests

* fix(test): expand assertion tests for POSTCONDITION, DCHECK, UNREACHABLE

* fix(ci): enable -Denable-dcheck=true in all test CI lanes

* fix(test): address assertion review findings

* refactor(daemon): isolate poller config assertion checks

* refactor(services): reduce deletion and poller complexity

* style(ci): normalize test workflow spacing

* refactor(services): extract list query helpers

* refactor(services): extract list hydration helpers

* refactor(daemon): extract metrics snapshot helpers

* refactor(daemon): extract metrics counter snapshot

* refactor(daemon): extract metrics resource snapshot

* feat(graph): add daemon-backed explore

Wire graph explore through daemon IPC, keep CLI rendering stable,
and fall back locally when daemon transport or protocol skew fails.

Also add graph context service/tests, improve graph follow-up hints,
and keep release builds working with the current JSON library.

* fix(services): prevent prune delete metadata orphans

* fix(cli): honor data-dir in graph explore

- bypass daemon explore when the CLI pins a data dir so\n  temp fixtures do not resolve against a long-lived global DB\n- guard snippet truncation and ratio math to remove Fold-\n  reported underflow and div-by-zero paths\n- clamp zero-based symbol lines before rendering graph\n  context

* fix(daemon): make lifecycle stop completion safe

* fix(daemon): gate db-open startup against shutdown

* fix(metadata): serialize FTS indexed counter updates

* test(daemon): simulate vector lock contention cross-process

* test(daemon): isolate post-ingest identity batch

* style(daemon): normalize test include order

* refactor(graph): split cli graph helpers

Keep graph command paths smaller and easier to evolve.

Lock graph explore/topology contracts into focused presenter,
dispatcher, and support tests so follow-up refactors keep the
current IPC and hot-path behavior intact.

* fix(config): harden expand_tilde and fix Windows XDG state path doubling

- expand_tilde: handle bare ~ and ~/ explicitly to avoid std::out_of_range
  on substr(2) for single-char input
- getXDGStateHome: on Windows, go up two levels from %LOCALAPPDATA%\yams\state
  to %LOCALAPPDATA% so callers' xdg/yams append doesn't double the yams segment

* fix(static-analysis): add logging to empty catch blocks and fix nosemgrep syntax

* fix(static-analysis): strip trailing text from nosemgrep comments

* fix(doctor): add logging to daemon_check empty catch blocks

* fix(db_repair): add logging to empty catch blocks

* fix(chunking): add precondition checks for rabin polynomial and windowSize

* fix(safety): add null precondition checks for Fold proved null-deref sites

* fix(daemon): widen init-failure dispatch, add null/dbz preconditions

InitializationFailedEvent races with shutdown — async init
fails after teardown begins. Widen canDispatch to accept
Ready/ShuttingDown. Add static_assert on SnapshotRing N>0.
Add YAMS_PRECONDITION null guards on hex/stages/data/bytes.

* ci(ubsan): use restricted build targets to avoid runner timeout

* fix(daemon): detach wake timers on Windows stop path to prevent SIGSEGV

* fix(config): cast parseTomlU32 to size_t for batch fields

* fix(ci): use static helper for Windows timer leak, restrict ubsan tests

* fix(storage): add MutexPool and hash-length precondition assertions

* fix(storage): add reference counter lifecycle assertions

- YAMS_PRECONDITION on increment size constraints
- YAMS_DCHECK on decrement zero-row detection
- YAMS_ASSERT on transaction AUTOINCREMENT ID
- YAMS_PRECONDITION on transaction active guard
- YAMS_POSTCONDITION on commit/rollback state

* fix(storage): add atomic-write and store postcondition assertions

* fix(ingestion): add batch concurrency and accounting assertions

* fix(storage): add remove counter and retrieve-read DCHECKs

* fix(daemon): add WorkCoordinator double-start precondition

* bench(storage): add store/retrieve throughput microbenchmark

* fix(daemon): replace WorkCoordinator double-start throw with precondition

* test(daemon): add fork-based double-start precondition test

* test(ingestion): add batch concurrency edge-case tests

* fix(storage): remove overly-aggressive decrement DCHECK, fix transaction test

* test(daemon): fork-based precondition test — known Catch2 SIGABRT artifact

Catch2 v3.12.0 signal handler catches child SIGABRT via abort(),
marking test as FAILED even though CHECK passes correctly.
Affects all fork-based assertion tests. Test logic is correct.

* refactor(test): remove duplicate fork test, defer to assert_catch2

* fix(review): downgrade transactionId to DCHECK, use unique benchmark hashes

* fix(review): document PRECONDITION+throw pattern, downgrade exists() to DCHECK

* test(ingestion): add maxConcurrent early-return branch coverage

* perf(storage): eliminate zero-init in retrieveRaw, fix benchmark build

retrieveRaw bottleneck: 73% of samples in vector<std::byte>(N) zero-init
that gets overwritten by file.read(). Use make_unique_for_overwrite (C++20)
to allocate uninitialized, then assign into vector.

Before/after (TSan, 1MB Retrieve): 41.7ms -> 9.0ms (4.6x faster)

Also fix Google Benchmark build: add benchmark_main_dep (provides main())
to all 7 benchmark targets. Fix storage benchmark to use valid hex hashes.

0 assertion violations across all benchmark runs.

* perf(storage): skip redundant temp-dir check, use hex table for names

Two micro-optimizations in getTempPath() hot path:
1. Drop ensureDirectories() call — temp dir already created at ctor,
   stat() on every store was wasted work.
2. Replace 16 fmt::format("{:x}") calls with pre-computed hex char
   table + push_back. Avoids format machinery per store.

0 opengrep findings.

* assert(storage): DCHECK temp dir, buffer, size invariants in hot paths

Coverage for the three changed functions:

getTempPath:
  - DCHECK temp dir still exists (catch external cleanup in dev/CI)
  - DCHECK temp name length matches TEMP_NAME_LENGTH

atomicWrite:
  - DCHECK temp file exists after successful ofstream write

retrieveRaw:
  - DCHECK buffer non-null for positive file sizes
  - DCHECK assigned vector size matches fileSize

All benchmarks pass with zero assertion violations (TSan + Release).

* style(storage): fmt wrap DCHECK lines

* fix(review): downgrade PRECONDITION→DCHECK for catchable throws, guard zero-size UB

getObjectPath: YAMS_PRECONDITION always aborts — throw below is dead code.
Use YAMS_DCHECK instead (aborts in dev/CI, compiles out in release so
throw fires for backward-compatible catchable std::invalid_argument).

retrieveRaw: fileSize==0 → make_unique_for_overwrite(0) may return null,
assign(null,null) is UB. Early-return empty RawObject for zero-size files.

meson.build: guard benchmark blocks with benchmark_main_dep.found() too.

* fix(review): drop PRECONDITION dead throw, fix test comments

reference_counter: downgrade YAMS_PRECONDITION→DCHECK for transaction
active_ guard (aborts dev/CI, throw fires in release). Add matching
DCHECKs to decrement() and pruneReference() for consistency.

Fix inaccurate test comments claiming fork-test coverage for
WorkCoordinator double-start invariant.

* fix(review): DCHECK store postcondition, cast fileSize, rm dup main

store: downgrade YAMS_POSTCONDITION→DCHECK for exists() check.
Always-on POSTCONDITION adds stat() syscall to hot store() path;
debug-only DCHECK is the correct tier.

retrieveRaw: explicit static_cast<size_t>(fileSize) in assign
to avoid signed/unsigned narrowing warnings.

meson.build: remove benchmark_main_dep from metadata_query_bench_deps.
metadata_path_query_bench.cpp and tree_list_filter_bench.cpp both
define BENCHMARK_MAIN() — linking benchmark_main would be duplicate.

WorkCoordinator.h: update @throws→@pre in doxygen comment to match
YAMS_PRECONDITION(!started_) behavior.

* fix(bench): move pre-store to SetUp, PauseTiming unreliable

PauseTiming before first benchmark iteration caused single-iteration
runs in some Google Benchmark versions. Move pre-store into Fixture::SetUp
where it runs outside the timing loop naturally.

* fix(bench): check store/retrieve results, fix comment copy-vs-move

SetUp: throw if pre-store fails instead of silently continuing.
Retrieve: SkipWithError if retrieveRaw returns error.
Comment: assign() copies, not moves — fix misleading wording.

* fix(bench): rm benchmark_main_dep from targets with own main

ingestion_e2e_bench, retrieval_quality_bench, fusion_strategy_experiment
all define their own main() — linking benchmark_main_dep would cause
duplicate symbol at link time.

* fix(test): rename work_coordinator section, no longer tests double-start

Section title "Rejects start when already running" implied
double-start coverage but only checked worker count after first
start. Renamed to "Starts with correct worker count".

* test(refcount): add closed-state throw coverage for increment/decrement

After commit, increment/decrement must fail:
- Dev/CI (DCHECK): YAMS_DCHECK aborts, covered by fork tests
- Release (NDEBUG): DCHECK compiles out, throw fires

Conditional REQUIRE_THROWS_AS for release path.

* fix(bench): split meson guard for own-main vs benchmark_main targets

Targets with BENCHMARK_MAIN (metadata_path_query, tree_list_filter,
retrieval_quality, fusion_experiment, ingestion_e2e) only need
benchmark_dep. Storage/retrieval benchmarks need benchmark_main_dep.

Split into two guarded blocks to avoid skipping own-main targets
when benchmark_main is unavailable.

* fix(refcount): downgrade increment size PRECONDITION to DCHECK+error return

compressedSize>0 and uncompressedSize>=compressedSize come from user
content (block sizes). PRECONDITION would abort the process on corrupted
or malicious blocks. Return ErrorCode::InvalidArgument instead so
callers can handle gracefully.

* fix(daemon): defer startup maintenance

Make metadata repository readiness the startup boundary while moving search build, recovery cleanup, salvage, and VACUUM into coordinated background or pre-publication paths.\n\nKeep status honest for operators by separating database readiness from maintenance phase and surfacing recovery artifacts.
2026-06-04 22:44:12 -04:00

7059 lines
295 KiB
C++

// Daemon metrics and status test suite (Catch2)
// Consolidates status/metrics tests: embedding status, plugin degradation, WAL metrics, FSM states
// Covers: DaemonMetrics, StatusResponse serialization, plugin degradation, WAL metrics
#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>
#include <algorithm>
#include <array>
#include <chrono>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <future>
#include <mutex>
#include <random>
#include <span>
#include <stdexcept>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include "../../common/test_helpers_catch2.h"
#include <yams/app/services/graph_query_service.hpp>
#include <yams/app/services/session_service.hpp>
#include <yams/crypto/hasher.h>
#include <yams/daemon/components/DaemonLifecycleFsm.h>
#include <yams/daemon/components/DaemonMetrics.h>
#include <yams/daemon/components/DatabaseManager.h>
#include <yams/daemon/components/RequestDispatcher.h>
#include <yams/daemon/components/ResourceGovernor.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/WorkCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/daemon_lifecycle.h>
#include <yams/daemon/ipc/fsm_metrics_registry.h>
#include <yams/daemon/ipc/ipc_protocol.h>
#include <yams/daemon/ipc/proto_serializer.h>
#include <yams/daemon/ipc/request_context.h>
#include <yams/daemon/ipc/request_context_registry.h>
#include <yams/daemon/metric_keys.h>
#include <yams/daemon/resource/external_plugin_host.h>
#include <yams/daemon/resource/model_provider.h>
#include <yams/daemon/resource/plugin_host.h>
#include <yams/metadata/knowledge_graph_store.h>
#include <yams/metadata/metadata_repository.h>
#include <yams/metadata/path_utils.h>
#include <yams/vector/vector_database.h>
#include <yams/version.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/use_future.hpp>
using namespace yams;
using namespace yams::daemon;
// =============================================================================
// Test Helpers
// =============================================================================
namespace {
class StubLifecycle : public IDaemonLifecycle {
public:
explicit StubLifecycle(LifecycleState state = LifecycleState::Ready) {
snapshot_.state = state;
}
LifecycleSnapshot getLifecycleSnapshot() const override {
++getLifecycleSnapshotCalls_;
if (throwOnGetLifecycleSnapshotCall_ > 0 &&
getLifecycleSnapshotCalls_ == throwOnGetLifecycleSnapshotCall_) {
throw std::runtime_error("getLifecycleSnapshot boom");
}
if (throwOnGetLifecycleSnapshotCount_ > 0) {
--throwOnGetLifecycleSnapshotCount_;
throw std::runtime_error("getLifecycleSnapshot boom");
}
return snapshot_;
}
void setSubsystemDegraded(const std::string& subsystem, bool degraded,
const std::string& reason) override {
if (throwOnSetSubsystemDegraded_) {
throw std::runtime_error("setSubsystemDegraded boom");
}
lastSubsystem_ = subsystem;
lastDegraded_ = degraded;
lastReason_ = reason;
setSubsystemDegradedCalls_++;
}
void onDocumentRemoved(const std::string& hash) override { removedHashes_.push_back(hash); }
void requestShutdown(bool, bool) override {}
const std::string& lastSubsystem() const { return lastSubsystem_; }
bool lastDegraded() const { return lastDegraded_; }
const std::string& lastReason() const { return lastReason_; }
int setSubsystemDegradedCalls() const { return setSubsystemDegradedCalls_; }
void resetDegradedState() {
lastSubsystem_.clear();
lastDegraded_ = false;
lastReason_.clear();
setSubsystemDegradedCalls_ = 0;
}
void setThrowOnSetSubsystemDegraded(bool enabled) { throwOnSetSubsystemDegraded_ = enabled; }
void setThrowOnGetLifecycleSnapshot(bool enabled) {
throwOnGetLifecycleSnapshotCount_ = enabled ? 1 : 0;
}
void setThrowOnGetLifecycleSnapshotCount(int count) {
throwOnGetLifecycleSnapshotCount_ = count;
}
void setThrowOnGetLifecycleSnapshotCall(int call) { throwOnGetLifecycleSnapshotCall_ = call; }
const std::vector<std::string>& removedHashes() const { return removedHashes_; }
private:
LifecycleSnapshot snapshot_{};
std::string lastSubsystem_;
bool lastDegraded_{false};
std::string lastReason_;
int setSubsystemDegradedCalls_{0};
bool throwOnSetSubsystemDegraded_{false};
mutable int getLifecycleSnapshotCalls_{0};
mutable int throwOnGetLifecycleSnapshotCount_{0};
int throwOnGetLifecycleSnapshotCall_{0};
std::vector<std::string> removedHashes_;
};
// Minimal stub provider for embedding tests
class StubModelProvider : public IModelProvider {
public:
explicit StubModelProvider(size_t dim, std::string path)
: dim_(dim), path_(std::move(path)), available_(true) {}
Result<std::vector<float>> generateEmbedding(const std::string& text) override {
lastTextArg_ = text;
if (auto it = singleTextErrors_.find(text); it != singleTextErrors_.end()) {
return it->second;
}
if (generateEmbeddingError_) {
return *generateEmbeddingError_;
}
if (!singleEmbeddingResult_.empty()) {
return singleEmbeddingResult_;
}
return std::vector<float>(dim_, 0.25f);
}
Result<std::vector<std::vector<float>>>
generateBatchEmbeddings(const std::vector<std::string>& texts) override {
lastBatchSizeArg_ = texts.size();
if (!texts.empty()) {
lastTextArg_ = texts.front();
}
if (generateBatchError_ && !isWarmupBatch(texts)) {
return *generateBatchError_;
}
if (!batchEmbeddingResult_.empty()) {
return batchEmbeddingResult_;
}
return std::vector<std::vector<float>>(lastBatchSizeArg_, std::vector<float>(dim_, 0.5f));
}
Result<std::vector<float>> generateEmbeddingFor(const std::string&,
const std::string& text) override {
lastTextArg_ = text;
return generateEmbedding(text);
}
Result<std::vector<std::vector<float>>>
generateBatchEmbeddingsFor(const std::string&, const std::vector<std::string>& texts) override {
lastBatchSizeArg_ = texts.size();
if (!texts.empty()) {
lastTextArg_ = texts.front();
}
return generateBatchEmbeddings(texts);
}
Result<void> loadModel(const std::string& modelName) override {
if (loadError_) {
return *loadError_;
}
lastOptionsModel_.clear();
lastOptionsJson_.clear();
loaded_.push_back(modelName);
return Result<void>();
}
Result<void> loadModelWithOptions(const std::string& modelName,
const std::string& optionsJson) override {
if (loadError_) {
return *loadError_;
}
lastOptionsModel_ = modelName;
lastOptionsJson_ = optionsJson;
loaded_.push_back(modelName);
return Result<void>();
}
Result<void> unloadModel(const std::string& modelName) override {
auto it = std::find(loaded_.begin(), loaded_.end(), modelName);
if (it == loaded_.end()) {
return ErrorCode::NotFound;
}
loaded_.erase(it);
return Result<void>();
}
bool isModelLoaded(const std::string& modelName) const override {
if (throwOnIsModelLoadedCount_ > 0) {
--throwOnIsModelLoadedCount_;
throw std::runtime_error("isModelLoaded boom");
}
return std::find(loaded_.begin(), loaded_.end(), modelName) != loaded_.end();
}
std::vector<std::string> getLoadedModels() const override { return loaded_; }
size_t getLoadedModelCount() const override { return loaded_.size(); }
Result<ModelInfo> getModelInfo(const std::string& modelName) const override {
if (failModelInfo_) {
return ErrorCode::NotFound;
}
if (!isModelLoaded(modelName))
return ErrorCode::NotFound;
ModelInfo mi;
mi.name = modelName;
mi.path = path_;
mi.embeddingDim = dim_;
return mi;
}
size_t getEmbeddingDim(const std::string&) const override { return dim_; }
std::shared_ptr<vector::EmbeddingGenerator>
getEmbeddingGenerator(const std::string& = "") override {
return nullptr;
}
std::string getProviderName() const override { return "StubProvider"; }
std::string getProviderVersion() const override { return "vtest"; }
bool isAvailable() const override { return available_; }
size_t getMemoryUsage() const override {
if (throwOnMemoryUsage_) {
throw std::runtime_error("memory boom");
}
return 0;
}
void releaseUnusedResources() override {}
void shutdown() override {}
void setAvailable(bool available) { available_ = available; }
const std::string& lastOptionsModel() const { return lastOptionsModel_; }
const std::string& lastOptionsJson() const { return lastOptionsJson_; }
void setLoadError(Error error) { loadError_ = std::move(error); }
void clearLoadError() { loadError_.reset(); }
void setThrowOnMemoryUsage(bool enabled) { throwOnMemoryUsage_ = enabled; }
void setFailModelInfo(bool enabled) { failModelInfo_ = enabled; }
void setThrowOnIsModelLoaded(bool enabled) { throwOnIsModelLoadedCount_ = enabled ? 1 : 0; }
void setThrowOnIsModelLoadedCount(int count) { throwOnIsModelLoadedCount_ = count; }
void setSingleEmbeddingResult(std::vector<float> embedding) {
singleEmbeddingResult_ = std::move(embedding);
}
void setBatchEmbeddingResult(std::vector<std::vector<float>> embeddings) {
batchEmbeddingResult_ = std::move(embeddings);
}
void setGenerateEmbeddingError(Error error) { generateEmbeddingError_ = std::move(error); }
void setGenerateBatchError(Error error) { generateBatchError_ = std::move(error); }
void setSingleTextError(const std::string& text, Error error) {
singleTextErrors_[text] = std::move(error);
}
void clearSingleTextErrors() { singleTextErrors_.clear(); }
private:
static bool isWarmupBatch(const std::vector<std::string>& texts) {
static const std::array<std::string, 4> kWarmupTexts = {
"warmup text one", "warmup text two", "warmup text three", "warmup text four"};
return texts.size() == kWarmupTexts.size() &&
std::equal(texts.begin(), texts.end(), kWarmupTexts.begin());
}
size_t dim_;
std::string path_;
bool available_;
std::vector<std::string> loaded_;
std::string lastOptionsModel_;
std::string lastOptionsJson_;
std::optional<Error> loadError_;
std::optional<Error> generateEmbeddingError_;
std::optional<Error> generateBatchError_;
bool throwOnMemoryUsage_{false};
bool failModelInfo_{false};
mutable int throwOnIsModelLoadedCount_{0};
std::vector<float> singleEmbeddingResult_;
std::vector<std::vector<float>> batchEmbeddingResult_;
std::unordered_map<std::string, Error> singleTextErrors_;
mutable std::string lastTextArg_;
mutable std::size_t lastBatchSizeArg_{0};
};
using yams::test::ScopedEnvVar;
class TuneAdvisorListGuard {
public:
TuneAdvisorListGuard(uint32_t inflightLimit, uint32_t admissionWaitMs)
: originalInflightLimit_(TuneAdvisor::listInflightLimit()),
originalAdmissionWaitMs_(TuneAdvisor::listAdmissionWaitMs()) {
TuneAdvisor::setListInflightLimit(inflightLimit);
TuneAdvisor::setListAdmissionWaitMs(admissionWaitMs);
}
~TuneAdvisorListGuard() {
TuneAdvisor::setListInflightLimit(originalInflightLimit_);
TuneAdvisor::setListAdmissionWaitMs(originalAdmissionWaitMs_);
}
private:
uint32_t originalInflightLimit_;
uint32_t originalAdmissionWaitMs_;
};
template <typename Fn> class ScopeExit {
public:
explicit ScopeExit(Fn fn) : fn_(std::move(fn)) {}
~ScopeExit() { fn_(); }
private:
Fn fn_;
};
std::shared_ptr<metadata::ConnectionPool> getPruneTestPool() {
static std::shared_ptr<metadata::ConnectionPool> pool = [] {
metadata::ConnectionPoolConfig cfg{};
cfg.enableWAL = false;
cfg.enableForeignKeys = false;
auto p = std::make_shared<metadata::ConnectionPool>(":memory:", cfg);
(void)p->initialize();
return p;
}();
return pool;
}
class StubPruneMetadataRepository : public metadata::MetadataRepository {
public:
StubPruneMetadataRepository() : metadata::MetadataRepository(*getPruneTestPool()) {}
void addDocument(const metadata::DocumentInfo& doc) {
std::lock_guard<std::mutex> lk(mu_);
docs_.push_back(doc);
docsByHash_[doc.sha256Hash] = doc;
docsById_[doc.id] = doc;
}
void setQueryError(Error error) { queryError_ = std::move(error); }
void setThrowOnQuery(std::string message) { throwOnQuery_ = std::move(message); }
void setMissingHash(const std::string& hash) { missingHashes_.insert(hash); }
void setThrowOnLookupHash(const std::string& hash, std::string message) {
throwOnLookup_[hash] = std::move(message);
}
void setDeleteError(int64_t id, Error error) { deleteErrors_[id] = std::move(error); }
void setSnapshots(std::vector<std::string> snapshots) { snapshots_ = std::move(snapshots); }
void setSnapshotError(Error error) { snapshotError_ = std::move(error); }
void setSnapshotDocuments(std::vector<metadata::DocumentInfo> docs) {
snapshotDocs_ = std::move(docs);
}
void setSnapshotDocumentsError(Error error) { snapshotDocsError_ = std::move(error); }
void setExactPathDocument(std::optional<metadata::DocumentInfo> doc) {
std::lock_guard<std::mutex> lk(mu_);
exactPathDocument_ = std::move(doc);
}
void setExactPathError(Error error) {
std::lock_guard<std::mutex> lk(mu_);
exactPathError_ = std::move(error);
}
void setAllMetadata(int64_t id,
std::unordered_map<std::string, metadata::MetadataValue> values) {
std::lock_guard<std::mutex> lk(mu_);
metadataById_[id] = std::move(values);
}
void setAllMetadataError(int64_t id, Error error) {
std::lock_guard<std::mutex> lk(mu_);
metadataByIdErrors_[id] = std::move(error);
}
void setPathTreeDocuments(std::vector<metadata::DocumentInfo> docs) {
std::lock_guard<std::mutex> lk(mu_);
pathTreeDocs_ = std::move(docs);
}
void setMetadataValueCounts(
std::unordered_map<std::string, std::vector<metadata::MetadataValueCount>> counts) {
metadataValueCounts_ = std::move(counts);
}
void setMetadataValueCountsError(Error error) { metadataValueCountsError_ = std::move(error); }
std::size_t deleteCallCount() const {
std::lock_guard<std::mutex> lk(mu_);
return deleteCallCount_;
}
std::vector<int64_t> deletedIds() const {
std::lock_guard<std::mutex> lk(mu_);
return deletedIds_;
}
Result<std::vector<metadata::DocumentInfo>>
queryDocuments(const metadata::DocumentQueryOptions&) override {
std::lock_guard<std::mutex> lk(mu_);
if (throwOnQuery_) {
throw std::runtime_error(*throwOnQuery_);
}
if (queryError_) {
return *queryError_;
}
return docs_;
}
Result<std::vector<metadata::DocumentInfo>>
findDocumentsBySnapshot(const std::string&) override {
std::lock_guard<std::mutex> lk(mu_);
if (snapshotDocsError_) {
return *snapshotDocsError_;
}
return snapshotDocs_;
}
Result<std::vector<std::string>> getSnapshots() override {
std::lock_guard<std::mutex> lk(mu_);
if (snapshotError_) {
return *snapshotError_;
}
return snapshots_;
}
Result<std::unordered_map<std::string, std::vector<metadata::MetadataValueCount>>>
getMetadataValueCounts(const std::vector<std::string>&,
const metadata::DocumentQueryOptions&) override {
std::lock_guard<std::mutex> lk(mu_);
if (metadataValueCountsError_) {
return *metadataValueCountsError_;
}
return metadataValueCounts_;
}
Result<std::optional<metadata::DocumentInfo>>
getDocumentByHash(const std::string& hash) override {
std::lock_guard<std::mutex> lk(mu_);
if (auto it = throwOnLookup_.find(hash); it != throwOnLookup_.end()) {
throw std::runtime_error(it->second);
}
if (missingHashes_.count(hash) > 0) {
return std::optional<metadata::DocumentInfo>(std::nullopt);
}
if (auto it = docsByHash_.find(hash); it != docsByHash_.end()) {
return std::optional<metadata::DocumentInfo>(it->second);
}
return std::optional<metadata::DocumentInfo>(std::nullopt);
}
Result<std::unordered_map<std::string, metadata::MetadataValue>>
getAllMetadata(int64_t documentId) override {
std::lock_guard<std::mutex> lk(mu_);
if (auto it = metadataByIdErrors_.find(documentId); it != metadataByIdErrors_.end()) {
return it->second;
}
if (auto it = metadataById_.find(documentId); it != metadataById_.end()) {
return it->second;
}
return std::unordered_map<std::string, metadata::MetadataValue>{};
}
Result<std::optional<metadata::DocumentInfo>>
findDocumentByExactPath(const std::string&) override {
std::lock_guard<std::mutex> lk(mu_);
if (exactPathError_) {
return *exactPathError_;
}
if (exactPathDocument_.has_value()) {
return exactPathDocument_;
}
return std::optional<metadata::DocumentInfo>(std::nullopt);
}
Result<std::vector<metadata::DocumentInfo>>
findDocumentsByPathTreePrefix(std::string_view, bool = true, int limit = 0) override {
std::lock_guard<std::mutex> lk(mu_);
if (limit > 0 && static_cast<std::size_t>(limit) < pathTreeDocs_.size()) {
return std::vector<metadata::DocumentInfo>(pathTreeDocs_.begin(),
pathTreeDocs_.begin() + limit);
}
return pathTreeDocs_;
}
Result<void> deleteDocument(int64_t id) override {
std::lock_guard<std::mutex> lk(mu_);
deleteCallCount_++;
if (auto it = deleteErrors_.find(id); it != deleteErrors_.end()) {
return it->second;
}
deletedIds_.push_back(id);
if (auto it = docsById_.find(id); it != docsById_.end()) {
docsByHash_.erase(it->second.sha256Hash);
docsById_.erase(it);
}
docs_.erase(
std::remove_if(docs_.begin(), docs_.end(),
[id](const metadata::DocumentInfo& doc) { return doc.id == id; }),
docs_.end());
return Result<void>();
}
private:
mutable std::mutex mu_;
std::vector<metadata::DocumentInfo> docs_;
std::unordered_map<std::string, metadata::DocumentInfo> docsByHash_;
std::unordered_map<int64_t, metadata::DocumentInfo> docsById_;
std::optional<Error> queryError_;
std::optional<Error> snapshotError_;
std::optional<Error> snapshotDocsError_;
std::optional<Error> metadataValueCountsError_;
std::optional<std::string> throwOnQuery_;
std::unordered_set<std::string> missingHashes_;
std::unordered_map<std::string, std::string> throwOnLookup_;
std::unordered_map<int64_t, Error> deleteErrors_;
std::optional<metadata::DocumentInfo> exactPathDocument_;
std::optional<Error> exactPathError_;
std::unordered_map<int64_t, std::unordered_map<std::string, metadata::MetadataValue>>
metadataById_;
std::unordered_map<int64_t, Error> metadataByIdErrors_;
std::vector<metadata::DocumentInfo> pathTreeDocs_;
std::vector<std::string> snapshots_;
std::vector<metadata::DocumentInfo> snapshotDocs_;
std::unordered_map<std::string, std::vector<metadata::MetadataValueCount>> metadataValueCounts_;
std::size_t deleteCallCount_{0};
std::vector<int64_t> deletedIds_;
};
class StubContentStore : public api::IContentStore {
public:
void setBlob(const std::string& hash, const std::string& content) {
blobs_[hash] = std::vector<std::byte>(reinterpret_cast<const std::byte*>(content.data()),
reinterpret_cast<const std::byte*>(content.data()) +
content.size());
}
void setRemoveResult(const std::string& hash, Result<bool> result) {
removeResults_[hash] = std::move(result);
}
bool hasBlob(const std::string& hash) const { return blobs_.contains(hash); }
std::vector<std::string> removedHashes() const { return removedHashes_; }
Result<api::StoreResult> storeBytes(std::span<const std::byte> data,
const api::ContentMetadata&) override {
std::string hash = "stub-hash-" + std::to_string(blobs_.size() + 1);
blobs_[hash] = std::vector<std::byte>(data.begin(), data.end());
api::StoreResult result;
result.contentHash = hash;
result.bytesStored = blobs_[hash].size();
return result;
}
Result<std::vector<std::byte>> retrieveBytes(const std::string& hash) override {
auto it = blobs_.find(hash);
if (it == blobs_.end()) {
return Error{ErrorCode::NotFound, "content not found"};
}
return it->second;
}
Result<std::vector<std::byte>> retrieveBytesPrefix(const std::string& hash,
std::size_t maxBytes) override {
auto bytes = retrieveBytes(hash);
if (!bytes) {
return bytes.error();
}
const auto prefixSize = std::min(maxBytes, bytes.value().size());
bytes.value().resize(prefixSize);
return bytes;
}
Result<api::IContentStore::RawContent> retrieveRaw(const std::string& hash) override {
auto bytes = retrieveBytes(hash);
if (!bytes) {
return bytes.error();
}
api::IContentStore::RawContent raw;
raw.data = std::move(bytes.value());
return raw;
}
std::future<Result<api::IContentStore::RawContent>>
retrieveRawAsync(const std::string& hash) override {
return std::async(std::launch::deferred, [this, hash]() { return retrieveRaw(hash); });
}
Result<api::StoreResult> store(const std::filesystem::path& path, const api::ContentMetadata&,
api::ProgressCallback) override {
std::ifstream ifs(path, std::ios::binary);
if (!ifs.good()) {
return Error{ErrorCode::FileNotFound, "stub file not found"};
}
std::vector<std::byte> bytes;
char ch = 0;
while (ifs.get(ch)) {
bytes.push_back(static_cast<std::byte>(ch));
}
if (!ifs.eof()) {
return Error{ErrorCode::IOError, "stub read failed"};
}
return storeBytes(bytes, {});
}
Result<api::RetrieveResult> retrieve(const std::string&, const std::filesystem::path&,
api::ProgressCallback) override {
return ErrorCode::NotImplemented;
}
Result<api::StoreResult> storeStream(std::istream&, const api::ContentMetadata&,
api::ProgressCallback) override {
return ErrorCode::NotImplemented;
}
Result<api::RetrieveResult> retrieveStream(const std::string&, std::ostream&,
api::ProgressCallback) override {
return ErrorCode::NotImplemented;
}
Result<bool> exists(const std::string&) const override { return ErrorCode::NotImplemented; }
Result<bool> remove(const std::string& hash) override {
if (auto it = removeResults_.find(hash); it != removeResults_.end()) {
return it->second;
}
removedHashes_.push_back(hash);
blobs_.erase(hash);
return true;
}
Result<api::ContentMetadata> getMetadata(const std::string&) const override {
return ErrorCode::NotImplemented;
}
Result<void> updateMetadata(const std::string&, const api::ContentMetadata&) override {
return ErrorCode::NotImplemented;
}
std::vector<Result<api::StoreResult>>
storeBatch(const std::vector<std::filesystem::path>&,
const std::vector<api::ContentMetadata>&) override {
return {};
}
std::vector<Result<bool>> removeBatch(const std::vector<std::string>&) override { return {}; }
api::ContentStoreStats getStats() const override { return {}; }
api::HealthStatus checkHealth() const override { return {}; }
Result<void> verify(api::ProgressCallback) override { return ErrorCode::NotImplemented; }
Result<void> compact(api::ProgressCallback) override { return ErrorCode::NotImplemented; }
Result<void> garbageCollect(api::ProgressCallback) override {
return ErrorCode::NotImplemented;
}
private:
std::unordered_map<std::string, std::vector<std::byte>> blobs_;
std::unordered_map<std::string, Result<bool>> removeResults_;
std::vector<std::string> removedHashes_;
};
class StubGraphQueryService : public app::services::IGraphQueryService {
public:
void addConnectedDocument(std::string hash, std::string relation, int distance = 1,
double relevance = 0.0) {
app::services::GraphConnectedNode node{};
node.nodeMetadata.documentHash = std::move(hash);
node.distance = distance;
node.relevanceScore = relevance;
app::services::GraphEdgeDescriptor edge{};
edge.relation = std::move(relation);
node.connectingEdges.push_back(std::move(edge));
response_.allConnectedNodes.push_back(std::move(node));
}
Result<app::services::GraphQueryResponse>
query(const app::services::GraphQueryRequest&) override {
return response_;
}
Result<app::services::ListSnapshotsResponse>
listSnapshots(const app::services::ListSnapshotsRequest&) override {
return app::services::ListSnapshotsResponse{};
}
Result<app::services::PathHistoryResponse>
getPathHistory(const app::services::PathHistoryRequest&) override {
return app::services::PathHistoryResponse{};
}
Result<std::optional<std::int64_t>> resolveToNodeId(const std::string&) override {
return std::optional<std::int64_t>(std::nullopt);
}
private:
app::services::GraphQueryResponse response_{};
};
std::filesystem::path makeTempDir(const std::string& prefix);
metadata::DocumentInfo makeGraphDispatcherDocument(const std::filesystem::path& path,
const std::string& hash) {
metadata::DocumentInfo info;
info.filePath = path.string();
info.fileName = path.filename().string();
info.fileExtension = path.extension().string();
info.fileSize = static_cast<std::int64_t>(std::filesystem::file_size(path));
info.sha256Hash = hash;
info.mimeType = "text/plain";
const auto now = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::now());
info.createdTime = now;
info.modifiedTime = now;
info.indexedTime = now;
info.contentExtracted = true;
info.extractionStatus = metadata::ExtractionStatus::Success;
metadata::populatePathDerivedFields(info);
return info;
}
struct GraphDispatcherFixture {
GraphDispatcherFixture() {
testDir = makeTempDir("yams_graph_dispatcher_");
cfg.dataDir = testDir;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
svc = std::make_unique<ServiceManager>(cfg, state, lifecycleFsm);
dispatcher = std::make_unique<RequestDispatcher>(&lifecycle, svc.get(), &state);
}
~GraphDispatcherFixture() {
dispatcher.reset();
svc.reset();
if (writeWork) {
writeWork.reset();
}
writeIo.stop();
if (writeThread.joinable()) {
writeThread.join();
}
kgStore.reset();
repo.reset();
pool.reset();
std::error_code ec;
std::filesystem::remove_all(testDir, ec);
}
void initMetadata(bool withKg = true) {
metadata::ConnectionPoolConfig poolConfig{};
poolConfig.minConnections = 1;
poolConfig.maxConnections = 2;
pool = std::make_shared<metadata::ConnectionPool>(
(testDir / "graph_dispatcher.db").string(), poolConfig);
REQUIRE(pool->initialize().has_value());
repo = std::make_shared<metadata::MetadataRepository>(*pool);
svc->__test_setMetadataRepo(repo);
if (withKg) {
metadata::KnowledgeGraphStoreConfig kgConfig{};
kgConfig.enable_alias_fts = true;
kgConfig.enable_wal = false;
auto kgResult = metadata::makeSqliteKnowledgeGraphStore(*pool, kgConfig);
REQUIRE(kgResult.has_value());
kgStore = std::shared_ptr<metadata::KnowledgeGraphStore>(std::move(kgResult).value());
repo->setKnowledgeGraphStore(kgStore);
writeIo.restart();
writeWork = std::make_unique<
boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
writeIo.get_executor());
writeThread = std::thread([this] { writeIo.run(); });
auto writeCoord = std::make_unique<WriteCoordinator>(writeIo, kgStore, repo);
writeCoord->start();
svc->__test_setWriteCoordinator(std::move(writeCoord));
}
}
int64_t upsertNode(const std::string& key, const std::string& label, const std::string& type,
const std::string& properties = {}) {
REQUIRE(kgStore);
metadata::KGNode node;
node.nodeKey = key;
node.label = label;
node.type = type;
if (!properties.empty()) {
node.properties = properties;
}
auto result = kgStore->upsertNode(node);
REQUIRE(result.has_value());
return result.value();
}
void addEdge(int64_t srcNodeId, int64_t dstNodeId, const std::string& relation,
float weight = 1.0f, const std::string& properties = {}) {
REQUIRE(kgStore);
metadata::KGEdge edge;
edge.srcNodeId = srcNodeId;
edge.dstNodeId = dstNodeId;
edge.relation = relation;
edge.weight = weight;
if (!properties.empty()) {
edge.properties = properties;
}
REQUIRE(kgStore->addEdge(edge).has_value());
}
void execSql(const std::string& sql) {
REQUIRE(pool);
auto result = pool->withConnection(
[&](metadata::Database& db) -> Result<void> { return db.execute(sql); });
REQUIRE(result.has_value());
}
std::filesystem::path testDir;
DaemonConfig cfg;
StubLifecycle lifecycle;
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
std::unique_ptr<ServiceManager> svc;
std::unique_ptr<RequestDispatcher> dispatcher;
boost::asio::io_context writeIo;
std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>
writeWork;
std::thread writeThread;
std::shared_ptr<metadata::ConnectionPool> pool;
std::shared_ptr<metadata::MetadataRepository> repo;
std::shared_ptr<metadata::KnowledgeGraphStore> kgStore;
};
std::filesystem::path makeTempDir(const std::string& prefix) {
auto base = std::filesystem::temp_directory_path();
auto dir = base / (prefix + std::to_string(std::random_device{}()));
std::filesystem::create_directories(dir);
return dir;
}
std::filesystem::path writeTempFile(const std::string& prefix, const std::string& name,
const std::string& content) {
auto path = makeTempDir(prefix) / name;
std::ofstream out(path, std::ios::binary);
REQUIRE(out.good());
out << content;
out.close();
REQUIRE(out.good());
return path;
}
std::filesystem::path createMockExternalPlugin(const std::filesystem::path& baseDir,
const std::string& name) {
auto pluginDir = baseDir / name;
std::filesystem::create_directories(pluginDir);
{
std::ofstream pluginFile(pluginDir / "plugin.py");
pluginFile << R"PY(#!/usr/bin/env python3
import json
import sys
def handle_request(req):
method = req.get("method", "")
if method == "handshake.manifest":
return {
"name": "dispatcher_external_plugin",
"version": "1.0.0",
"interfaces": ["content_extractor_v1"]
}
if method == "plugin.init":
return {"status": "initialized"}
if method == "plugin.health":
return {"status": "ok"}
if method == "plugin.shutdown":
return {"status": "ok"}
raise ValueError(f"Method not found: {method}")
for line in sys.stdin:
line = line.strip()
if not line:
continue
req_id = None
try:
req = json.loads(line)
req_id = req.get("id")
response = {"jsonrpc": "2.0", "id": req_id, "result": handle_request(req)}
except Exception as exc:
response = {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32603, "message": str(exc)}
}
print(json.dumps(response), flush=True)
)PY";
}
{
std::ofstream manifestFile(pluginDir / "yams-plugin.json");
manifestFile << R"JSON({
"name": "dispatcher_external_plugin",
"version": "1.0.0",
"interfaces": ["content_extractor_v1"],
"entry": {
"fallback_cmd": )JSON";
#ifdef _WIN32
manifestFile << R"JSON(["python", "-u", "${plugin_dir}/plugin.py"])JSON";
#else
manifestFile << R"JSON(["/usr/bin/env", "python3", "-u", "${plugin_dir}/plugin.py"])JSON";
#endif
manifestFile << R"JSON(,
"env": {
"PYTHONUNBUFFERED": "1"
}
}
})JSON";
}
return pluginDir;
}
std::string makePythonShimPath() {
auto shimDir = makeTempDir("yams_python_shim_");
auto shim = shimDir / "python";
{
std::ofstream shimFile(shim);
shimFile << "#!/bin/sh\n"
"exec /usr/bin/env python3 \"$@\"\n";
}
std::filesystem::permissions(shim,
std::filesystem::perms::owner_read |
std::filesystem::perms::owner_write |
std::filesystem::perms::owner_exec,
std::filesystem::perm_options::replace);
if (const char* currentPath = std::getenv("PATH"); currentPath && *currentPath) {
return shimDir.string() + ":" + currentPath;
}
return shimDir.string() + ":/usr/bin:/bin";
}
Response dispatchRequest(RequestDispatcher& dispatcher, const Request& req) {
boost::asio::io_context ioc;
auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(req), boost::asio::use_future);
ioc.run();
return fut.get();
}
std::shared_ptr<SpscQueue<InternalEventBus::StoreDocumentTask>>
getStoreDocumentTaskChannel(std::size_t capacity = 0) {
auto existing = InternalEventBus::instance().get_channel<InternalEventBus::StoreDocumentTask>(
"store_document_tasks");
if (existing) {
return existing;
}
if (capacity == 0) {
capacity = static_cast<std::size_t>(TuneAdvisor::storeDocumentChannelCapacity());
}
return InternalEventBus::instance().get_or_create_channel<InternalEventBus::StoreDocumentTask>(
"store_document_tasks", capacity);
}
void drainStoreDocumentTasks() {
auto channel = InternalEventBus::instance().get_channel<InternalEventBus::StoreDocumentTask>(
"store_document_tasks");
if (!channel) {
return;
}
InternalEventBus::StoreDocumentTask task;
while (channel->try_pop(task)) {
}
}
template <typename Predicate> bool waitForCondition(Predicate&& predicate, int attempts = 100) {
for (int i = 0; i < attempts; ++i) {
if (predicate()) {
return true;
}
std::this_thread::sleep_for(std::chrono::milliseconds(20));
}
return predicate();
}
} // namespace
// =============================================================================
// Embedding Status Tests
// =============================================================================
TEST_CASE("DaemonMetrics: Embedding provider status", "[daemon][metrics][embedding]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_metrics_embed_");
ServiceManager svc(cfg, state, lifecycleFsm);
SECTION("Snapshot includes embedding backend information") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/test-model.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setAdoptedProviderPluginName("stub");
// Model provider is now directly used - no need to ensure generator
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
// Backend should be "plugin:<name>" format or "unknown"
bool validBackend =
(snap->embeddingBackend == "unknown" || snap->embeddingBackend.starts_with("plugin:"));
REQUIRE(validBackend);
}
SECTION("Metrics snapshot doesn't crash without provider") {
// No provider set
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
REQUIRE(!snap->embeddingBackend.empty());
}
}
// =============================================================================
// WAL Metrics Tests
// =============================================================================
TEST_CASE("DaemonMetrics: WAL metrics in GetStats", "[daemon][metrics][wal]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_metrics_wal_");
YamsDaemon daemon(cfg);
DaemonLifecycleAdapter lifecycleAdapter(&daemon);
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);
SECTION("GetStats includes WAL metric keys") {
GetStatsRequest req;
Request r = req;
boost::asio::io_context ioc;
auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
ioc.run();
auto resp = fut.get();
REQUIRE(std::holds_alternative<GetStatsResponse>(resp));
auto stats = std::get<GetStatsResponse>(resp);
// WAL metrics should be present
REQUIRE(stats.additionalStats.count("wal_active_transactions") > 0);
REQUIRE(stats.additionalStats.count("wal_pending_entries") > 0);
}
}
TEST_CASE("RequestDispatcher: status includes canonical readiness flags",
"[daemon][status][readiness]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_readiness_flags_");
YamsDaemon daemon(cfg);
DaemonLifecycleAdapter lifecycleAdapter(&daemon);
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);
StatusRequest req;
req.detailed = false;
Request r = req;
boost::asio::io_context ioc;
auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
ioc.run();
auto resp = fut.get();
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& status = std::get<StatusResponse>(resp);
const std::vector<std::string_view> requiredCoreReadinessKeys = {
readiness::kIpcServer,
readiness::kContentStore,
readiness::kDatabase,
readiness::kMetadataRepo,
readiness::kSearchEngine,
readiness::kModelProvider,
readiness::kVectorIndex,
readiness::kVectorDb,
readiness::kPlugins,
readiness::kVectorDbInitAttempted,
readiness::kVectorDbReady,
readiness::kVectorDbDim,
readiness::kEmbeddingReady,
readiness::kEmbeddingDegraded,
readiness::kPluginsReady,
readiness::kPluginsDegraded,
readiness::kContentExtractorsReady,
readiness::kSymbolExtractorsReady,
readiness::kEntityExtractorsReady,
readiness::kTitleExtractorReady,
readiness::kRepairService,
readiness::kSearchEngineBuildReasonInitial,
readiness::kSearchEngineBuildReasonRebuild,
readiness::kSearchEngineBuildReasonDegraded,
readiness::kSearchEngineLexicalReady,
readiness::kSearchEngineVectorUsable,
readiness::kSearchEngineHybridUsable,
readiness::kSearchEngineLexicalEnhancementConfigured,
readiness::kSearchEngineLexicalEnhancementReady,
readiness::kSearchEngineLexicalEnhancementBuilding,
readiness::kSearchEngineFragmentGeometryReady};
for (const auto key : requiredCoreReadinessKeys) {
INFO("missing readiness key: " << key);
REQUIRE(status.readinessStates.count(std::string(key)) > 0);
}
const std::array<std::string_view, 5> requiredCapabilityMetricKeys = {
metrics::kContentExtractorsLoaded, metrics::kSymbolExtractorsLoaded,
metrics::kEntityExtractorsLoaded, metrics::kTitleExtractorEnabled,
metrics::kPluginSkippedCount,
};
for (const auto key : requiredCapabilityMetricKeys) {
INFO("missing capability metric key: " << key);
REQUIRE(status.requestCounts.count(std::string(key)) > 0);
}
}
TEST_CASE("RequestDispatcher: compact and detailed status agree on FSM readiness",
"[daemon][status][readiness]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_fsm_readiness_parity_");
YamsDaemon daemon(cfg);
DaemonLifecycleAdapter lifecycleAdapter(&daemon);
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);
auto fetch = [&](bool detailed) {
StatusRequest req;
req.detailed = detailed;
Request r = req;
boost::asio::io_context ioc;
auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
ioc.run();
auto resp = fut.get();
REQUIRE(std::holds_alternative<StatusResponse>(resp));
return std::get<StatusResponse>(resp);
};
const auto compact = fetch(false);
const auto detailed = fetch(true);
for (const auto key : {readiness::kEmbeddingReady, readiness::kEmbeddingDegraded,
readiness::kPluginsReady, readiness::kPluginsDegraded}) {
const std::string k(key);
INFO("readiness key must agree between compact and detailed: " << k);
REQUIRE(compact.readinessStates.count(k) > 0);
REQUIRE(detailed.readinessStates.count(k) > 0);
REQUIRE(compact.readinessStates.at(k) == detailed.readinessStates.at(k));
}
}
TEST_CASE("RequestDispatcher: status includes repair metrics and flags",
"[daemon][status][repair]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_repair_metrics_");
YamsDaemon daemon(cfg);
DaemonLifecycleAdapter lifecycleAdapter(&daemon);
StateComponent state;
state.stats.repairQueueDepth.store(7);
state.stats.repairBatchesAttempted.store(11);
state.stats.repairEmbeddingsGenerated.store(5);
state.stats.repairEmbeddingsSkipped.store(2);
state.stats.repairFailedOperations.store(1);
state.stats.repairIdleTicks.store(99);
state.stats.repairBusyTicks.store(33);
state.stats.repairCurrentOperationCode.store(10);
state.stats.repairCurrentOperationStartedMs.store(
static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now().time_since_epoch())
.count()) -
1500);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);
StatusRequest req;
req.detailed = true;
Request r = req;
boost::asio::io_context ioc;
auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
ioc.run();
auto resp = fut.get();
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& status = std::get<StatusResponse>(resp);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairQueueDepth)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairBatchesAttempted)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairEmbeddingsGenerated)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairEmbeddingsSkipped)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairFailedOperations)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairIdleTicks)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairBusyTicks)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairRunning)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairInProgress)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairCurrentOperationCode)) > 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairCurrentOperationElapsedMs)) > 0);
CHECK(status.requestCounts.at(std::string(metrics::kRepairCurrentOperationCode)) == 10);
CHECK(status.requestCounts.at(std::string(metrics::kRepairCurrentOperationElapsedMs)) >= 1);
}
TEST_CASE("RequestDispatcher: status includes the current request in requestsProcessed",
"[daemon][status][requests]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_request_count_");
YamsDaemon daemon(cfg);
DaemonLifecycleAdapter lifecycleAdapter(&daemon);
StateComponent state;
state.stats.requestsProcessed.store(41, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);
auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& status = std::get<StatusResponse>(resp);
CHECK(status.requestsProcessed == 42);
CHECK(state.stats.requestsProcessed.load(std::memory_order_relaxed) == 42);
}
TEST_CASE("RequestDispatcher: status responds even when lifecycle not ready",
"[daemon][status][readiness]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_not_ready_");
YamsDaemon daemon(cfg);
DaemonLifecycleAdapter lifecycleAdapter(&daemon);
// Daemon is not started; lifecycle state is expected to be non-ready.
{
const auto snap = daemon.getLifecycle().snapshot();
REQUIRE(snap.state != LifecycleState::Ready);
REQUIRE(snap.state != LifecycleState::Degraded);
}
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycleAdapter, &svc, &state);
StatusRequest req;
req.detailed = true;
Request r = req;
boost::asio::io_context ioc;
auto fut = boost::asio::co_spawn(ioc, dispatcher.dispatch(r), boost::asio::use_future);
ioc.run();
auto resp = fut.get();
REQUIRE(std::holds_alternative<StatusResponse>(resp));
}
TEST_CASE("RequestDispatcher: tree diff handler validates inputs and missing metadata repo",
"[daemon][tree-diff][dispatcher]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_tree_diff_dispatcher_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
SECTION("returns invalid argument when snapshot ids are missing") {
ListTreeDiffRequest req;
req.targetSnapshotId = "target-only";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message.find("baseSnapshotId") != std::string::npos);
}
SECTION("returns internal error when metadata repository is unavailable") {
ListTreeDiffRequest req;
req.baseSnapshotId = "base";
req.targetSnapshotId = "target";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Metadata repository not available") != std::string::npos);
}
}
TEST_CASE("RequestDispatcher: prune handler reports missing metadata repo",
"[daemon][prune][dispatcher]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_prune_dispatcher_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
PruneRequest req;
req.dryRun = true;
req.extensions = {"log"};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
}
TEST_CASE("RequestDispatcher: prune handler covers parsing and execution branches",
"[daemon][prune][dispatcher]") {
auto makeDoc = [](int64_t id, std::string filePath, std::string ext, int64_t size,
std::string hash, int64_t ageDays) {
metadata::DocumentInfo doc{};
doc.id = id;
doc.filePath = std::move(filePath);
doc.fileName = std::filesystem::path(doc.filePath).filename().string();
doc.fileExtension = std::move(ext);
doc.fileSize = size;
doc.sha256Hash = std::move(hash);
doc.mimeType = "text/plain";
const auto now =
std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());
doc.modifiedTime = now - std::chrono::hours(ageDays * 24);
doc.indexedTime = now;
return doc;
};
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_prune_dispatcher_live_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto store = std::make_shared<StubContentStore>();
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
repo->addDocument(makeDoc(1, "build/output.log", "log", 2048, std::string(64, 'a'), 10));
repo->addDocument(
makeDoc(2, "logs/server.log", ".log", 3 * 1024 * 1024, std::string(64, 'b'), 14));
repo->addDocument(makeDoc(3, "notes/readme.txt", "txt", 64, std::string(64, 'c'), 2));
repo->addDocument(makeDoc(4, "cache/old.tmp", "tmp", 4096, std::string(64, 'd'), 500));
SECTION("dry run parses age and size filters across units") {
PruneRequest req;
req.dryRun = true;
req.olderThan = "1w";
req.largerThan = "1KB";
req.smallerThan = "5MB";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.filesDeleted == 0);
CHECK(pruneResp.filesFailed == 0);
CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 4096));
REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
REQUIRE(pruneResp.categoryCounts.count("none") == 1);
REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
CHECK(pruneResp.categoryCounts.at("build-object") == 1);
CHECK(pruneResp.categoryCounts.at("none") == 1);
CHECK(pruneResp.categoryCounts.at("temp") == 1);
}
SECTION("dry run accepts month and year age units") {
PruneRequest monthReq;
monthReq.dryRun = true;
monthReq.olderThan = "1m";
auto monthResp = dispatchRequest(dispatcher, Request{monthReq});
REQUIRE(std::holds_alternative<PruneResponse>(monthResp));
const auto& monthPruneResp = std::get<PruneResponse>(monthResp);
REQUIRE(monthPruneResp.categoryCounts.count("temp") == 1);
CHECK(monthPruneResp.categoryCounts.at("temp") == 1);
CHECK(monthPruneResp.categoryCounts.count("logs") == 0);
PruneRequest yearReq;
yearReq.dryRun = true;
yearReq.olderThan = "1y";
auto yearResp = dispatchRequest(dispatcher, Request{yearReq});
REQUIRE(std::holds_alternative<PruneResponse>(yearResp));
const auto& yearPruneResp = std::get<PruneResponse>(yearResp);
CHECK(yearPruneResp.filesDeleted == 0);
CHECK(yearPruneResp.totalBytesFreed == 4096);
REQUIRE(yearPruneResp.categoryCounts.count("temp") == 1);
CHECK(yearPruneResp.categoryCounts.at("temp") == 1);
}
SECTION("dry run accepts gigabyte size units") {
PruneRequest req;
req.dryRun = true;
req.largerThan = "1GB";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.totalBytesFreed == 0);
CHECK(pruneResp.categoryCounts.empty());
}
SECTION("dry run falls back on unknown age unit and raw byte sizes") {
PruneRequest req;
req.dryRun = true;
req.olderThan = "3q";
req.largerThan = "2000";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 4096));
REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
REQUIRE(pruneResp.categoryCounts.count("none") == 1);
REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
CHECK(pruneResp.categoryCounts.at("build-object") == 1);
CHECK(pruneResp.categoryCounts.at("none") == 1);
CHECK(pruneResp.categoryCounts.at("temp") == 1);
}
SECTION("dry run filters by extension and categories") {
PruneRequest req;
req.dryRun = true;
req.categories = {"build"};
req.extensions = {"log"};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.totalBytesFreed == 2048);
CHECK(pruneResp.categoryCounts.size() == 1);
REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
CHECK(pruneResp.categoryCounts.at("build-object") == 1);
}
SECTION("dry run matches dotted extensions when category is none") {
PruneRequest req;
req.dryRun = true;
req.categories = {"none"};
req.extensions = {"log"};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.totalBytesFreed == 3 * 1024 * 1024);
CHECK(pruneResp.categoryCounts.size() == 1);
REQUIRE(pruneResp.categoryCounts.count("none") == 1);
CHECK(pruneResp.categoryCounts.at("none") == 1);
}
SECTION("dry run handles query failures from repository") {
repo->setQueryError(Error{ErrorCode::InternalError, "query boom"});
auto resp = dispatchRequest(dispatcher, Request{PruneRequest{}});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.errorMessage.empty());
CHECK(pruneResp.totalBytesFreed == 0);
CHECK(pruneResp.categoryCounts.empty());
}
SECTION("dry run handles repository exceptions") {
repo->setThrowOnQuery("query exception");
auto resp = dispatchRequest(dispatcher, Request{PruneRequest{}});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.errorMessage == "Failed to query candidates: query exception");
}
SECTION("apply mode counts deleted missing failed and exception cases") {
repo->setMissingHash(std::string(64, 'b'));
repo->setDeleteError(4, Error{ErrorCode::InternalError, "delete failed"});
repo->setThrowOnLookupHash(std::string(64, 'c'), "lookup boom");
PruneRequest req;
req.dryRun = false;
req.olderThan = "1d";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.filesDeleted == 2);
CHECK(pruneResp.filesFailed == 2);
CHECK(pruneResp.totalBytesFreed == 2048);
REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
REQUIRE(pruneResp.categoryCounts.count("none") == 1);
REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
CHECK(pruneResp.categoryCounts.at("build-object") == 1);
CHECK(pruneResp.categoryCounts.at("none") == 2);
CHECK(pruneResp.categoryCounts.at("temp") == 1);
CHECK(repo->deleteCallCount() == 2);
CHECK(repo->deletedIds() == std::vector<int64_t>{1});
const auto removed = store->removedHashes();
CHECK(std::find(removed.begin(), removed.end(), std::string(64, 'd')) == removed.end());
}
SECTION("apply mode removes content through the document deletion service") {
const auto targetHash = std::string(64, 'a');
store->setBlob(targetHash, "indexed log bytes");
PruneRequest req;
req.dryRun = false;
req.categories = {"build"};
req.extensions = {"log"};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.filesDeleted == 1);
CHECK(pruneResp.filesFailed == 0);
CHECK(pruneResp.totalBytesFreed == 2048);
CHECK_FALSE(store->hasBlob(targetHash));
CHECK(store->removedHashes() == std::vector<std::string>{targetHash});
CHECK(repo->deletedIds() == std::vector<int64_t>{1});
}
SECTION("apply mode yields during large prune batches") {
auto fullHashFor = [](int i) {
static constexpr char kHex[] = "0123456789abcdef";
std::string hash(64, 'e');
hash[62] = kHex[(i / 16) & 0x0f];
hash[63] = kHex[i & 0x0f];
return hash;
};
for (int i = 0; i < 105; ++i) {
repo->addDocument(makeDoc(1000 + i, "cache/bulk_" + std::to_string(i) + ".tmp", "tmp",
2048, fullHashFor(i), 30));
}
PruneRequest req;
req.dryRun = false;
req.categories = {"temp"};
req.extensions = {"tmp"};
req.olderThan = "1d";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.filesDeleted == 106);
CHECK(pruneResp.filesFailed == 0);
CHECK(pruneResp.totalBytesFreed == (4096 + 105 * 2048));
REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
CHECK(pruneResp.categoryCounts.at("temp") == 106);
CHECK(repo->deleteCallCount() == 106);
CHECK(repo->deletedIds().size() == 106);
}
SECTION("dry run leaves empty size filters as zero") {
PruneRequest req;
req.dryRun = true;
req.largerThan = "";
req.smallerThan = "";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PruneResponse>(resp));
const auto& pruneResp = std::get<PruneResponse>(resp);
CHECK(pruneResp.totalBytesFreed == (2048 + 3 * 1024 * 1024 + 64 + 4096));
REQUIRE(pruneResp.categoryCounts.count("build-object") == 1);
REQUIRE(pruneResp.categoryCounts.count("none") == 1);
REQUIRE(pruneResp.categoryCounts.count("temp") == 1);
CHECK(pruneResp.categoryCounts.at("build-object") == 1);
CHECK(pruneResp.categoryCounts.at("none") == 2);
CHECK(pruneResp.categoryCounts.at("temp") == 1);
}
}
TEST_CASE("RequestDispatcher: model handlers cover validation and status branches",
"[daemon][model][dispatcher]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_model_dispatcher_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
SECTION("load model rejects missing name when provider exists") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
svc.__test_setModelProvider(provider);
LoadModelRequest req;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidData);
CHECK(err.message == "modelName is required");
}
SECTION("load model with options succeeds and records provider state") {
auto provider = std::make_shared<StubModelProvider>(768, "/tmp/provider-model.onnx");
svc.__test_setModelProvider(provider);
LoadModelRequest req;
req.modelName = "dispatcher-model";
req.optionsJson = R"({"mode":"offline"})";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
const auto& loadResp = std::get<ModelLoadResponse>(resp);
CHECK(loadResp.success);
CHECK(loadResp.modelName == "dispatcher-model");
CHECK(provider->isModelLoaded("dispatcher-model"));
CHECK(provider->lastOptionsModel() == "dispatcher-model");
CHECK(provider->lastOptionsJson() == req.optionsJson);
}
SECTION("load model failure marks embedding degraded") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/failing-model.onnx");
provider->setLoadError(Error{ErrorCode::InvalidState, "provider refused load"});
svc.__test_setModelProvider(provider);
lifecycle.resetDegradedState();
LoadModelRequest req;
req.modelName = "failing-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "provider refused load");
CHECK(lifecycle.setSubsystemDegradedCalls() == 1);
CHECK(lifecycle.lastSubsystem() == "embedding");
CHECK(lifecycle.lastDegraded());
CHECK(lifecycle.lastReason() == "provider_load_failed");
}
SECTION("load model sanitizes invalid utf8 in provider errors") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/utf8-model.onnx");
std::string invalidMessage = "prefix ";
invalidMessage.push_back(static_cast<char>(0xC2));
invalidMessage.push_back(static_cast<char>(0xA9));
invalidMessage.push_back(static_cast<char>(0xE2));
invalidMessage.push_back(static_cast<char>(0x82));
invalidMessage.push_back(static_cast<char>(0xAC));
invalidMessage.push_back(static_cast<char>(0xF0));
invalidMessage.push_back(static_cast<char>(0x9F));
invalidMessage.push_back(static_cast<char>(0x92));
invalidMessage.push_back(static_cast<char>(0xA9));
invalidMessage.push_back(static_cast<char>(0xC0));
invalidMessage.push_back(static_cast<char>(0xFF));
provider->setLoadError(Error{ErrorCode::InvalidState, invalidMessage});
svc.__test_setModelProvider(provider);
LoadModelRequest req;
req.modelName = "utf8-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
std::string expected = "prefix ";
expected.push_back(static_cast<char>(0xC2));
expected.push_back(static_cast<char>(0xA9));
expected.push_back(static_cast<char>(0xE2));
expected.push_back(static_cast<char>(0x82));
expected.push_back(static_cast<char>(0xAC));
expected.push_back(static_cast<char>(0xF0));
expected.push_back(static_cast<char>(0x9F));
expected.push_back(static_cast<char>(0x92));
expected.push_back(static_cast<char>(0xA9));
expected += "\xEF\xBF\xBD\xEF\xBF\xBD";
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == expected);
}
SECTION("load model returns internal error when provider throws after load") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/throwing-model.onnx");
provider->setThrowOnMemoryUsage(true);
svc.__test_setModelProvider(provider);
LoadModelRequest req;
req.modelName = "throwing-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Load model failed: memory boom") != std::string::npos);
}
SECTION("load model skips rebuild when model is already loaded") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/already-loaded-model.onnx");
REQUIRE(provider->loadModel("already-loaded-model").has_value());
svc.__test_setModelProvider(provider);
LoadModelRequest req;
req.modelName = "already-loaded-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
const auto& loadResp = std::get<ModelLoadResponse>(resp);
CHECK(loadResp.success);
CHECK(loadResp.modelName == "already-loaded-model");
CHECK(provider->getLoadedModelCount() == 1);
}
SECTION("load model honors timeout env floor") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/timeout-model.onnx");
svc.__test_setModelProvider(provider);
ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "5");
struct ShutdownBeforeEnvRestore {
ServiceManager& svc;
~ShutdownBeforeEnvRestore() { svc.shutdown(); }
} shutdownBeforeEnvRestore{svc};
LoadModelRequest req;
req.modelName = "timeout-clamped-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
CHECK(std::get<ModelLoadResponse>(resp).success);
CHECK(provider->isModelLoaded("timeout-clamped-model"));
}
SECTION("load model survives invalid timeout env value") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/timeout-invalid.onnx");
svc.__test_setModelProvider(provider);
ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "not-a-number");
struct ShutdownBeforeEnvRestore {
ServiceManager& svc;
~ShutdownBeforeEnvRestore() { svc.shutdown(); }
} shutdownBeforeEnvRestore{svc};
LoadModelRequest req;
req.modelName = "timeout-invalid-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
CHECK(std::get<ModelLoadResponse>(resp).success);
CHECK(provider->isModelLoaded("timeout-invalid-model"));
}
SECTION("load model rebuilds when cached search engine reports missing embedding generator") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/rebuild-health.onnx");
REQUIRE(provider->loadModel("already-loaded-health-model").has_value());
svc.__test_setModelProvider(provider);
metadata::ConnectionPoolConfig poolCfg{};
poolCfg.enableWAL = false;
poolCfg.enableForeignKeys = false;
auto pool = std::make_shared<metadata::ConnectionPool>(":memory:", poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
search::SearchEngineConfig config;
config.vectorWeight = 1.0f;
auto unhealthyEngine =
std::make_shared<search::SearchEngine>(repo, nullptr, nullptr, nullptr, config);
svc.__test_setCachedSearchEngine(unhealthyEngine, true);
LoadModelRequest req;
req.modelName = "already-loaded-health-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
CHECK(std::get<ModelLoadResponse>(resp).success);
CHECK(provider->getLoadedModelCount() == 1);
}
SECTION("load model ignores isModelLoaded exceptions before loading") {
auto provider = std::make_shared<StubModelProvider>(4, "/tmp/isloaded-throw.onnx");
svc.__test_setModelProvider(provider);
metadata::ConnectionPoolConfig poolCfg{};
poolCfg.enableWAL = false;
poolCfg.enableForeignKeys = false;
auto pool = std::make_shared<metadata::ConnectionPool>(":memory:", poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
yams::vector::VectorDatabaseConfig vectorCfg{};
vectorCfg.database_path = ":memory:";
vectorCfg.embedding_dim = 4;
vectorCfg.create_if_missing = true;
vectorCfg.use_in_memory = true;
auto vectorDb = std::make_shared<yams::vector::VectorDatabase>(vectorCfg);
REQUIRE(vectorDb->initialize());
search::SearchEngineConfig config;
config.vectorWeight = 1.0f;
auto healthyEngine = std::make_shared<search::SearchEngine>(
repo, vectorDb, provider->getEmbeddingGenerator(), nullptr, config);
svc.__test_setCachedSearchEngine(healthyEngine, true);
provider->setThrowOnIsModelLoadedCount(1);
LoadModelRequest req;
req.modelName = "isloaded-throw-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
const auto& loadResp = std::get<ModelLoadResponse>(resp);
CHECK(loadResp.success);
CHECK(loadResp.modelName == "isloaded-throw-model");
provider->setThrowOnIsModelLoaded(false);
CHECK(provider->isModelLoaded("isloaded-throw-model"));
}
SECTION("load model survives lifecycle degradation callback exceptions") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/lifecycle-throw.onnx");
svc.__test_setModelProvider(provider);
lifecycle.setThrowOnSetSubsystemDegraded(true);
LoadModelRequest req;
req.modelName = "lifecycle-throw-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelLoadResponse>(resp));
const auto& loadResp = std::get<ModelLoadResponse>(resp);
CHECK(loadResp.success);
CHECK(loadResp.modelName == "lifecycle-throw-model");
CHECK(provider->isModelLoaded("lifecycle-throw-model"));
lifecycle.setThrowOnSetSubsystemDegraded(false);
}
SECTION("unload model rejects missing name") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
svc.__test_setModelProvider(provider);
UnloadModelRequest req;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidData);
CHECK(err.message == "modelName is required");
}
SECTION("unload model propagates provider missing-model error") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
svc.__test_setModelProvider(provider);
UnloadModelRequest req;
req.modelName = "missing-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
}
SECTION("unload model reports provider not ready when no provider is available") {
auto resp = dispatchRequest(dispatcher, Request{UnloadModelRequest{"missing-provider"}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message.find("Plugin system") != std::string::npos);
}
SECTION("model status returns loaded model details and filters by name") {
auto provider = std::make_shared<StubModelProvider>(512, "/tmp/status-model.onnx");
svc.__test_setModelProvider(provider);
REQUIRE(provider->loadModel("status-model").has_value());
ModelStatusRequest req;
req.modelName = "status-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ModelStatusResponse>(resp));
const auto& statusResp = std::get<ModelStatusResponse>(resp);
REQUIRE(statusResp.models.size() == 1);
CHECK(statusResp.models.front().name == "status-model");
CHECK(statusResp.models.front().loaded);
CHECK(statusResp.models.front().embeddingDim == 512);
}
SECTION("model status handles model info lookup failure") {
auto provider = std::make_shared<StubModelProvider>(256, "/tmp/status-fallback.onnx");
REQUIRE(provider->loadModel("status-fallback").has_value());
provider->setFailModelInfo(true);
svc.__test_setModelProvider(provider);
auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}});
REQUIRE(std::holds_alternative<ModelStatusResponse>(resp));
const auto& statusResp = std::get<ModelStatusResponse>(resp);
REQUIRE(statusResp.models.size() == 1);
CHECK(statusResp.models.front().name == "status-fallback");
CHECK(statusResp.models.front().memoryMb == 0);
CHECK(statusResp.models.front().maxSequenceLength == 0);
CHECK(statusResp.models.front().embeddingDim == 256);
}
SECTION("model status returns empty response when provider unavailable") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
provider->setAvailable(false);
svc.__test_setModelProvider(provider);
auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}});
REQUIRE(std::holds_alternative<ModelStatusResponse>(resp));
const auto& statusResp = std::get<ModelStatusResponse>(resp);
CHECK(statusResp.models.empty());
CHECK(statusResp.totalMemoryMb == 0);
}
SECTION("model status converts provider exceptions into internal errors") {
auto provider = std::make_shared<StubModelProvider>(384, "/tmp/model.onnx");
REQUIRE(provider->loadModel("status-throw").has_value());
provider->setThrowOnMemoryUsage(true);
svc.__test_setModelProvider(provider);
auto resp = dispatchRequest(dispatcher, Request{ModelStatusRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("memory boom") != std::string::npos);
}
}
TEST_CASE("RequestDispatcher: document handlers cover direct helper and error branches",
"[daemon][documents][dispatcher]") {
auto makeDoc = [](int64_t id, std::string filePath, std::string hash) {
metadata::DocumentInfo doc{};
doc.id = id;
doc.filePath = std::move(filePath);
doc.fileName = std::filesystem::path(doc.filePath).filename().string();
doc.fileExtension = std::filesystem::path(doc.filePath).extension().string();
doc.fileSize = 12;
doc.sha256Hash = std::move(hash);
doc.mimeType = "text/plain";
doc.setIndexedTime(1710000000 + id);
return doc;
};
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_documents_dispatcher_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
SECTION("prepareSession returns missing-session error for unknown session") {
ScopedEnvVar stateHome("XDG_STATE_HOME",
makeTempDir("yams_prepare_missing_session_").string().c_str());
RequestDispatcher::PrepareSessionOptions opts;
opts.sessionName = "missing";
CHECK(dispatcher.prepareSession(opts) == -2);
}
SECTION("prepareSession warms current session documents") {
auto stateRoot = makeTempDir("yams_prepare_session_");
ScopedEnvVar stateHome("XDG_STATE_HOME", stateRoot.string().c_str());
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(7, "src/demo.cpp", "warm-hash");
repo->setPathTreeDocuments({doc});
store->setBlob("warm-hash", "hello snippet world");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
auto sessionSvc = app::services::makeSessionService(nullptr);
REQUIRE(sessionSvc != nullptr);
sessionSvc->init("warm-session", "coverage");
sessionSvc->addPathSelector("src/**/*.cpp", {}, {});
RequestDispatcher::PrepareSessionOptions opts;
opts.sessionName = "warm-session";
opts.limit = 5;
opts.snippetLen = 5;
CHECK(dispatcher.prepareSession(opts) == 1);
auto materialized = sessionSvc->listMaterialized("warm-session");
REQUIRE(materialized.size() == 1);
CHECK(materialized.front().hash == "warm-hash");
}
SECTION("cancel request succeeds when context is registered") {
auto ctx = std::make_shared<RequestContext>();
RequestContextRegistry::instance().register_context(4242, ctx);
auto resp = dispatchRequest(dispatcher, Request{CancelRequest{4242}});
RequestContextRegistry::instance().deregister_context(4242);
REQUIRE(std::holds_alternative<SuccessResponse>(resp));
CHECK(std::get<SuccessResponse>(resp).message == "Cancel accepted");
CHECK(ctx->canceled.load(std::memory_order_relaxed));
}
SECTION("get init rejects missing hash and name") {
auto resp = dispatchRequest(dispatcher, Request{GetInitRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message.find("hash or name required") != std::string::npos);
}
SECTION("get init reports missing content store") {
GetInitRequest req;
req.hash = "missing-store-hash";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("content store unavailable") != std::string::npos);
}
SECTION("get init reports name resolution failure") {
GetInitRequest req;
req.name = "missing-name";
req.byName = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("Metadata repository not available") != std::string::npos);
}
SECTION("get init reports retrieve bytes failure") {
auto store = std::make_shared<StubContentStore>();
svc.__test_setContentStore(store);
GetInitRequest req;
req.hash = "missing-bytes";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("retrieveBytes failed") != std::string::npos);
}
SECTION("get init reports missing retrieval session manager") {
auto store = std::make_shared<StubContentStore>();
store->setBlob("needs-session-hash", "hello retrieval session");
svc.__test_setContentStore(store);
GetInitRequest req;
req.hash = "needs-session-hash";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("retrieval session manager unavailable") != std::string::npos);
}
SECTION("get init chunk and end succeed across retrieval session lifecycle") {
auto store = std::make_shared<StubContentStore>();
store->setBlob("session-flow-hash", "hello retrieval flow");
svc.__test_setContentStore(store);
svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());
GetInitRequest initReq;
initReq.hash = "session-flow-hash";
initReq.chunkSize = 5;
auto initResp = dispatchRequest(dispatcher, Request{initReq});
REQUIRE(std::holds_alternative<GetInitResponse>(initResp));
const auto& init = std::get<GetInitResponse>(initResp);
CHECK(init.transferId > 0);
CHECK(init.totalSize == 20);
CHECK(init.chunkSize == 5);
CHECK(init.metadata.at("hash") == "session-flow-hash");
auto chunkResp =
dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 0, 5}});
REQUIRE(std::holds_alternative<GetChunkResponse>(chunkResp));
const auto& chunk = std::get<GetChunkResponse>(chunkResp);
CHECK(chunk.data == "hello");
CHECK(chunk.bytesRemaining == 15);
auto endResp = dispatchRequest(dispatcher, Request{GetEndRequest{init.transferId}});
REQUIRE(std::holds_alternative<SuccessResponse>(endResp));
CHECK(std::get<SuccessResponse>(endResp).message == "OK");
auto postEndChunk =
dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 0, 5}});
REQUIRE(std::holds_alternative<ErrorResponse>(postEndChunk));
const auto& postEndErr = std::get<ErrorResponse>(postEndChunk);
CHECK(postEndErr.code == ErrorCode::NotFound);
CHECK(postEndErr.message.find("Invalid transferId") != std::string::npos);
}
SECTION("get chunk rejects offsets beyond total size") {
auto store = std::make_shared<StubContentStore>();
store->setBlob("offset-error-hash", "hello retrieval flow");
svc.__test_setContentStore(store);
svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());
GetInitRequest initReq;
initReq.hash = "offset-error-hash";
auto initResp = dispatchRequest(dispatcher, Request{initReq});
REQUIRE(std::holds_alternative<GetInitResponse>(initResp));
const auto& init = std::get<GetInitResponse>(initResp);
auto chunkResp =
dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 99, 5}});
REQUIRE(std::holds_alternative<ErrorResponse>(chunkResp));
const auto& err = std::get<ErrorResponse>(chunkResp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message.find("Offset beyond total size") != std::string::npos);
}
SECTION("get chunk respects max bytes cap and can return empty chunk") {
auto store = std::make_shared<StubContentStore>();
store->setBlob("cap-flow-hash", "hello retrieval flow");
svc.__test_setContentStore(store);
svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());
GetInitRequest initReq;
initReq.hash = "cap-flow-hash";
initReq.chunkSize = 10;
initReq.maxBytes = 7;
auto initResp = dispatchRequest(dispatcher, Request{initReq});
REQUIRE(std::holds_alternative<GetInitResponse>(initResp));
const auto& init = std::get<GetInitResponse>(initResp);
auto chunkResp =
dispatchRequest(dispatcher, Request{GetChunkRequest{init.transferId, 7, 5}});
REQUIRE(std::holds_alternative<GetChunkResponse>(chunkResp));
const auto& chunk = std::get<GetChunkResponse>(chunkResp);
CHECK(chunk.data.empty());
CHECK(chunk.bytesRemaining == 0);
}
SECTION("get chunk reports missing retrieval session manager") {
auto resp = dispatchRequest(dispatcher, Request{GetChunkRequest{9999, 0, 4}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("retrieval session manager unavailable") != std::string::npos);
}
SECTION("get chunk reports invalid transfer id") {
svc.__test_setRetrievalSessionManager(std::make_unique<RetrievalSessionManager>());
auto resp = dispatchRequest(dispatcher, Request{GetChunkRequest{9999, 0, 4}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message.find("Invalid transferId") != std::string::npos);
}
SECTION("get end succeeds even without retrieval session manager") {
auto resp = dispatchRequest(dispatcher, Request{GetEndRequest{777}});
REQUIRE(std::holds_alternative<SuccessResponse>(resp));
CHECK(std::get<SuccessResponse>(resp).message == "OK");
}
SECTION("delete rejects directory removal without recursive flag") {
DeleteRequest req;
req.directory = "cache";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message.find("recursive flag") != std::string::npos);
}
SECTION("delete recursive directory returns not found when no documents match") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto store = std::make_shared<StubContentStore>();
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
DeleteRequest req;
req.directory = "cache";
req.recursive = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message.find("No documents found matching criteria") != std::string::npos);
}
SECTION("delete reports mixed success and failure results") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto store = std::make_shared<StubContentStore>();
auto successDoc = makeDoc(21, "/tmp/delete/ok.txt", "delete-ok-hash");
auto failedDoc = makeDoc(22, "/tmp/delete/fail.txt", "delete-fail-hash");
repo->addDocument(successDoc);
repo->addDocument(failedDoc);
store->setBlob(successDoc.sha256Hash, "ok");
store->setBlob(failedDoc.sha256Hash, "fail");
store->setRemoveResult(failedDoc.sha256Hash,
Error{ErrorCode::IOError, "store remove failed"});
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
DeleteRequest req;
req.pattern = "*.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<DeleteResponse>(resp));
const auto& deleteResp = std::get<DeleteResponse>(resp);
CHECK_FALSE(deleteResp.dryRun);
CHECK(deleteResp.successCount == 1);
CHECK(deleteResp.failureCount == 1);
REQUIRE(deleteResp.results.size() == 2);
CHECK(lifecycle.removedHashes().size() == 1);
CHECK(lifecycle.removedHashes().front() == successDoc.sha256Hash);
}
SECTION("delete reports missing content store") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
svc.__test_setMetadataRepo(repo);
DeleteRequest req;
req.hash = "no-store-hash";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("Content store not available") != std::string::npos);
}
SECTION("get request maps related documents when graph view is enabled") {
auto repoPath = makeTempDir("yams_get_graph_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto graphQuery = std::make_shared<StubGraphQueryService>();
auto mainDoc = makeDoc(69, "/tmp/get-graph/main.txt",
"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff");
auto relatedDoc =
makeDoc(70, "/tmp/get-graph/related.txt",
"1111111111111111111111111111111111111111111111111111111111111111");
auto mainId = repo->insertDocument(mainDoc);
auto relatedId = repo->insertDocument(relatedDoc);
REQUIRE(mainId.has_value());
REQUIRE(relatedId.has_value());
REQUIRE(repo->upsertPathTreeForDocument(mainDoc, mainId.value(), true, {}).has_value());
REQUIRE(
repo->upsertPathTreeForDocument(relatedDoc, relatedId.value(), true, {}).has_value());
graphQuery->addConnectedDocument(relatedDoc.sha256Hash, "same_content", 1, 0.8);
store->setBlob(mainDoc.sha256Hash, "main content");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
svc.__test_setGraphQueryService(graphQuery);
GetRequest req;
req.hash = mainDoc.sha256Hash;
req.showGraph = true;
req.graphDepth = 1;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<GetResponse>(resp));
const auto& getResp = std::get<GetResponse>(resp);
CHECK(getResp.graphEnabled);
CHECK(getResp.hash == mainDoc.sha256Hash);
REQUIRE(getResp.related.size() == 1);
CHECK(getResp.related.front().hash == relatedDoc.sha256Hash);
CHECK(getResp.related.front().relationship == "same_content");
}
SECTION("get request still succeeds without query trace env") {
auto repoPath = makeTempDir("yams_get_query_trace_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(71, "/tmp/get/query-trace.txt",
"9999999999999999999999999999999999999999999999999999999999999999");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "query trace content");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
GetRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<GetResponse>(resp));
const auto& getResp = std::get<GetResponse>(resp);
CHECK(getResp.hash == doc.sha256Hash);
CHECK(getResp.name == doc.fileName);
}
SECTION("get request maps fallback documents vector response") {
auto repoPath = makeTempDir("yams_get_documents_vector_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(72, "/tmp/get/documents-vector.txt",
"abababababababababababababababababababababababababababababababab");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "documents vector content");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher::__test_forceGetResponseDocumentsVectorOnce();
GetRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<GetResponse>(resp));
const auto& getResp = std::get<GetResponse>(resp);
CHECK(getResp.hash == doc.sha256Hash);
CHECK(getResp.name == doc.fileName);
CHECK(getResp.path == doc.filePath);
}
SECTION("get request returns not found when service result is forced empty") {
auto repoPath = makeTempDir("yams_get_empty_result_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(73, "/tmp/get/empty-result.txt",
"cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "empty result content");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher::__test_forceGetEmptyResultOnce();
GetRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "No documents found matching criteria");
}
SECTION("file history reports missing metadata repository") {
FileHistoryRequest req;
req.filepath = "missing.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("Metadata repository not available") != std::string::npos);
}
SECTION("file history falls back to filename query and reports no snapshot versions") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto doc = makeDoc(9, "/tmp/archive/report.txt", "history-hash");
repo->addDocument(doc);
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = "report.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
CHECK_FALSE(history.found);
CHECK(history.totalVersions == 0);
CHECK(history.versions.empty());
CHECK(history.message == "File found in index but not in any snapshot");
CHECK(history.filepath.find("report.txt") != std::string::npos);
}
SECTION("file history reports file not found in index") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = "ghost.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
CHECK_FALSE(history.found);
CHECK(history.totalVersions == 0);
CHECK(history.message == "File not found in index");
}
SECTION("file history returns sorted snapshot versions from exact path") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto doc = makeDoc(11, "/tmp/archive/sorted-report.txt", "sorted-history-hash");
repo->setExactPathDocument(doc);
repo->setAllMetadata(
doc.id, {{"snapshot_id", metadata::MetadataValue("snap-alpha")},
{"snapshot_id:snap-beta", metadata::MetadataValue("")},
{"snapshot_time:snap-alpha", metadata::MetadataValue("1710000000000000")}});
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
REQUIRE(history.found);
REQUIRE(history.totalVersions == 2);
REQUIRE(history.versions.size() == 2);
CHECK(history.message == "Found 2 version(s) across snapshots");
CHECK(history.versions[0].snapshotId == "snap-beta");
CHECK(history.versions[1].snapshotId == "snap-alpha");
CHECK(history.versions[0].indexedTimestamp > history.versions[1].indexedTimestamp);
}
SECTION(
"file history falls back to document indexed time when snapshot-specific time is missing") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto doc = makeDoc(42, "/tmp/archive/fallback-time.txt", "fallback-time-hash");
repo->setExactPathDocument(doc);
repo->setAllMetadata(doc.id, {{"snapshot_id", metadata::MetadataValue("snap-fallback")}});
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
REQUIRE(history.found);
REQUIRE(history.totalVersions == 1);
REQUIRE(history.versions.size() == 1);
CHECK(history.versions.front().snapshotId == "snap-fallback");
CHECK(history.versions.front().indexedTimestamp == 1710000042);
}
SECTION("file history ignores malformed snapshot timestamps and falls back to indexed time") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto doc = makeDoc(64, "/tmp/archive/bad-snapshot-time.txt", "bad-snapshot-time-hash");
repo->setExactPathDocument(doc);
repo->setAllMetadata(doc.id, {{"snapshot_id", metadata::MetadataValue("snap-bad-time")},
{"snapshot_time", metadata::MetadataValue("not-a-number")},
{"snapshot_time:snap-bad-time",
metadata::MetadataValue("still-not-a-number")}});
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
REQUIRE(history.found);
REQUIRE(history.totalVersions == 1);
REQUIRE(history.versions.size() == 1);
CHECK(history.versions.front().snapshotId == "snap-bad-time");
CHECK(history.versions.front().indexedTimestamp == 1710000064);
}
SECTION("list request paths-only mode trims snippets metadata and tags") {
auto repoPath = makeTempDir("yams_list_dispatcher_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto doc = makeDoc(55, "/tmp/list/paths-only.md", "list-paths-hash");
REQUIRE(repo->insertDocument(doc).has_value());
svc.__test_setMetadataRepo(repo);
ListRequest req;
req.limit = 5;
req.pathsOnly = true;
req.showMetadata = true;
req.showTags = true;
req.showSnippets = true;
req.recent = false;
req.recentCount = 0;
req.namePattern = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ListResponse>(resp));
const auto& listResp = std::get<ListResponse>(resp);
REQUIRE(listResp.items.size() == 1);
CHECK(listResp.totalCount == 1);
CHECK(listResp.items.front().path == doc.filePath);
CHECK(listResp.items.front().snippet.empty());
CHECK(listResp.items.front().metadata.empty());
CHECK(listResp.items.front().tags.empty());
}
SECTION("list request emits dispatch timing stats by default") {
auto repoPath = makeTempDir("yams_list_query_trace_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto doc = makeDoc(56, "/tmp/list/query-trace.md", "list-query-trace-hash");
REQUIRE(repo->insertDocument(doc).has_value());
svc.__test_setMetadataRepo(repo);
ListRequest req;
req.limit = 3;
req.recent = false;
req.recentCount = 0;
req.namePattern = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ListResponse>(resp));
const auto& listResp = std::get<ListResponse>(resp);
REQUIRE(listResp.items.size() == 1);
CHECK(listResp.listStats.contains("phase_dispatch_total_ms"));
CHECK(listResp.listStats.contains("phase_dispatch_map_us"));
}
SECTION("list request parses filter tags and recent count") {
auto repoPath = makeTempDir("yams_list_filter_tags_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto doc = makeDoc(67, "/tmp/list/filter-tags.md", "list-filter-tags-hash");
REQUIRE(repo->insertDocument(doc).has_value());
svc.__test_setMetadataRepo(repo);
ListRequest req;
req.limit = 5;
req.offset = 0;
req.recent = false;
req.recentCount = 3;
req.filterTags = " alpha, beta ";
req.namePattern = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ListResponse>(resp));
const auto& listResp = std::get<ListResponse>(resp);
CHECK(listResp.totalCount >= 0);
}
SECTION("list request works when inflight limit is disabled") {
TuneAdvisorListGuard tuneGuard(0, 1);
auto repoPath = makeTempDir("yams_list_no_inflight_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto doc = makeDoc(68, "/tmp/list/no-inflight.md", "list-no-inflight-hash");
REQUIRE(repo->insertDocument(doc).has_value());
svc.__test_setMetadataRepo(repo);
ListRequest req;
req.limit = 5;
req.offset = 0;
req.recent = true;
req.showSnippets = true;
req.noSnippets = true;
req.namePattern = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ListResponse>(resp));
const auto& listResp = std::get<ListResponse>(resp);
REQUIRE(listResp.items.size() == 1);
CHECK(listResp.items.front().snippet.empty());
}
SECTION("list request reports missing metadata repository") {
ListRequest req;
req.limit = 2;
req.recent = false;
req.recentCount = 0;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message.find("Metadata repository not available") != std::string::npos);
}
SECTION("list request reports resource exhaustion when inflight limit is saturated") {
TuneAdvisorListGuard tuneGuard(1, 1);
auto repoPath = makeTempDir("yams_list_exhaustion_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto doc = makeDoc(60, "/tmp/list/exhausted.md", "list-exhausted-hash");
REQUIRE(repo->insertDocument(doc).has_value());
svc.__test_setMetadataRepo(repo);
RequestDispatcher::__test_setListInflightRequests(1);
auto resetGuard = ScopeExit([] { RequestDispatcher::__test_setListInflightRequests(0); });
ListRequest req;
req.limit = 1;
req.recent = false;
req.recentCount = 0;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::ResourceExhausted);
CHECK(err.message.find("List concurrency limit reached") != std::string::npos);
}
SECTION("list request catches std exceptions") {
RequestDispatcher::__test_forceListExceptionOnce("forced list exception");
ListRequest req;
req.limit = 2;
req.recent = false;
req.recentCount = 0;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "List failed: forced list exception");
}
SECTION("list request catches unknown exceptions") {
RequestDispatcher::__test_forceListUnknownExceptionOnce();
ListRequest req;
req.limit = 2;
req.recent = false;
req.recentCount = 0;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "List failed: unknown error");
}
SECTION("add document queues deferred work when system is under pressure") {
const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setEnableResourceGovernor(originalGovernor);
TuneAdvisor::setEnableAdmissionControl(originalAdmission);
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
drainStoreDocumentTasks();
});
TuneAdvisor::setEnableResourceGovernor(true);
TuneAdvisor::setEnableAdmissionControl(true);
TuneAdvisor::setStoreDocumentChannelCapacity(64);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
drainStoreDocumentTasks();
ScopedEnvVar retriesGuard("YAMS_INGEST_ENQUEUE_RETRIES", "not-a-number");
ScopedEnvVar baseDelayGuard("YAMS_INGEST_ENQUEUE_BASE_DELAY_MS", "-5");
ScopedEnvVar maxDelayGuard("YAMS_INGEST_ENQUEUE_MAX_DELAY_MS", "999999");
AddDocumentRequest req;
req.name = "pressure.txt";
req.content = "queued under pressure";
auto resp = dispatchRequest(dispatcher, Request{req});
INFO([&] {
if (std::holds_alternative<ErrorResponse>(resp)) {
return std::get<ErrorResponse>(resp).message;
}
return std::string("not an error");
}());
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
auto hasher = yams::crypto::createSHA256Hasher();
REQUIRE(hasher != nullptr);
CHECK(addResp.path == "pressure.txt");
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash == hasher->hash(req.content));
CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask task;
REQUIRE(channel->try_pop(task));
CHECK(task.request.name == req.name);
CHECK(task.request.content == req.content);
}
SECTION("add document clears deferred hash when pressure-path hashing fails") {
const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setEnableResourceGovernor(originalGovernor);
TuneAdvisor::setEnableAdmissionControl(originalAdmission);
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
drainStoreDocumentTasks();
});
TuneAdvisor::setEnableResourceGovernor(true);
TuneAdvisor::setEnableAdmissionControl(true);
TuneAdvisor::setStoreDocumentChannelCapacity(64);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
drainStoreDocumentTasks();
auto filePath = writeTempFile("yams_documents_pressure_file_", "pressure-file.txt",
"pressure file content");
RequestDispatcher::__test_forceDocumentsHashFailureOnce();
AddDocumentRequest req;
req.path = filePath.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash.empty());
CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
}
SECTION("add document clears deferred hash for recursive pressure path") {
const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setEnableResourceGovernor(originalGovernor);
TuneAdvisor::setEnableAdmissionControl(originalAdmission);
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
drainStoreDocumentTasks();
});
TuneAdvisor::setEnableResourceGovernor(true);
TuneAdvisor::setEnableAdmissionControl(true);
TuneAdvisor::setStoreDocumentChannelCapacity(64);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
drainStoreDocumentTasks();
AddDocumentRequest req;
req.path = makeTempDir("yams_documents_pressure_recursive_").string();
req.recursive = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash.empty());
CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
}
SECTION("add document reports resource exhaustion when pressure queue is full") {
const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setEnableResourceGovernor(originalGovernor);
TuneAdvisor::setEnableAdmissionControl(originalAdmission);
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
drainStoreDocumentTasks();
});
TuneAdvisor::setEnableResourceGovernor(true);
TuneAdvisor::setEnableAdmissionControl(true);
TuneAdvisor::setStoreDocumentChannelCapacity(64);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
drainStoreDocumentTasks();
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask filler;
while (channel->try_push(filler)) {
}
AddDocumentRequest req;
req.name = "queue-full.txt";
req.content = "full";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::ResourceExhausted);
CHECK((err.message == "Ingestion queue is full. Please try again later." ||
err.message == "Ingestion queue is full; retry shortly"));
}
SECTION("add document queues async work while daemon is initializing") {
DaemonConfig notReadyCfg;
notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_");
StubLifecycle notReadyLifecycle(LifecycleState::Starting);
StateComponent notReadyState;
notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm notReadyLifecycleFsm;
ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
AddDocumentRequest req;
req.name = "initializing.txt";
req.content = "booting daemon";
auto resp = dispatchRequest(notReadyDispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
auto hasher = yams::crypto::createSHA256Hasher();
REQUIRE(hasher != nullptr);
CHECK(addResp.path == "initializing.txt");
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash == hasher->hash(req.content));
CHECK(addResp.message ==
"Ingestion accepted for asynchronous processing (daemon initializing).");
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask task;
REQUIRE(channel->try_pop(task));
CHECK(task.request.name == req.name);
CHECK(task.request.content == req.content);
}
SECTION("add document reports queue full while daemon is initializing") {
DaemonConfig notReadyCfg;
notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_full_");
StubLifecycle notReadyLifecycle(LifecycleState::Starting);
StateComponent notReadyState;
notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm notReadyLifecycleFsm;
ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
drainStoreDocumentTasks();
});
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
TuneAdvisor::setStoreDocumentChannelCapacity(64);
drainStoreDocumentTasks();
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask filler;
while (channel->try_push(filler)) {
}
AddDocumentRequest req;
req.name = "initializing-full.txt";
req.content = "booting daemon full";
auto resp = dispatchRequest(notReadyDispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::ResourceExhausted);
CHECK((err.message == "Ingestion queue is full. Please try again later." ||
err.message == "Ingestion queue is full; retry shortly"));
}
SECTION("add document clears async hash when daemon-initializing hashing fails") {
DaemonConfig notReadyCfg;
notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_hash_fail_");
StubLifecycle notReadyLifecycle(LifecycleState::Starting);
StateComponent notReadyState;
notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm notReadyLifecycleFsm;
ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
auto filePath =
writeTempFile("yams_documents_init_hash_fail_", "init-file.txt", "init file content");
RequestDispatcher::__test_forceDocumentsHashFailureOnce();
AddDocumentRequest req;
req.path = filePath.string();
auto resp = dispatchRequest(notReadyDispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash.empty());
CHECK(addResp.message ==
"Ingestion accepted for asynchronous processing (daemon initializing).");
}
SECTION("add document hashes file path while daemon is initializing") {
DaemonConfig notReadyCfg;
notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_not_ready_file_hash_");
StubLifecycle notReadyLifecycle(LifecycleState::Starting);
StateComponent notReadyState;
notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm notReadyLifecycleFsm;
ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
auto filePath = writeTempFile("yams_documents_init_file_hash_", "init-file-hash.txt",
"init file hash content");
AddDocumentRequest req;
req.path = filePath.string();
auto resp = dispatchRequest(notReadyDispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
auto hasher = yams::crypto::createSHA256Hasher();
REQUIRE(hasher != nullptr);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash == hasher->hashFile(req.path));
CHECK(addResp.message ==
"Ingestion accepted for asynchronous processing (daemon initializing).");
}
SECTION("add document treats directory path as async directory ingestion") {
auto dirPath = makeTempDir("yams_documents_directory_ingest_");
drainStoreDocumentTasks();
AddDocumentRequest req;
req.path = dirPath.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.path == dirPath.string());
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash.empty());
CHECK(addResp.message == "Directory ingestion accepted for asynchronous processing.");
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask task;
REQUIRE(channel->try_pop(task));
CHECK(task.request.path == req.path);
}
SECTION("add document rejects missing path and content") {
AddDocumentRequest req;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message == "Provide either 'path' or 'content' + 'name'");
}
SECTION("add document uses async single-file path by default when ready") {
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
AddDocumentRequest req;
req.name = "async-default.txt";
req.content = "ready but async";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
auto hasher = yams::crypto::createSHA256Hasher();
REQUIRE(hasher != nullptr);
CHECK(addResp.path == "async-default.txt");
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash == hasher->hash(req.content));
CHECK(addResp.extractionStatus == "pending");
CHECK(addResp.message == "Ingestion accepted for asynchronous processing.");
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask task;
REQUIRE(channel->try_pop(task));
CHECK(task.request.name == req.name);
CHECK(task.request.content == req.content);
}
SECTION("add document hashes file paths on async path when ready") {
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
auto filePath =
writeTempFile("yams_documents_async_file_", "async-file.txt", "async file content");
AddDocumentRequest req;
req.path = filePath.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
auto hasher = yams::crypto::createSHA256Hasher();
REQUIRE(hasher != nullptr);
CHECK(addResp.path == req.path);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash == hasher->hashFile(req.path));
CHECK(addResp.message == "Ingestion accepted for asynchronous processing.");
}
SECTION("add document hashes file path on pressure path after one retry") {
const bool originalGovernor = TuneAdvisor::enableResourceGovernor();
const bool originalAdmission = TuneAdvisor::enableAdmissionControl();
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setEnableResourceGovernor(originalGovernor);
TuneAdvisor::setEnableAdmissionControl(originalAdmission);
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Normal);
RequestDispatcher::__test_setDocumentsEnqueueFailuresBeforeSuccess(0);
drainStoreDocumentTasks();
});
TuneAdvisor::setEnableResourceGovernor(true);
TuneAdvisor::setEnableAdmissionControl(true);
TuneAdvisor::setStoreDocumentChannelCapacity(64);
ResourceGovernor::instance().testing_updateScalingCaps(ResourcePressureLevel::Emergency);
drainStoreDocumentTasks();
ScopedEnvVar retriesGuard("YAMS_INGEST_ENQUEUE_RETRIES", "3");
ScopedEnvVar baseDelayGuard("YAMS_INGEST_ENQUEUE_BASE_DELAY_MS", "1");
ScopedEnvVar maxDelayGuard("YAMS_INGEST_ENQUEUE_MAX_DELAY_MS", "1");
RequestDispatcher::__test_setDocumentsEnqueueFailuresBeforeSuccess(1);
auto filePath = writeTempFile("yams_documents_pressure_retry_", "pressure-retry.txt",
"pressure retry content");
AddDocumentRequest req;
req.path = filePath.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
auto hasher = yams::crypto::createSHA256Hasher();
REQUIRE(hasher != nullptr);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash == hasher->hashFile(req.path));
CHECK(addResp.message == "Queued for deferred processing (system under pressure).");
}
SECTION("add document falls back to initializing path when later lifecycle snapshot throws") {
DaemonConfig notReadyCfg;
notReadyCfg.dataDir = makeTempDir("yams_documents_dispatcher_lifecycle_throw_late_");
StubLifecycle notReadyLifecycle(LifecycleState::Ready);
notReadyLifecycle.setThrowOnGetLifecycleSnapshotCall(2);
StateComponent notReadyState;
notReadyState.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
notReadyState.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm notReadyLifecycleFsm;
ServiceManager notReadySvc(notReadyCfg, notReadyState, notReadyLifecycleFsm);
RequestDispatcher notReadyDispatcher(&notReadyLifecycle, &notReadySvc, &notReadyState);
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
AddDocumentRequest req;
req.name = "late-throw.txt";
req.content = "late throw";
auto resp = dispatchRequest(notReadyDispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.message ==
"Ingestion accepted for asynchronous processing (daemon initializing).");
}
SECTION("add document reports queue full on async single-file path") {
const uint32_t originalCapacity = TuneAdvisor::storeDocumentChannelCapacity();
auto restore = ScopeExit([&] {
TuneAdvisor::setStoreDocumentChannelCapacity(originalCapacity);
drainStoreDocumentTasks();
});
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
TuneAdvisor::setStoreDocumentChannelCapacity(64);
drainStoreDocumentTasks();
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask filler;
while (channel->try_push(filler)) {
}
AddDocumentRequest req;
req.name = "async-full.txt";
req.content = "full async queue";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::ResourceExhausted);
CHECK((err.message == "Ingestion queue is full. Please try again later." ||
err.message == "Ingestion queue is full; retry shortly"));
}
SECTION("add document clears async hash when hashing fails on ready path") {
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "0");
drainStoreDocumentTasks();
auto filePath = writeTempFile("yams_documents_async_hash_fail_", "async-hash-fail.txt",
"async hash fail content");
RequestDispatcher::__test_forceDocumentsHashFailureOnce();
AddDocumentRequest req;
req.path = filePath.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.documentsAdded == 0);
CHECK(addResp.hash.empty());
CHECK(addResp.extractionStatus == "pending");
CHECK(addResp.message == "Ingestion accepted for asynchronous processing.");
}
SECTION("add document uses sync fallback when explicitly enabled") {
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "1");
drainStoreDocumentTasks();
svc.__test_setContentStore(std::make_shared<StubContentStore>());
AddDocumentRequest req;
req.name = "sync-add.txt";
req.content = "sync path content";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.path == "sync-add.txt");
CHECK(addResp.documentsAdded == 1);
CHECK(addResp.size == req.content.size());
CHECK(addResp.extractionStatus == "pending");
CHECK(addResp.message == "Document stored successfully.");
CHECK_FALSE(addResp.hash.empty());
auto channel = getStoreDocumentTaskChannel();
InternalEventBus::StoreDocumentTask task;
CHECK_FALSE(channel->try_pop(task));
}
SECTION("add document sync fallback forwards metadata and path inputs") {
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "true");
drainStoreDocumentTasks();
svc.__test_setContentStore(std::make_shared<StubContentStore>());
auto filePath =
writeTempFile("yams_documents_sync_file_", "sync-file.txt", "sync file content");
AddDocumentRequest req;
req.path = filePath.string();
req.mimeType = "text/custom";
req.disableAutoMime = true;
req.tags = {"alpha", "beta"};
req.metadata["task"] = "documents";
req.collection = "focus";
req.snapshotId = "snap-1";
req.snapshotLabel = "label-1";
req.sessionId = "session-1";
req.noEmbeddings = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<AddDocumentResponse>(resp));
const auto& addResp = std::get<AddDocumentResponse>(resp);
CHECK(addResp.path == req.path);
CHECK(addResp.documentsAdded == 1);
CHECK(addResp.size == std::filesystem::file_size(filePath));
CHECK_FALSE(addResp.hash.empty());
}
SECTION("add document sync fallback returns store errors") {
ScopedEnvVar syncGuard("YAMS_SYNC_SINGLE_FILE_ADD", "1");
drainStoreDocumentTasks();
svc.__test_setContentStore(std::shared_ptr<api::IContentStore>{});
AddDocumentRequest req;
req.name = "sync-error.txt";
req.content = "sync failure";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message == "Content store not available");
}
SECTION("file history skips documents when metadata lookup fails") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto failingDoc = makeDoc(57, "/tmp/archive/history-mixed.txt", "metadata-fail-hash");
auto okDoc = makeDoc(58, "/var/tmp/archive/history-mixed.txt", "metadata-ok-hash");
repo->addDocument(failingDoc);
repo->addDocument(okDoc);
repo->setAllMetadataError(failingDoc.id,
Error{ErrorCode::DatabaseError, "metadata lookup failed"});
repo->setAllMetadata(okDoc.id,
{{"snapshot_id", metadata::MetadataValue("snap-ok")},
{"snapshot_time", metadata::MetadataValue("1710000000000000")}});
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = "history-mixed.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
REQUIRE(history.found);
CHECK(history.totalVersions == 1);
REQUIRE(history.versions.size() == 1);
CHECK(history.versions.front().snapshotId == "snap-ok");
}
SECTION("file history falls back when exact path lookup errors") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto doc =
makeDoc(63, "/tmp/archive/fallback-exact-error.txt", "fallback-exact-error-hash");
repo->addDocument(doc);
repo->setExactPathError(Error{ErrorCode::DatabaseError, "exact lookup failed"});
repo->setAllMetadata(doc.id,
{{"snapshot_id", metadata::MetadataValue("snap-fallback-error")}});
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = doc.filePath;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
REQUIRE(history.found);
CHECK(history.totalVersions == 1);
REQUIRE(history.versions.size() == 1);
CHECK(history.versions.front().snapshotId == "snap-fallback-error");
}
SECTION("file history caps oversized filename matches to first hundred documents") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
for (int i = 0; i < 101; ++i) {
auto doc = makeDoc(1000 + i,
"/tmp/archive/subdir_" + std::to_string(i) + "/history-overflow.txt",
"history-overflow-hash-" + std::to_string(i));
doc.fileName = "history-overflow.txt";
repo->addDocument(doc);
repo->setAllMetadata(
doc.id, {{"snapshot_id", metadata::MetadataValue("snap-" + std::to_string(i))}});
}
svc.__test_setMetadataRepo(repo);
FileHistoryRequest req;
req.filepath = "history-overflow.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<FileHistoryResponse>(resp));
const auto& history = std::get<FileHistoryResponse>(resp);
REQUIRE(history.found);
CHECK(history.totalVersions == 100);
CHECK(history.versions.size() == 100);
}
SECTION("file history reports invalid filepath when current directory is missing") {
#ifdef _WIN32
SKIP("Windows keeps the current working directory locked, so missing-cwd simulation is "
"unreliable");
#endif
auto repo = std::make_shared<StubPruneMetadataRepository>();
svc.__test_setMetadataRepo(repo);
auto originalCwd = std::filesystem::current_path();
auto doomedDir = makeTempDir("yams_missing_cwd_");
std::filesystem::current_path(doomedDir);
std::filesystem::remove_all(doomedDir);
FileHistoryRequest req;
req.filepath = "history-from-missing-cwd.txt";
auto resp = dispatchRequest(dispatcher, Request{req});
std::filesystem::current_path(originalCwd);
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message.find("Invalid filepath") != std::string::npos);
}
SECTION("cat request reports missing content") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(58, "/tmp/cat/missing.txt", "cat-missing-hash");
repo->addDocument(doc);
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
CatRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message.find("Document not found") != std::string::npos);
}
SECTION("cat request reports missing content payload for indexed document") {
auto repoPath = makeTempDir("yams_cat_missing_content_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(59, "/tmp/cat/metadata-only.txt",
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa");
REQUIRE(repo->insertDocument(doc).has_value());
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
CatRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message.find("Document content not found") != std::string::npos);
}
SECTION("cat request hits missing document branch after successful retrieval") {
auto repoPath = makeTempDir("yams_cat_force_missing_doc_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(61, "/tmp/cat/force-missing-doc.txt",
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "payload");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher::__test_forceCatMissingDocumentOnce();
CatRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "Document not found");
}
SECTION("cat request hits content unavailable branch after successful retrieval") {
auto repoPath = makeTempDir("yams_cat_force_missing_content_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(62, "/tmp/cat/force-missing-content.txt",
"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "payload");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher::__test_forceCatMissingContentOnce();
CatRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Content unavailable");
}
SECTION("cat request hits native missing document branch") {
auto repoPath = makeTempDir("yams_cat_native_missing_doc_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(65, "/tmp/cat/native-missing-doc.txt",
"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "payload");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher::__test_forceCatNativeMissingDocumentOnce();
CatRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "Document not found");
}
SECTION("cat request hits native missing content branch") {
auto repoPath = makeTempDir("yams_cat_native_missing_content_repo_") / "metadata.db";
metadata::ConnectionPoolConfig poolCfg{};
auto pool = std::make_shared<metadata::ConnectionPool>(repoPath.string(), poolCfg);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<metadata::MetadataRepository>(*pool);
auto store = std::make_shared<StubContentStore>();
auto doc = makeDoc(66, "/tmp/cat/native-missing-content.txt",
"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee");
REQUIRE(repo->insertDocument(doc).has_value());
store->setBlob(doc.sha256Hash, "payload");
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(store);
RequestDispatcher::__test_forceCatNativeMissingContentOnce();
CatRequest req;
req.hash = doc.sha256Hash;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Content unavailable");
}
}
TEST_CASE("RequestDispatcher: download handler enforces daemon policy",
"[daemon][documents][dispatcher][download]") {
ScopedEnvVar daemonDownloadEnv("YAMS_ENABLE_DAEMON_DOWNLOAD", "0");
auto dispatchDownload = [](RequestDispatcher& dispatcher, const std::string& url,
std::string outputPath = {}, std::string checksum = {}) {
DownloadRequest req;
req.url = url;
req.outputPath = std::move(outputPath);
req.checksum = std::move(checksum);
return dispatchRequest(dispatcher, Request{req});
};
auto waitForDownloadJob = [](RequestDispatcher& dispatcher, const std::string& jobId,
std::chrono::milliseconds timeout = std::chrono::seconds(2)) {
const auto deadline = std::chrono::steady_clock::now() + timeout;
DownloadResponse latest;
while (std::chrono::steady_clock::now() < deadline) {
auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{jobId}});
if (std::holds_alternative<DownloadResponse>(resp)) {
latest = std::get<DownloadResponse>(resp);
if (latest.state != "queued" && latest.state != "running") {
return latest;
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
return latest;
};
auto waitForFollowRedirectsCapture = [](std::chrono::milliseconds timeout =
std::chrono::seconds(2)) {
const auto deadline = std::chrono::steady_clock::now() + timeout;
std::optional<bool> latest;
while (std::chrono::steady_clock::now() < deadline) {
latest = RequestDispatcher::__test_lastDownloadServiceFollowRedirects();
if (latest.has_value()) {
return latest;
}
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
return latest;
};
RequestDispatcher::__test_resetDownloadServiceRequestCapture();
SECTION("disabled policy returns safe reminder") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_disabled_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("Daemon download is disabled") != std::string::npos);
}
SECTION("disallowed scheme is rejected before download starts") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_scheme_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "http://example.com/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("scheme 'http'") != std::string::npos);
}
SECTION("invalid URL shape is rejected before policy host checks") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_invalid_url_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "not-a-url");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error == "Invalid download URL");
}
SECTION("download URL rejects missing authority after scheme") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_missing_authority_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error == "Invalid download URL");
}
SECTION("download URL rejects empty host after credentials stripping") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_empty_host_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://user@/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error == "Invalid download URL");
}
SECTION("wildcard host policy accepts matching subdomains before checksum gate") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_wildcard_host_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = true;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*.example.com"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://api.example.com/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("checksum is required") != std::string::npos);
}
SECTION("IPv6 authority parses and matches exact allowed host") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_ipv6_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = true;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"::1"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://[::1]/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("checksum is required") != std::string::npos);
}
SECTION("invalid IPv6 authority is rejected") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_ipv6_invalid_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://[]/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error == "Invalid download URL");
}
SECTION("disallowed host is rejected before download starts") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_host_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"downloads.example.com"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("host 'example.com'") != std::string::npos);
}
SECTION("store-only policy rejects output path") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_store_only_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.storeOnly = true;
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", "/tmp/export.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("store_only=true") != std::string::npos);
}
SECTION("invalid checksum format is rejected before download starts") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_bad_checksum_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp =
dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha1:not-allowed");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("invalid checksum format") != std::string::npos);
}
SECTION("unsupported checksum algorithm with valid hex is rejected") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_bad_checksum_algo_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp =
dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha1:abcdef12");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("invalid checksum format") != std::string::npos);
}
SECTION("malformed checksum missing separator is rejected") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_bad_checksum_shape_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("invalid checksum format") != std::string::npos);
}
SECTION("checksum-required policy rejects missing checksum") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_checksum_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = true;
cfg.downloadPolicy.allowedSchemes = {"https"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.error.find("checksum is required") != std::string::npos);
}
SECTION("valid checksum enters job tracking and handles unavailable service") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_valid_checksum_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();
auto resp =
dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK_FALSE(dl.success);
CHECK(dl.state == "queued");
CHECK_FALSE(dl.jobId.empty());
const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
CHECK(finalStatus.jobId == dl.jobId);
CHECK(finalStatus.state == "failed");
CHECK(finalStatus.error == "Download service not available in daemon");
}
SECTION("sha512 checksum enters job tracking and handles unavailable service") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_sha512_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", {},
"sha512:abcdef1234567890");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.state == "queued");
CHECK_FALSE(dl.jobId.empty());
const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
CHECK(finalStatus.jobId == dl.jobId);
CHECK(finalStatus.state == "failed");
CHECK(finalStatus.error == "Download service not available in daemon");
}
SECTION("md5 checksum enters job tracking and handles unavailable service") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_md5_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();
auto resp = dispatchDownload(dispatcher, "https://example.com/file.bin", {},
"md5:abcdef1234567890");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.state == "queued");
CHECK_FALSE(dl.jobId.empty());
const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
CHECK(finalStatus.jobId == dl.jobId);
CHECK(finalStatus.state == "failed");
CHECK(finalStatus.error == "Download service not available in daemon");
}
SECTION("forced successful download completes the tracked job") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_success_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
RequestDispatcher::__test_forceDownloadServiceSuccessOnce();
auto resp =
dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.state == "queued");
CHECK_FALSE(dl.jobId.empty());
const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
CHECK(finalStatus.jobId == dl.jobId);
CHECK(finalStatus.state == "completed");
CHECK(finalStatus.success);
CHECK(finalStatus.error.empty());
CHECK(finalStatus.hash == "sha256:forced-download-success");
CHECK(finalStatus.localPath.find("forced-download-success.bin") != std::string::npos);
CHECK(finalStatus.size == 64);
}
SECTION("daemon download disables redirects and surfaces blocked redirect failures") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_policy_redirect_blocked_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*"};
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
RequestDispatcher::__test_forceDownloadServiceFailureOnce(
"Redirect blocked by policy during probe (follow_redirects=false)");
auto resp =
dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.state == "queued");
CHECK_FALSE(dl.jobId.empty());
const auto lastFollowRedirects = waitForFollowRedirectsCapture();
REQUIRE(lastFollowRedirects.has_value());
CHECK_FALSE(*lastFollowRedirects);
const auto finalStatus = waitForDownloadJob(dispatcher, dl.jobId);
CHECK(finalStatus.jobId == dl.jobId);
CHECK(finalStatus.state == "failed");
CHECK(finalStatus.error ==
"Redirect blocked by policy during probe (follow_redirects=false)");
}
}
TEST_CASE("RequestDispatcher: download job status and list use in-memory registry",
"[daemon][documents][dispatcher][download-jobs]") {
RequestDispatcher::__test_resetDownloadJobs();
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_job_registry_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
DownloadResponse first;
first.url = "https://example.com/a.bin";
first.jobId = "download-1";
first.state = "completed";
first.createdAtMs = 100;
first.updatedAtMs = 200;
first.success = true;
DownloadResponse second;
second.url = "https://example.com/b.bin";
second.jobId = "download-2";
second.state = "failed";
second.createdAtMs = 300;
second.updatedAtMs = 400;
second.success = false;
second.error = "network boom";
RequestDispatcher::__test_seedDownloadJob(first);
RequestDispatcher::__test_seedDownloadJob(second);
SECTION("download status returns the seeded job") {
auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{"download-2"}});
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.jobId == "download-2");
CHECK(dl.state == "failed");
CHECK(dl.error == "network boom");
}
SECTION("download status returns not found for unknown job") {
auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{"missing-job"}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "Download job not found");
}
SECTION("list download jobs returns most recent first and honors limit") {
auto resp = dispatchRequest(dispatcher, Request{ListDownloadJobsRequest{1}});
REQUIRE(std::holds_alternative<ListDownloadJobsResponse>(resp));
const auto& jobs = std::get<ListDownloadJobsResponse>(resp);
REQUIRE(jobs.jobs.size() == 1);
CHECK(jobs.jobs.front().jobId == "download-2");
CHECK(jobs.jobs.front().state == "failed");
}
SECTION("list download jobs clamps zero limit to one") {
auto resp = dispatchRequest(dispatcher, Request{ListDownloadJobsRequest{0}});
REQUIRE(std::holds_alternative<ListDownloadJobsResponse>(resp));
const auto& jobs = std::get<ListDownloadJobsResponse>(resp);
REQUIRE(jobs.jobs.size() == 1);
CHECK(jobs.jobs.front().jobId == "download-2");
}
SECTION("cancel download job marks running job as canceled") {
DownloadResponse running;
running.url = "https://example.com/c.bin";
running.jobId = "download-3";
running.state = "running";
running.createdAtMs = 500;
running.updatedAtMs = 500;
running.success = false;
RequestDispatcher::__test_seedDownloadJob(running);
auto resp = dispatchRequest(dispatcher, Request{CancelDownloadJobRequest{"download-3"}});
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.jobId == "download-3");
CHECK(dl.state == "canceled");
CHECK(dl.error == "Canceled by client");
CHECK_FALSE(dl.success);
}
SECTION("cancel download job keeps terminal jobs terminal") {
auto resp = dispatchRequest(dispatcher, Request{CancelDownloadJobRequest{"download-2"}});
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
CHECK(dl.jobId == "download-2");
CHECK(dl.state == "failed");
CHECK(dl.error == "network boom");
}
SECTION("cancel download job returns not found for unknown job") {
auto resp = dispatchRequest(dispatcher, Request{CancelDownloadJobRequest{"missing-job"}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "Download job not found");
}
}
TEST_CASE("RequestDispatcher: download jobs reload from persisted registry",
"[daemon][documents][dispatcher][download-jobs]") {
RequestDispatcher::__test_resetDownloadJobs();
auto dispatchDownload = [](RequestDispatcher& dispatcher, const std::string& url,
std::string outputPath = {}, std::string checksum = {}) {
DownloadRequest req;
req.url = url;
req.outputPath = std::move(outputPath);
req.checksum = std::move(checksum);
return dispatchRequest(dispatcher, Request{req});
};
auto waitForDownloadJob = [](RequestDispatcher& dispatcher, const std::string& jobId,
std::chrono::milliseconds timeout = std::chrono::seconds(2)) {
const auto deadline = std::chrono::steady_clock::now() + timeout;
DownloadResponse latest;
while (std::chrono::steady_clock::now() < deadline) {
auto resp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{jobId}});
if (std::holds_alternative<DownloadResponse>(resp)) {
latest = std::get<DownloadResponse>(resp);
if (latest.state != "queued" && latest.state != "running") {
return latest;
}
}
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
return latest;
};
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_download_job_registry_persist_");
cfg.downloadPolicy.enable = true;
cfg.downloadPolicy.requireChecksum = false;
cfg.downloadPolicy.allowedSchemes = {"https"};
cfg.downloadPolicy.allowedHosts = {"*"};
std::string jobId;
{
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
RequestDispatcher::__test_forceDownloadServiceUnavailableOnce();
auto resp =
dispatchDownload(dispatcher, "https://example.com/file.bin", {}, "sha256:abcd1234");
REQUIRE(std::holds_alternative<DownloadResponse>(resp));
const auto& dl = std::get<DownloadResponse>(resp);
REQUIRE_FALSE(dl.jobId.empty());
CHECK_FALSE(dl.success);
CHECK(dl.state == "queued");
jobId = dl.jobId;
const auto finalStatus = waitForDownloadJob(dispatcher, jobId);
CHECK(finalStatus.state == "failed");
CHECK(finalStatus.error == "Download service not available in daemon");
}
RequestDispatcher::__test_forgetDownloadJobsCache();
{
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
auto statusResp = dispatchRequest(dispatcher, Request{DownloadStatusRequest{jobId}});
REQUIRE(std::holds_alternative<DownloadResponse>(statusResp));
const auto& status = std::get<DownloadResponse>(statusResp);
CHECK(status.jobId == jobId);
CHECK(status.state == "failed");
CHECK(status.error == "Download service not available in daemon");
auto listResp = dispatchRequest(dispatcher, Request{ListDownloadJobsRequest{1}});
REQUIRE(std::holds_alternative<ListDownloadJobsResponse>(listResp));
const auto& list = std::get<ListDownloadJobsResponse>(listResp);
REQUIRE(list.jobs.size() == 1);
CHECK(list.jobs.front().jobId == jobId);
CHECK(list.jobs.front().state == "failed");
}
}
TEST_CASE("RequestDispatcher: embedding handlers cover generation and repair branches",
"[daemon][embedding][dispatcher]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_embedding_dispatcher_");
cfg.configFilePath = cfg.dataDir / "missing-config.toml";
ScopedEnvVar preferredModelGuard("YAMS_PREFERRED_MODEL", std::nullopt);
ScopedEnvVar embedBackendGuard("YAMS_EMBED_BACKEND", std::nullopt);
ScopedEnvVar configPathGuard("YAMS_CONFIG", cfg.configFilePath.string());
ScopedEnvVar xdgConfigHomeGuard("XDG_CONFIG_HOME", cfg.dataDir.string());
ScopedEnvVar homeGuard("HOME", cfg.dataDir.string());
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
SECTION("generate embedding marks provider unavailable as degraded") {
auto resp = dispatchRequest(
dispatcher, Request{GenerateEmbeddingRequest{"hello world", "embed-model", true}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message.find("Plugin system") != std::string::npos);
CHECK(lifecycle.setSubsystemDegradedCalls() == 1);
CHECK(lifecycle.lastSubsystem() == "embedding");
CHECK(lifecycle.lastDegraded());
CHECK(lifecycle.lastReason() == "provider_unavailable");
}
SECTION("generate embedding auto-loads unloaded model and returns vector") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-model.onnx");
provider->setSingleEmbeddingResult({0.1f, 0.2f, 0.3f});
svc.__test_setModelProvider(provider);
GenerateEmbeddingRequest req{"hello world", "embed-model", true};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<EmbeddingResponse>(resp));
const auto& embedResp = std::get<EmbeddingResponse>(resp);
CHECK(embedResp.embedding == std::vector<float>{0.1f, 0.2f, 0.3f});
CHECK(embedResp.dimensions == 3);
CHECK(embedResp.modelUsed == "embed-model");
CHECK(provider->isModelLoaded("embed-model"));
}
SECTION("generate embedding forwards generator errors") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-error.onnx");
provider->setGenerateEmbeddingError(
Error{ErrorCode::InvalidState, "generator unavailable"});
svc.__test_setModelProvider(provider);
GenerateEmbeddingRequest req{"hello world", "embed-error", true};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "generator unavailable");
}
SECTION("generate embedding sanitizes utf8 in load failures") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-utf8.onnx");
std::string invalidMessage = "prefix ";
invalidMessage.push_back(static_cast<char>(0xC2));
invalidMessage.push_back(static_cast<char>(0xA9));
invalidMessage.push_back(static_cast<char>(0xE2));
invalidMessage.push_back(static_cast<char>(0x82));
invalidMessage.push_back(static_cast<char>(0xAC));
invalidMessage.push_back(static_cast<char>(0xF0));
invalidMessage.push_back(static_cast<char>(0x9F));
invalidMessage.push_back(static_cast<char>(0x92));
invalidMessage.push_back(static_cast<char>(0xA9));
invalidMessage.push_back(static_cast<char>(0xC0));
invalidMessage.push_back(static_cast<char>(0xFF));
provider->setLoadError(Error{ErrorCode::InvalidState, invalidMessage});
svc.__test_setModelProvider(provider);
GenerateEmbeddingRequest req{"hello world", "embed-utf8", true};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
std::string expected = "prefix ";
expected.push_back(static_cast<char>(0xC2));
expected.push_back(static_cast<char>(0xA9));
expected.push_back(static_cast<char>(0xE2));
expected.push_back(static_cast<char>(0x82));
expected.push_back(static_cast<char>(0xAC));
expected.push_back(static_cast<char>(0xF0));
expected.push_back(static_cast<char>(0x9F));
expected.push_back(static_cast<char>(0x92));
expected.push_back(static_cast<char>(0xA9));
expected += "\xEF\xBF\xBD\xEF\xBF\xBD";
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == expected);
}
SECTION("generate embedding honors timeout env floor while auto-loading") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-timeout.onnx");
provider->setSingleEmbeddingResult({0.7f, 0.8f, 0.9f});
svc.__test_setModelProvider(provider);
ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "5");
GenerateEmbeddingRequest req{"hello world", "embed-timeout", true};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<EmbeddingResponse>(resp));
const auto& embedResp = std::get<EmbeddingResponse>(resp);
CHECK(embedResp.embedding == std::vector<float>{0.7f, 0.8f, 0.9f});
CHECK(provider->isModelLoaded("embed-timeout"));
}
SECTION("generate embedding converts isModelLoaded exceptions into internal errors") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/embed-throw.onnx");
provider->setThrowOnIsModelLoadedCount(2);
svc.__test_setModelProvider(provider);
GenerateEmbeddingRequest req{"hello world", "embed-throw", true};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Embedding generation failed: isModelLoaded boom") !=
std::string::npos);
}
SECTION("batch embedding falls back to per-item generation") {
auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-fallback.onnx");
provider->setGenerateBatchError(Error{ErrorCode::NotImplemented, "batch unsupported"});
provider->setSingleEmbeddingResult({0.4f, 0.6f});
svc.__test_setModelProvider(provider);
BatchEmbeddingRequest req;
req.texts = {"one", "two"};
req.modelName = "batch-fallback";
req.normalize = false;
req.batchSize = 2;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<BatchEmbeddingResponse>(resp));
const auto& batchResp = std::get<BatchEmbeddingResponse>(resp);
REQUIRE(batchResp.embeddings.size() == 2);
CHECK(batchResp.embeddings[0] == std::vector<float>{0.4f, 0.6f});
CHECK(batchResp.embeddings[1] == std::vector<float>{0.4f, 0.6f});
CHECK(batchResp.successCount == 2);
CHECK(batchResp.failureCount == 0);
CHECK(batchResp.dimensions == 2);
CHECK(batchResp.modelUsed == "batch-fallback");
}
SECTION("batch embedding counts per-item failures during fallback") {
auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-partial.onnx");
provider->setGenerateBatchError(Error{ErrorCode::NotImplemented, "batch unsupported"});
provider->setSingleEmbeddingResult({0.9f, 0.1f});
provider->setSingleTextError("bad", Error{ErrorCode::InternalError, "single failed"});
svc.__test_setModelProvider(provider);
BatchEmbeddingRequest req;
req.texts = {"good", "bad", "good-again"};
req.modelName = "batch-partial";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<BatchEmbeddingResponse>(resp));
const auto& batchResp = std::get<BatchEmbeddingResponse>(resp);
REQUIRE(batchResp.embeddings.size() == 2);
CHECK(batchResp.successCount == 2);
CHECK(batchResp.failureCount == 1);
CHECK(batchResp.dimensions == 2);
CHECK(batchResp.modelUsed == "batch-partial");
}
SECTION("batch embedding forwards non-fallback errors") {
auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-error.onnx");
provider->setGenerateBatchError(Error{ErrorCode::InvalidState, "batch failed"});
svc.__test_setModelProvider(provider);
BatchEmbeddingRequest req;
req.texts = {"one", "two"};
req.modelName = "batch-error";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "batch failed");
}
SECTION("batch embedding auto-loads model and returns direct batch results") {
auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-direct.onnx");
provider->setBatchEmbeddingResult({{0.2f, 0.8f}});
svc.__test_setModelProvider(provider);
ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "not-a-number");
BatchEmbeddingRequest req;
req.texts = {"one", "two"};
req.modelName = "batch-direct";
req.normalize = true;
req.batchSize = 2;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<BatchEmbeddingResponse>(resp));
const auto& batchResp = std::get<BatchEmbeddingResponse>(resp);
REQUIRE(batchResp.embeddings.size() == 1);
CHECK(batchResp.embeddings[0] == std::vector<float>{0.2f, 0.8f});
CHECK(batchResp.successCount == 1);
CHECK(batchResp.failureCount == 1);
CHECK(batchResp.dimensions == 2);
CHECK(batchResp.modelUsed == "batch-direct");
CHECK(provider->isModelLoaded("batch-direct"));
}
SECTION("batch embedding returns load failures for unloaded models") {
auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-load-fail.onnx");
provider->setLoadError(Error{ErrorCode::InvalidState, "batch load refused"});
svc.__test_setModelProvider(provider);
ScopedEnvVar timeoutGuard("YAMS_MODEL_LOAD_TIMEOUT_MS", "5");
BatchEmbeddingRequest req;
req.texts = {"one", "two"};
req.modelName = "batch-load-fail";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "batch load refused");
}
SECTION("batch embedding converts isModelLoaded exceptions into internal errors") {
auto provider = std::make_shared<StubModelProvider>(2, "/tmp/batch-throw.onnx");
provider->setThrowOnIsModelLoadedCount(2);
svc.__test_setModelProvider(provider);
BatchEmbeddingRequest req;
req.texts = {"one"};
req.modelName = "batch-throw";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Batch embedding failed: isModelLoaded boom") != std::string::npos);
}
SECTION("embed documents rejects missing service manager") {
RequestDispatcher noServiceDispatcher(&lifecycle, nullptr, &state);
auto resp = dispatchRequest(noServiceDispatcher, Request{EmbedDocumentsRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message == "ServiceManager not available");
}
SECTION("embed documents reports unavailable model provider") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-provider.onnx");
provider->setAvailable(false);
svc.__test_setModelProvider(provider);
EmbedDocumentsRequest req;
req.modelName = "repair-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message == "Model provider not available");
}
SECTION("embed documents returns success when repair finds no work") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-empty.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
EmbedDocumentsRequest req;
req.modelName = "repair-model";
req.batchSize = 4;
req.skipExisting = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<EmbedDocumentsResponse>(resp));
const auto& embedResp = std::get<EmbedDocumentsResponse>(resp);
CHECK(embedResp.requested == 0);
CHECK(embedResp.embedded == 0);
CHECK(embedResp.skipped == 0);
CHECK(embedResp.failed == 0);
}
SECTION("embed documents rejects non-empty hash lists that resolve to no daemon documents") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-missing-hashes.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
EmbedDocumentsRequest req;
req.modelName = "repair-model";
req.documentHashes = {std::string(64, 'a')};
req.batchSize = 4;
req.skipExisting = false;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message.find("requested documents were found") != std::string::npos);
}
SECTION("embed documents rejects degraded providers") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-degraded.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setModelProviderDegraded(true, "simulated degraded state");
auto resp = dispatchRequest(dispatcher, Request{EmbedDocumentsRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "Embedding generation disabled: provider degraded");
svc.__test_setModelProviderDegraded(false);
}
SECTION("embed documents reports missing content store") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-content.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
EmbedDocumentsRequest req;
req.modelName = "repair-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message == "Content store not available");
}
SECTION("embed documents reports missing metadata repo") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-meta.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
EmbedDocumentsRequest req;
req.modelName = "repair-model";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message == "Metadata repository not available");
}
SECTION("embed documents reports missing configured model name") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-empty-model.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
EmbedDocumentsRequest req;
req.modelName = "";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotInitialized);
CHECK(err.message == "No embedding model configured");
}
SECTION("embed documents uses daemon config resolver when request model is empty") {
auto provider = std::make_shared<StubModelProvider>(3, "/tmp/repair-config-model.onnx");
svc.__test_setModelProvider(provider);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
svc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
const auto configPath = cfg.dataDir / "config.toml";
{
std::ofstream out(configPath);
REQUIRE(out.is_open());
out << "[embeddings]\n";
out << "preferred_model = \"config-repair-model\"\n";
}
cfg.configFilePath = configPath;
ScopedEnvVar sectionConfigPathGuard("YAMS_CONFIG", configPath.string());
DaemonLifecycleFsm configLifecycleFsm;
ServiceManager configSvc(cfg, state, configLifecycleFsm);
configSvc.__test_setModelProvider(provider);
configSvc.__test_setContentStore(std::make_shared<StubContentStore>());
configSvc.__test_setMetadataRepo(std::make_shared<StubPruneMetadataRepository>());
RequestDispatcher configDispatcher(&lifecycle, &configSvc, &state);
EmbedDocumentsRequest req;
req.modelName.clear();
req.batchSize = 4;
req.skipExisting = true;
auto resp = dispatchRequest(configDispatcher, Request{req});
REQUIRE(std::holds_alternative<EmbedDocumentsResponse>(resp));
CHECK(provider->isModelLoaded("config-repair-model"));
}
}
TEST_CASE("RequestDispatcher: graph maintenance handlers report missing graph component",
"[daemon][graph][dispatcher]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_graph_dispatcher_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
SECTION("graph repair returns internal error when graph component is unavailable") {
GraphRepairRequest req;
req.dryRun = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("GraphComponent not available") != std::string::npos);
}
SECTION("graph validate returns internal error when graph component is unavailable") {
auto resp = dispatchRequest(dispatcher, Request{GraphValidateRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("GraphComponent not available") != std::string::npos);
}
}
TEST_CASE("RequestDispatcher: graph query and ingest handlers cover dispatcher branches",
"[daemon][graph][dispatcher][query]") {
GraphDispatcherFixture fixture;
SECTION("graph query reports unavailable metadata repository") {
auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphQueryRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Metadata repository unavailable");
}
SECTION("graph query reports unavailable knowledge graph store") {
fixture.initMetadata(false);
auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphQueryRequest{}});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK_FALSE(graphResp.kgAvailable);
CHECK(graphResp.warning == "Knowledge graph not available");
}
SECTION("graph path history reports unavailable metadata repository") {
GraphPathHistoryRequest req;
req.path = "/repo/file.cpp";
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Metadata repository unavailable");
}
SECTION("graph explore reports unavailable metadata repository") {
auto resp =
dispatchRequest(*fixture.dispatcher, Request{GraphExploreRequest{.query = "entry"}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Metadata repository unavailable");
}
SECTION("graph explore reports unavailable knowledge graph store") {
fixture.initMetadata(false);
auto resp =
dispatchRequest(*fixture.dispatcher, Request{GraphExploreRequest{.query = "entry"}});
REQUIRE(std::holds_alternative<GraphExploreResponse>(resp));
const auto& graphResp = std::get<GraphExploreResponse>(resp);
CHECK(graphResp.query == "entry");
CHECK_FALSE(graphResp.kgAvailable);
REQUIRE(graphResp.warnings.size() == 1);
CHECK(graphResp.warnings.front() == "Knowledge graph not available");
}
SECTION("graph query traversal requires an origin") {
fixture.initMetadata();
auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphQueryRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message.find("No valid origin specified") != std::string::npos);
}
SECTION("graph explore returns full and omitted snippet modes through dispatcher") {
fixture.initMetadata();
const auto sourcePath = fixture.testDir / "src" / "fallback.cpp";
std::filesystem::create_directories(sourcePath.parent_path());
{
std::ofstream out(sourcePath);
REQUIRE(out.good());
out << "int fallbackEntry() {\n return 7;\n}\n";
}
REQUIRE(
fixture.repo->insertDocument(makeGraphDispatcherDocument(sourcePath, "hash-fallback"))
.has_value());
metadata::KGNode node;
node.nodeKey = "function:demo::fallbackEntry@" + sourcePath.string();
node.label = "fallbackEntry";
node.type = "function";
REQUIRE(fixture.kgStore->upsertNode(node).has_value());
GraphExploreRequest fullReq;
fullReq.query = "fallbackEntry";
auto fullRespRaw = dispatchRequest(*fixture.dispatcher, Request{fullReq});
REQUIRE(std::holds_alternative<GraphExploreResponse>(fullRespRaw));
const auto& fullResp = std::get<GraphExploreResponse>(fullRespRaw);
REQUIRE(fullResp.files.size() == 1);
CHECK(fullResp.files.front().mode == "full");
CHECK(fullResp.files.front().content.find("fallbackEntry") != std::string::npos);
CHECK(fullResp.entrySymbols.size() == 1);
CHECK(fullResp.entrySymbols.front().label == "fallbackEntry");
GraphExploreRequest omittedReq;
omittedReq.query = "fallbackEntry";
omittedReq.includeCode = false;
auto omittedRespRaw = dispatchRequest(*fixture.dispatcher, Request{omittedReq});
REQUIRE(std::holds_alternative<GraphExploreResponse>(omittedRespRaw));
const auto& omittedResp = std::get<GraphExploreResponse>(omittedRespRaw);
REQUIRE(omittedResp.files.size() == 1);
CHECK(omittedResp.files.front().mode == "omitted");
CHECK(omittedResp.files.front().content.empty());
}
SECTION("graph query lists node types") {
fixture.initMetadata();
fixture.upsertNode("fn:list:1", "alpha", "function");
fixture.upsertNode("fn:list:2", "beta", "function");
fixture.upsertNode("path:list:1", "file.cpp", "file");
GraphQueryRequest req;
req.listTypes = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.kgAvailable);
CHECK(graphResp.originNode.label == "listTypes");
REQUIRE(graphResp.nodeTypeCounts.size() == 2);
CHECK(graphResp.nodeTypeCounts[0] == std::pair<std::string, uint64_t>{"function", 2});
CHECK(graphResp.nodeTypeCounts[1] == std::pair<std::string, uint64_t>{"file", 1});
}
SECTION("graph query list types reports store errors") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");
GraphQueryRequest req;
req.listTypes = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code != ErrorCode::Success);
CHECK_FALSE(err.message.empty());
}
SECTION("graph query lists relation counts") {
fixture.initMetadata();
auto callerId = fixture.upsertNode("fn:rel:caller", "caller", "function");
auto calleeId = fixture.upsertNode("fn:rel:callee", "callee", "function");
auto fileId = fixture.upsertNode("path:rel:file", "file.cpp", "file");
fixture.addEdge(callerId, calleeId, "calls");
fixture.addEdge(callerId, fileId, "includes");
GraphQueryRequest req;
req.listRelations = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.originNode.label == "listRelations");
REQUIRE(graphResp.relationTypeCounts.size() == 2);
CHECK(graphResp.relationTypeCounts[0] == std::pair<std::string, uint64_t>{"calls", 1});
CHECK(graphResp.relationTypeCounts[1] == std::pair<std::string, uint64_t>{"includes", 1});
}
SECTION("graph query list relations reports store errors") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_edges RENAME TO kg_edges_broken");
GraphQueryRequest req;
req.listRelations = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code != ErrorCode::Success);
CHECK_FALSE(err.message.empty());
}
SECTION("graph query search mode requires a pattern") {
fixture.initMetadata();
GraphQueryRequest req;
req.searchMode = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message == "searchPattern is required for search mode");
}
SECTION("graph query search mode returns matching nodes with properties") {
fixture.initMetadata();
fixture.upsertNode("fn:search:alpha", "AlphaNode", "function", R"({"rank":1})");
fixture.upsertNode("fn:search:beta", "BetaNode", "function");
GraphQueryRequest req;
req.searchMode = true;
req.searchPattern = "%Alpha%";
req.includeNodeProperties = true;
req.limit = 10;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.originNode.label == "search:%Alpha%");
REQUIRE(graphResp.connectedNodes.size() == 1);
CHECK(graphResp.connectedNodes[0].nodeKey == "fn:search:alpha");
CHECK(graphResp.connectedNodes[0].properties == R"({"rank":1})");
}
SECTION("graph query search mode reports store errors") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");
GraphQueryRequest req;
req.searchMode = true;
req.searchPattern = "%Alpha%";
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code != ErrorCode::Success);
CHECK_FALSE(err.message.empty());
}
SECTION("graph query list by type requires node type") {
fixture.initMetadata();
GraphQueryRequest req;
req.listByType = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message == "nodeType is required for listByType mode");
}
SECTION("graph query list by type returns paginated nodes") {
fixture.initMetadata();
fixture.upsertNode("fn:type:1", "TypeOne", "function", R"({"kind":"a"})");
fixture.upsertNode("fn:type:2", "TypeTwo", "function", R"({"kind":"b"})");
fixture.upsertNode("path:type:1", "file", "file");
GraphQueryRequest req;
req.listByType = true;
req.nodeType = "function";
req.limit = 1;
req.offset = 0;
req.includeNodeProperties = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.originNode.label == "listByType:function");
CHECK(graphResp.totalNodesFound == 2);
CHECK(graphResp.truncated);
REQUIRE(graphResp.connectedNodes.size() == 1);
CHECK(graphResp.connectedNodes[0].type == "function");
CHECK_FALSE(graphResp.connectedNodes[0].properties.empty());
}
SECTION("graph query list by type reports store errors") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");
GraphQueryRequest req;
req.listByType = true;
req.nodeType = "function";
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code != ErrorCode::Success);
CHECK_FALSE(err.message.empty());
}
SECTION("graph query isolated mode canonicalizes relation name") {
fixture.initMetadata();
auto calledId = fixture.upsertNode("fn:isolated:called", "called", "function");
fixture.upsertNode("fn:isolated:one", "isolated1", "function");
fixture.upsertNode("fn:isolated:two", "isolated2", "function");
auto callerId = fixture.upsertNode("fn:isolated:caller", "caller", "function");
fixture.addEdge(callerId, calledId, "calls");
GraphQueryRequest req;
req.isolatedMode = true;
req.nodeType = "function";
req.isolatedRelation = " CALL ";
req.limit = 10;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.originNode.label == "isolated:function:calls");
CHECK(graphResp.totalNodesFound == 3);
for (const auto& node : graphResp.connectedNodes) {
CHECK(node.nodeKey != "fn:isolated:called");
}
}
SECTION("graph query isolated mode uses default node type and relation") {
fixture.initMetadata();
auto calledId = fixture.upsertNode("fn:isolated:default:called", "called", "function");
fixture.upsertNode("fn:isolated:default:free", "free", "function");
auto callerId = fixture.upsertNode("fn:isolated:default:caller", "caller", "function");
fixture.addEdge(callerId, calledId, "calls");
GraphQueryRequest req;
req.isolatedMode = true;
req.limit = 10;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.originNode.label == "isolated:function:calls");
REQUIRE(graphResp.connectedNodes.size() == 2);
}
SECTION("graph query isolated mode reports store errors") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");
GraphQueryRequest req;
req.isolatedMode = true;
req.limit = 10;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code != ErrorCode::Success);
CHECK_FALSE(err.message.empty());
}
SECTION("graph query node key lookup returns not found") {
fixture.initMetadata();
GraphQueryRequest req;
req.nodeKey = "missing-node";
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "Node not found: missing-node");
}
SECTION("graph query traverses from node key with canonical relation filters") {
fixture.initMetadata();
auto callerId = fixture.upsertNode("fn:key:caller", "caller", "function");
auto calleeId = fixture.upsertNode("fn:key:callee", "callee", "function");
auto fileId = fixture.upsertNode("path:key:file", "header", "file");
fixture.addEdge(callerId, calleeId, "calls", 0.8f, R"({"source":"unit"})");
fixture.addEdge(callerId, fileId, "includes");
GraphQueryRequest req;
req.nodeKey = "fn:key:caller";
req.maxDepth = 1;
req.maxResults = 10;
req.relationFilters = {" Call "};
req.includeEdgeProperties = true;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
CHECK(graphResp.kgAvailable);
CHECK(graphResp.originNode.nodeKey == "fn:key:caller");
REQUIRE(graphResp.connectedNodes.size() == 1);
CHECK(graphResp.connectedNodes[0].nodeKey == "fn:key:callee");
REQUIRE(graphResp.edges.size() == 1);
CHECK(graphResp.edges[0].relation == "calls");
CHECK(graphResp.edges[0].properties == R"({"source":"unit"})");
}
SECTION("graph query canonicalizes relation aliases and skips blank filters") {
fixture.initMetadata();
auto originId = fixture.upsertNode("fn:key:origin", "origin", "function");
auto includeId = fixture.upsertNode("path:key:include", "header", "file");
auto inheritId = fixture.upsertNode("cls:key:inherit", "Base", "class");
auto implementId = fixture.upsertNode("iface:key:impl", "Iface", "interface");
auto referenceId = fixture.upsertNode("sym:key:ref", "Ref", "symbol");
auto renameToId = fixture.upsertNode("path:key:rename_to", "new.cpp", "file");
auto renameFromId = fixture.upsertNode("path:key:rename_from", "old.cpp", "file");
auto customId = fixture.upsertNode("sym:key:custom", "Custom", "symbol");
fixture.addEdge(originId, includeId, "includes");
fixture.addEdge(originId, inheritId, "inherits");
fixture.addEdge(originId, implementId, "implements");
fixture.addEdge(originId, referenceId, "references");
fixture.addEdge(originId, renameToId, "renamed_to");
fixture.addEdge(originId, renameFromId, "renamed_from");
fixture.addEdge(originId, customId, "custom_rel");
GraphQueryRequest req;
req.nodeKey = "fn:key:origin";
req.maxDepth = 1;
req.maxResults = 20;
req.relationFilters = {" ", " Include ", "inherit", "implement",
"reference", "rename_to", "rename_from", "custom-rel"};
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphQueryResponse>(resp));
const auto& graphResp = std::get<GraphQueryResponse>(resp);
REQUIRE(graphResp.edges.size() == 7);
std::unordered_set<std::string> relations;
for (const auto& edge : graphResp.edges) {
relations.insert(edge.relation);
}
CHECK(relations.count("includes") == 1);
CHECK(relations.count("inherits") == 1);
CHECK(relations.count("implements") == 1);
CHECK(relations.count("references") == 1);
CHECK(relations.count("renamed_to") == 1);
CHECK(relations.count("renamed_from") == 1);
CHECK(relations.count("custom_rel") == 1);
}
SECTION("graph path history requires path") {
fixture.initMetadata();
auto resp = dispatchRequest(*fixture.dispatcher, Request{GraphPathHistoryRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidArgument);
CHECK(err.message == "Path is required for path history query");
}
SECTION("graph path history returns empty response when kg store is unavailable") {
fixture.initMetadata(false);
GraphPathHistoryRequest req;
req.path = "/repo/file.cpp";
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphPathHistoryResponse>(resp));
const auto& historyResp = std::get<GraphPathHistoryResponse>(resp);
CHECK(historyResp.queryPath == "/repo/file.cpp");
CHECK(historyResp.history.empty());
CHECK_FALSE(historyResp.hasMore);
}
SECTION("graph path history returns snapshot entries") {
fixture.initMetadata();
fixture.upsertNode("path:snap1:/repo/file.cpp", "/repo/file.cpp", "path",
R"({"snapshot_id":"snap1","path":"/repo/file.cpp"})");
fixture.upsertNode("path:snap2:/repo/file.cpp", "/repo/file.cpp", "path",
R"({"snapshot_id":"snap2","path":"/repo/file.cpp"})");
GraphPathHistoryRequest req;
req.path = "/repo/file.cpp";
req.limit = 10;
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<GraphPathHistoryResponse>(resp));
const auto& historyResp = std::get<GraphPathHistoryResponse>(resp);
CHECK(historyResp.queryPath == "/repo/file.cpp");
REQUIRE(historyResp.history.size() == 2);
CHECK(historyResp.history[0].snapshotId == "snap2");
CHECK(historyResp.history[0].blobHash.empty());
CHECK(historyResp.history[0].changeType == "unknown");
CHECK(historyResp.history[1].snapshotId == "snap1");
CHECK_FALSE(historyResp.hasMore);
}
SECTION("graph path history reports store errors") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");
GraphPathHistoryRequest req;
req.path = "/repo/file.cpp";
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code != ErrorCode::Success);
CHECK_FALSE(err.message.empty());
}
SECTION("kg ingest reports unavailable metadata repository") {
auto resp = dispatchRequest(*fixture.dispatcher, Request{KgIngestRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Metadata repository unavailable");
}
SECTION("kg ingest reports unavailable knowledge graph store") {
fixture.initMetadata(false);
auto resp = dispatchRequest(*fixture.dispatcher, Request{KgIngestRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Knowledge graph store unavailable");
}
SECTION("kg ingest records node edge and alias resolution errors") {
fixture.initMetadata();
auto existingId = fixture.upsertNode("fn:existing", "existing", "function");
(void)existingId;
KgIngestRequest req;
req.nodes.push_back(KgIngestNode{"fn:existing", "existing updated", "function", ""});
req.edges.push_back(KgIngestEdge{"fn:existing", "fn:missing", "calls", 1.0f, ""});
req.aliases.push_back(KgIngestAlias{"fn:missing", "missing-alias", "unit", 0.5f});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK(ingestResp.success);
CHECK(ingestResp.nodesInserted == 1);
CHECK(ingestResp.edgesSkipped == 1);
CHECK(ingestResp.aliasesSkipped == 1);
REQUIRE(ingestResp.errors.size() == 2);
CHECK(ingestResp.errors[0] == "Edge destination node not found: fn:missing");
CHECK(ingestResp.errors[1] == "Alias node not found: fn:missing");
}
SECTION("kg ingest inserts nodes edges and aliases") {
fixture.initMetadata();
KgIngestRequest req;
req.nodes.push_back(
KgIngestNode{"fn:ingest:src", "Source", "function", R"({"role":"src"})"});
req.nodes.push_back(KgIngestNode{"fn:ingest:dst", "Dest", "function", ""});
req.edges.push_back(KgIngestEdge{"fn:ingest:src", "fn:ingest:dst", "calls", 0.7f,
R"({"origin":"dispatcher"})"});
req.aliases.push_back(KgIngestAlias{"fn:ingest:src", "sourceAlias", "unit", 0.9f});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK(ingestResp.success);
CHECK(ingestResp.nodesInserted == 2);
CHECK(ingestResp.edgesInserted == 1);
CHECK(ingestResp.aliasesInserted == 1);
CHECK(ingestResp.errors.empty());
auto sourceNode = fixture.kgStore->getNodeByKey("fn:ingest:src");
REQUIRE(sourceNode.has_value());
REQUIRE(sourceNode.value().has_value());
auto edges =
fixture.kgStore->getEdgesFrom(sourceNode.value()->id, std::string_view("calls"), 10, 0);
REQUIRE(edges.has_value());
REQUIRE(edges.value().size() == 1);
auto aliases = fixture.kgStore->resolveAliasExact("sourceAlias", 10);
REQUIRE(aliases.has_value());
REQUIRE(aliases.value().size() == 1);
}
SECTION("kg ingest reports node upsert failures") {
fixture.initMetadata();
fixture.execSql("ALTER TABLE kg_nodes RENAME TO kg_nodes_broken");
KgIngestRequest req;
req.nodes.push_back(KgIngestNode{"fn:broken:node", "Broken", "function", ""});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK_FALSE(ingestResp.success);
CHECK(ingestResp.nodesInserted == 0);
REQUIRE(ingestResp.errors.size() == 1);
CHECK(ingestResp.errors[0].find("Node upsert failed:") == 0);
}
SECTION("kg ingest inserts edges without de-duplication when requested") {
fixture.initMetadata();
fixture.upsertNode("fn:addedges:src", "Source", "function");
fixture.upsertNode("fn:addedges:dst", "Dest", "function");
KgIngestRequest req;
req.skipExistingEdges = false;
req.edges.push_back(KgIngestEdge{"fn:addedges:src", "fn:addedges:dst", "calls", 1.0f, ""});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK(ingestResp.success);
CHECK(ingestResp.edgesInserted == 1);
CHECK(ingestResp.errors.empty());
}
SECTION("kg ingest reports edge insert failures") {
fixture.initMetadata();
fixture.upsertNode("fn:edgefail:src", "Source", "function");
fixture.upsertNode("fn:edgefail:dst", "Dest", "function");
fixture.execSql("ALTER TABLE kg_edges RENAME TO kg_edges_broken");
KgIngestRequest req;
req.skipExistingEdges = false;
req.edges.push_back(KgIngestEdge{"fn:edgefail:src", "fn:edgefail:dst", "calls", 1.0f, ""});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK_FALSE(ingestResp.success);
CHECK(ingestResp.edgesInserted == 0);
REQUIRE(ingestResp.errors.size() == 1);
CHECK(ingestResp.errors[0].find("Edge insert failed:") == 0);
}
SECTION("kg ingest reports alias insert failures") {
fixture.initMetadata();
fixture.upsertNode("fn:aliasfail:node", "Node", "function");
fixture.execSql("ALTER TABLE kg_aliases RENAME TO kg_aliases_broken");
KgIngestRequest req;
req.aliases.push_back(KgIngestAlias{"fn:aliasfail:node", "brokenAlias", "unit", 0.4f});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK(ingestResp.success);
CHECK(ingestResp.aliasesInserted == 0);
REQUIRE(ingestResp.errors.size() == 1);
CHECK(ingestResp.errors[0].find("Alias insert failed:") == 0);
}
SECTION("kg ingest resolves existing nodes for edges and aliases") {
fixture.initMetadata();
auto srcId = fixture.upsertNode("fn:existing:src", "Source", "function");
auto dstId = fixture.upsertNode("fn:existing:dst", "Dest", "function");
auto aliasNodeId = fixture.upsertNode("fn:existing:alias", "AliasTarget", "function");
(void)srcId;
(void)dstId;
(void)aliasNodeId;
KgIngestRequest req;
req.edges.push_back(KgIngestEdge{"fn:existing:src", "fn:existing:dst", "calls", 1.0f, ""});
req.aliases.push_back(KgIngestAlias{"fn:existing:alias", "existingAlias", "", 0.6f});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK(ingestResp.success);
CHECK(ingestResp.nodesInserted == 0);
CHECK(ingestResp.edgesInserted == 1);
CHECK(ingestResp.aliasesInserted == 1);
CHECK(ingestResp.errors.empty());
auto aliases = fixture.kgStore->resolveAliasExact("existingAlias", 10);
REQUIRE(aliases.has_value());
REQUIRE(aliases.value().size() == 1);
}
SECTION("kg ingest reports missing edge source nodes") {
fixture.initMetadata();
fixture.upsertNode("fn:existing:dst_only", "Dest", "function");
KgIngestRequest req;
req.edges.push_back(
KgIngestEdge{"fn:missing:src", "fn:existing:dst_only", "calls", 1.0f, ""});
auto resp = dispatchRequest(*fixture.dispatcher, Request{req});
REQUIRE(std::holds_alternative<KgIngestResponse>(resp));
const auto& ingestResp = std::get<KgIngestResponse>(resp);
CHECK(ingestResp.success);
CHECK(ingestResp.edgesInserted == 0);
CHECK(ingestResp.edgesSkipped == 1);
REQUIRE(ingestResp.errors.size() == 1);
CHECK(ingestResp.errors[0] == "Edge source node not found: fn:missing:src");
}
}
TEST_CASE("RequestDispatcher: plugin handlers cover readiness and error branches",
"[daemon][plugin][dispatcher]") {
auto makeReadyService = []() {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_plugin_dispatcher_");
auto state = std::make_unique<StateComponent>();
state->readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state->readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
auto lifecycleFsm = std::make_unique<DaemonLifecycleFsm>();
auto svc = std::make_unique<ServiceManager>(cfg, *state, *lifecycleFsm);
svc->__test_pluginScanComplete(0);
REQUIRE(svc->getPluginHostFsmSnapshot().state == PluginHostState::Ready);
return std::tuple{std::move(state), std::move(lifecycleFsm), std::move(svc)};
};
SECTION("plugin handlers reject non-ready host state") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
svc->__test_pluginLoadFailed("simulated plugin init failure");
auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "Plugin subsystem not ready (state=failed)");
}
SECTION("plugin handlers reject missing service manager and non-ready FSM states") {
auto state = std::make_unique<StateComponent>();
state->readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state->readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
StubLifecycle lifecycle;
{
RequestDispatcher dispatcher(&lifecycle, nullptr, state.get());
auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "Plugin host unavailable");
}
{
auto [readyState, lifecycleFsm, svc] = makeReadyService();
RequestDispatcher dispatcher(&lifecycle, svc.get(), readyState.get());
svc->__test_pluginScanStarted(1);
auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "Plugin subsystem not ready (state=scanning)");
}
{
auto [readyState, lifecycleFsm, svc] = makeReadyService();
RequestDispatcher dispatcher(&lifecycle, svc.get(), readyState.get());
svc->__test_pluginLoaded("mock-loaded-plugin");
svc->__test_pluginScanComplete(1);
svc->__test_pluginLoaded("mock-loaded-plugin");
auto resp = dispatchRequest(dispatcher, Request{PluginTrustListRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InvalidState);
CHECK(err.message == "Plugin subsystem not ready (state=loading)");
}
}
SECTION("plugin handlers report missing hosts and trust failures") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
svc->__test_setAbiHost({});
svc->__test_setExternalPluginHost({});
auto scanResp = dispatchRequest(dispatcher, Request{PluginScanRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(scanResp));
const auto& scanErr = std::get<ErrorResponse>(scanResp);
CHECK(scanErr.code == ErrorCode::NotImplemented);
CHECK(scanErr.message == "No plugin host available");
PluginLoadRequest loadReq;
loadReq.pathOrName = "missing-plugin";
auto loadResp = dispatchRequest(dispatcher, Request{loadReq});
REQUIRE(std::holds_alternative<ErrorResponse>(loadResp));
const auto& loadErr = std::get<ErrorResponse>(loadResp);
CHECK(loadErr.code == ErrorCode::NotImplemented);
CHECK(loadErr.message == "No plugin host available");
auto unloadResp =
dispatchRequest(dispatcher, Request{PluginUnloadRequest{"missing-plugin"}});
REQUIRE(std::holds_alternative<ErrorResponse>(unloadResp));
const auto& unloadErr = std::get<ErrorResponse>(unloadResp);
CHECK(unloadErr.code == ErrorCode::NotFound);
CHECK(unloadErr.message == "Plugin not found or unload failed");
auto trustAddResp =
dispatchRequest(dispatcher, Request{PluginTrustAddRequest{"/tmp/missing-host"}});
REQUIRE(std::holds_alternative<ErrorResponse>(trustAddResp));
const auto& trustAddErr = std::get<ErrorResponse>(trustAddResp);
CHECK(trustAddErr.code == ErrorCode::Unknown);
CHECK(trustAddErr.message == "Trust add failed");
auto trustRemoveResp =
dispatchRequest(dispatcher, Request{PluginTrustRemoveRequest{"/tmp/missing-host"}});
REQUIRE(std::holds_alternative<ErrorResponse>(trustRemoveResp));
const auto& trustRemoveErr = std::get<ErrorResponse>(trustRemoveResp);
CHECK(trustRemoveErr.code == ErrorCode::Unknown);
CHECK(trustRemoveErr.message == "Trust remove failed");
}
SECTION("plugin load dry run reports missing plugin") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
PluginLoadRequest req;
req.pathOrName = (svc->getResolvedDataDir() / "missing_plugin.py").string();
req.dryRun = true;
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::NotFound);
CHECK(err.message == "Plugin not found");
}
SECTION("plugin scan and load search default directories") {
auto homeDir = makeTempDir("yams_plugin_home_");
auto homeText = homeDir.string();
ScopedEnvVar homeGuard("HOME", homeText.c_str());
#ifdef _WIN32
ScopedEnvVar localAppDataGuard("LOCALAPPDATA", homeText.c_str());
#endif
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
#ifdef _WIN32
auto abiDir = homeDir / "yams" / "plugins";
#else
auto abiDir = homeDir / ".local" / "lib" / "yams" / "plugins";
#endif
std::filesystem::create_directories(abiDir);
auto fakePlugin = abiDir /
#ifdef _WIN32
"yams_default_scan.dll";
#else
"yams_default_scan.so";
#endif
{
std::ofstream out(fakePlugin, std::ios::binary);
out << "not a valid plugin";
}
#ifdef _WIN32
auto externalDir = homeDir / "yams" / "external-plugins";
#else
auto externalDir = homeDir / ".local" / "lib" / "yams" / "external-plugins";
#endif
std::filesystem::create_directories(externalDir);
auto externalPluginDir =
createMockExternalPlugin(externalDir, "dispatcher_external_default");
auto missingResp = dispatchRequest(
dispatcher, Request{PluginLoadRequest{"missing_default_plugin.so", "", false}});
REQUIRE(std::holds_alternative<ErrorResponse>(missingResp));
const auto& missingErr = std::get<ErrorResponse>(missingResp);
CHECK(missingErr.code == ErrorCode::NotFound);
CHECK(missingErr.message == "Plugin not found");
auto scanResp = dispatchRequest(dispatcher, Request{PluginScanRequest{}});
REQUIRE(std::holds_alternative<PluginScanResponse>(scanResp));
const auto& scan = std::get<PluginScanResponse>(scanResp);
bool foundDefaultPlugin = false;
bool foundDefaultExternalPlugin = false;
for (const auto& record : scan.plugins) {
if (record.path == fakePlugin.string()) {
foundDefaultPlugin = true;
CHECK(record.name == "yams_default_scan");
}
if (record.path == externalPluginDir.string()) {
foundDefaultExternalPlugin = true;
CHECK(record.name == "dispatcher_external_plugin");
}
}
CHECK(foundDefaultPlugin);
CHECK(foundDefaultExternalPlugin);
auto trustResp =
dispatchRequest(dispatcher, Request{PluginTrustAddRequest{abiDir.string()}});
REQUIRE(std::holds_alternative<SuccessResponse>(trustResp));
auto* external = svc->getExternalPluginHost();
REQUIRE(external != nullptr);
REQUIRE(external->trustAdd(externalDir).has_value());
PluginLoadRequest req;
req.pathOrName = fakePlugin.filename().string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Plugin load failed for: " + fakePlugin.string());
PluginLoadRequest externalReq;
externalReq.pathOrName = externalPluginDir.filename().string();
auto externalResp = dispatchRequest(dispatcher, Request{externalReq});
REQUIRE(std::holds_alternative<PluginLoadResponse>(externalResp));
const auto& externalLoad = std::get<PluginLoadResponse>(externalResp);
CHECK(externalLoad.loaded);
CHECK(externalLoad.record.name == "dispatcher_external_plugin");
CHECK(externalLoad.record.path == externalPluginDir.string());
auto externalUnloadResp =
dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
REQUIRE(std::holds_alternative<SuccessResponse>(externalUnloadResp));
}
SECTION("plugin handlers route external plugins when ABI host is absent") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
svc->__test_setAbiHost({});
auto pluginDir = createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external");
PluginScanRequest scanReq;
scanReq.target = pluginDir.string();
auto scanResp = dispatchRequest(dispatcher, Request{scanReq});
REQUIRE(std::holds_alternative<PluginScanResponse>(scanResp));
const auto& scan = std::get<PluginScanResponse>(scanResp);
REQUIRE(scan.plugins.size() == 1);
CHECK(scan.plugins.front().name == "dispatcher_external_plugin");
PluginLoadRequest dryRunReq;
dryRunReq.pathOrName = pluginDir.string();
dryRunReq.dryRun = true;
auto dryRunResp = dispatchRequest(dispatcher, Request{dryRunReq});
REQUIRE(std::holds_alternative<PluginLoadResponse>(dryRunResp));
const auto& dryRun = std::get<PluginLoadResponse>(dryRunResp);
CHECK_FALSE(dryRun.loaded);
CHECK(dryRun.message == "dry-run");
CHECK(dryRun.record.name == "dispatcher_external_plugin");
auto* external = svc->getExternalPluginHost();
REQUIRE(external != nullptr);
REQUIRE(external->trustAdd(pluginDir).has_value());
PluginLoadRequest loadReq;
loadReq.pathOrName = pluginDir.string();
auto loadResp = dispatchRequest(dispatcher, Request{loadReq});
REQUIRE(std::holds_alternative<PluginLoadResponse>(loadResp));
const auto& load = std::get<PluginLoadResponse>(loadResp);
CHECK(load.loaded);
CHECK(load.message == "loaded");
CHECK(load.record.name == "dispatcher_external_plugin");
CHECK(external->listLoaded().size() == 1);
auto unloadResp =
dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
CHECK(std::get<SuccessResponse>(unloadResp).message == "unloaded");
CHECK(external->listLoaded().empty());
}
SECTION("plugin load routes direct external files before ABI fallback") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
auto shimPath = makePythonShimPath();
ScopedEnvVar pythonGuard("PATH", shimPath.c_str());
auto pluginDir =
createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_py");
auto pluginFile = pluginDir / "plugin.py";
auto* external = svc->getExternalPluginHost();
REQUIRE(external != nullptr);
REQUIRE(external->trustAdd(pluginFile).has_value());
PluginLoadRequest req;
req.pathOrName = pluginFile.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PluginLoadResponse>(resp));
const auto& load = std::get<PluginLoadResponse>(resp);
CHECK(load.loaded);
CHECK(load.record.name == "dispatcher_external_plugin");
CHECK(load.record.path == pluginFile.string());
auto unloadResp =
dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
}
SECTION("plugin load recognizes manifest-adjacent executable files") {
#ifdef _WIN32
SKIP("manifest-adjacent executable file semantics are POSIX-specific");
#else
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
auto pluginDir =
createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_exec");
auto executable = pluginDir / "plugin_runner";
std::filesystem::copy_file(pluginDir / "plugin.py", executable,
std::filesystem::copy_options::overwrite_existing);
std::filesystem::permissions(executable,
std::filesystem::perms::owner_read |
std::filesystem::perms::owner_write |
std::filesystem::perms::owner_exec,
std::filesystem::perm_options::replace);
auto* external = svc->getExternalPluginHost();
REQUIRE(external != nullptr);
REQUIRE(external->trustAdd(executable).has_value());
PluginLoadRequest req;
req.pathOrName = executable.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<PluginLoadResponse>(resp));
const auto& load = std::get<PluginLoadResponse>(resp);
CHECK(load.loaded);
CHECK(load.record.name == "dispatcher_external_plugin");
CHECK(load.record.path == executable.string());
auto unloadResp =
dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
#endif
}
SECTION("plugin load falls back after untrusted external file fails") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
auto pluginDir =
createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_untrusted");
auto pluginFile = pluginDir / "plugin.py";
PluginLoadRequest req;
req.pathOrName = pluginFile.string();
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message == "Plugin load failed for: " + pluginFile.string());
}
SECTION("plugin trust add loads regular files and reuses loaded path metadata") {
auto [state, lifecycleFsm, svc] = makeReadyService();
StubLifecycle lifecycle;
RequestDispatcher dispatcher(&lifecycle, svc.get(), state.get());
auto shimPath = makePythonShimPath();
ScopedEnvVar pythonGuard("PATH", shimPath.c_str());
auto pluginDir =
createMockExternalPlugin(svc->getResolvedDataDir(), "dispatcher_external_trust_file");
auto pluginFile = pluginDir / "plugin.py";
auto trustAddResp =
dispatchRequest(dispatcher, Request{PluginTrustAddRequest{pluginFile.string()}});
REQUIRE(std::holds_alternative<SuccessResponse>(trustAddResp));
auto* external = svc->getExternalPluginHost();
REQUIRE(external != nullptr);
REQUIRE(waitForCondition([&] { return external->listLoaded().size() == 1; }));
CHECK(external->listLoaded().front().path == pluginFile);
auto secondTrustResp =
dispatchRequest(dispatcher, Request{PluginTrustAddRequest{pluginDir.string()}});
REQUIRE(std::holds_alternative<SuccessResponse>(secondTrustResp));
CHECK(waitForCondition([&] { return external->listLoaded().size() == 1; }));
auto unloadResp =
dispatchRequest(dispatcher, Request{PluginUnloadRequest{"dispatcher_external_plugin"}});
REQUIRE(std::holds_alternative<SuccessResponse>(unloadResp));
}
}
TEST_CASE("RequestDispatcher: collection handlers report missing dependencies and batch defaults",
"[daemon][collections][dispatcher]") {
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_collections_dispatcher_");
StubLifecycle lifecycle;
StateComponent state;
state.readiness.contentStoreReady.store(true, std::memory_order_relaxed);
state.readiness.metadataRepoReady.store(true, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
ServiceManager svc(cfg, state, lifecycleFsm);
RequestDispatcher dispatcher(&lifecycle, &svc, &state);
struct SeededDocument {
int64_t id{0};
std::string hash;
std::string relativePath;
};
auto makeReadyService = [&](StateComponent& readyState, DaemonLifecycleFsm& readyLifecycleFsm,
const std::string& prefix) {
ScopedEnvVar disableVectors("YAMS_DISABLE_VECTORS", "1");
ScopedEnvVar disableVectorDb("YAMS_DISABLE_VECTOR_DB", "1");
ScopedEnvVar skipModelLoading("YAMS_SKIP_MODEL_LOADING", "1");
ScopedEnvVar safeSingleInstance("YAMS_TEST_SAFE_SINGLE_INSTANCE", "1");
DaemonConfig readyCfg = cfg;
readyCfg.dataDir = makeTempDir(prefix);
auto readySvc = std::make_shared<ServiceManager>(readyCfg, readyState, readyLifecycleFsm);
auto initResult = readySvc->initialize();
REQUIRE(initResult.has_value());
readySvc->startAsyncInit();
auto serviceSnap = readySvc->waitForServiceManagerTerminalState(30);
REQUIRE(serviceSnap.state == ServiceManagerState::Ready);
REQUIRE(readySvc->getMetadataRepo() != nullptr);
REQUIRE(readySvc->getContentStore() != nullptr);
return readySvc;
};
auto seedDocument = [&](const std::shared_ptr<ServiceManager>& readySvc,
const std::string& relativePath, const std::string& text,
const std::string& collectionName, const std::string& snapshotId) {
auto meta = readySvc->getMetadataRepo();
auto store = readySvc->getContentStore();
REQUIRE(meta != nullptr);
REQUIRE(store != nullptr);
const auto textBytes = std::as_bytes(std::span<const char>(text.data(), text.size()));
auto storeRes = store->storeBytes(textBytes);
REQUIRE(storeRes.has_value());
std::filesystem::path relativeDocPath(relativePath);
std::string extension = relativeDocPath.extension().string();
if (!extension.empty() && extension.front() == '.') {
extension.erase(0, 1);
}
metadata::DocumentInfo doc{};
doc.fileName = relativeDocPath.filename().string();
doc.filePath = relativePath;
doc.fileExtension = extension;
doc.fileSize = static_cast<int64_t>(text.size());
doc.sha256Hash = storeRes.value().contentHash;
doc.mimeType = "text/plain";
const auto now =
std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());
doc.createdTime = now;
doc.modifiedTime = now;
doc.indexedTime = now;
auto idRes = meta->insertDocument(doc);
REQUIRE(idRes.has_value());
REQUIRE(
meta->setMetadata(idRes.value(), "collection", metadata::MetadataValue(collectionName))
.has_value());
REQUIRE(meta->setMetadata(idRes.value(), "snapshot_id", metadata::MetadataValue(snapshotId))
.has_value());
return SeededDocument{idRes.value(), storeRes.value().contentHash, relativePath};
};
SECTION("list snapshots returns internal error when metadata repo is unavailable") {
auto resp = dispatchRequest(dispatcher, Request{ListSnapshotsRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
}
SECTION("list snapshots forwards repository errors") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
repo->setSnapshotError(Error{ErrorCode::DatabaseError, "snapshot lookup failed"});
svc.__test_setMetadataRepo(repo);
auto resp = dispatchRequest(dispatcher, Request{ListSnapshotsRequest{}});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::DatabaseError);
CHECK(err.message == "snapshot lookup failed");
}
SECTION("restore collection returns internal error before dependencies are initialized") {
RestoreCollectionRequest req;
req.collection = "dispatcher-collection";
req.outputDirectory = "out";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
}
SECTION("restore collection returns internal error when content store is unavailable") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
svc.__test_setMetadataRepo(repo);
RestoreCollectionRequest req;
req.collection = "dispatcher-collection";
req.outputDirectory = "out";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Content store unavailable") != std::string::npos);
}
SECTION("restore collection forwards repository query errors") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
repo->setQueryError(Error{ErrorCode::DatabaseError, "collection query failed"});
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
RestoreCollectionRequest req;
req.collection = "dispatcher-collection";
req.outputDirectory = "out";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::DatabaseError);
CHECK(err.message == "Failed to find collection documents: collection query failed");
}
SECTION("restore snapshot returns internal error when metadata repo is unavailable") {
RestoreSnapshotRequest req;
req.snapshotId = "dispatcher-snapshot";
req.outputDirectory = "out";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
}
SECTION("restore snapshot returns internal error when content store is unavailable") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
svc.__test_setMetadataRepo(repo);
RestoreSnapshotRequest req;
req.snapshotId = "dispatcher-snapshot";
req.outputDirectory = "out";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Content store unavailable") != std::string::npos);
}
SECTION("restore snapshot forwards repository query errors") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
repo->setSnapshotDocumentsError(Error{ErrorCode::DatabaseError, "snapshot query failed"});
svc.__test_setMetadataRepo(repo);
svc.__test_setContentStore(std::make_shared<StubContentStore>());
RestoreSnapshotRequest req;
req.snapshotId = "dispatcher-snapshot";
req.outputDirectory = "out";
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::DatabaseError);
CHECK(err.message == "Failed to find snapshot documents: snapshot query failed");
}
SECTION("metadata value counts returns internal error when metadata repo is unavailable") {
MetadataValueCountsRequest req;
req.keys = {"collection"};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::InternalError);
CHECK(err.message.find("Metadata repository unavailable") != std::string::npos);
}
SECTION("metadata value counts forwards repository errors") {
auto repo = std::make_shared<StubPruneMetadataRepository>();
repo->setMetadataValueCountsError(Error{ErrorCode::DatabaseError, "counts failed"});
svc.__test_setMetadataRepo(repo);
MetadataValueCountsRequest req;
req.keys = {"collection"};
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<ErrorResponse>(resp));
const auto& err = std::get<ErrorResponse>(resp);
CHECK(err.code == ErrorCode::DatabaseError);
CHECK(err.message == "counts failed");
}
SECTION("batch request returns not implemented items for each sequence") {
BatchRequest req;
req.items.push_back(BatchItem{.sequenceId = 7, .request = SearchRequest{.query = "one"}});
req.items.push_back(BatchItem{.sequenceId = 8, .request = ListRequest{}});
auto resp = dispatchRequest(dispatcher, Request{req});
REQUIRE(std::holds_alternative<BatchResponse>(resp));
const auto& batchResp = std::get<BatchResponse>(resp);
CHECK(batchResp.totalCount == 2);
CHECK(batchResp.successCount == 0);
CHECK(batchResp.errorCount == 2);
REQUIRE(batchResp.items.size() == 2);
CHECK(batchResp.items[0].sequenceId == 7);
CHECK_FALSE(batchResp.items[0].success);
REQUIRE(std::holds_alternative<ErrorResponse>(batchResp.items[0].response));
CHECK(std::get<ErrorResponse>(batchResp.items[0].response).code ==
ErrorCode::NotImplemented);
CHECK(batchResp.items[1].sequenceId == 8);
CHECK_FALSE(batchResp.items[1].success);
}
SECTION("restore collection exercises regex, layout, and io branches") {
StubLifecycle readyLifecycle;
StateComponent readyState;
DaemonLifecycleFsm readyLifecycleFsm;
auto readySvc =
makeReadyService(readyState, readyLifecycleFsm, "yams_collections_live_dispatcher_");
RequestDispatcher readyDispatcher(&readyLifecycle, readySvc.get(), &readyState);
const std::string collectionName = "dispatcher-live-collection";
const std::string snapshotId = "dispatcher-live-snapshot";
const auto seeded = seedDocument(readySvc, "nested/alpha.txt", "alpha collection text\n",
collectionName, snapshotId);
auto includeOut = makeTempDir("yams_collection_include_");
RestoreCollectionRequest includeReq;
includeReq.collection = collectionName;
includeReq.outputDirectory = includeOut.string();
includeReq.includePatterns = {"*alpha.txt"};
includeReq.dryRun = true;
auto includeResp = dispatchRequest(readyDispatcher, Request{includeReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(includeResp));
const auto& included = std::get<RestoreCollectionResponse>(includeResp);
REQUIRE(included.files.size() == 1);
CHECK(included.filesRestored == 1);
CHECK(included.files.front().path == (includeOut / seeded.relativePath).string());
RestoreCollectionRequest invalidPatternReq;
invalidPatternReq.collection = collectionName;
invalidPatternReq.outputDirectory = includeOut.string();
invalidPatternReq.includePatterns = {"["};
invalidPatternReq.dryRun = true;
auto invalidPatternResp = dispatchRequest(readyDispatcher, Request{invalidPatternReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(invalidPatternResp));
const auto& invalidPattern = std::get<RestoreCollectionResponse>(invalidPatternResp);
CHECK(invalidPattern.filesRestored == 0);
CHECK(invalidPattern.files.empty());
RestoreCollectionRequest excludeReq;
excludeReq.collection = collectionName;
excludeReq.outputDirectory = includeOut.string();
excludeReq.excludePatterns = {"*alpha.txt"};
excludeReq.dryRun = true;
auto excludeResp = dispatchRequest(readyDispatcher, Request{excludeReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(excludeResp));
const auto& excluded = std::get<RestoreCollectionResponse>(excludeResp);
CHECK(excluded.filesRestored == 0);
CHECK(excluded.files.empty());
auto writeOut = makeTempDir("yams_collection_write_");
RestoreCollectionRequest writeReq;
writeReq.collection = collectionName;
writeReq.outputDirectory = writeOut.string();
writeReq.layoutTemplate = "deep/{hash}{ext}";
writeReq.overwrite = true;
writeReq.dryRun = false;
auto writeResp = dispatchRequest(readyDispatcher, Request{writeReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(writeResp));
const auto& written = std::get<RestoreCollectionResponse>(writeResp);
REQUIRE(written.files.size() == 1);
CHECK(written.filesRestored == 1);
const auto writtenPath = writeOut / "deep" / (seeded.hash + ".txt");
CHECK(std::filesystem::path(written.files.front().path).lexically_normal() ==
writtenPath.lexically_normal());
REQUIRE(std::filesystem::exists(writtenPath));
std::ifstream writtenFile(writtenPath, std::ios::binary);
REQUIRE(writtenFile.good());
std::string writtenText((std::istreambuf_iterator<char>(writtenFile)),
std::istreambuf_iterator<char>());
CHECK(writtenText == "alpha collection text\n");
auto createDirBlocker = makeTempDir("yams_collection_create_dir_blocker_") / "blocked";
{
std::ofstream blocker(createDirBlocker);
REQUIRE(blocker.good());
blocker << "blocker";
}
RestoreCollectionRequest createDirFailReq;
createDirFailReq.collection = collectionName;
createDirFailReq.outputDirectory = createDirBlocker.string();
createDirFailReq.createDirs = true;
createDirFailReq.overwrite = true;
createDirFailReq.dryRun = false;
auto createDirFailResp = dispatchRequest(readyDispatcher, Request{createDirFailReq});
REQUIRE(std::holds_alternative<ErrorResponse>(createDirFailResp));
CHECK(std::get<ErrorResponse>(createDirFailResp).code == ErrorCode::IOError);
CHECK(std::get<ErrorResponse>(createDirFailResp)
.message.find("Failed to create output directory") != std::string::npos);
auto parentFailOut = makeTempDir("yams_collection_parent_fail_");
auto occupiedParent = parentFailOut / "occupied";
{
std::ofstream blocker(occupiedParent);
REQUIRE(blocker.good());
blocker << "occupied";
}
RestoreCollectionRequest parentFailReq;
parentFailReq.collection = collectionName;
parentFailReq.outputDirectory = parentFailOut.string();
parentFailReq.layoutTemplate = "occupied/restored.txt";
parentFailReq.overwrite = true;
parentFailReq.dryRun = false;
auto parentFailResp = dispatchRequest(readyDispatcher, Request{parentFailReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(parentFailResp));
const auto& parentFailed = std::get<RestoreCollectionResponse>(parentFailResp);
REQUIRE(parentFailed.files.size() == 1);
CHECK(parentFailed.filesRestored == 0);
CHECK(parentFailed.files.front().skipped);
CHECK(parentFailed.files.front().skipReason.find("Failed to create parent directory") !=
std::string::npos);
auto openFailOut = makeTempDir("yams_collection_open_fail_");
RestoreCollectionRequest openFailReq;
openFailReq.collection = collectionName;
openFailReq.outputDirectory = openFailOut.string();
openFailReq.layoutTemplate = ".";
openFailReq.overwrite = true;
openFailReq.dryRun = false;
auto openFailResp = dispatchRequest(readyDispatcher, Request{openFailReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(openFailResp));
const auto& openFailed = std::get<RestoreCollectionResponse>(openFailResp);
REQUIRE(openFailed.files.size() == 1);
CHECK(openFailed.filesRestored == 0);
CHECK(openFailed.files.front().skipped);
CHECK(openFailed.files.front().skipReason == "Failed to open output file");
auto writeFailOut = makeTempDir("yams_collection_write_fail_");
RestoreCollectionRequest writeFailReq;
writeFailReq.collection = collectionName;
writeFailReq.outputDirectory = writeFailOut.string();
writeFailReq.layoutTemplate = "write-fail/{name}{ext}";
writeFailReq.overwrite = true;
writeFailReq.dryRun = false;
RequestDispatcher::__test_forceRestoreWriteFailureOnce();
auto writeFailResp = dispatchRequest(readyDispatcher, Request{writeFailReq});
REQUIRE(std::holds_alternative<RestoreCollectionResponse>(writeFailResp));
const auto& writeFailed = std::get<RestoreCollectionResponse>(writeFailResp);
REQUIRE(writeFailed.files.size() == 1);
CHECK(writeFailed.filesRestored == 0);
CHECK(writeFailed.files.front().skipped);
CHECK(writeFailed.files.front().skipReason == "Failed to write file content");
readySvc->shutdown();
}
SECTION("restore snapshot exercises regex, layout, and io branches") {
StubLifecycle readyLifecycle;
StateComponent readyState;
DaemonLifecycleFsm readyLifecycleFsm;
auto readySvc =
makeReadyService(readyState, readyLifecycleFsm, "yams_snapshots_live_dispatcher_");
RequestDispatcher readyDispatcher(&readyLifecycle, readySvc.get(), &readyState);
const std::string collectionName = "dispatcher-live-collection";
const std::string snapshotId = "dispatcher-live-snapshot";
const auto seeded = seedDocument(readySvc, "nested/alpha.txt", "alpha snapshot text\n",
collectionName, snapshotId);
auto defaultOut = makeTempDir("yams_snapshot_default_");
RestoreSnapshotRequest defaultReq;
defaultReq.snapshotId = snapshotId;
defaultReq.outputDirectory = defaultOut.string();
defaultReq.includePatterns = {"*alpha.txt"};
defaultReq.dryRun = true;
auto defaultResp = dispatchRequest(readyDispatcher, Request{defaultReq});
REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(defaultResp));
const auto& defaultRestore = std::get<RestoreSnapshotResponse>(defaultResp);
REQUIRE(defaultRestore.files.size() == 1);
CHECK(defaultRestore.filesRestored == 1);
CHECK(defaultRestore.files.front().path == (defaultOut / seeded.relativePath).string());
RestoreSnapshotRequest invalidPatternReq;
invalidPatternReq.snapshotId = snapshotId;
invalidPatternReq.outputDirectory = defaultOut.string();
invalidPatternReq.includePatterns = {"["};
invalidPatternReq.dryRun = true;
auto invalidPatternResp = dispatchRequest(readyDispatcher, Request{invalidPatternReq});
REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(invalidPatternResp));
const auto& invalidPattern = std::get<RestoreSnapshotResponse>(invalidPatternResp);
CHECK(invalidPattern.filesRestored == 0);
CHECK(invalidPattern.files.empty());
auto writeOut = makeTempDir("yams_snapshot_write_");
RestoreSnapshotRequest writeReq;
writeReq.snapshotId = snapshotId;
writeReq.outputDirectory = writeOut.string();
writeReq.layoutTemplate = "deep/{name}{ext}";
writeReq.overwrite = true;
writeReq.dryRun = false;
auto writeResp = dispatchRequest(readyDispatcher, Request{writeReq});
REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(writeResp));
const auto& written = std::get<RestoreSnapshotResponse>(writeResp);
REQUIRE(written.files.size() == 1);
CHECK(written.filesRestored == 1);
const auto writtenPath = writeOut / "deep" / "alpha.txt";
CHECK(std::filesystem::path(written.files.front().path).lexically_normal() ==
writtenPath.lexically_normal());
REQUIRE(std::filesystem::exists(writtenPath));
std::ifstream writtenFile(writtenPath, std::ios::binary);
REQUIRE(writtenFile.good());
std::string writtenText((std::istreambuf_iterator<char>(writtenFile)),
std::istreambuf_iterator<char>());
CHECK(writtenText == "alpha snapshot text\n");
auto createDirBlocker = makeTempDir("yams_snapshot_create_dir_blocker_") / "blocked";
{
std::ofstream blocker(createDirBlocker);
REQUIRE(blocker.good());
blocker << "blocker";
}
RestoreSnapshotRequest createDirFailReq;
createDirFailReq.snapshotId = snapshotId;
createDirFailReq.outputDirectory = createDirBlocker.string();
createDirFailReq.createDirs = true;
createDirFailReq.overwrite = true;
createDirFailReq.dryRun = false;
auto createDirFailResp = dispatchRequest(readyDispatcher, Request{createDirFailReq});
REQUIRE(std::holds_alternative<ErrorResponse>(createDirFailResp));
CHECK(std::get<ErrorResponse>(createDirFailResp).code == ErrorCode::IOError);
CHECK(std::get<ErrorResponse>(createDirFailResp)
.message.find("Failed to create output directory") != std::string::npos);
auto parentFailOut = makeTempDir("yams_snapshot_parent_fail_");
auto occupiedParent = parentFailOut / "occupied";
{
std::ofstream blocker(occupiedParent);
REQUIRE(blocker.good());
blocker << "occupied";
}
RestoreSnapshotRequest parentFailReq;
parentFailReq.snapshotId = snapshotId;
parentFailReq.outputDirectory = parentFailOut.string();
parentFailReq.layoutTemplate = "occupied/restored.txt";
parentFailReq.overwrite = true;
parentFailReq.dryRun = false;
auto parentFailResp = dispatchRequest(readyDispatcher, Request{parentFailReq});
REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(parentFailResp));
const auto& parentFailed = std::get<RestoreSnapshotResponse>(parentFailResp);
REQUIRE(parentFailed.files.size() == 1);
CHECK(parentFailed.filesRestored == 0);
CHECK(parentFailed.files.front().skipped);
CHECK(parentFailed.files.front().skipReason.find("Failed to create parent directory") !=
std::string::npos);
auto openFailOut = makeTempDir("yams_snapshot_open_fail_");
RestoreSnapshotRequest openFailReq;
openFailReq.snapshotId = snapshotId;
openFailReq.outputDirectory = openFailOut.string();
openFailReq.layoutTemplate = ".";
openFailReq.overwrite = true;
openFailReq.dryRun = false;
auto openFailResp = dispatchRequest(readyDispatcher, Request{openFailReq});
REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(openFailResp));
const auto& openFailed = std::get<RestoreSnapshotResponse>(openFailResp);
REQUIRE(openFailed.files.size() == 1);
CHECK(openFailed.filesRestored == 0);
CHECK(openFailed.files.front().skipped);
CHECK(openFailed.files.front().skipReason == "Failed to open output file");
auto writeFailOut = makeTempDir("yams_snapshot_write_fail_");
RestoreSnapshotRequest writeFailReq;
writeFailReq.snapshotId = snapshotId;
writeFailReq.outputDirectory = writeFailOut.string();
writeFailReq.layoutTemplate = "write-fail/{hash}{ext}";
writeFailReq.overwrite = true;
writeFailReq.dryRun = false;
RequestDispatcher::__test_forceRestoreWriteFailureOnce();
auto writeFailResp = dispatchRequest(readyDispatcher, Request{writeFailReq});
REQUIRE(std::holds_alternative<RestoreSnapshotResponse>(writeFailResp));
const auto& writeFailed = std::get<RestoreSnapshotResponse>(writeFailResp);
REQUIRE(writeFailed.files.size() == 1);
CHECK(writeFailed.filesRestored == 0);
CHECK(writeFailed.files.front().skipped);
CHECK(writeFailed.files.front().skipReason == "Failed to write file content");
readySvc->shutdown();
}
}
TEST_CASE("DaemonMetrics: snapshot includes canonical readiness flags",
"[daemon][metrics][readiness]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_metrics_readiness_flags_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
const std::array<std::string_view, 12> requiredCoreReadinessKeys = {
readiness::kIpcServer, readiness::kContentStore,
readiness::kDatabase, readiness::kMetadataRepo,
readiness::kSearchEngine, readiness::kModelProvider,
readiness::kVectorIndex, readiness::kVectorDb,
readiness::kPlugins, readiness::kVectorDbInitAttempted,
readiness::kVectorDbReady, readiness::kVectorDbDim};
for (const auto key : requiredCoreReadinessKeys) {
INFO("missing readiness key: " << key);
REQUIRE(snap->readinessStates.count(std::string(key)) > 0);
}
}
TEST_CASE("DaemonMetrics: snapshot reports generated build version", "[daemon][metrics][version]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_metrics_version_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
CHECK(snap->version == yams::version::string_v);
}
TEST_CASE("DaemonMetrics: cold snapshot refreshes synchronously", "[daemon][metrics][freshness]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_metrics_freshness_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
CHECK_FALSE(snap->statusSnapshotStale);
CHECK(snap->statusSnapshotAgeMs == 0);
CHECK(snap->statusDetailAgeMs == 0);
CHECK(snap->statusDetailStale);
CHECK_FALSE(snap->statusDetailAvailable);
}
TEST_CASE("RequestDispatcher: status refreshes cold snapshot without retry hint",
"[daemon][status][freshness]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_busy_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
RequestDispatcher dispatcher(nullptr, &svc, &state, &metrics);
auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& status = std::get<StatusResponse>(resp);
CHECK(status.running);
CHECK(status.retryAfterMs == 0);
}
TEST_CASE("RequestDispatcher: cold status still reports active repair work",
"[daemon][status][freshness]") {
StateComponent state;
state.stats.repairRunning.store(true, std::memory_order_relaxed);
state.stats.repairQueueDepth.store(7, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_busy_repair_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
RequestDispatcher dispatcher(nullptr, &svc, &state, &metrics);
auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& status = std::get<StatusResponse>(resp);
CHECK(status.retryAfterMs == 0);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairRunning)) == 1);
REQUIRE(status.requestCounts.count(std::string(metrics::kRepairQueueDepth)) == 1);
CHECK(status.requestCounts.at(std::string(metrics::kRepairRunning)) == 1);
CHECK(status.requestCounts.at(std::string(metrics::kRepairQueueDepth)) == 7);
}
TEST_CASE("RequestDispatcher: status mirrors memory breakdown into requestCounts",
"[daemon][status][memory]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_status_mem_breakdown_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
RequestDispatcher dispatcher(nullptr, &svc, &state, &metrics);
auto resp = dispatchRequest(dispatcher, Request{StatusRequest{}});
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& status = std::get<StatusResponse>(resp);
REQUIRE(status.requestCounts.count("status_mem_rss_bytes") == 1);
CHECK(status.requestCounts.at("status_mem_rss_bytes") > 0);
#if defined(__APPLE__)
REQUIRE(status.requestCounts.count("status_mem_phys_footprint_bytes") == 1);
const auto footprintBytes = status.requestCounts.at("status_mem_phys_footprint_bytes");
CHECK(footprintBytes > 0);
const auto expectedMb = static_cast<double>(footprintBytes) / (1024.0 * 1024.0);
CHECK(std::abs(status.memoryUsageMb - expectedMb) < std::max(1.0, expectedMb * 0.05));
#endif
}
TEST_CASE("DaemonMetrics: repair idle and busy counters export from state",
"[daemon][metrics][repair]") {
StateComponent state;
state.stats.repairIdleTicks.store(3, std::memory_order_relaxed);
state.stats.repairBusyTicks.store(9, std::memory_order_relaxed);
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_repair_counter_export_");
ServiceManager svc(cfg, state, lifecycleFsm);
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
CHECK(snap->repairIdleTicks == 3);
CHECK(snap->repairBusyTicks == 9);
}
TEST_CASE("StatusResponse: post_ingest_rpc requestCounts keys round-trip",
"[daemon][status][protocol][post_ingest]") {
StatusResponse s{};
s.requestCounts["post_ingest_rpc_queued"] = 3;
s.requestCounts["post_ingest_rpc_capacity"] = 64;
s.requestCounts["post_ingest_rpc_max_per_batch"] = 8;
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const auto& resp = std::get<Response>(dec.value().payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& decoded = std::get<StatusResponse>(resp);
REQUIRE(decoded.requestCounts.at("post_ingest_rpc_queued") == 3);
REQUIRE(decoded.requestCounts.at("post_ingest_rpc_capacity") == 64);
REQUIRE(decoded.requestCounts.at("post_ingest_rpc_max_per_batch") == 8);
}
TEST_CASE("StatusResponse: maintenance phase fields round-trip",
"[daemon][status][protocol][maintenance]") {
StatusResponse s{};
s.databasePhase = "ready";
s.databasePhaseElapsedMs = 11;
s.maintenancePhase = "salvaging";
s.maintenancePhaseElapsedMs = 222;
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const auto& resp = std::get<Response>(dec.value().payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& decoded = std::get<StatusResponse>(resp);
CHECK(decoded.databasePhase == "ready");
CHECK(decoded.databasePhaseElapsedMs == 11);
CHECK(decoded.maintenancePhase == "salvaging");
CHECK(decoded.maintenancePhaseElapsedMs == 222);
}
TEST_CASE("StatusResponse: freshness requestCounts keys round-trip",
"[daemon][status][protocol][freshness]") {
StatusResponse s{};
s.requestCounts["status_snapshot_age_ms"] = 42;
s.requestCounts["status_snapshot_stale"] = 0;
s.requestCounts["status_detail_age_ms"] = 17;
s.requestCounts["status_detail_stale"] = 1;
s.requestCounts["status_detail_available"] = 1;
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const auto& resp = std::get<Response>(dec.value().payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& decoded = std::get<StatusResponse>(resp);
REQUIRE(decoded.requestCounts.at("status_snapshot_age_ms") == 42);
REQUIRE(decoded.requestCounts.at("status_snapshot_stale") == 0);
REQUIRE(decoded.requestCounts.at("status_detail_age_ms") == 17);
REQUIRE(decoded.requestCounts.at("status_detail_stale") == 1);
REQUIRE(decoded.requestCounts.at("status_detail_available") == 1);
}
TEST_CASE("StatusResponse: backpressure requestCounts keys round-trip",
"[daemon][status][protocol][post_ingest]") {
StatusResponse s{};
s.requestCounts["post_ingest_backpressure_rejects"] = 123;
s.requestCounts["kg_jobs_depth"] = 3880;
s.requestCounts["kg_jobs_capacity"] = 4096;
s.requestCounts["kg_jobs_fill_pct"] = 95;
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const auto& resp = std::get<Response>(dec.value().payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& decoded = std::get<StatusResponse>(resp);
REQUIRE(decoded.requestCounts.at("post_ingest_backpressure_rejects") == 123);
REQUIRE(decoded.requestCounts.at("kg_jobs_depth") == 3880);
REQUIRE(decoded.requestCounts.at("kg_jobs_capacity") == 4096);
REQUIRE(decoded.requestCounts.at("kg_jobs_fill_pct") == 95);
}
TEST_CASE("StatusResponse: memory breakdown requestCounts keys round-trip",
"[daemon][status][protocol][memory]") {
StatusResponse s{};
s.requestCounts["status_mem_rss_bytes"] = 123456;
s.requestCounts["status_mem_phys_footprint_bytes"] = 654321;
s.requestCounts["status_mem_malloc_allocated_bytes"] = 777777;
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const auto& resp = std::get<Response>(dec.value().payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& decoded = std::get<StatusResponse>(resp);
REQUIRE(decoded.requestCounts.at("status_mem_rss_bytes") == 123456);
REQUIRE(decoded.requestCounts.at("status_mem_phys_footprint_bytes") == 654321);
REQUIRE(decoded.requestCounts.at("status_mem_malloc_allocated_bytes") == 777777);
}
TEST_CASE("AbiPluginHost: failed plugin loads are retained in last scan skips",
"[daemon][plugin][status][load-failure]") {
#if defined(__APPLE__)
const std::filesystem::path pluginPath{"/opt/homebrew/lib/yams/plugins/yams_glint.dylib"};
if (!std::filesystem::exists(pluginPath)) {
SKIP("Installed yams_glint plugin not present");
}
AbiPluginHost host(nullptr);
const auto trustFile = makeTempDir("yams_glint_skip_status_") / "plugins.trust";
host.setTrustFile(trustFile);
REQUIRE(host.trustAdd(pluginPath.parent_path()));
const auto loadResult = host.load(pluginPath, "{}");
if (loadResult) {
SUCCEED("yams_glint loaded successfully on this machine");
REQUIRE(host.unload(loadResult.value().name));
return;
}
const auto skips = host.getLastScanSkips();
REQUIRE_FALSE(skips.empty());
const auto it = std::find_if(skips.begin(), skips.end(),
[&](const auto& entry) { return entry.first == pluginPath; });
REQUIRE(it != skips.end());
CHECK((it->second.find("dlopen failed") != std::string::npos ||
it->second.find("preflight failed") != std::string::npos ||
it->second.find("Plugin init failed") != std::string::npos));
#else
SUCCEED("macOS-specific plugin load retention test");
#endif
}
// =============================================================================
// StatusResponse Protocol Serialization Tests
// =============================================================================
TEST_CASE("StatusResponse: Protocol serialization", "[daemon][status][protocol]") {
SECTION("Round-trip with lifecycle state and error") {
StatusResponse s{};
s.running = true;
s.ready = false;
s.uptimeSeconds = 42;
s.requestsProcessed = 7;
s.activeConnections = 1;
s.memoryUsageMb = 12.5;
s.cpuUsagePercent = 3.2;
s.version = "test";
s.overallStatus = "failed";
s.lifecycleState = "starting";
s.lastError = "boom";
s.dataDir = "/tmp/daemon-data-dir";
s.metadataDbPath = "/tmp/daemon-data-dir/yams.db";
s.vectorDbPath = "/tmp/daemon-data-dir/vectors.db";
s.requestCounts["worker_threads"] = 4;
// FSM-exported fields
s.requestCounts["service_fsm_state"] = 3;
s.requestCounts["embedding_state"] = 2;
s.requestCounts["plugin_host_state"] = 1;
// Readiness states
s.readinessStates["embedding_ready"] = false;
s.readinessStates["plugins_ready"] = true;
s.readinessStates["plugins_degraded"] = true;
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
// Encode
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
// Decode
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const Message& out = dec.value();
REQUIRE(std::holds_alternative<Response>(out.payload));
const auto& resp = std::get<Response>(out.payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
const auto& decoded = std::get<StatusResponse>(resp);
// Verify key fields survived round-trip
REQUIRE(decoded.running == true);
REQUIRE(decoded.ready == false);
REQUIRE(decoded.uptimeSeconds == 42);
REQUIRE(decoded.version == "test");
REQUIRE(decoded.overallStatus == "failed");
REQUIRE(decoded.lastError == "boom");
REQUIRE(decoded.dataDir == "/tmp/daemon-data-dir");
REQUIRE(decoded.metadataDbPath == "/tmp/daemon-data-dir/yams.db");
REQUIRE(decoded.vectorDbPath == "/tmp/daemon-data-dir/vectors.db");
// Verify FSM state counts
REQUIRE(decoded.requestCounts.at("worker_threads") == 4);
REQUIRE(decoded.requestCounts.at("service_fsm_state") == 3);
// Verify readiness flags
REQUIRE(decoded.readinessStates.at("embedding_ready") == false);
REQUIRE(decoded.readinessStates.at("plugins_ready") == true);
REQUIRE(decoded.readinessStates.at("plugins_degraded") == true);
}
SECTION("Empty status response serializes correctly") {
StatusResponse s{};
Message m{};
m.payload = Response{std::in_place_type<StatusResponse>, s};
auto enc = ProtoSerializer::encode_payload(m);
REQUIRE(enc.has_value());
auto dec = ProtoSerializer::decode_payload(enc.value());
REQUIRE(dec.has_value());
const auto& resp = std::get<Response>(dec.value().payload);
REQUIRE(std::holds_alternative<StatusResponse>(resp));
}
}
// =============================================================================
// Plugin Degradation Tests
// =============================================================================
// Note: Full integration tests (DISABLED in original) are skipped here
// These would test daemon start with trust failures and recovery hooks
// They belong in integration tests due to full daemon lifecycle requirements
TEST_CASE("ServiceManager: Plugin degradation tracking", "[daemon][status][plugins]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_plugin_degrade_");
SECTION("Plugin degradation flag is tracked in FSM") {
// This is a unit test of the FSM tracking mechanism
// Full daemon integration tests are separate
ServiceManager svc(cfg, state, lifecycleFsm);
// ServiceManager should track plugin states via FSM
// Actual degradation testing requires daemon start, which is integration level
// Basic verification: ServiceManager exists and has FSM snapshot capability
auto snapshot = svc.getServiceManagerFsmSnapshot();
// ServiceManager starts uninitialized until explicitly initialized
REQUIRE(snapshot.state == ServiceManagerState::Uninitialized);
}
}
// =============================================================================
// FSM State Metrics
// =============================================================================
TEST_CASE("DaemonMetrics: FSM state export", "[daemon][metrics][fsm]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_fsm_metrics_");
ServiceManager svc(cfg, state, lifecycleFsm);
SECTION("Lifecycle FSM state is tracked") {
lifecycleFsm.dispatch(BootstrappedEvent{});
auto snapshot = lifecycleFsm.snapshot();
REQUIRE(snapshot.state != LifecycleState::Unknown);
}
SECTION("Service manager FSM state is accessible") {
auto snapshot = svc.getServiceManagerFsmSnapshot();
// ServiceManager starts in Uninitialized state until initialized
REQUIRE(snapshot.state == ServiceManagerState::Uninitialized);
}
SECTION("Embedding provider FSM state is accessible") {
auto snapshot = svc.getEmbeddingProviderFsmSnapshot();
// Embedding provider starts in Unavailable state until plugin loaded
REQUIRE(snapshot.state == EmbeddingProviderState::Unavailable);
}
SECTION("Plugin host FSM state is accessible") {
auto snapshot = svc.getPluginHostFsmSnapshot();
// When ABI plugins are disabled, PluginManager initializes successfully with
// an empty plugin list, transitioning to Ready state. When plugins are enabled
// but PluginManager hasn't been created yet, the fallback is NotInitialized.
REQUIRE((snapshot.state == PluginHostState::NotInitialized ||
snapshot.state == PluginHostState::Ready));
}
}
// =============================================================================
// DatabaseManager Metrics Tests
// =============================================================================
TEST_CASE("DatabaseManager: Metrics tracking", "[daemon][metrics][database]") {
StateComponent state;
DatabaseManager::Dependencies deps{.state = &state};
DatabaseManager dbm(deps);
SECTION("Stats start at zero") {
const auto& stats = dbm.getStats();
REQUIRE(stats.openDurationMs.load() == 0);
REQUIRE(stats.migrationDurationMs.load() == 0);
REQUIRE(stats.openErrors.load() == 0);
REQUIRE(stats.migrationErrors.load() == 0);
REQUIRE(stats.repositoryInitErrors.load() == 0);
}
SECTION("Stats are returned by reference") {
// Verify we can access stats multiple times
const auto& stats1 = dbm.getStats();
const auto& stats2 = dbm.getStats();
REQUIRE(stats1.openErrors.load() == stats2.openErrors.load());
}
}
TEST_CASE("DaemonMetrics: DatabaseManager metrics export", "[daemon][metrics][database]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_db_metrics_");
ServiceManager svc(cfg, state, lifecycleFsm);
SECTION("DatabaseManager metrics are included in snapshot") {
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
// DatabaseManager metrics should be present (may be 0 if not initialized)
REQUIRE(snap->dbOpenErrors == 0); // Default value when no errors
}
}
// =============================================================================
// WorkCoordinator Metrics Tests
// =============================================================================
TEST_CASE("WorkCoordinator: Metrics tracking", "[daemon][metrics][work]") {
WorkCoordinator wc;
SECTION("Stats before start") {
auto stats = wc.getStats();
REQUIRE(stats.workerCount == 0);
REQUIRE(stats.activeWorkers == 0);
REQUIRE(stats.isRunning == false);
}
SECTION("Stats accessors don't crash") {
REQUIRE(wc.getWorkerCount() == 0);
REQUIRE(wc.getActiveWorkerCount() == 0);
REQUIRE(wc.isRunning() == false);
}
}
TEST_CASE("DaemonMetrics: WorkCoordinator metrics export", "[daemon][metrics][work]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_wc_metrics_");
ServiceManager svc(cfg, state, lifecycleFsm);
SECTION("WorkCoordinator metrics are included in snapshot") {
DaemonMetrics metrics(nullptr, &state, &svc, svc.getWorkCoordinator());
auto snap = metrics.getSnapshot();
REQUIRE(snap != nullptr);
// WorkCoordinator metrics should be present
// ServiceManager creates WorkCoordinator with hardware_concurrency() workers
REQUIRE(snap->workerThreads > 0); // Should have workers (worker_threads)
REQUIRE(snap->workCoordinatorRunning == true); // Should be running
}
}
// =============================================================================
// Search Load Metrics Tests
// =============================================================================
TEST_CASE("ServiceManager: Search load metrics", "[daemon][metrics][search]") {
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
DaemonConfig cfg;
cfg.dataDir = makeTempDir("yams_search_metrics_");
ServiceManager svc(cfg, state, lifecycleFsm);
SECTION("Search load metrics return valid structure") {
auto metrics = svc.getSearchLoadMetrics();
REQUIRE(metrics.active == 0);
REQUIRE(metrics.queued == 0);
REQUIRE(metrics.concurrencyLimit > 0);
}
SECTION("Search lifecycle counters update load metrics") {
svc.onSearchRequestQueued();
auto queued = svc.getSearchLoadMetrics();
REQUIRE(queued.queued == 1);
REQUIRE(queued.active == 0);
const auto cap = queued.concurrencyLimit;
REQUIRE(svc.tryStartSearchRequest(cap));
auto started = svc.getSearchLoadMetrics();
REQUIRE(started.queued == 0);
REQUIRE(started.active == 1);
svc.onSearchRequestFinished();
auto finished = svc.getSearchLoadMetrics();
REQUIRE(finished.active == 0);
REQUIRE(finished.queued == 0);
}
SECTION("Failed search start keeps queued count until rejection") {
// Pin cap to 1 so the second start is guaranteed to fail regardless of the
// machine-derived concurrencyLimit (varies with CPU cores).
constexpr std::uint32_t cap = 1;
svc.onSearchRequestQueued();
REQUIRE(svc.tryStartSearchRequest(cap));
svc.onSearchRequestQueued();
REQUIRE_FALSE(svc.tryStartSearchRequest(cap));
auto waiting = svc.getSearchLoadMetrics();
REQUIRE(waiting.active == 1);
REQUIRE(waiting.queued == 1);
svc.onSearchRequestRejected();
auto rejected = svc.getSearchLoadMetrics();
REQUIRE(rejected.active == 1);
REQUIRE(rejected.queued == 0);
svc.onSearchRequestFinished();
auto finished = svc.getSearchLoadMetrics();
REQUIRE(finished.active == 0);
REQUIRE(finished.queued == 0);
}
}
// =============================================================================
// FsmMetricsRegistry Integration Tests
// =============================================================================
TEST_CASE("FsmMetricsRegistry: Metrics collection", "[daemon][metrics][fsm]") {
// Reset registry to known state
FsmMetricsRegistry::instance().reset();
SECTION("Registry starts with zeros") {
auto snap = FsmMetricsRegistry::instance().snapshot();
REQUIRE(snap.transitions == 0);
REQUIRE(snap.headerReads == 0);
REQUIRE(snap.payloadReads == 0);
REQUIRE(snap.payloadWrites == 0);
REQUIRE(snap.bytesSent == 0);
REQUIRE(snap.bytesReceived == 0);
}
SECTION("Registry increments correctly") {
FsmMetricsRegistry::instance().incrementTransitions(5);
FsmMetricsRegistry::instance().incrementHeaderReads(3);
FsmMetricsRegistry::instance().incrementPayloadReads(2);
FsmMetricsRegistry::instance().addBytesSent(100);
FsmMetricsRegistry::instance().addBytesReceived(200);
auto snap = FsmMetricsRegistry::instance().snapshot();
REQUIRE(snap.transitions == 5);
REQUIRE(snap.headerReads == 3);
REQUIRE(snap.payloadReads == 2);
REQUIRE(snap.bytesSent == 100);
REQUIRE(snap.bytesReceived == 200);
}
SECTION("Registry reset works") {
FsmMetricsRegistry::instance().incrementTransitions(10);
FsmMetricsRegistry::instance().reset();
auto snap = FsmMetricsRegistry::instance().snapshot();
REQUIRE(snap.transitions == 0);
}
}