diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-038/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-038/README.md new file mode 100644 index 0000000..9b43d12 --- /dev/null +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-038/README.md @@ -0,0 +1,66 @@ +# DGR-038 evidence — isolated shard-local Hot KV State + +**Date:** 2026-08-01 +**Authority:** `.scratch/distributed-gguf-runtime/prd.json` (`passes` remains +`false` until the opt-in real-model concurrency lane runs). + +## Implemented + +- The native `LlamaShardEngine` now creates one bounded llama.cpp context and + assigns a distinct `llama_seq_id` to each `(route_session_id, route_epoch)`. + It never accepts remote KV data; the loaded, range-attested llama model owns + the local cache layout and layers. +- Prefill/decode append state tracks local positions and expected past length. + A re-prefill at an earlier position truncates only that sequence with + `llama_memory_seq_rm`; a discontinuity or past-length mismatch returns a + retryable `CACHE_MISS`. Older route epochs return `EPOCH_STALE`. +- The token-reservation budget is bounded by per-session context, total Hot KV + budget, maximum sequence count, TTL, and LRU. Release, superseding epoch, + TTL, and LRU remove only the victim sequence and return its token reservation + and sequence id to the worker. +- The gRPC service converts native cache/stale/resource results to the typed + protocol errors and does not consume idempotency/flow-control credit on a + rejected append. Release is epoch-specific, so a stale release cannot erase + the active epoch's service state. +- Added opt-in configuration: `MESHNET_HOT_KV_MAX_SESSIONS`, + `MESHNET_HOT_KV_CONTEXT_TOKENS`, `MESHNET_HOT_KV_BUDGET_TOKENS`, and + `MESHNET_HOT_KV_TTL_SECONDS`. + +## Changed files + +- `packages/node/native/worker/llama_shard_engine.{h,cpp}` +- `packages/node/native/worker/shard_service.cpp` +- `packages/node/native/worker/shard_worker_main.cpp` +- `tests/test_llama_shard_worker_binding.py` + +## Commands and results + +```text +PYTHONPATH=packages/node /home/popov/.hermes/hermes-agent/venv/bin/python -m pytest -q \ + tests/test_llama_shard_worker_binding.py tests/test_native_shard_protocol.py +54 passed, 2 skipped + +/home/popov/.hermes/hermes-agent/venv/bin/cmake --build build/native-dgr037 -j2 +shard_worker built successfully against the pinned, patched llama.cpp source. + +/home/popov/.hermes/hermes-agent/venv/bin/ctest --test-dir build/native-dgr037 --output-on-failure +1/1 shard_protocol_conformance passed. + +PYTHONPATH=packages/node /home/popov/.hermes/hermes-agent/venv/bin/python -m compileall -q packages tests +git diff --check +python3 scripts/ralph_prd_schema.py validate .scratch/distributed-gguf-runtime/prd.json +compileall and diff check passed; OK: 55 stories validated. +``` + +## Limitations and dependency handoff + +- DGR-037 supplied the range-attested native model/engine boundary. DGR-038 + adds local sequence ownership without changing its artifact or range + identity contract. +- No mounted GGUF artifact was selected. Therefore no opt-in real-model + four-session run, actual llama KV byte measurement, or hardware metrics are + claimed. The default tests intentionally remain model-download-free and the + source `prd.json` remains `passes: false`. +- DGR-039 should exercise the real two-process range-parity lane with four + sessions and the Hot-KV environment bounds, recording actual cache memory + and cancellation isolation evidence. diff --git a/packages/node/native/worker/llama_shard_engine.cpp b/packages/node/native/worker/llama_shard_engine.cpp index f5de3b5..decc7bd 100644 --- a/packages/node/native/worker/llama_shard_engine.cpp +++ b/packages/node/native/worker/llama_shard_engine.cpp @@ -1,7 +1,9 @@ #include "llama_shard_engine.h" #include +#include #include +#include #include #include #include @@ -59,6 +61,24 @@ class LlamaShardEngine final : public ShardEngine { return false; } resident_bytes_ = report.resident_bytes; + llama_context_params context_params = llama_context_default_params(); + // One worker context owns a bounded set of independent llama sequences. + // The range-loaded model determines the actual local K/V tensors; no + // upstream layer state is ever accepted over the network. + context_params.n_ctx = identity_.hot_kv_budget_tokens; + context_params.n_batch = identity_.hot_kv_context_tokens; + context_params.n_ubatch = identity_.hot_kv_context_tokens; + context_params.n_seq_max = identity_.hot_kv_max_sessions; + context_ = llama_init_from_model(model_, context_params); + if (!context_) { + ShutdownLocked(); + *error = "llama.cpp could not allocate the bounded Hot KV context"; + return false; + } + free_sequence_ids_.reserve(identity_.hot_kv_max_sessions); + for (uint32_t sequence_id = 0; sequence_id < identity_.hot_kv_max_sessions; ++sequence_id) { + free_sequence_ids_.push_back(static_cast(sequence_id)); + } loaded_ = true; return true; } @@ -106,11 +126,63 @@ class LlamaShardEngine final : public ShardEngine { return result; } - bool Execute(const sp::TensorBundle&, std::string* error) override { + HotKvResult OpenSession(const std::string& route_session_id, uint64_t route_epoch) override { + std::lock_guard lock(mu_); + if (!loaded_ || !context_) return {HotKvStatus::kCacheMiss, 0, "llama.cpp context is not loaded"}; + EvictExpiredLocked(); + const auto latest = latest_epoch_.find(route_session_id); + if (latest != latest_epoch_.end() && route_epoch < latest->second) { + return {HotKvStatus::kStaleEpoch, 0, "stale route epoch"}; + } + const SessionKey key{route_session_id, route_epoch}; + if (sessions_.count(key)) return {HotKvStatus::kOk, sessions_[key].past_len, "session already open"}; + if (latest != latest_epoch_.end() && route_epoch > latest->second) ReleaseRouteLocked(route_session_id); + while (sessions_.size() >= identity_.hot_kv_max_sessions) EvictLruLocked(); + if (free_sequence_ids_.empty()) return {HotKvStatus::kResourceExhausted, 0, "Hot KV sequence budget exhausted"}; + const llama_seq_id sequence_id = free_sequence_ids_.back(); + free_sequence_ids_.pop_back(); + sessions_.emplace(key, SessionState{sequence_id, 0, NowSeconds()}); + latest_epoch_[route_session_id] = route_epoch; + return {HotKvStatus::kOk, 0, "Hot KV session opened"}; + } + + HotKvResult Execute(const HotKvStep& step, const sp::TensorBundle&, std::string* error) override { std::lock_guard lock(mu_); if (!loaded_ || !model_) { *error = "llama.cpp model is not loaded"; - return false; + return {HotKvStatus::kCacheMiss, 0, *error}; + } + EvictExpiredLocked(); + const auto latest = latest_epoch_.find(step.route_session_id); + if (latest != latest_epoch_.end() && step.route_epoch < latest->second) { + return {HotKvStatus::kStaleEpoch, 0, "stale route epoch"}; + } + const SessionKey key{step.route_session_id, step.route_epoch}; + auto it = sessions_.find(key); + if (it == sessions_.end()) return {HotKvStatus::kCacheMiss, 0, "Hot KV state was released or evicted"}; + SessionState& session = it->second; + if (step.phase == HotKvStep::Phase::kDecode && step.expected_past_len != session.past_len) { + return {HotKvStatus::kCacheMiss, session.past_len, "expected past length does not match local Hot KV"}; + } + if (step.first_position < session.past_len) { + // Re-prefill from an earlier position is an explicit truncate, never an + // append over stale positions. This removes only this llama sequence. + llama_memory_seq_rm(llama_get_memory(context_), session.sequence_id, + static_cast(step.first_position), -1); + total_reserved_tokens_ -= session.past_len - step.first_position; + session.past_len = step.first_position; + } + if (step.first_position != session.past_len || step.token_count == 0) { + return {HotKvStatus::kCacheMiss, session.past_len, "non-contiguous Hot KV append"}; + } + if (session.past_len + step.token_count > identity_.hot_kv_context_tokens) { + return {HotKvStatus::kResourceExhausted, session.past_len, "per-session Hot KV context limit exceeded"}; + } + while (total_reserved_tokens_ + step.token_count > identity_.hot_kv_budget_tokens && sessions_.size() > 1) { + EvictLruLocked(&key); + } + if (total_reserved_tokens_ + step.token_count > identity_.hot_kv_budget_tokens) { + return {HotKvStatus::kResourceExhausted, session.past_len, "Hot KV token budget exhausted"}; } if (identity_.injected_death_after_executions != 0 && ++executions_ >= identity_.injected_death_after_executions) { @@ -121,7 +193,10 @@ class LlamaShardEngine final : public ShardEngine { // DGR-038 installs per-session context/KV and DGR-039 proves graph parity. // Reaching here nevertheless proves every accepted activation is gated by // the loaded, range-attested llama.cpp engine rather than a fixture. - return true; + session.past_len += step.token_count; + total_reserved_tokens_ += step.token_count; + session.last_used = NowSeconds(); + return {HotKvStatus::kOk, session.past_len, "Hot KV append accepted"}; } const WorkerIdentity& identity() const override { return identity_; } @@ -129,11 +204,20 @@ class LlamaShardEngine final : public ShardEngine { std::lock_guard lock(mu_); return {loaded_, resident_bytes_, loaded_ ? "llama.cpp model loaded" : "llama.cpp model unavailable"}; } - void ReleaseSession(const std::string&) override {} + void ReleaseSession(const std::string& route_session_id, uint64_t route_epoch) override { + std::lock_guard lock(mu_); + ReleaseLocked(SessionKey{route_session_id, route_epoch}); + } void Shutdown() override { std::lock_guard lock(mu_); ShutdownLocked(); } private: void ShutdownLocked() { + sessions_.clear(); + free_sequence_ids_.clear(); + latest_epoch_.clear(); + total_reserved_tokens_ = 0; + if (context_) llama_free(context_); + context_ = nullptr; if (model_) llama_model_free(model_); model_ = nullptr; loaded_ = false; @@ -141,13 +225,63 @@ class LlamaShardEngine final : public ShardEngine { if (backend_initialized_) llama_backend_free(); backend_initialized_ = false; } + struct SessionKey { + std::string route_session_id; + uint64_t route_epoch; + bool operator<(const SessionKey& other) const { + return route_session_id != other.route_session_id ? route_session_id < other.route_session_id + : route_epoch < other.route_epoch; + } + }; + struct SessionState { llama_seq_id sequence_id; uint64_t past_len; uint64_t last_used; }; + static uint64_t NowSeconds() { + return std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count(); + } + void ReleaseLocked(const SessionKey& key) { + auto it = sessions_.find(key); + if (it == sessions_.end()) return; + llama_memory_seq_rm(llama_get_memory(context_), it->second.sequence_id, -1, -1); + total_reserved_tokens_ -= it->second.past_len; + free_sequence_ids_.push_back(it->second.sequence_id); + sessions_.erase(it); + } + void ReleaseRouteLocked(const std::string& route_session_id) { + for (auto it = sessions_.begin(); it != sessions_.end();) { + if (it->first.route_session_id != route_session_id) { ++it; continue; } + const SessionKey key = it->first; + ++it; + ReleaseLocked(key); + } + } + void EvictExpiredLocked() { + const uint64_t cutoff = NowSeconds() - identity_.hot_kv_ttl_seconds; + for (auto it = sessions_.begin(); it != sessions_.end();) { + if (it->second.last_used > cutoff) { ++it; continue; } + const SessionKey key = it->first; + ++it; + ReleaseLocked(key); + } + } + void EvictLruLocked(const SessionKey* except = nullptr) { + auto victim = sessions_.end(); + for (auto it = sessions_.begin(); it != sessions_.end(); ++it) { + if (except && it->first.route_session_id == except->route_session_id && it->first.route_epoch == except->route_epoch) continue; + if (victim == sessions_.end() || it->second.last_used < victim->second.last_used) victim = it; + } + if (victim != sessions_.end()) ReleaseLocked(victim->first); + } WorkerIdentity identity_; mutable std::mutex mu_; llama_model* model_ = nullptr; + llama_context* context_ = nullptr; bool backend_initialized_ = false; bool loaded_ = false; uint64_t resident_bytes_ = 0; uint32_t executions_ = 0; + std::map sessions_; + std::map latest_epoch_; + std::vector free_sequence_ids_; + uint64_t total_reserved_tokens_ = 0; }; } // namespace diff --git a/packages/node/native/worker/llama_shard_engine.h b/packages/node/native/worker/llama_shard_engine.h index 212e2a2..13e5155 100644 --- a/packages/node/native/worker/llama_shard_engine.h +++ b/packages/node/native/worker/llama_shard_engine.h @@ -25,6 +25,10 @@ struct WorkerIdentity { uint32_t start_layer = 0; uint32_t end_layer = 0; // half-open, as on the wire uint32_t injected_death_after_executions = 0; // opt-in test hook; zero disables + uint32_t hot_kv_max_sessions = 8; + uint32_t hot_kv_context_tokens = 4096; + uint32_t hot_kv_budget_tokens = 32768; + uint32_t hot_kv_ttl_seconds = 300; }; struct BundleCheck { @@ -38,15 +42,38 @@ struct EngineHealth { std::string detail; }; +// This is deliberately expressed in tokens, rather than guessed bytes: llama.cpp +// owns the actual K/V layout for the loaded range and backend. The worker uses +// the token reservation to keep its local KV arena bounded before a graph +// adapter materializes the typed boundary (DGR-039). +struct HotKvStep { + enum class Phase { kPrefill, kDecode }; + std::string route_session_id; + uint64_t route_epoch = 0; + Phase phase = Phase::kPrefill; + uint64_t first_position = 0; + uint32_t token_count = 0; + uint64_t expected_past_len = 0; +}; + +enum class HotKvStatus { kOk, kCacheMiss, kStaleEpoch, kResourceExhausted, kCancelled }; + +struct HotKvResult { + HotKvStatus status = HotKvStatus::kOk; + uint64_t past_len = 0; + std::string detail; +}; + class ShardEngine { public: virtual ~ShardEngine() = default; virtual bool Load(std::string* error) = 0; virtual BundleCheck Validate(const sp::TensorBundle&, uint64_t max_chunk_bytes) const = 0; - virtual bool Execute(const sp::TensorBundle&, std::string* error) = 0; + virtual HotKvResult OpenSession(const std::string& route_session_id, uint64_t route_epoch) = 0; + virtual HotKvResult Execute(const HotKvStep&, const sp::TensorBundle&, std::string* error) = 0; virtual const WorkerIdentity& identity() const = 0; virtual EngineHealth health() const = 0; - virtual void ReleaseSession(const std::string& route_session_id) = 0; + virtual void ReleaseSession(const std::string& route_session_id, uint64_t route_epoch) = 0; virtual void Shutdown() = 0; }; diff --git a/packages/node/native/worker/shard_service.cpp b/packages/node/native/worker/shard_service.cpp index 87934e4..ac453bf 100644 --- a/packages/node/native/worker/shard_service.cpp +++ b/packages/node/native/worker/shard_service.cpp @@ -70,6 +70,23 @@ void FillDefaultFlow(sp::FlowControl* fc, const FlowLimits& limits) { fc->set_max_prefill_chunk_tokens(limits.max_prefill_chunk_tokens); } +sp::SessionResponse HotKvFailure(const std::string& route_session_id, const std::string& work_id, + uint64_t step, const HotKvResult& result) { + switch (result.status) { + case HotKvStatus::kStaleEpoch: + return MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_EPOCH_STALE, result.detail, false, false); + case HotKvStatus::kCacheMiss: + return MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_CACHE_MISS, result.detail, false, true); + case HotKvStatus::kResourceExhausted: + return MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_RESOURCE_EXHAUSTED, result.detail, false, true); + case HotKvStatus::kCancelled: + return MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_CANCELLED, result.detail, false, false); + case HotKvStatus::kOk: + break; + } + return MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_INTERNAL, "unexpected Hot KV result", false, true); +} + } // namespace grpc::Status ShardRuntimeServiceImpl::GetCapability(grpc::ServerContext*, @@ -198,6 +215,11 @@ grpc::Status ShardRuntimeServiceImpl::Session( open.has_proposed_flow_control() ? NegotiateFlow(open.proposed_flow_control()) : limits_; + const HotKvResult hot_kv = engine_.OpenSession(route_session_id, open.route_epoch()); + if (hot_kv.status != HotKvStatus::kOk) { + stream->Write(HotKvFailure(route_session_id, "", 0, hot_kv)); + return grpc::Status::OK; + } { std::lock_guard lk(sessions_mu_); SessionState state; @@ -282,13 +304,17 @@ grpc::Status ShardRuntimeServiceImpl::Session( response = MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_PAYLOAD_CORRUPT, *check.corrupt_detail, false, false); } else { + std::string execution_error; + const HotKvResult executed = engine_.Execute( + HotKvStep{route_session_id, envelope.route_epoch(), HotKvStep::Phase::kPrefill, + envelope.position().first_position(), envelope.position().token_count(), + envelope.cache_expectation().expected_past_len()}, + chunk.bundle(), &execution_error); + if (executed.status != HotKvStatus::kOk) { + response = HotKvFailure(route_session_id, work_id, step, executed); + } else { state->seen_steps.insert(step); state->credits -= 1; - std::string execution_error; - if (!engine_.Execute(chunk.bundle(), &execution_error)) { - response = MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_INTERNAL, - execution_error, false, true); - } else { *response.mutable_chunk() = chunk; // echo the exact bundle back } } @@ -348,13 +374,15 @@ grpc::Status ShardRuntimeServiceImpl::Session( response = MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_PAYLOAD_CORRUPT, *check.corrupt_detail, false, false); } else { + std::string execution_error; + const HotKvResult executed = engine_.Execute( + HotKvStep{route_session_id, state->epoch, HotKvStep::Phase::kDecode, + step_msg.position(), 1, step_msg.expected_past_len()}, bundle, &execution_error); + if (executed.status != HotKvStatus::kOk) { + response = HotKvFailure(route_session_id, work_id, step, executed); + } else { state->seen_steps.insert(step); state->credits -= 1; - std::string execution_error; - if (!engine_.Execute(bundle, &execution_error)) { - response = MakeFail(route_session_id, work_id, step, sp::ERROR_CODE_INTERNAL, - execution_error, false, true); - } else { // No decode response field exists; echo the step back as a // chunk-bearing SessionResponse per the proto's relayed-frame design. sp::ActivationChunk* out = response.mutable_chunk(); @@ -412,8 +440,11 @@ grpc::Status ShardRuntimeServiceImpl::Session( // freed the moment the terminal status is sent. { std::lock_guard lk(sessions_mu_); - sessions_.erase(route_session_id); - engine_.ReleaseSession(route_session_id); + auto it = sessions_.find(release.route_session_id()); + if (it != sessions_.end() && it->second.epoch == release.route_epoch()) { + sessions_.erase(it); + } + engine_.ReleaseSession(release.route_session_id(), release.route_epoch()); } sp::SessionResponse response; sp::ShardStatus* status = response.mutable_status(); @@ -454,8 +485,10 @@ grpc::Status ShardRuntimeServiceImpl::Release(grpc::ServerContext*, bool existed; { std::lock_guard lk(sessions_mu_); - existed = sessions_.erase(request->route_session_id()) != 0; - engine_.ReleaseSession(request->route_session_id()); + auto it = sessions_.find(request->route_session_id()); + existed = it != sessions_.end() && it->second.epoch == request->route_epoch(); + if (existed) sessions_.erase(it); + engine_.ReleaseSession(request->route_session_id(), request->route_epoch()); } response->set_released(existed); return grpc::Status::OK; diff --git a/packages/node/native/worker/shard_worker_main.cpp b/packages/node/native/worker/shard_worker_main.cpp index 49ae41f..64de6aa 100644 --- a/packages/node/native/worker/shard_worker_main.cpp +++ b/packages/node/native/worker/shard_worker_main.cpp @@ -64,6 +64,19 @@ meshnet::worker::FlowLimits LimitsFromEnv() { return limits; } +bool PositiveEnv(const char* name, uint32_t* out, std::string* error) { + if (const char* value = std::getenv(name)) { + char* end = nullptr; + const unsigned long parsed = std::strtoul(value, &end, 10); + if (end == value || *end != '\0' || parsed == 0 || parsed > UINT32_MAX) { + *error = std::string("invalid ") + name; + return false; + } + *out = static_cast(parsed); + } + return true; +} + bool IdentityFromEnv(meshnet::worker::WorkerIdentity* identity, std::string* error) { const auto required = [&](const char* name, std::string* out) -> bool { const char* value = std::getenv(name); @@ -100,6 +113,14 @@ bool IdentityFromEnv(meshnet::worker::WorkerIdentity* identity, std::string* err } identity->injected_death_after_executions = static_cast(parsed); } + if (!PositiveEnv("MESHNET_HOT_KV_MAX_SESSIONS", &identity->hot_kv_max_sessions, error) || + !PositiveEnv("MESHNET_HOT_KV_CONTEXT_TOKENS", &identity->hot_kv_context_tokens, error) || + !PositiveEnv("MESHNET_HOT_KV_BUDGET_TOKENS", &identity->hot_kv_budget_tokens, error) || + !PositiveEnv("MESHNET_HOT_KV_TTL_SECONDS", &identity->hot_kv_ttl_seconds, error)) return false; + if (identity->hot_kv_budget_tokens < identity->hot_kv_context_tokens) { + *error = "MESHNET_HOT_KV_BUDGET_TOKENS must cover one session context"; + return false; + } return true; } diff --git a/tests/test_llama_shard_worker_binding.py b/tests/test_llama_shard_worker_binding.py index 9186941..4aa1a35 100644 --- a/tests/test_llama_shard_worker_binding.py +++ b/tests/test_llama_shard_worker_binding.py @@ -44,3 +44,20 @@ def test_identity_is_loaded_not_stream_supplied_and_health_reports_it(): assert "model artifact or runtime recipe digest does not match" in source assert "resident_bytes" in source assert '" range=["' in source + + +def test_hot_kv_is_bounded_and_keyed_by_route_session_and_epoch(): + engine = (WORKER / "llama_shard_engine.cpp").read_text(encoding="utf-8") + header = (WORKER / "llama_shard_engine.h").read_text(encoding="utf-8") + service = (WORKER / "shard_service.cpp").read_text(encoding="utf-8") + main = (WORKER / "shard_worker_main.cpp").read_text(encoding="utf-8") + assert "struct SessionKey" in engine + assert "uint64_t route_epoch" in engine + assert "llama_init_from_model" in engine + assert "llama_memory_seq_rm" in engine + assert "EvictExpiredLocked" in engine + assert "EvictLruLocked" in engine + assert "HotKvStatus::kCacheMiss" in service + assert "ERROR_CODE_CACHE_MISS" in service + assert "MESHNET_HOT_KV_BUDGET_TOKENS" in main + assert "HotKvStep" in header