Files
neuron-tai/packages/node/native/worker/fake_engine.h
Dobromir Popov 7473bb7e44 fix: DGR-033 repair native worker protocol per cross-review BLOCK
Address the Codex GPT-5.5 review of the standalone fake C++ gRPC Shard
worker. Four root protocol defects fixed:

- Fail closed before SessionOpen: a per-session `opened` flag gates
  chunk/decode so no activation bypasses lifecycle, cancellation, epoch
  or flow-control state (terminal ERROR_CODE_INTERNAL), even when an
  out-of-band Cancel created placeholder state.
- Strict flow-control negotiation: NegotiateFlow takes the strictest of
  peer-vs-worker bounds (mirrors codec.negotiate_flow_control) and the
  negotiated per-session max_chunk_bytes is enforced on every bundle
  instead of trusting the peer proposal.
- In-stream ReleaseSignal now erases session state immediately.
- SessionOpen rejects incompatible schema, fingerprint, and shard-range
  identity and reports the worker's own served fingerprint rather than
  echoing the caller.

Adds 9 regression tests (worker suite 18 -> 27). Real gates on the
rebuilt pinned-gRPC binary: cmake build exit 0; ctest 2/2; worker
pytest 27 passed; harness+protocol 63 passed; compileall 0; diff --check
clean; ldd/nm show 0 llama/ggml linkage. DGR-033 passes -> true.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-26 22:57:03 +03:00

166 lines
6.9 KiB
C++

// Deterministic, model-free fake ShardEngine for the native worker (DGR-033).
//
// This is the C++ analogue of `meshnet_node.fake_shard_engine.FakeShardEngine`
// (DGR-032): a pure fixture that performs a *bounded real forward* over the
// bytes it received off the socket and never links, loads, or dispatches to
// llama.cpp. It exists to prove the standalone worker process, stream,
// lifecycle, and supervision shape before any real engine is bound (DGR-037).
//
// The "forward" is deliberately transport-verifiable rather than semantic: it
// reassembles a tensor's fragments, checks they tile exactly, and derives a
// CRC32C over the uncompressed bytes — the same rule the schema's `Checksum`
// declares and the same bounded forward the DGR-024 Python surface performs.
// Feeding the same bytes back (echo) lets a client prove the payload truly
// traversed the wire and returned unmodified; a direct hop and an opaque relay
// of the identical frames therefore yield byte-identical responses.
//
// There is no arbitrary-graph entry point here and no llama.cpp RPC: the engine
// only knows how to reassemble/checksum a bundle. That is the whole point of a
// fixture worker (acceptance criterion 4).
#ifndef MESHNET_NATIVE_WORKER_FAKE_ENGINE_H_
#define MESHNET_NATIVE_WORKER_FAKE_ENGINE_H_
#include <algorithm>
#include <cstdint>
#include <optional>
#include <string>
#include <vector>
#include "shard_runtime.pb.h"
namespace meshnet::worker {
namespace sp = ::meshnet::shard::v1;
// Standard CRC-32 (ISO-HDLC / zlib polynomial 0xEDB88320, reflected).
//
// The schema's `Checksum` field is labelled CRC32C, but the DGR-024 Python
// runtime surface (`shard_runtime_server.py`) computes it with `zlib.crc32`
// (standard CRC-32, not the Castagnoli CRC32C). This worker deliberately mirrors
// that exact computation so its checksum acceptance is byte-for-byte identical
// to the existing Python gRPC surface and to a relayed frame's expectations.
inline 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;
}
// Outcome of validating one bundle before the bounded forward runs.
struct BundleCheck {
// Set when the bundle is malformed/corrupt (maps to PAYLOAD_CORRUPT).
std::optional<std::string> corrupt_detail;
// Set when the declared payload exceeds the negotiated per-chunk ceiling
// (maps to RESOURCE_EXHAUSTED) — the worker refuses unbounded messages.
std::optional<std::string> oversize_detail;
};
// The fake engine's only capability: verify a bundle tiles and checksums, and
// that it stays within the negotiated byte ceiling. Mirrors `_validate_bundle`
// in `shard_runtime_server.py` plus the bounded-message rule DGR-033 adds.
class FakeShardEngine {
public:
// Marker mirroring `FakeShardEngine.EVIDENCE_CLASS` so a future parity check
// (DGR-036) can assert this is a fixture, not a real engine.
static constexpr const char* kEvidenceClass = "fixture";
FakeShardEngine() = default;
// `max_chunk_bytes` is the per-session *negotiated* ceiling (the strictest of
// the worker's own limit and the peer's proposal), passed in on every call so
// the engine enforces exactly what the SessionOpen handshake settled — never a
// value the peer proposed unilaterally.
BundleCheck Validate(const sp::TensorBundle& bundle, uint64_t max_chunk_bytes) const {
BundleCheck result;
for (const auto& tensor : bundle.tensors()) {
// Bounded message: a declared payload larger than the ceiling is refused
// before any reassembly work is done.
if (max_chunk_bytes != 0 && tensor.total_bytes() > max_chunk_bytes) {
result.oversize_detail =
"tensor '" + tensor.name() + "': declared total_bytes " +
std::to_string(tensor.total_bytes()) + " exceeds max_chunk_bytes " +
std::to_string(max_chunk_bytes);
return result;
}
// Fragments must tile the wire body exactly: no hole, no overlap.
std::vector<const sp::TensorFragment*> ordered;
ordered.reserve(tensor.fragments_size());
for (const auto& fragment : tensor.fragments()) {
ordered.push_back(&fragment);
}
std::sort(ordered.begin(), ordered.end(),
[](const sp::TensorFragment* a, const sp::TensorFragment* b) {
return a->byte_offset() < b->byte_offset();
});
uint64_t expected_offset = 0;
std::string payload;
for (const auto* fragment : ordered) {
if (fragment->byte_offset() != expected_offset) {
result.corrupt_detail =
"tensor '" + tensor.name() + "': fragment at offset " +
std::to_string(fragment->byte_offset()) +
" does not tile the preceding " + std::to_string(expected_offset) +
" bytes (gap or overlap)";
return result;
}
payload.append(fragment->payload());
expected_offset += fragment->payload().size();
}
if (tensor.compression() == sp::COMPRESSION_NONE &&
expected_offset != tensor.total_bytes()) {
result.corrupt_detail =
"tensor '" + tensor.name() + "': fragments cover " +
std::to_string(expected_offset) + " bytes, declared total_bytes is " +
std::to_string(tensor.total_bytes());
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();
std::string actual_be(4, '\0');
actual_be[0] = static_cast<char>((actual >> 24) & 0xFF);
actual_be[1] = static_cast<char>((actual >> 16) & 0xFF);
actual_be[2] = static_cast<char>((actual >> 8) & 0xFF);
actual_be[3] = static_cast<char>(actual & 0xFF);
if (declared != actual_be) {
result.corrupt_detail = "tensor '" + tensor.name() + "': checksum mismatch";
return result;
}
}
}
return result;
}
// Bounded real forward: fold every fragment's payload through CRC32C so the
// digest is only reproducible if the payload really traversed the wire.
uint32_t BoundedForward(const sp::TensorBundle& bundle) const {
uint32_t digest = 0;
for (const auto& tensor : bundle.tensors()) {
for (const auto& fragment : tensor.fragments()) {
digest = Crc32(fragment.payload(), digest);
}
}
return digest;
}
};
} // namespace meshnet::worker
#endif // MESHNET_NATIVE_WORKER_FAKE_ENGINE_H_