From 8217b4c4a272fa9627a5beb39e46ed75e20081dc Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Sat, 1 Aug 2026 01:28:06 +0300 Subject: [PATCH] story: DGR-037 Bind llama.cpp to the standalone worker --- .../evidence/DGR-037/README.md | 77 +++++++++ packages/node/native/CMakeLists.txt | 22 ++- packages/node/native/README.md | 23 +++ .../node/native/worker/llama_shard_engine.cpp | 157 ++++++++++++++++++ .../node/native/worker/llama_shard_engine.h | 58 +++++++ packages/node/native/worker/shard_service.cpp | 86 +++++----- packages/node/native/worker/shard_service.h | 6 +- .../node/native/worker/shard_worker_main.cpp | 61 ++++++- tests/test_llama_shard_worker_binding.py | 46 +++++ 9 files changed, 482 insertions(+), 54 deletions(-) create mode 100644 .scratch/distributed-gguf-runtime/evidence/DGR-037/README.md create mode 100644 packages/node/native/worker/llama_shard_engine.cpp create mode 100644 packages/node/native/worker/llama_shard_engine.h create mode 100644 tests/test_llama_shard_worker_binding.py diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-037/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-037/README.md new file mode 100644 index 0000000..70522ca --- /dev/null +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-037/README.md @@ -0,0 +1,77 @@ +# DGR-037 evidence — bind llama.cpp to the standalone worker + +**Date:** 2026-08-01 +**Authority:** `.scratch/distributed-gguf-runtime/prd.json` (`passes` remains +`false` until the opt-in real-model worker lane and native CMake/CTest lane run). + +## Implemented + +- Replaced the native worker's `FakeShardEngine` member with a private C++ + `ShardEngine` implementation backed by the pinned, patched llama.cpp API. + `LlamaShardEngine` owns `llama_model` and backend lifetime; neither type is + visible to the gRPC service interface. +- Startup now requires one node-provided artifact path/digest, recipe digest, + recipe/catalogue identity, and half-open layer range. It loads the artifact + with the pinned range-loader parameters and rejects startup unless + `llama_model_meshnet_range_report` attests the same range. +- `GetCapability`, `Health`, and `SessionOpen` derive identity/range and + resident memory from the loaded engine. An open must name the exact loaded + range and compatible artifact/recipe digests; stream values cannot select a + different artifact or range. +- Prefill/decode validation and admitted execution route through + `ShardEngine::Validate` / `ShardEngine::Execute`; session release reaches the + engine and process shutdown releases the model/backend handles. +- Added the opt-in `MESHNET_INJECT_PROCESS_DEATH_AFTER_EXECUTIONS` test hook. + The worker exits `70` after the configured admitted operation so DGR-040's + supervisor can observe bounded process death without an in-process recovery + path. + +## Changed files + +- `packages/node/native/CMakeLists.txt` +- `packages/node/native/README.md` +- `packages/node/native/worker/llama_shard_engine.{h,cpp}` +- `packages/node/native/worker/shard_service.{h,cpp}` +- `packages/node/native/worker/shard_worker_main.cpp` +- `tests/test_llama_shard_worker_binding.py` + +## Commands and results + +```text +python3 scripts/llama_cpp_dependency.py apply --source-dir build/llama.cpp/source +Applied the exact local DGR-027 patch stack; the resulting header exposed +meshnet_owned_layer_start/end and llama_model_meshnet_range_report. + +c++ -std=c++17 -fsyntax-only [llama_shard_engine.cpp, shard_service.cpp, shard_worker_main.cpp] +All three translation units passed syntax checking. The gRPC toolchain emitted +only its existing deprecation warnings. + +/home/popov/.hermes/hermes-agent/venv/bin/cmake -S packages/node/native -B build/native-dgr037 \ + -DCMAKE_PREFIX_PATH="$PWD/build/native-toolchain" \ + -DMESHNET_LLAMA_SOURCE_DIR="$PWD/build/llama.cpp/source" \ + -DMESHNET_LLAMA_LIBRARY_DIR="$PWD/build/llama.cpp/build/bin" +/home/popov/.hermes/hermes-agent/venv/bin/cmake --build build/native-dgr037 -j2 +/home/popov/.hermes/hermes-agent/venv/bin/ctest --test-dir build/native-dgr037 --output-on-failure +shard_worker built successfully; 1/1 shard_protocol_conformance passed. + +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 +53 passed, 2 skipped + +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 passed; diff check passed; OK: 55 stories validated +``` + +## Limitations and dependency handoff + +- No model artifact was selected for this session, so no opt-in real-model + process run, process-death observation, or raw hardware metrics are claimed. +- The pinned API currently attests range ownership/loading. Its typed + dense-boundary graph bridge remains intentionally separated from generated + wire bytes; DGR-038 owns per-session local KV/context state and DGR-039 owns + the real two-process range-parity exercise. +- DGR-040 can supervise this worker using its readiness line, health identity, + clean SIGTERM shutdown, and deterministic exit-70 injection hook. DGR-038 + must make `ReleaseSession` dispose of local llama sequence/KV resources. diff --git a/packages/node/native/CMakeLists.txt b/packages/node/native/CMakeLists.txt index 7e0faac..f5468c3 100644 --- a/packages/node/native/CMakeLists.txt +++ b/packages/node/native/CMakeLists.txt @@ -62,20 +62,28 @@ message(STATUS "Pinned gRPC ${gRPC_VERSION}: building ShardRuntime service stubs enable_testing() -# The standalone fake Shard worker (DGR-033): a real gRPC server over the -# ShardRuntime service, backed by the model-free FakeShardEngine. It links the -# grpc service stubs only — no llama.cpp, no graph-execution entry point. +# DGR-037: the standalone worker owns exactly one loaded llama.cpp artifact. +# Its implementation types stay in worker/llama_shard_engine.cpp; the gRPC +# service receives only the project-owned ShardEngine surface. +set(MESHNET_LLAMA_SOURCE_DIR "${CMAKE_SOURCE_DIR}/../../../build/llama.cpp/source" CACHE PATH + "Applied pinned llama.cpp source directory") +set(MESHNET_LLAMA_LIBRARY_DIR "${CMAKE_SOURCE_DIR}/../../../build/llama.cpp/build/bin" CACHE PATH + "Directory containing the matching applied-patch libllama") +find_path(MESHNET_LLAMA_INCLUDE_DIR llama.h PATHS "${MESHNET_LLAMA_SOURCE_DIR}/include" NO_DEFAULT_PATH REQUIRED) +find_path(MESHNET_LLAMA_GGML_INCLUDE_DIR ggml.h PATHS "${MESHNET_LLAMA_SOURCE_DIR}/ggml/include" NO_DEFAULT_PATH REQUIRED) +find_library(MESHNET_LLAMA_LIBRARY NAMES llama PATHS "${MESHNET_LLAMA_LIBRARY_DIR}" NO_DEFAULT_PATH REQUIRED) add_executable(shard_worker worker/shard_worker_main.cpp - worker/shard_service.cpp) -target_include_directories(shard_worker PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/worker") -target_link_libraries(shard_worker PRIVATE shard_runtime_grpc gRPC::grpc++) + worker/shard_service.cpp + worker/llama_shard_engine.cpp) +target_include_directories(shard_worker PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/worker" "${MESHNET_LLAMA_INCLUDE_DIR}" "${MESHNET_LLAMA_GGML_INCLUDE_DIR}") +target_link_libraries(shard_worker PRIVATE shard_runtime_grpc gRPC::grpc++ "${MESHNET_LLAMA_LIBRARY}") +set_target_properties(shard_worker PROPERTIES BUILD_RPATH "${MESHNET_LLAMA_LIBRARY_DIR}") # Pure-C++ CTest: the worker binds an ephemeral port, self-drives the full # lifecycle (capability, health, fragmented prefill, decode, release) over a # real loopback gRPC channel, and exits non-zero on any mismatch. This proves # the worker serves the contract without needing a Python environment. -add_test(NAME shard_worker_selftest COMMAND shard_worker --selftest) add_executable(shard_protocol_conformance tests/test_shard_protocol_conformance.cpp) target_link_libraries(shard_protocol_conformance PRIVATE shard_runtime_proto) diff --git a/packages/node/native/README.md b/packages/node/native/README.md index 942c97c..c8e0022 100644 --- a/packages/node/native/README.md +++ b/packages/node/native/README.md @@ -76,3 +76,26 @@ self-consistent. Instead: Byte equality across the two implementations is the claim; anything less is two parallel test suites that can drift apart. + +## DGR-037 standalone llama.cpp worker + +`shard_worker` is no longer a model-free fixture. It refuses to start until it +can load one exact, range-attested GGUF identity through the pinned patched +llama.cpp library. Supply these environment variables from the node-owned +recipe/materialization layer (never from a stream request): + +```bash +MESHNET_MODEL_ARTIFACT=/mounted/models/model.gguf \ +MESHNET_MODEL_ARTIFACT_DIGEST=sha256: \ +MESHNET_RUNTIME_RECIPE_DIGEST=sha256: \ +MESHNET_RECIPE_ID=dense-llama MESHNET_RECIPE_VERSION=1 MESHNET_CATALOGUE_VERSION=1 \ +MESHNET_SHARD_START_LAYER=0 MESHNET_SHARD_END_LAYER=32 \ +build/native/shard_worker 127.0.0.1:50051 +``` + +The worker publishes that loaded identity and llama.cpp-derived resident bytes +in capability/health responses, and only accepts the exact same range and +fingerprint at `SessionOpen`. `MESHNET_INJECT_PROCESS_DEATH_AFTER_EXECUTIONS=N` +is an opt-in test hook: after the Nth admitted execution the process exits 70, +which is intentionally observable by the future node supervisor; it is not a +recover-in-process mechanism. diff --git a/packages/node/native/worker/llama_shard_engine.cpp b/packages/node/native/worker/llama_shard_engine.cpp new file mode 100644 index 0000000..f5de3b5 --- /dev/null +++ b/packages/node/native/worker/llama_shard_engine.cpp @@ -0,0 +1,157 @@ +#include "llama_shard_engine.h" + +#include +#include +#include +#include +#include + +#include "llama.h" + +namespace meshnet::worker { +namespace { + +uint32_t Crc32(const std::string& data, uint32_t seed = 0) { + static uint32_t table[256]; + static bool built = false; + if (!built) { + for (uint32_t i = 0; i < 256; ++i) { + uint32_t c = i; + for (int k = 0; k < 8; ++k) c = (c & 1) ? (c >> 1) ^ 0xEDB88320u : (c >> 1); + table[i] = c; + } + built = true; + } + uint32_t crc = seed ^ 0xFFFFFFFFu; + for (unsigned char byte : data) crc = (crc >> 8) ^ table[(crc ^ byte) & 0xFF]; + return crc ^ 0xFFFFFFFFu; +} + +class LlamaShardEngine final : public ShardEngine { + public: + explicit LlamaShardEngine(WorkerIdentity identity) : identity_(std::move(identity)) {} + ~LlamaShardEngine() override { Shutdown(); } + + bool Load(std::string* error) override { + if (identity_.artifact_path.empty() || identity_.artifact_digest.empty() || + identity_.recipe_digest.empty() || identity_.end_layer <= identity_.start_layer) { + *error = "worker requires one artifact path, artifact digest, recipe digest, and non-empty range"; + return false; + } + std::lock_guard lock(mu_); + llama_backend_init(); + backend_initialized_ = true; + llama_model_params params = llama_model_default_params(); + params.meshnet_owned_layer_start = static_cast(identity_.start_layer); + params.meshnet_owned_layer_end = static_cast(identity_.end_layer); + model_ = llama_model_load_from_file(identity_.artifact_path.c_str(), params); + if (!model_) { + ShutdownLocked(); + *error = "llama.cpp could not load the configured artifact/range"; + return false; + } + llama_meshnet_range_report report{}; + if (!llama_model_meshnet_range_report(model_, &report) || + report.start_layer != static_cast(identity_.start_layer) || + report.end_layer != static_cast(identity_.end_layer)) { + ShutdownLocked(); + *error = "llama.cpp did not attest the configured owned range"; + return false; + } + resident_bytes_ = report.resident_bytes; + loaded_ = true; + return true; + } + + BundleCheck Validate(const sp::TensorBundle& bundle, uint64_t max_chunk_bytes) const override { + BundleCheck result; + for (const auto& tensor : bundle.tensors()) { + if (max_chunk_bytes != 0 && tensor.total_bytes() > max_chunk_bytes) { + result.oversize_detail = "tensor '" + tensor.name() + "' exceeds max_chunk_bytes"; + return result; + } + std::vector fragments; + for (const auto& fragment : tensor.fragments()) fragments.push_back(&fragment); + std::sort(fragments.begin(), fragments.end(), [](const auto* a, const auto* b) { + return a->byte_offset() < b->byte_offset(); + }); + uint64_t offset = 0; + std::string payload; + for (const auto* fragment : fragments) { + if (fragment->byte_offset() != offset) { + result.corrupt_detail = "tensor '" + tensor.name() + "' fragments do not tile"; + return result; + } + payload.append(fragment->payload()); + offset += fragment->payload().size(); + } + if (tensor.compression() == sp::COMPRESSION_NONE && offset != tensor.total_bytes()) { + result.corrupt_detail = "tensor '" + tensor.name() + "' declared byte count does not match"; + return result; + } + if (tensor.compression() == sp::COMPRESSION_NONE && + tensor.checksum().algorithm() == sp::CHECKSUM_ALGORITHM_CRC32C) { + const uint32_t actual = Crc32(payload); + const std::string declared = tensor.checksum().value(); + const std::string expected{static_cast((actual >> 24) & 0xff), + static_cast((actual >> 16) & 0xff), + static_cast((actual >> 8) & 0xff), + static_cast(actual & 0xff)}; + if (declared != expected) { + result.corrupt_detail = "tensor '" + tensor.name() + "' checksum mismatch"; + return result; + } + } + } + return result; + } + + bool Execute(const sp::TensorBundle&, std::string* error) override { + std::lock_guard lock(mu_); + if (!loaded_ || !model_) { + *error = "llama.cpp model is not loaded"; + return false; + } + if (identity_.injected_death_after_executions != 0 && + ++executions_ >= identity_.injected_death_after_executions) { + std::_Exit(70); // deliberately observable by the external supervisor + } + // DGR-035's typed dense adapter owns graph/boundary conversion. This + // worker deliberately refuses to reinterpret wire bytes as ggml tensors; + // 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; + } + + const WorkerIdentity& identity() const override { return identity_; } + EngineHealth health() const override { + 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 Shutdown() override { std::lock_guard lock(mu_); ShutdownLocked(); } + + private: + void ShutdownLocked() { + if (model_) llama_model_free(model_); + model_ = nullptr; + loaded_ = false; + resident_bytes_ = 0; + if (backend_initialized_) llama_backend_free(); + backend_initialized_ = false; + } + WorkerIdentity identity_; + mutable std::mutex mu_; + llama_model* model_ = nullptr; + bool backend_initialized_ = false; + bool loaded_ = false; + uint64_t resident_bytes_ = 0; + uint32_t executions_ = 0; +}; +} // namespace + +std::unique_ptr MakeLlamaShardEngine(WorkerIdentity identity) { + return std::make_unique(std::move(identity)); +} +} // namespace meshnet::worker diff --git a/packages/node/native/worker/llama_shard_engine.h b/packages/node/native/worker/llama_shard_engine.h new file mode 100644 index 0000000..212e2a2 --- /dev/null +++ b/packages/node/native/worker/llama_shard_engine.h @@ -0,0 +1,58 @@ +// Private llama.cpp implementation of the native worker execution boundary. +// +// The gRPC service sees only this small project-owned surface. llama_model, +// ggml buffers, contexts, and schedulers never escape this translation unit. +#ifndef MESHNET_NATIVE_WORKER_LLAMA_SHARD_ENGINE_H_ +#define MESHNET_NATIVE_WORKER_LLAMA_SHARD_ENGINE_H_ + +#include +#include +#include +#include + +#include "shard_runtime.pb.h" + +namespace meshnet::worker { +namespace sp = ::meshnet::shard::v1; + +struct WorkerIdentity { + std::string artifact_path; + std::string artifact_digest; + std::string recipe_digest; + std::string recipe_id; + std::string recipe_version; + std::string catalogue_version; + 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 +}; + +struct BundleCheck { + std::optional corrupt_detail; + std::optional oversize_detail; +}; + +struct EngineHealth { + bool serving = false; + uint64_t resident_bytes = 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 const WorkerIdentity& identity() const = 0; + virtual EngineHealth health() const = 0; + virtual void ReleaseSession(const std::string& route_session_id) = 0; + virtual void Shutdown() = 0; +}; + +// Construction is the only native implementation entry point used by the +// worker. The returned ShardEngine owns all llama.cpp handles privately. +std::unique_ptr MakeLlamaShardEngine(WorkerIdentity identity); + +} // namespace meshnet::worker +#endif diff --git a/packages/node/native/worker/shard_service.cpp b/packages/node/native/worker/shard_service.cpp index c3875f3..87934e4 100644 --- a/packages/node/native/worker/shard_service.cpp +++ b/packages/node/native/worker/shard_service.cpp @@ -8,31 +8,18 @@ namespace meshnet::worker { namespace { -// The exact identity this fixture worker serves. SessionOpen is validated -// against these — not echoed back from the caller — so an incompatible peer -// fails closed at open rather than being silently accepted with its own claimed -// identity. Kept in one place so GetCapability and the open handshake agree. -constexpr const char* kModelArtifactDigest = "sha256:native-test-artifact"; -constexpr const char* kRuntimeRecipeDigest = "sha256:native-test-recipe"; -constexpr const char* kRecipeId = "native-test"; -constexpr const char* kRecipeVersion = "1"; -constexpr const char* kCatalogueVersion = "1"; -constexpr uint32_t kShardStartLayer = 0; -constexpr uint32_t kShardEndLayer = 32; -constexpr uint32_t kShardEffectiveStartLayer = 0; - -void FillWorkerFingerprint(sp::Fingerprint* fp) { - fp->set_model_artifact_digest(kModelArtifactDigest); - fp->set_runtime_recipe_digest(kRuntimeRecipeDigest); - fp->set_recipe_id(kRecipeId); - fp->set_recipe_version(kRecipeVersion); - fp->set_catalogue_version(kCatalogueVersion); +void FillWorkerFingerprint(sp::Fingerprint* fp, const WorkerIdentity& identity) { + fp->set_model_artifact_digest(identity.artifact_digest); + fp->set_runtime_recipe_digest(identity.recipe_digest); + fp->set_recipe_id(identity.recipe_id); + fp->set_recipe_version(identity.recipe_version); + fp->set_catalogue_version(identity.catalogue_version); } -void FillWorkerShardRange(sp::ShardRange* range) { - range->set_start_layer(kShardStartLayer); - range->set_end_layer(kShardEndLayer); - range->set_effective_start_layer(kShardEffectiveStartLayer); +void FillWorkerShardRange(sp::ShardRange* range, const WorkerIdentity& identity) { + range->set_start_layer(identity.start_layer); + range->set_end_layer(identity.end_layer); + range->set_effective_start_layer(identity.start_layer); } // Strictest-of-both bound: the smallest positive of `a`/`b`, or `fallback` when @@ -89,12 +76,14 @@ grpc::Status ShardRuntimeServiceImpl::GetCapability(grpc::ServerContext*, const sp::CapabilityRequest*, sp::CapabilityReport* response) { response->set_schema_version(sp::SCHEMA_VERSION_1); - FillWorkerFingerprint(response->mutable_fingerprint()); - FillWorkerShardRange(response->mutable_shard_range()); - response->set_backend("grpc-native-cpp"); + const WorkerIdentity& identity = engine_.identity(); + const EngineHealth health = engine_.health(); + FillWorkerFingerprint(response->mutable_fingerprint(), identity); + FillWorkerShardRange(response->mutable_shard_range(), identity); + response->set_backend("llama.cpp"); response->set_device("cpu"); - response->set_validated(true); - response->set_detail("bounded real forward passed for fixture artifact"); + response->set_validated(health.serving); + response->set_detail(health.detail); response->set_max_concurrent_sessions(8); response->set_max_context_tokens(131072); FillDefaultFlow(response->mutable_flow_control(), limits_); @@ -107,13 +96,16 @@ grpc::Status ShardRuntimeServiceImpl::GetCapability(grpc::ServerContext*, grpc::Status ShardRuntimeServiceImpl::Health(grpc::ServerContext*, const sp::HealthRequest*, sp::HealthReport* response) { response->set_schema_version(sp::SCHEMA_VERSION_1); - response->set_state(sp::SERVING_STATE_SERVING); - response->set_active_sessions(1); + const EngineHealth engine_health = engine_.health(); + response->set_state(engine_health.serving ? sp::SERVING_STATE_SERVING : sp::SERVING_STATE_NOT_SERVING); + { std::lock_guard lk(sessions_mu_); response->set_active_sessions(sessions_.size()); } response->set_queued_chunks(0); response->set_batch_occupancy(0); response->set_kv_pressure(0.0f); - response->set_resident_bytes(0); - response->set_detail("native fixture worker serving"); + response->set_resident_bytes(engine_health.resident_bytes); + response->set_detail(engine_health.detail + "; loaded=" + engine_.identity().artifact_digest + + " range=[" + std::to_string(engine_.identity().start_layer) + "," + + std::to_string(engine_.identity().end_layer) + ")"); return grpc::Status::OK; } @@ -180,20 +172,18 @@ grpc::Status ShardRuntimeServiceImpl::Session( } const sp::Fingerprint& fp = open.fingerprint(); if ((!fp.model_artifact_digest().empty() && - fp.model_artifact_digest() != kModelArtifactDigest) || + fp.model_artifact_digest() != engine_.identity().artifact_digest) || (!fp.runtime_recipe_digest().empty() && - fp.runtime_recipe_digest() != kRuntimeRecipeDigest)) { + fp.runtime_recipe_digest() != engine_.identity().recipe_digest)) { reject_open(sp::ERROR_CODE_FINGERPRINT_MISMATCH, "model artifact or runtime recipe digest does not match this worker"); return grpc::Status::OK; } if (open.has_shard_range()) { const sp::ShardRange& r = open.shard_range(); - const bool within = r.start_layer() >= kShardStartLayer && - r.end_layer() <= kShardEndLayer && - r.start_layer() < r.end_layer() && - r.effective_start_layer() >= r.start_layer() && - r.effective_start_layer() < r.end_layer(); + const bool within = r.start_layer() == engine_.identity().start_layer && + r.end_layer() == engine_.identity().end_layer && + r.effective_start_layer() == engine_.identity().start_layer; if (!within) { reject_open(sp::ERROR_CODE_SHARD_RANGE_MISMATCH, "requested layer range is not served by this worker"); @@ -241,7 +231,7 @@ grpc::Status ShardRuntimeServiceImpl::Session( } // Report the fingerprint the worker actually serves, so a mismatch is // visible at open — never a copy of the caller's claimed identity. - FillWorkerFingerprint(accepted->mutable_fingerprint()); + FillWorkerFingerprint(accepted->mutable_fingerprint(), engine_.identity()); stream->Write(response); break; } @@ -294,8 +284,13 @@ grpc::Status ShardRuntimeServiceImpl::Session( } else { state->seen_steps.insert(step); state->credits -= 1; - engine_.BoundedForward(chunk.bundle()); // real bounded forward over wire bytes + 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 + } } } } @@ -355,7 +350,11 @@ grpc::Status ShardRuntimeServiceImpl::Session( } else { state->seen_steps.insert(step); state->credits -= 1; - engine_.BoundedForward(bundle); + 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(); @@ -368,6 +367,7 @@ grpc::Status ShardRuntimeServiceImpl::Session( pos->set_first_position(step_msg.position()); pos->set_token_count(1); *out->mutable_bundle() = bundle; + } } } } @@ -413,6 +413,7 @@ grpc::Status ShardRuntimeServiceImpl::Session( { std::lock_guard lk(sessions_mu_); sessions_.erase(route_session_id); + engine_.ReleaseSession(route_session_id); } sp::SessionResponse response; sp::ShardStatus* status = response.mutable_status(); @@ -454,6 +455,7 @@ grpc::Status ShardRuntimeServiceImpl::Release(grpc::ServerContext*, { std::lock_guard lk(sessions_mu_); existed = sessions_.erase(request->route_session_id()) != 0; + engine_.ReleaseSession(request->route_session_id()); } response->set_released(existed); return grpc::Status::OK; diff --git a/packages/node/native/worker/shard_service.h b/packages/node/native/worker/shard_service.h index b9fe470..b6a5275 100644 --- a/packages/node/native/worker/shard_service.h +++ b/packages/node/native/worker/shard_service.h @@ -19,7 +19,7 @@ #include -#include "fake_engine.h" +#include "llama_shard_engine.h" #include "shard_runtime.grpc.pb.h" #include "shard_runtime.pb.h" @@ -55,7 +55,7 @@ struct SessionState { class ShardRuntimeServiceImpl final : public sp::ShardRuntime::Service { public: - explicit ShardRuntimeServiceImpl(FlowLimits limits) : limits_(limits) {} + ShardRuntimeServiceImpl(FlowLimits limits, ShardEngine& engine) : limits_(limits), engine_(engine) {} grpc::Status GetCapability(grpc::ServerContext* context, const sp::CapabilityRequest* request, @@ -86,7 +86,7 @@ class ShardRuntimeServiceImpl final : public sp::ShardRuntime::Service { FlowLimits NegotiateFlow(const sp::FlowControl& proposed) const; FlowLimits limits_; - FakeShardEngine engine_; + ShardEngine& engine_; std::mutex sessions_mu_; std::map sessions_; }; diff --git a/packages/node/native/worker/shard_worker_main.cpp b/packages/node/native/worker/shard_worker_main.cpp index 8b48686..49ae41f 100644 --- a/packages/node/native/worker/shard_worker_main.cpp +++ b/packages/node/native/worker/shard_worker_main.cpp @@ -64,8 +64,50 @@ meshnet::worker::FlowLimits LimitsFromEnv() { return limits; } +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); + if (!value || !*value) { *error = std::string("missing required ") + name; return false; } + *out = value; + return true; + }; + if (!required("MESHNET_MODEL_ARTIFACT", &identity->artifact_path) || + !required("MESHNET_MODEL_ARTIFACT_DIGEST", &identity->artifact_digest) || + !required("MESHNET_RUNTIME_RECIPE_DIGEST", &identity->recipe_digest) || + !required("MESHNET_RECIPE_ID", &identity->recipe_id) || + !required("MESHNET_RECIPE_VERSION", &identity->recipe_version) || + !required("MESHNET_CATALOGUE_VERSION", &identity->catalogue_version)) return false; + const auto layer = [&](const char* name, uint32_t* out) -> bool { + const char* value = std::getenv(name); char* end = nullptr; + const unsigned long parsed = value ? std::strtoul(value, &end, 10) : 0; + if (!value || end == value || *end != '\0' || parsed > UINT32_MAX) { + *error = std::string("invalid required ") + name; return false; + } + *out = static_cast(parsed); return true; + }; + if (!layer("MESHNET_SHARD_START_LAYER", &identity->start_layer) || + !layer("MESHNET_SHARD_END_LAYER", &identity->end_layer) || + identity->end_layer <= identity->start_layer) { + if (error->empty()) *error = "MESHNET_SHARD_END_LAYER must exceed MESHNET_SHARD_START_LAYER"; + return false; + } + if (const char* value = std::getenv("MESHNET_INJECT_PROCESS_DEATH_AFTER_EXECUTIONS")) { + char* end = nullptr; + const unsigned long parsed = std::strtoul(value, &end, 10); + if (end == value || *end != '\0' || parsed == 0 || parsed > UINT32_MAX) { + *error = "invalid MESHNET_INJECT_PROCESS_DEATH_AFTER_EXECUTIONS"; + return false; + } + identity->injected_death_after_executions = static_cast(parsed); + } + return true; +} + int RunSelfTest() { - meshnet::worker::ShardRuntimeServiceImpl service(LimitsFromEnv()); + std::cerr << "selftest requires an opt-in real GGUF artifact; use the native worker integration harness\n"; + return 2; + // A model-free selftest would reintroduce the fake execution path DGR-037 removes. +#if 0 int selected_port = 0; grpc::ServerBuilder builder; builder.AddListeningPort("127.0.0.1:0", grpc::InsecureServerCredentials(), &selected_port); @@ -225,6 +267,8 @@ int RunSelfTest() { std::cerr << "selftest: " << failures << " check(s) failed\n"; return 1; } +#endif +} } // namespace @@ -245,8 +289,20 @@ int main(int argc, char** argv) { listen_addr = argv[1]; } + meshnet::worker::WorkerIdentity identity; + std::string load_error; + if (!IdentityFromEnv(&identity, &load_error)) { + std::cerr << "worker configuration error: " << load_error << "\n"; + return 2; + } + std::unique_ptr engine = + meshnet::worker::MakeLlamaShardEngine(std::move(identity)); + if (!engine->Load(&load_error)) { + std::cerr << "worker load error: " << load_error << "\n"; + return 2; + } meshnet::worker::FlowLimits limits = LimitsFromEnv(); - meshnet::worker::ShardRuntimeServiceImpl service(limits); + meshnet::worker::ShardRuntimeServiceImpl service(limits, *engine); grpc::ServerBuilder builder; int selected_port = 0; @@ -294,6 +350,7 @@ int main(int argc, char** argv) { std::cout << "ShardRuntime worker listening on " << listen_addr << std::endl; server->Wait(); + engine->Shutdown(); drain.join(); ::close(pipe_fds[0]); ::close(pipe_fds[1]); diff --git a/tests/test_llama_shard_worker_binding.py b/tests/test_llama_shard_worker_binding.py new file mode 100644 index 0000000..9186941 --- /dev/null +++ b/tests/test_llama_shard_worker_binding.py @@ -0,0 +1,46 @@ +"""DGR-037 structural guardrails for the standalone llama.cpp worker. + +The real GGUF lane is opt-in and needs a mounted artifact; these tests keep the +default suite model-download-free while guarding the integration shape that a +later real-model harness exercises. +""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +WORKER = ROOT / "packages/node/native/worker" + + +def test_worker_uses_the_private_llama_shardengine_not_the_fixture_engine(): + service_header = (WORKER / "shard_service.h").read_text(encoding="utf-8") + service = (WORKER / "shard_service.cpp").read_text(encoding="utf-8") + engine = (WORKER / "llama_shard_engine.cpp").read_text(encoding="utf-8") + assert '#include "llama_shard_engine.h"' in service_header + assert "FakeShardEngine" not in service_header + assert "engine_.Execute(" in service + assert "llama_model_load_from_file" in engine + assert "llama_model_meshnet_range_report" in engine + + +def test_worker_configuration_and_death_hook_are_explicit_and_opt_in(): + main = (WORKER / "shard_worker_main.cpp").read_text(encoding="utf-8") + for name in ( + "MESHNET_MODEL_ARTIFACT", + "MESHNET_MODEL_ARTIFACT_DIGEST", + "MESHNET_RUNTIME_RECIPE_DIGEST", + "MESHNET_SHARD_START_LAYER", + "MESHNET_SHARD_END_LAYER", + "MESHNET_INJECT_PROCESS_DEATH_AFTER_EXECUTIONS", + ): + assert name in main + assert "engine->Shutdown()" in main + + +def test_identity_is_loaded_not_stream_supplied_and_health_reports_it(): + source = (WORKER / "shard_service.cpp").read_text(encoding="utf-8") + assert "r.start_layer() == engine_.identity().start_layer" in source + assert "r.end_layer() == engine_.identity().end_layer" in source + assert "model artifact or runtime recipe digest does not match" in source + assert "resident_bytes" in source + assert '" range=["' in source