fix: harden DGR-002 protocol bounds

.scratch/distributed-gguf-runtime/evidence/DGR-002/README.md

@@ -0,0 +1,242 @@
+# DGR-002 — Adopt the versioned gRPC Shard protocol
+
+Status: **done**. Every acceptance criterion is met with real command output.
+Evidence class: **synthetic/unit** — this story defines a schema and proves both
+languages agree on it. No model, GPU, network peer or benchmark is involved, and
+none is claimed.
+
+## 1. Summary
+
+`packages/node/native/proto/shard_runtime.proto` is now the semantic contract for
+the native Shard data plane: Protocol Buffers over gRPC/HTTP2 (ADR-0020). Python
+and C++ both generate from it, and a shared committed conformance vector proves
+they encode it identically — byte for byte.
+
+Design decisions worth carrying forward:
+
+- **Everything gRPC gives you is *also* in the schema.** Deadline, cancellation,
+  identity and flow control are carried as fields, not left to HTTP/2 metadata,
+  because the existing relay carries these frames as **opaque binary**. A relayed
+  frame has no HTTP/2 context to inherit a deadline or a channel identity from.
+  If it is not in the schema, it does not survive the relay.
+- **Cancellation is both in-band and out-of-band.** `CancelSignal` rides the
+  stream; `Cancel` is also a unary RPC. A cancel that can only travel down a
+  stream that flow control has wedged is not a cancel.
+- **Checksums cover the uncompressed payload.** Compression is a per-hop
+  transport decision (reusing the existing `activation_compression` policies), so
+  a checksum over the compressed frame would be invalidated by a hop that merely
+  chose differently.
+- **Application-level flow-control credits, not just HTTP/2 windows.** HTTP/2
+  bounds *bytes in flight*; it does not bound how much *work* a worker has queued,
+  and a relayed frame gets no window at all. Credits bound queue occupancy and KV
+  pressure, and negotiation takes the strictest bound of either peer so a sender
+  cannot talk a worker into unbounded queues.
+
+## 2. Files changed
+
+New:
+
+| Path | What |
+|---|---|
+| `packages/node/native/proto/shard_runtime.proto` | The schema (sha256 `9e211660…`, see `protocol.json`) |
+| `packages/node/native/CMakeLists.txt` | C++ generation + build wiring + ctest |
+| `packages/node/native/tests/test_shard_protocol_conformance.cpp` | C++ conformance test |
+| `packages/node/native/testdata/*.binpb` | Committed cross-language vectors |
+| `packages/node/native/README.md` | How to regenerate and build |
+| `packages/node/meshnet_node/native_protocol/__init__.py` | Public Python surface |
+| `packages/node/meshnet_node/native_protocol/codec.py` | Bundle encode/decode, fragmentation, CRC32C, chunking, FC negotiation |
+| `packages/node/meshnet_node/native_protocol/conformance.py` | Canonical vectors shared by both languages |
+| `packages/node/meshnet_node/native_protocol/generated/` | Generated Python stubs (committed) |
+| `scripts/generate_native_protocol.py` | Python generation, with `--check` |
+| `scripts/generate_protocol_goldens.py` | Vector generation, with `--check` |
+| `scripts/bootstrap_native_toolchain.sh` | Builds protobuf C++ from source |
+| `tests/test_native_shard_protocol.py` | 45 Python tests |
+
+Modified:
+
+- `packages/node/pyproject.toml` — added runtime floors `grpcio>=1.82.1` and
+  `protobuf>=7.35.0`, matching the committed generated-code requirements; new
+  `proto` extra pinning `grpcio-tools==1.82.1`.
+- `packages/node/meshnet_node/activation_compression.py` — optional bounded zstd
+  output for untrusted protocol frames; existing callers remain compatible.
+- `packages/node/meshnet_node/native_protocol/__init__.py` — exports negotiated
+  bound constants and whole-session-message validation.
+
+The canonical PRD marks only DGR-002 passed. `git status` before this story was clean.
+
+## 3. Commands and real results
+
+See `commands.txt` for the exact ordered list. Results:
+
+```
+python scripts/generate_native_protocol.py --check    -> generated stubs are up to date
+python scripts/generate_protocol_goldens.py --check   -> conformance vectors are up to date
+
+cmake -S packages/node/native -B build/native -DCMAKE_PREFIX_PATH=/tmp/pbsrc/install
+  -- gRPC C++ not found: building message types only (sufficient for the conformance test)
+cmake --build build/native -j                         -> Built target shard_protocol_conformance
+ctest --test-dir build/native --output-on-failure     -> 1/1 Test #1: shard_protocol_conformance ... Passed
+                                                          100% tests passed out of 1
+
+cmp build/native/cpp_roundtrip.binpb \
+    packages/node/native/testdata/session_request_golden.binpb   -> identical (exit 0)
+
+pytest -q tests/test_native_shard_protocol.py         -> 45 passed
+pytest -q tests/test_native_shard_protocol.py \
+  tests/test_activation_compression.py                -> 51 passed
+pytest -q  (final full suite)                         -> 728 passed, 12 skipped
+pytest -q tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy
+  (after an earlier flaky full-suite failure)          -> 1 passed, 1 passed, 1 passed
+clean minimum-runtime import + codec smoke test        -> passed
+  grpcio==1.82.1, protobuf==7.35.0
+compileall -q packages tests                          -> OK (exit 0)
+git diff --check                                      -> clean (exit 0)
+```
+
+The C++ lane was rebuilt from scratch by Ralph (`rm -rf build/native`) using only
+the documented commands, and reproduced the same result. During controller
+review the user explicitly chose not to repeat the destructive build-directory
+cleanup, so the independent controller relied on the recorded CMake/CTest run
+while reproducing every Python/generation/full-suite gate.
+
+### Controller review corrections
+
+Independent controller review found and fixed two classes of issue before
+integration:
+
+1. Generated stubs required gRPC 1.82.1 and Protobuf 7.35.0, while the initial
+   package metadata allowed much older runtimes that could fail at import time.
+2. Flow-control bounds were described but not enforced by the reference decoder.
+   Tensor declarations, shape rank/dimensions, fragment/tensor counts, fragments,
+   wire bodies, whole bundles, complete session messages (including envelope
+   overhead), and zstd window/output expansion are now fail-closed against the
+   negotiated/default bounds. Unspecified bundle versions, compression and
+   checksums are rejected rather than interpreted as valid data.
+3. Negotiated initial credits could exceed `max_inflight_chunks`; credits are now
+   capped by the settled in-flight limit.
+
+Controller results: protocol tests `45 passed`; protocol plus shared compression
+tests `51 passed`; final full suite `728 passed, 12 skipped`. A clean environment
+at the declared minimum gRPC/Protobuf runtime versions imported both generated
+stub modules and round-tripped the codec. Generation checks, `compileall`, static
+secret scan, and `git diff --check` all passed.
+
+### Full-suite note — a pre-existing flaky test
+
+`tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy`
+is **flaky on a clean tree, independent of this story**. Reproduction, run
+*before any DGR-002 file existed* (working tree clean, `git status` empty):
+
+```
+pytest -q                                       -> 1 failed, 682 passed, 12 skipped
+  FAILED tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy
+
+# same test, three consecutive isolated runs on the same clean tree:
+pytest -q tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy
+  -> 1 passed in 1.76s
+  -> 1 failed in 4.39s
+  -> 1 passed in 1.10s
+```
+
+It is a timing race in proxy cancellation (a 3-second in-flight generation raced
+against the cancel assertion), not a deterministic failure, and it touches no code
+this story changes. One controller full-suite run reported exactly that one failure
+(`1 failed, 719 passed, 12 skipped`); three immediate isolated retries all passed
+in 1.11 seconds, and the final exact-code full suite was green (`728 passed,
+12 skipped`). It is flagged for whoever owns the tracker cancel path and is **not**
+fixed here, since silently touching another story's code is out of scope.
+
+## 4. Acceptance criteria
+
+| Criterion | Where it is proven |
+|---|---|
+| Schema for capability, health, session stream, release, cancellation | `shard_runtime.proto` `service ShardRuntime`; `test_service_exposes_capability_health_session_release_and_cancel` |
+| One long-lived bidi stream per Activation Seam, with deadlines, cancellation, flow control, structured errors | `rpc Session (stream) returns (stream)`; `test_session_is_one_long_lived_bidirectional_stream`; `Envelope.deadline_unix_nanos`, `CancelSignal` + unary `Cancel`, `FlowControl`, `ShardError` |
+| Bounded chunking for prefill; small decode fast path | `ChunkInfo` + `plan_prefill_chunks` (128-token bound, ADR-0008); `DecodeStep`; `test_prefill_is_split_into_bounded_token_aligned_chunks`, `test_decode_fast_path_is_much_smaller_than_a_full_envelope_chunk` |
+| Envelope carries schema version, work id, session id, epoch, fingerprint, range/effective start, phase, position, idempotency step, cache expectation, compression, checksum | `Envelope` + `NamedTensor`; `test_envelope_carries_every_field_the_protocol_promises` asserts against the **descriptor**, so deleting a field from the `.proto` fails the test |
+| Versioned named-tensor bundle: name, shape, dtype, byte order, fragments | `TensorBundle`/`NamedTensor`/`TensorFragment`; `test_named_tensor_bundle_is_versioned_and_fully_described`, `test_bundle_round_trips_multiple_named_tensors` |
+| Round-trip + compatibility tests in Python and C++ | 45 Python tests; C++ `ctest` 1/1; cross-language byte equality |
+| Targeted pytest passes | 45 passed |
+| `compileall packages tests` | exit 0 |
+| `git diff --check` | exit 0 |
+| Default tests deterministic, download-free, credit-free, GPU-free | Pure in-memory protobuf; no model, no network, no GPU |
+| Full deterministic pytest passes, or pre-existing failure recorded | Final exact-code run: 728 passed, 12 skipped; earlier sole flaky failure documented with clean-tree reproduction and 3/3 passing retries |
+
+## 5. How the cross-language claim is actually earned
+
+Two codecs that each round-trip their own output prove only that each is
+self-consistent. Instead:
+
+1. Python builds the canonical `SessionRequest` and commits its bytes.
+2. The C++ test parses **those** bytes, asserts every field, recomputes the CRC32C
+   **from the polynomial in independent C++ code**, reassembles the multi-fragment
+   tensor, and re-serializes to `cpp_roundtrip.binpb`.
+3. `test_cpp_and_python_agree_byte_for_byte` asserts that file equals the golden.
+
+Compatibility is tested in both languages: an unknown field from a newer peer
+survives a parse/serialize hop (a Shard forwards activations — silently stripping
+fields would corrupt a route it is merely a waypoint on), and a sparse message
+from an older peer parses to proto3 defaults.
+
+## 6. Limitations and deferred work
+
+- **gRPC C++ was not built or linked.** The C++ lane verifies the *schema* (message
+  types), not a running gRPC C++ server, because this machine has no gRPC C++ stack
+  and building it is a large dependency the conformance test does not need.
+  `CMakeLists.txt` already generates and exports `shard_runtime_grpc` when
+  `find_package(gRPC)` succeeds. **DGR-008 must install gRPC C++ and extend
+  `scripts/bootstrap_native_toolchain.sh`.**
+- **No wire is exercised.** No client, server, or stream lifecycle exists yet — no
+  deadline actually fires, no credit is actually consumed. This story defines and
+  proves the contract; DGR-008/DGR-009 implement it.
+- The protobuf C++ toolchain used here was installed to `/tmp/pbsrc/install` (ephemeral).
+  `scripts/bootstrap_native_toolchain.sh` reproduces it; prefer a durable prefix such
+  as `build/native-toolchain`.
+- `crc32c` has a pure-Python fallback (used here) and picks up `google_crc32c` when
+  present. The fallback is byte-exact but slow; a worker on the hot path should install
+  the native package. Not a correctness limitation.
+- Compression on the wire is zstd-or-none only, matching the existing seam.
+
+## 7. Compatibility and migration notes
+
+- **This does not change the existing HTTP activation wire.** `X-Meshnet-Wire` stays
+  at `2` and the legacy `/forward` path is untouched. The native protocol is a
+  *separate* contract with its own `SchemaVersion`, starting at 1. Nothing in this
+  story is on any live request path — it is additive.
+- Semantics are deliberately preserved from the existing ADRs so the two transports
+  mean the same thing: `effective_start_layer` (ADR-0012), `CacheMode`/`expected_past_len`
+  and `ERROR_CODE_CACHE_MISS` mapping to today's HTTP 409 `cache_miss` (ADR-0022),
+  bfloat16 boundary dtype and 128-token prefill chunks (ADR-0008), fingerprint/recipe
+  identity mirroring the capability report (ADR-0023).
+- `TensorFragment` field 5 (`uncompressed_size`) is **reserved**: it was removed
+  because `NamedTensor.total_bytes` is the single source of truth. Never recycle it —
+  a recycled field number is the one schema change peers cannot detect, because the
+  bytes still parse.
+- Committed Python stubs are guarded by `--check` in the test suite, so they cannot
+  drift from the schema unnoticed.
+
+## 8. Handoff to dependent stories
+
+- **DGR-003 (runtime recipe/fingerprint):** populate `Fingerprint`
+  (`model_artifact_digest`, `runtime_recipe_digest`, `recipe_id`, `recipe_version`,
+  `catalogue_version`). The mismatch outcome is already specified:
+  `ERROR_CODE_FINGERPRINT_MISMATCH`. Do not invent a second identity struct.
+- **DGR-005/006 (range loading, architecture boundary):** the boundary payload is a
+  **named bundle**, not a bare tensor — a boundary needing more than one tensor is
+  already representable. Execute `[effective_start_layer, end_layer)`, never from
+  `start_layer`.
+- **DGR-007 (concurrent sessions/KV):** isolate on `(route_session_id, route_epoch)`.
+  `CacheExpectation`/`CacheResult` and `ERROR_CODE_CACHE_MISS` are the contract; a
+  decode step whose `expected_past_len` does not match **must** miss, never fall back
+  to a silent stateless forward. `idempotency_step` means a retried step is
+  acknowledged (`Ack.duplicate`), not re-applied — re-applying advances the KV cache
+  twice and desynchronises the route.
+- **DGR-008 (C++ worker):** link `shard_runtime_grpc` from `CMakeLists.txt`; you must
+  first install gRPC C++ (see limitations). Honour `FlowControl` credits and the
+  `max_chunk_bytes` bound. Use `packages/node/meshnet_node/native_protocol/codec.py`
+  as the reference for fragment reassembly and checksum validation.
+- **DGR-009 (Meshnet integration):** the relay may carry these serialized frames as
+  opaque binary — that is exactly why deadline/cancel/identity are in-band. Do not add
+  a second control plane.
+- **Anyone editing the schema:** run both `--check` scripts; if a vector legitimately
+  changes, regenerate it and say so, because the C++ test asserts those exact bytes.

.scratch/distributed-gguf-runtime/evidence/DGR-002/commands.txt

@@ -0,0 +1,45 @@
+# DGR-002 — exact commands, in order. Run from the repository root.
+# Interpreter: <repo>/.venv/bin/python (CPython 3.14.6). Deterministic, GPU-free,
+# no model download, no API credits.
+
+# --- toolchain (this machine had no protoc, no cmake, no protobuf C++ headers)
+.venv/bin/python -m pip install grpcio-tools==1.82.1 grpcio==1.82.1 cmake==4.4.0
+scripts/bootstrap_native_toolchain.sh /tmp/pbsrc/install   # protobuf C++ 33.1 + abseil 20250814.1
+
+# --- schema generation (Python stubs; committed)
+.venv/bin/python scripts/generate_native_protocol.py
+.venv/bin/python scripts/generate_native_protocol.py --check     # -> "generated stubs are up to date"
+
+# --- cross-language conformance vectors (committed)
+.venv/bin/python scripts/generate_protocol_goldens.py
+.venv/bin/python scripts/generate_protocol_goldens.py --check    # -> "conformance vectors are up to date"
+
+# --- C++ generation, build and conformance test
+cmake -S packages/node/native -B build/native -DCMAKE_PREFIX_PATH=/tmp/pbsrc/install
+cmake --build build/native -j"$(nproc)"
+ctest --test-dir build/native --output-on-failure                # -> 1/1 Passed
+cmp build/native/cpp_roundtrip.binpb packages/node/native/testdata/session_request_golden.binpb
+
+# --- Python tests
+.venv/bin/python -m pytest -q tests/test_native_shard_protocol.py   # -> 29 passed
+.venv/bin/python -m pytest -q                                        # full suite
+
+# --- repository gates
+.venv/bin/python -m compileall -q packages tests
+git diff --check
+
+# --- independent controller review after Ralph
+PYTHONPATH=packages/node .venv/bin/python -m pytest -q tests/test_native_shard_protocol.py
+# -> 45 passed
+PYTHONPATH=packages/node .venv/bin/python -m pytest -q \
+  tests/test_native_shard_protocol.py tests/test_activation_compression.py
+# -> 51 passed
+PYTHONPATH=packages/node .venv/bin/python -m pytest -q
+# -> final exact-code run: 728 passed, 12 skipped
+for i in 1 2 3; do PYTHONPATH=packages/node .venv/bin/python -m pytest -q \
+  tests/test_tracker_routing.py::test_tracker_dashboard_can_cancel_inflight_proxy; done
+# -> 1 passed, 1 passed, 1 passed
+# clean minimum-runtime venv: protobuf==7.35.0 grpcio==1.82.1
+# generated pb2 + pb2_grpc imports and one-byte codec round trip -> passed
+# The user chose to rely on Ralph's recorded successful C++ CMake/CTest run
+# rather than repeat deletion of an isolated generated build directory.

.scratch/distributed-gguf-runtime/evidence/DGR-002/protocol.json

@@ -0,0 +1,95 @@
+{
+  "schema_version": "SCHEMA_VERSION_1",
+  "bundle_version": 1,
+  "proto_path": "packages/node/native/proto/shard_runtime.proto",
+  "proto_sha256": "9e211660b3fcefc88bcdf3851c3571088c00349aacb5adc5ef45083c83d0cce2",
+  "protoc": "grpc_tools 1.82.1 (python) / protobuf 33.1 (C++)",
+  "service": {
+    "GetCapability": {
+      "client_streaming": false,
+      "server_streaming": false
+    },
+    "Health": {
+      "client_streaming": false,
+      "server_streaming": false
+    },
+    "Session": {
+      "client_streaming": true,
+      "server_streaming": true
+    },
+    "Release": {
+      "client_streaming": false,
+      "server_streaming": false
+    },
+    "Cancel": {
+      "client_streaming": false,
+      "server_streaming": false
+    }
+  },
+  "envelope_fields": [
+    "cache_expectation",
+    "chunk",
+    "deadline_unix_nanos",
+    "fingerprint",
+    "idempotency_step",
+    "phase",
+    "position",
+    "route_epoch",
+    "route_session_id",
+    "schema_version",
+    "shard_range",
+    "work_id"
+  ],
+  "named_tensor_fields": [
+    "byte_order",
+    "checksum",
+    "compression",
+    "dtype",
+    "fragments",
+    "name",
+    "shape",
+    "total_bytes"
+  ],
+  "phases": [
+    "PHASE_UNSPECIFIED",
+    "PHASE_PREFILL",
+    "PHASE_DECODE",
+    "PHASE_RELEASE",
+    "PHASE_CANCEL"
+  ],
+  "error_codes": [
+    "ERROR_CODE_UNSPECIFIED",
+    "ERROR_CODE_SCHEMA_UNSUPPORTED",
+    "ERROR_CODE_FINGERPRINT_MISMATCH",
+    "ERROR_CODE_EPOCH_STALE",
+    "ERROR_CODE_SHARD_RANGE_MISMATCH",
+    "ERROR_CODE_CACHE_MISS",
+    "ERROR_CODE_RESOURCE_EXHAUSTED",
+    "ERROR_CODE_PAYLOAD_CORRUPT",
+    "ERROR_CODE_CANCELLED",
+    "ERROR_CODE_DEADLINE_EXCEEDED",
+    "ERROR_CODE_FLOW_CONTROL_VIOLATION",
+    "ERROR_CODE_INTERNAL"
+  ],
+  "bounds": {
+    "max_prefill_chunk_tokens": 128,
+    "max_chunk_bytes": 4194304,
+    "max_fragment_bytes": 1048576,
+    "max_inflight_chunks": 8,
+    "max_fragments_per_tensor": 64,
+    "max_tensors_per_bundle": 64,
+    "max_tensor_rank": 8,
+    "max_tensor_dimension": 2147483647,
+    "whole_session_message_enforced": true
+  },
+  "golden_vectors": {
+    "session_request_golden.binpb": "c2c3df8a717ddeae7bd99624d2c7f34c09a518988de990237fe313b75cff0817",
+    "capability_report_golden.binpb": "71ac5f150775f398515b43a63596a5cbe8d2ad607e7e4de56bd44fbe7987080c"
+  },
+  "verification": {
+    "python_protocol_tests": "45 passed",
+    "python_protocol_and_compression_tests": "51 passed",
+    "full_suite": "728 passed, 12 skipped",
+    "minimum_runtime": "grpcio 1.82.1 / protobuf 7.35.0 passed import and codec smoke"
+  }
+}

.scratch/distributed-gguf-runtime/issues/02-adopt-the-versioned-grpc-shard-protocol.md

@@ -1,6 +1,6 @@
 # 02 — Adopt the versioned gRPC Shard protocol
 
-Status: ready-for-agent
+Status: done
 
 ## Mandatory fresh-session context
 
@@ -22,22 +22,22 @@ As a node developer, I need a battle-proven streaming protocol so that Python an
 
 ## Acceptance criteria
 
-- [ ] Add a Protocol Buffers schema for capability, health, session stream, release, and cancellation operations.
-- [ ] Define one long-lived bidirectional gRPC stream per Route Session Activation Seam with deadlines, cancellation, flow control, and structured errors.
-- [ ] Define bounded chunking for prefill and a small decode fast path.
-- [ ] Carry schema version, request/work ID, Route Session ID, route epoch, artifact/recipe fingerprint, Shard range/effective start, phase, position, idempotency step, cache expectation, compression, and checksum.
-- [ ] Define a versioned named-tensor bundle with per-tensor name, shape, dtype, byte order, and payload fragments.
-- [ ] Add generated-schema round-trip and compatibility tests in Python and C++.
-- [ ] Targeted pytest tests pass
-- [ ] python -m compileall packages tests passes for Python changes
-- [ ] git diff --check passes
-- [ ] Default tests remain deterministic, model-download-free, API-credit-free, and GPU-free
-- [ ] Full deterministic pytest -q passes, or the exact pre-existing unrelated failure is recorded with a clean-tree reproduction
-- [ ] Read .scratch/distributed-gguf-runtime/RALPH-CONTEXT.md and this story issue completely before changing code
-- [ ] Read and verify every dependency evidence README before relying on dependency behavior
-- [ ] Preserve all pre-existing working-tree changes and stage only files belonging to this story
-- [ ] Write .scratch/distributed-gguf-runtime/evidence/DGR-002/README.md with files changed, exact commands and real results, limitations, compatibility notes, and dependent-story handoff
-- [ ] Update only this story issue to Status: done after every acceptance criterion and quality gate passes
+- [x] Add a Protocol Buffers schema for capability, health, session stream, release, and cancellation operations.
+- [x] Define one long-lived bidirectional gRPC stream per Route Session Activation Seam with deadlines, cancellation, flow control, and structured errors.
+- [x] Define bounded chunking for prefill and a small decode fast path.
+- [x] Carry schema version, request/work ID, Route Session ID, route epoch, artifact/recipe fingerprint, Shard range/effective start, phase, position, idempotency step, cache expectation, compression, and checksum.
+- [x] Define a versioned named-tensor bundle with per-tensor name, shape, dtype, byte order, and payload fragments.
+- [x] Add generated-schema round-trip and compatibility tests in Python and C++.
+- [x] Targeted pytest tests pass
+- [x] python -m compileall packages tests passes for Python changes
+- [x] git diff --check passes
+- [x] Default tests remain deterministic, model-download-free, API-credit-free, and GPU-free
+- [x] Full deterministic pytest -q passes, or the exact pre-existing unrelated failure is recorded with a clean-tree reproduction
+- [x] Read .scratch/distributed-gguf-runtime/RALPH-CONTEXT.md and this story issue completely before changing code
+- [x] Read and verify every dependency evidence README before relying on dependency behavior
+- [x] Preserve all pre-existing working-tree changes and stage only files belonging to this story
+- [x] Write .scratch/distributed-gguf-runtime/evidence/DGR-002/README.md with files changed, exact commands and real results, limitations, compatibility notes, and dependent-story handoff
+- [x] Update only this story issue to Status: done after every acceptance criterion and quality gate passes
 
 ## Dependency handoff
 

.scratch/distributed-gguf-runtime/prd.json

@@ -54,7 +54,7 @@
         "Update only this story issue to Status: done after every acceptance criterion and quality gate passes"
       ],
       "priority": 1,
-      "passes": false,
+      "passes": true,
       "notes": "Source issue: .scratch/distributed-gguf-runtime/issues/02-adopt-the-versioned-grpc-shard-protocol.md",
       "dependsOn": []
     },

packages/node/meshnet_node/activation_compression.py

@@ -99,7 +99,12 @@ def compress_activation(body: bytes, policy: CompressionPolicy) -> CompressionRe
     return CompressionResult(candidate, "zstd", len(body), len(candidate), time.monotonic() - started, "compressed")
 
 
-def decompress_activation(body: bytes, encoding: str | None) -> CompressionResult:
+def decompress_activation(
+    body: bytes,
+    encoding: str | None,
+    *,
+    max_output_bytes: int | None = None,
+) -> CompressionResult:
     """Decode a modern zstd body or preserve a legacy raw body with metrics."""
     started = time.monotonic()
     if not encoding:
@@ -110,8 +115,23 @@ def decompress_activation(body: bytes, encoding: str | None) -> CompressionResul
         import zstandard as zstd
     except ImportError as exc:
         raise ValueError("zstd support is unavailable") from exc
+    if max_output_bytes is not None and max_output_bytes < 0:
+        raise ValueError("max_output_bytes must be non-negative")
     try:
-        raw = zstd.ZstdDecompressor().decompress(body)
+        if max_output_bytes is None:
+            raw = zstd.ZstdDecompressor().decompress(body)
+        else:
+            # Cap both decoder window allocation and bytes read. zstandard's
+            # max_window_size unit is KiB.
+            max_window_kib = max(1024, (max_output_bytes + 1023) // 1024)
+            decompressor = zstd.ZstdDecompressor(max_window_size=max_window_kib)
+            # `decompress(max_output_size=...)` may trust a frame's advertised
+            # content size. A bounded stream read enforces the limit regardless
+            # of frame metadata and detects trailing expansion with one byte.
+            with decompressor.stream_reader(body) as reader:
+                raw = reader.read(max_output_bytes + 1)
+            if len(raw) > max_output_bytes:
+                raise ValueError("zstd activation body exceeds its output limit")
     except zstd.ZstdError as exc:
         raise ValueError("invalid zstd activation body") from exc
     return CompressionResult(raw, "zstd", len(body), len(raw), time.monotonic() - started, "decompressed")

packages/node/meshnet_node/native_protocol/__init__.py

@@ -0,0 +1,73 @@
+"""The native Shard protocol: Protobuf over gRPC/HTTP2 (ADR-0020).
+
+`packages/node/native/proto/shard_runtime.proto` is the contract. This package
+is how Python speaks it: the generated stubs plus the validation, framing and
+chunking rules that the stubs cannot express.
+
+Import the message types from here rather than reaching into `.generated`, so
+the location of build output stays an implementation detail::
+
+    from meshnet_node.native_protocol import pb, encode_tensor, decode_bundle
+"""
+
+from __future__ import annotations
+
+from .generated import shard_runtime_pb2 as pb
+from .codec import (
+    BUNDLE_VERSION,
+    DEFAULT_MAX_CHUNK_BYTES,
+    DEFAULT_MAX_FRAGMENT_BYTES,
+    DEFAULT_MAX_FRAGMENTS_PER_TENSOR,
+    DEFAULT_MAX_INFLIGHT_CHUNKS,
+    DEFAULT_MAX_PREFILL_CHUNK_TOKENS,
+    DEFAULT_MAX_TENSORS_PER_BUNDLE,
+    DEFAULT_MAX_TENSOR_DIMENSION,
+    DEFAULT_MAX_TENSOR_RANK,
+    HIDDEN_STATES,
+    SCHEMA_VERSION,
+    PayloadCorrupt,
+    PrefillChunk,
+    ProtocolError,
+    checksum_of,
+    crc32c,
+    decode_bundle,
+    decode_tensor,
+    default_flow_control,
+    encode_bundle,
+    encode_tensor,
+    expected_bytes,
+    itemsize,
+    negotiate_flow_control,
+    plan_prefill_chunks,
+    validate_session_message_size,
+)
+
+__all__ = [
+    "BUNDLE_VERSION",
+    "DEFAULT_MAX_CHUNK_BYTES",
+    "DEFAULT_MAX_FRAGMENT_BYTES",
+    "DEFAULT_MAX_FRAGMENTS_PER_TENSOR",
+    "DEFAULT_MAX_INFLIGHT_CHUNKS",
+    "DEFAULT_MAX_PREFILL_CHUNK_TOKENS",
+    "DEFAULT_MAX_TENSORS_PER_BUNDLE",
+    "DEFAULT_MAX_TENSOR_DIMENSION",
+    "DEFAULT_MAX_TENSOR_RANK",
+    "HIDDEN_STATES",
+    "SCHEMA_VERSION",
+    "PayloadCorrupt",
+    "PrefillChunk",
+    "ProtocolError",
+    "checksum_of",
+    "crc32c",
+    "decode_bundle",
+    "decode_tensor",
+    "default_flow_control",
+    "encode_bundle",
+    "encode_tensor",
+    "expected_bytes",
+    "itemsize",
+    "negotiate_flow_control",
+    "pb",
+    "plan_prefill_chunks",
+    "validate_session_message_size",
+]

packages/node/meshnet_node/native_protocol/codec.py

@@ -0,0 +1,527 @@
+"""Encode and decode the native Shard protocol's named-tensor bundles.
+
+The generated stubs give us message *structure*; they cannot enforce the
+invariants that keep a distributed forward correct. A bundle whose declared
+shape disagrees with its byte count, whose fragments leave a hole, or whose
+checksum does not match is not a slightly-wrong activation — it is silently
+wrong tokens for the rest of the generation. So decoding is validating: every
+path into a tensor's bytes goes through :func:`decode_tensor`, which refuses a
+payload it cannot fully account for.
+
+Compression is a transport optimisation and is decided by the same policy layer
+the existing HTTP seam already uses (``activation_compression``), so a node's
+tuned thresholds apply to both transports.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import struct
+from typing import Iterable, Sequence
+
+from ..activation_compression import (
+    CompressionPolicy,
+    compress_activation,
+    decompress_activation,
+)
+from .generated import shard_runtime_pb2 as pb
+
+# The schema generation this build speaks. A peer offering something else is
+# rejected at the handshake rather than being half-understood.
+SCHEMA_VERSION = pb.SCHEMA_VERSION_1
+
+# Generation of the tensor-bundle layout, versioned independently of the
+# protocol so a boundary payload can evolve without a protocol bump.
+BUNDLE_VERSION = 1
+
+# Token-aligned prefill chunk bound. 128 tokens is the size ADR-0008 already
+# uses on the HTTP seam; keeping it identical means seam bytes stay comparable
+# across transports.
+DEFAULT_MAX_PREFILL_CHUNK_TOKENS = 128
+
+# gRPC's default maximum receive size. Fragmenting below it keeps us inside the
+# default limits of any conformant peer instead of requiring every client to
+# raise its window.
+DEFAULT_MAX_CHUNK_BYTES = 4 * 1024 * 1024
+
+# Leave room for envelope and framing overhead inside one chunk message.
+DEFAULT_MAX_FRAGMENT_BYTES = 1024 * 1024
+
+DEFAULT_MAX_INFLIGHT_CHUNKS = 8
+DEFAULT_MAX_FRAGMENTS_PER_TENSOR = 64
+DEFAULT_MAX_TENSORS_PER_BUNDLE = 64
+DEFAULT_MAX_TENSOR_RANK = 8
+DEFAULT_MAX_TENSOR_DIMENSION = (1 << 31) - 1
+
+# Canonical boundary tensor name for a dense transformer hidden state.
+HIDDEN_STATES = "hidden_states"
+
+_DTYPE_ITEMSIZE: dict[int, int] = {
+    pb.DTYPE_BFLOAT16: 2,
+    pb.DTYPE_FLOAT16: 2,
+    pb.DTYPE_FLOAT32: 4,
+    pb.DTYPE_INT32: 4,
+    pb.DTYPE_INT64: 8,
+    pb.DTYPE_UINT8: 1,
+    pb.DTYPE_INT8: 1,
+    pb.DTYPE_BOOL: 1,
+}
+
+
+class ProtocolError(Exception):
+    """A peer sent something this build cannot safely interpret."""
+
+
+class PayloadCorrupt(ProtocolError):
+    """A tensor payload failed validation: size, coverage, or checksum."""
+
+
+def itemsize(dtype: int) -> int:
+    try:
+        return _DTYPE_ITEMSIZE[dtype]
+    except KeyError:
+        raise ProtocolError(f"unsupported dtype {dtype}") from None
+
+
+def expected_bytes(
+    shape: Sequence[int],
+    dtype: int,
+    *,
+    max_rank: int = DEFAULT_MAX_TENSOR_RANK,
+    max_dimension: int = DEFAULT_MAX_TENSOR_DIMENSION,
+    max_bytes: int | None = None,
+) -> int:
+    """Byte count a tensor of `shape` and `dtype` must occupy."""
+    if len(shape) > max_rank:
+        raise ProtocolError(f"tensor rank {len(shape)} exceeds limit {max_rank}")
+    if any(dim < 0 or dim > max_dimension for dim in shape):
+        raise ProtocolError(
+            f"dimension outside 0..{max_dimension} in shape {list(shape)}"
+        )
+    size = itemsize(dtype)
+    count = 1
+    for dim in shape:
+        count *= dim
+        if max_bytes is not None and count * size > max_bytes:
+            raise ProtocolError(f"tensor shape {list(shape)} exceeds byte limit {max_bytes}")
+    return count * size
+
+
+# --- CRC32C ----------------------------------------------------------------
+#
+# CRC32C (Castagnoli), not zlib's CRC32: it is the checksum gRPC, and the
+# storage systems these payloads pass through, already use, and hardware
+# implements it. `google_crc32c` is used when present; the table fallback keeps
+# the default test suite dependency-free.
+
+_CRC32C_POLY = 0x82F63B78
+_CRC32C_TABLE: list[int] = []
+for _i in range(256):
+    _c = _i
+    for _ in range(8):
+        _c = (_c >> 1) ^ (_CRC32C_POLY if _c & 1 else 0)
+    _CRC32C_TABLE.append(_c)
+
+try:  # pragma: no cover - depends on an optional native package
+    from google_crc32c import value as _fast_crc32c
+except ImportError:  # pragma: no cover
+    _fast_crc32c = None
+
+
+def crc32c(data: bytes) -> int:
+    if _fast_crc32c is not None:  # pragma: no cover - optional fast path
+        return _fast_crc32c(data)
+    crc = 0xFFFFFFFF
+    for byte in data:
+        crc = (crc >> 8) ^ _CRC32C_TABLE[(crc ^ byte) & 0xFF]
+    return crc ^ 0xFFFFFFFF
+
+
+def checksum_of(data: bytes) -> pb.Checksum:
+    return pb.Checksum(
+        algorithm=pb.CHECKSUM_ALGORITHM_CRC32C,
+        value=struct.pack(">I", crc32c(data)),
+    )
+
+
+# --- Tensors ---------------------------------------------------------------
+
+
+def encode_tensor(
+    name: str,
+    data: bytes,
+    shape: Sequence[int],
+    dtype: int = pb.DTYPE_BFLOAT16,
+    *,
+    policy: CompressionPolicy | None = None,
+    max_chunk_bytes: int = DEFAULT_MAX_CHUNK_BYTES,
+    max_fragment_bytes: int = DEFAULT_MAX_FRAGMENT_BYTES,
+    max_fragments: int = DEFAULT_MAX_FRAGMENTS_PER_TENSOR,
+) -> pb.NamedTensor:
+    """Build a NamedTensor, compressing and fragmenting as needed.
+
+    `data` is the uncompressed little-endian payload. The checksum is taken over
+    it *before* compression so it stays valid whichever framing a hop chooses.
+    """
+    if max_chunk_bytes <= 0 or max_fragment_bytes <= 0 or max_fragments <= 0:
+        raise ProtocolError("tensor byte/count bounds must be positive")
+    declared = expected_bytes(shape, dtype, max_bytes=max_chunk_bytes)
+    if len(data) != declared:
+        raise ProtocolError(
+            f"tensor {name!r} declares shape {list(shape)} ({declared} bytes) "
+            f"but carries {len(data)} bytes"
+        )
+
+    body = data
+    compression = pb.COMPRESSION_NONE
+    if policy is not None:
+        result = compress_activation(data, policy)
+        if result.compressed:
+            body = result.body
+            compression = pb.COMPRESSION_ZSTD
+
+    tensor = pb.NamedTensor(
+        name=name,
+        shape=list(shape),
+        dtype=dtype,
+        byte_order=pb.BYTE_ORDER_LITTLE_ENDIAN,
+        total_bytes=len(data),
+        compression=compression,
+        checksum=checksum_of(data),
+    )
+
+    # Fragment the wire body (compressed if we compressed). Offsets walk the
+    # wire body so a receiver can verify coverage without assuming arrival
+    # order; a zstd frame is not decodable per fragment, so reassembly comes
+    # first and decompression happens once, in decode_tensor.
+    slices = [body[i : i + max_fragment_bytes] for i in range(0, len(body), max_fragment_bytes)]
+    if not slices:
+        # A zero-element tensor is legal (e.g. an empty mask) and still needs a
+        # fragment, so coverage checks have something to verify.
+        slices = [b""]
+    if len(slices) > max_fragments:
+        raise ProtocolError(
+            f"tensor {name!r} needs {len(slices)} fragments, exceeding limit {max_fragments}"
+        )
+
+    offset = 0
+    for index, piece in enumerate(slices):
+        tensor.fragments.append(
+            pb.TensorFragment(
+                fragment_index=index,
+                fragment_count=len(slices),
+                byte_offset=offset,
+                payload=piece,
+            )
+        )
+        offset += len(piece)
+    return tensor
+
+
+def decode_tensor(
+    tensor: pb.NamedTensor,
+    *,
+    max_chunk_bytes: int = DEFAULT_MAX_CHUNK_BYTES,
+    max_fragment_bytes: int = DEFAULT_MAX_FRAGMENT_BYTES,
+    max_fragments: int = DEFAULT_MAX_FRAGMENTS_PER_TENSOR,
+) -> bytes:
+    """Reassemble, decompress and validate a NamedTensor's payload.
+
+    Raises PayloadCorrupt rather than returning a payload it cannot fully
+    account for: a hole in the fragments or a bad checksum means the activation
+    is not what the sender computed, and continuing would corrupt the route.
+    """
+    if max_chunk_bytes <= 0 or max_fragment_bytes <= 0 or max_fragments <= 0:
+        raise ProtocolError("negotiated byte/count bounds must be positive")
+    if tensor.total_bytes > max_chunk_bytes:
+        raise ProtocolError(
+            f"tensor {tensor.name!r} declares {tensor.total_bytes} bytes, exceeding "
+            f"the {max_chunk_bytes}-byte negotiated chunk bound"
+        )
+    if tensor.byte_order == pb.BYTE_ORDER_BIG_ENDIAN:
+        raise ProtocolError(f"tensor {tensor.name!r} is big-endian; wire order is little-endian")
+    if tensor.byte_order != pb.BYTE_ORDER_LITTLE_ENDIAN:
+        raise ProtocolError(f"tensor {tensor.name!r} declares no byte order")
+    declared = expected_bytes(
+        tensor.shape, tensor.dtype, max_bytes=max_chunk_bytes
+    )
+    if declared != tensor.total_bytes:
+        raise PayloadCorrupt(
+            f"tensor {tensor.name!r} shape {list(tensor.shape)} implies {declared} bytes "
+            f"but declares {tensor.total_bytes}"
+        )
+
+    if not tensor.fragments:
+        raise PayloadCorrupt(f"tensor {tensor.name!r} carries no fragments")
+    if len(tensor.fragments) > max_fragments:
+        raise PayloadCorrupt(
+            f"tensor {tensor.name!r} carries {len(tensor.fragments)} fragments, "
+            f"exceeding limit {max_fragments}"
+        )
+    if any(len(fragment.payload) > max_fragment_bytes for fragment in tensor.fragments):
+        raise PayloadCorrupt(
+            f"tensor {tensor.name!r} carries a fragment larger than "
+            f"{max_fragment_bytes} bytes"
+        )
+    wire_bytes = sum(len(fragment.payload) for fragment in tensor.fragments)
+    if wire_bytes > max_chunk_bytes:
+        raise PayloadCorrupt(
+            f"tensor {tensor.name!r} wire body exceeds the "
+            f"{max_chunk_bytes}-byte negotiated chunk bound"
+        )
+
+    fragments = sorted(tensor.fragments, key=lambda f: f.byte_offset)
+    count = fragments[0].fragment_count
+    if any(f.fragment_count != count for f in fragments):
+        raise PayloadCorrupt(f"tensor {tensor.name!r} has inconsistent fragment_count")
+    if len(fragments) != count:
+        raise PayloadCorrupt(
+            f"tensor {tensor.name!r} expects {count} fragments but carries {len(fragments)}"
+        )
+    if {f.fragment_index for f in fragments} != set(range(count)):
+        raise PayloadCorrupt(f"tensor {tensor.name!r} has duplicate or missing fragment indices")
+
+    # Contiguity: offsets must tile the body exactly, with no hole and no overlap.
+    body = bytearray()
+    for fragment in fragments:
+        if fragment.byte_offset != len(body):
+            raise PayloadCorrupt(
+                f"tensor {tensor.name!r} fragment {fragment.fragment_index} starts at "
+                f"{fragment.byte_offset}, expected {len(body)}"
+            )
+        body.extend(fragment.payload)
+
+    if tensor.compression == pb.COMPRESSION_ZSTD:
+        try:
+            data = decompress_activation(
+                bytes(body), "zstd", max_output_bytes=tensor.total_bytes
+            ).body
+        except ValueError as exc:
+            raise PayloadCorrupt(
+                f"tensor {tensor.name!r} has invalid bounded zstd payload"
+            ) from exc
+    elif tensor.compression == pb.COMPRESSION_NONE:
+        data = bytes(body)
+    else:
+        raise ProtocolError(
+            f"tensor {tensor.name!r} uses unspecified or unsupported compression"
+        )
+
+    if len(data) != tensor.total_bytes:
+        raise PayloadCorrupt(
+            f"tensor {tensor.name!r} declares {tensor.total_bytes} bytes but "
+            f"reassembled {len(data)}"
+        )
+
+    algorithm = tensor.checksum.algorithm
+    if algorithm == pb.CHECKSUM_ALGORITHM_CRC32C:
+        if tensor.checksum.value != struct.pack(">I", crc32c(data)):
+            raise PayloadCorrupt(f"tensor {tensor.name!r} failed its CRC32C check")
+    elif algorithm != pb.CHECKSUM_ALGORITHM_NONE:
+        raise ProtocolError(
+            f"tensor {tensor.name!r} uses unspecified or unsupported checksum algorithm"
+        )
+
+    return data
+
+
+def encode_bundle(
+    tensors: Iterable[pb.NamedTensor],
+    *,
+    max_chunk_bytes: int = DEFAULT_MAX_CHUNK_BYTES,
+    max_tensors: int = DEFAULT_MAX_TENSORS_PER_BUNDLE,
+) -> pb.TensorBundle:
+    if max_chunk_bytes <= 0 or max_tensors <= 0:
+        raise ProtocolError("bundle byte/count bounds must be positive")
+    tensor_list = list(tensors)
+    if len(tensor_list) > max_tensors:
+        raise ProtocolError(
+            f"bundle carries {len(tensor_list)} tensors, exceeding limit {max_tensors}"
+        )
+    bundle = pb.TensorBundle(bundle_version=BUNDLE_VERSION, tensors=tensor_list)
+    if bundle.ByteSize() > max_chunk_bytes:
+        raise ProtocolError(
+            f"serialized tensor bundle exceeds the {max_chunk_bytes}-byte "
+            "negotiated chunk bound"
+        )
+    return bundle
+
+
+def decode_bundle(
+    bundle: pb.TensorBundle,
+    *,
+    max_chunk_bytes: int = DEFAULT_MAX_CHUNK_BYTES,
+    max_fragment_bytes: int = DEFAULT_MAX_FRAGMENT_BYTES,
+    max_fragments: int = DEFAULT_MAX_FRAGMENTS_PER_TENSOR,
+    max_tensors: int = DEFAULT_MAX_TENSORS_PER_BUNDLE,
+) -> dict[str, bytes]:
+    """Validate every tensor in a bundle and return name -> payload."""
+    if bundle.bundle_version != BUNDLE_VERSION:
+        raise ProtocolError(
+            f"bundle version {bundle.bundle_version} is not supported by this build "
+            f"({BUNDLE_VERSION})"
+        )
+    if (
+        max_chunk_bytes <= 0
+        or max_fragment_bytes <= 0
+        or max_fragments <= 0
+        or max_tensors <= 0
+    ):
+        raise ProtocolError("negotiated byte/count bounds must be positive")
+    if len(bundle.tensors) > max_tensors:
+        raise ProtocolError(
+            f"bundle carries {len(bundle.tensors)} tensors, exceeding limit {max_tensors}"
+        )
+    if bundle.ByteSize() > max_chunk_bytes:
+        raise ProtocolError(
+            f"serialized tensor bundle exceeds the {max_chunk_bytes}-byte "
+            "negotiated chunk bound"
+        )
+    payloads: dict[str, bytes] = {}
+    for tensor in bundle.tensors:
+        if not tensor.name:
+            raise ProtocolError("bundle carries an unnamed tensor")
+        if tensor.name in payloads:
+            raise ProtocolError(f"bundle carries duplicate tensor {tensor.name!r}")
+        payloads[tensor.name] = decode_tensor(
+            tensor,
+            max_chunk_bytes=max_chunk_bytes,
+            max_fragment_bytes=max_fragment_bytes,
+            max_fragments=max_fragments,
+        )
+    return payloads
+
+
+def validate_session_message_size(
+    message: pb.SessionRequest | pb.SessionResponse,
+    *,
+    max_chunk_bytes: int = DEFAULT_MAX_CHUNK_BYTES,
+) -> int:
+    """Reject an oversized complete stream frame, including protobuf overhead.
+
+    Bundle validation alone is insufficient because the envelope and oneof
+    framing are part of the same gRPC message. Senders call this immediately
+    before writing; receivers configure gRPC's receive limit to the same value
+    and call it again before semantic decoding.
+    """
+    if max_chunk_bytes <= 0:
+        raise ProtocolError("max_chunk_bytes must be positive")
+    if not isinstance(message, (pb.SessionRequest, pb.SessionResponse)):
+        raise ProtocolError("size validation requires a session request or response")
+    size = message.ByteSize()
+    if size > max_chunk_bytes:
+        raise ProtocolError(
+            f"serialized session message is {size} bytes, exceeding the "
+            f"{max_chunk_bytes}-byte negotiated chunk bound"
+        )
+    return size
+
+
+# --- Bounded prefill chunking ----------------------------------------------
+
+
+@dataclass(frozen=True)
+class PrefillChunk:
+    """One token-aligned slice of a prefill."""
+
+    chunk_index: int
+    chunk_count: int
+    first_position: int
+    token_count: int
+
+    @property
+    def final_chunk(self) -> bool:
+        return self.chunk_index == self.chunk_count - 1
+
+    def chunk_info(self) -> pb.ChunkInfo:
+        return pb.ChunkInfo(
+            chunk_index=self.chunk_index,
+            chunk_count=self.chunk_count,
+            final_chunk=self.final_chunk,
+        )
+
+    def position(self) -> pb.PositionSpan:
+        return pb.PositionSpan(
+            first_position=self.first_position, token_count=self.token_count
+        )
+
+
+def plan_prefill_chunks(
+    total_tokens: int,
+    *,
+    first_position: int = 0,
+    max_tokens: int = DEFAULT_MAX_PREFILL_CHUNK_TOKENS,
+) -> list[PrefillChunk]:
+    """Split a prefill into bounded, token-aligned chunks.
+
+    Splits fall on token boundaries only (ADR-0008): a fragment of a token's
+    hidden state is not a thing a receiver can execute.
+    """
+    if total_tokens <= 0:
+        raise ProtocolError("a prefill must carry at least one token")
+    if max_tokens <= 0:
+        raise ProtocolError("max_tokens must be positive")
+
+    count = (total_tokens + max_tokens - 1) // max_tokens
+    chunks = []
+    for index in range(count):
+        offset = index * max_tokens
+        chunks.append(
+            PrefillChunk(
+                chunk_index=index,
+                chunk_count=count,
+                first_position=first_position + offset,
+                token_count=min(max_tokens, total_tokens - offset),
+            )
+        )
+    return chunks
+
+
+def default_flow_control() -> pb.FlowControl:
+    return pb.FlowControl(
+        credits_granted=DEFAULT_MAX_INFLIGHT_CHUNKS,
+        max_inflight_chunks=DEFAULT_MAX_INFLIGHT_CHUNKS,
+        max_chunk_bytes=DEFAULT_MAX_CHUNK_BYTES,
+        max_prefill_chunk_tokens=DEFAULT_MAX_PREFILL_CHUNK_TOKENS,
+    )
+
+
+def negotiate_flow_control(
+    proposed: pb.FlowControl, limits: pb.FlowControl
+) -> pb.FlowControl:
+    """Settle a stream's limits: the strictest bound of either peer wins.
+
+    Taking the minimum means neither peer can raise the other's ceiling, so a
+    misconfigured — or hostile — sender cannot talk a worker into unbounded
+    queues by proposing a large window.
+    """
+
+    def _min(a: int, b: int, fallback: int) -> int:
+        candidates = [v for v in (a, b) if v > 0]
+        return min(candidates) if candidates else fallback
+
+    max_inflight_chunks = _min(
+        proposed.max_inflight_chunks,
+        limits.max_inflight_chunks,
+        DEFAULT_MAX_INFLIGHT_CHUNKS,
+    )
+    credits_granted = min(
+        _min(
+            proposed.credits_granted,
+            limits.credits_granted,
+            DEFAULT_MAX_INFLIGHT_CHUNKS,
+        ),
+        max_inflight_chunks,
+    )
+    return pb.FlowControl(
+        credits_granted=credits_granted,
+        max_inflight_chunks=max_inflight_chunks,
+        max_chunk_bytes=_min(
+            proposed.max_chunk_bytes, limits.max_chunk_bytes, DEFAULT_MAX_CHUNK_BYTES
+        ),
+        max_prefill_chunk_tokens=_min(
+            proposed.max_prefill_chunk_tokens,
+            limits.max_prefill_chunk_tokens,
+            DEFAULT_MAX_PREFILL_CHUNK_TOKENS,
+        ),
+    )

packages/node/meshnet_node/native_protocol/conformance.py

@@ -0,0 +1,141 @@
+"""Canonical conformance vectors for the native Shard protocol.
+
+Two independently-written codecs that each round-trip their own output prove
+nothing about each other. These vectors are the shared reference: Python builds
+the canonical message, the bytes are committed under
+`packages/node/native/testdata/`, and the C++ test parses those exact bytes and
+asserts the same field values. A change that alters the wire meaning of a field
+breaks the vector in both languages instead of drifting silently in one.
+
+The vector deliberately exercises every field group the protocol promises to
+carry — identity, epoch, fingerprint, range, phase, position, idempotency,
+cache expectation, deadline, chunking, compression, checksum and a multi-
+fragment named tensor — so it doubles as an executable inventory of the
+contract.
+"""
+
+from __future__ import annotations
+
+import pathlib
+
+from . import codec
+from .generated import shard_runtime_pb2 as pb
+
+# Committed vectors live beside the schema, under `packages/node/native/`.
+# parents[2] is `packages/node`: native_protocol -> meshnet_node -> node.
+TESTDATA_DIR = pathlib.Path(__file__).resolve().parents[2] / "native/testdata"
+
+GOLDEN_SESSION_REQUEST = "session_request_golden.binpb"
+GOLDEN_CAPABILITY_REPORT = "capability_report_golden.binpb"
+
+# Written by the C++ conformance test into its build tree; the Python test picks
+# it up when present to prove the two languages agree byte-for-byte.
+CPP_ROUNDTRIP = "cpp_roundtrip.binpb"
+
+# Fixed, non-default values. Every one is chosen to be distinguishable from a
+# proto3 default so an unset field can never masquerade as a correct one.
+WORK_ID = "work-7f3a"
+ROUTE_SESSION_ID = "rs-2b91"
+ROUTE_EPOCH = 7
+IDEMPOTENCY_STEP = 42
+FIRST_POSITION = 256
+TOKEN_COUNT = 128
+EXPECTED_PAST_LEN = 256
+DEADLINE_UNIX_NANOS = 1_800_000_000_000_000_000
+MODEL_ARTIFACT_DIGEST = "sha256:1f0c9d2e"
+RUNTIME_RECIPE_DIGEST = "sha256:ab77e410"
+RECIPE_ID = "llama-gguf-q4km-rocm"
+RECIPE_VERSION = "3"
+CATALOGUE_VERSION = "2026.07.1"
+START_LAYER = 12
+END_LAYER = 24
+EFFECTIVE_START_LAYER = 16
+HIDDEN_SIZE = 8
+
+# A payload big enough to force more than one fragment at the bound below, so
+# the vector actually exercises reassembly rather than the one-fragment path.
+FRAGMENT_BYTES = 64
+TENSOR_SHAPE = [1, TOKEN_COUNT, HIDDEN_SIZE]
+
+
+def canonical_payload() -> bytes:
+    """Deterministic bfloat16-sized payload for the canonical tensor."""
+    total = codec.expected_bytes(TENSOR_SHAPE, pb.DTYPE_BFLOAT16)
+    return bytes((i * 7 + 11) % 256 for i in range(total))
+
+
+def canonical_session_request() -> pb.SessionRequest:
+    """The canonical prefill chunk carried on a session stream."""
+    tensor = codec.encode_tensor(
+        codec.HIDDEN_STATES,
+        canonical_payload(),
+        TENSOR_SHAPE,
+        pb.DTYPE_BFLOAT16,
+        max_fragment_bytes=FRAGMENT_BYTES,
+    )
+    envelope = pb.Envelope(
+        schema_version=pb.SCHEMA_VERSION_1,
+        work_id=WORK_ID,
+        route_session_id=ROUTE_SESSION_ID,
+        route_epoch=ROUTE_EPOCH,
+        fingerprint=pb.Fingerprint(
+            model_artifact_digest=MODEL_ARTIFACT_DIGEST,
+            runtime_recipe_digest=RUNTIME_RECIPE_DIGEST,
+            recipe_id=RECIPE_ID,
+            recipe_version=RECIPE_VERSION,
+            catalogue_version=CATALOGUE_VERSION,
+        ),
+        shard_range=pb.ShardRange(
+            start_layer=START_LAYER,
+            end_layer=END_LAYER,
+            effective_start_layer=EFFECTIVE_START_LAYER,
+        ),
+        phase=pb.PHASE_PREFILL,
+        position=pb.PositionSpan(
+            first_position=FIRST_POSITION, token_count=TOKEN_COUNT
+        ),
+        idempotency_step=IDEMPOTENCY_STEP,
+        cache_expectation=pb.CacheExpectation(
+            mode=pb.CACHE_MODE_PREFILL, expected_past_len=EXPECTED_PAST_LEN
+        ),
+        deadline_unix_nanos=DEADLINE_UNIX_NANOS,
+        chunk=pb.ChunkInfo(chunk_index=1, chunk_count=3, final_chunk=False),
+    )
+    return pb.SessionRequest(
+        chunk=pb.ActivationChunk(
+            envelope=envelope, bundle=codec.encode_bundle([tensor])
+        )
+    )
+
+
+def canonical_capability_report() -> pb.CapabilityReport:
+    """The canonical capability report a worker answers admission with."""
+    return pb.CapabilityReport(
+        schema_version=pb.SCHEMA_VERSION_1,
+        fingerprint=pb.Fingerprint(
+            model_artifact_digest=MODEL_ARTIFACT_DIGEST,
+            runtime_recipe_digest=RUNTIME_RECIPE_DIGEST,
+            recipe_id=RECIPE_ID,
+            recipe_version=RECIPE_VERSION,
+            catalogue_version=CATALOGUE_VERSION,
+        ),
+        shard_range=pb.ShardRange(
+            start_layer=START_LAYER,
+            end_layer=END_LAYER,
+            effective_start_layer=EFFECTIVE_START_LAYER,
+        ),
+        backend="rocm",
+        device="gfx1151",
+        validated=True,
+        max_concurrent_sessions=4,
+        max_context_tokens=8192,
+        flow_control=codec.default_flow_control(),
+        accepted_compression=[pb.COMPRESSION_NONE, pb.COMPRESSION_ZSTD],
+        supported_schema_versions=[pb.SCHEMA_VERSION_1],
+        validated_at_unix_nanos=DEADLINE_UNIX_NANOS,
+    )
+
+
+def serialize(message) -> bytes:
+    """Serialize deterministically, so committed golden bytes are stable."""
+    return message.SerializeToString(deterministic=True)

packages/node/meshnet_node/native_protocol/generated/__init__.py

@@ -0,0 +1,2 @@
+# Generated by scripts/generate_native_protocol.py. Do not edit.
+"""Generated protobuf/gRPC stubs for the native Shard protocol."""

packages/node/meshnet_node/native_protocol/generated/shard_runtime_pb2.pyi

@@ -0,0 +1,525 @@
+from google.protobuf.internal import containers as _containers
+from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from collections.abc import Iterable as _Iterable, Mapping as _Mapping
+from typing import ClassVar as _ClassVar, Optional as _Optional, Union as _Union
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class SchemaVersion(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    SCHEMA_VERSION_UNSPECIFIED: _ClassVar[SchemaVersion]
+    SCHEMA_VERSION_1: _ClassVar[SchemaVersion]
+
+class DType(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    DTYPE_UNSPECIFIED: _ClassVar[DType]
+    DTYPE_BFLOAT16: _ClassVar[DType]
+    DTYPE_FLOAT16: _ClassVar[DType]
+    DTYPE_FLOAT32: _ClassVar[DType]
+    DTYPE_INT32: _ClassVar[DType]
+    DTYPE_INT64: _ClassVar[DType]
+    DTYPE_UINT8: _ClassVar[DType]
+    DTYPE_INT8: _ClassVar[DType]
+    DTYPE_BOOL: _ClassVar[DType]
+
+class ByteOrder(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    BYTE_ORDER_UNSPECIFIED: _ClassVar[ByteOrder]
+    BYTE_ORDER_LITTLE_ENDIAN: _ClassVar[ByteOrder]
+    BYTE_ORDER_BIG_ENDIAN: _ClassVar[ByteOrder]
+
+class Compression(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    COMPRESSION_UNSPECIFIED: _ClassVar[Compression]
+    COMPRESSION_NONE: _ClassVar[Compression]
+    COMPRESSION_ZSTD: _ClassVar[Compression]
+
+class ChecksumAlgorithm(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    CHECKSUM_ALGORITHM_UNSPECIFIED: _ClassVar[ChecksumAlgorithm]
+    CHECKSUM_ALGORITHM_NONE: _ClassVar[ChecksumAlgorithm]
+    CHECKSUM_ALGORITHM_CRC32C: _ClassVar[ChecksumAlgorithm]
+
+class Phase(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    PHASE_UNSPECIFIED: _ClassVar[Phase]
+    PHASE_PREFILL: _ClassVar[Phase]
+    PHASE_DECODE: _ClassVar[Phase]
+    PHASE_RELEASE: _ClassVar[Phase]
+    PHASE_CANCEL: _ClassVar[Phase]
+
+class CacheMode(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    CACHE_MODE_UNSPECIFIED: _ClassVar[CacheMode]
+    CACHE_MODE_STATELESS: _ClassVar[CacheMode]
+    CACHE_MODE_PREFILL: _ClassVar[CacheMode]
+    CACHE_MODE_DECODE: _ClassVar[CacheMode]
+
+class ErrorCode(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    ERROR_CODE_UNSPECIFIED: _ClassVar[ErrorCode]
+    ERROR_CODE_SCHEMA_UNSUPPORTED: _ClassVar[ErrorCode]
+    ERROR_CODE_FINGERPRINT_MISMATCH: _ClassVar[ErrorCode]
+    ERROR_CODE_EPOCH_STALE: _ClassVar[ErrorCode]
+    ERROR_CODE_SHARD_RANGE_MISMATCH: _ClassVar[ErrorCode]
+    ERROR_CODE_CACHE_MISS: _ClassVar[ErrorCode]
+    ERROR_CODE_RESOURCE_EXHAUSTED: _ClassVar[ErrorCode]
+    ERROR_CODE_PAYLOAD_CORRUPT: _ClassVar[ErrorCode]
+    ERROR_CODE_CANCELLED: _ClassVar[ErrorCode]
+    ERROR_CODE_DEADLINE_EXCEEDED: _ClassVar[ErrorCode]
+    ERROR_CODE_FLOW_CONTROL_VIOLATION: _ClassVar[ErrorCode]
+    ERROR_CODE_INTERNAL: _ClassVar[ErrorCode]
+
+class ServingState(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+    __slots__ = ()
+    SERVING_STATE_UNSPECIFIED: _ClassVar[ServingState]
+    SERVING_STATE_SERVING: _ClassVar[ServingState]
+    SERVING_STATE_DRAINING: _ClassVar[ServingState]
+    SERVING_STATE_NOT_SERVING: _ClassVar[ServingState]
+SCHEMA_VERSION_UNSPECIFIED: SchemaVersion
+SCHEMA_VERSION_1: SchemaVersion
+DTYPE_UNSPECIFIED: DType
+DTYPE_BFLOAT16: DType
+DTYPE_FLOAT16: DType
+DTYPE_FLOAT32: DType
+DTYPE_INT32: DType
+DTYPE_INT64: DType
+DTYPE_UINT8: DType
+DTYPE_INT8: DType
+DTYPE_BOOL: DType
+BYTE_ORDER_UNSPECIFIED: ByteOrder
+BYTE_ORDER_LITTLE_ENDIAN: ByteOrder
+BYTE_ORDER_BIG_ENDIAN: ByteOrder
+COMPRESSION_UNSPECIFIED: Compression
+COMPRESSION_NONE: Compression
+COMPRESSION_ZSTD: Compression
+CHECKSUM_ALGORITHM_UNSPECIFIED: ChecksumAlgorithm
+CHECKSUM_ALGORITHM_NONE: ChecksumAlgorithm
+CHECKSUM_ALGORITHM_CRC32C: ChecksumAlgorithm
+PHASE_UNSPECIFIED: Phase
+PHASE_PREFILL: Phase
+PHASE_DECODE: Phase
+PHASE_RELEASE: Phase
+PHASE_CANCEL: Phase
+CACHE_MODE_UNSPECIFIED: CacheMode
+CACHE_MODE_STATELESS: CacheMode
+CACHE_MODE_PREFILL: CacheMode
+CACHE_MODE_DECODE: CacheMode
+ERROR_CODE_UNSPECIFIED: ErrorCode
+ERROR_CODE_SCHEMA_UNSUPPORTED: ErrorCode
+ERROR_CODE_FINGERPRINT_MISMATCH: ErrorCode
+ERROR_CODE_EPOCH_STALE: ErrorCode
+ERROR_CODE_SHARD_RANGE_MISMATCH: ErrorCode
+ERROR_CODE_CACHE_MISS: ErrorCode
+ERROR_CODE_RESOURCE_EXHAUSTED: ErrorCode
+ERROR_CODE_PAYLOAD_CORRUPT: ErrorCode
+ERROR_CODE_CANCELLED: ErrorCode
+ERROR_CODE_DEADLINE_EXCEEDED: ErrorCode
+ERROR_CODE_FLOW_CONTROL_VIOLATION: ErrorCode
+ERROR_CODE_INTERNAL: ErrorCode
+SERVING_STATE_UNSPECIFIED: ServingState
+SERVING_STATE_SERVING: ServingState
+SERVING_STATE_DRAINING: ServingState
+SERVING_STATE_NOT_SERVING: ServingState
+
+class Checksum(_message.Message):
+    __slots__ = ("algorithm", "value")
+    ALGORITHM_FIELD_NUMBER: _ClassVar[int]
+    VALUE_FIELD_NUMBER: _ClassVar[int]
+    algorithm: ChecksumAlgorithm
+    value: bytes
+    def __init__(self, algorithm: _Optional[_Union[ChecksumAlgorithm, str]] = ..., value: _Optional[bytes] = ...) -> None: ...
+
+class TensorFragment(_message.Message):
+    __slots__ = ("fragment_index", "fragment_count", "byte_offset", "payload")
+    FRAGMENT_INDEX_FIELD_NUMBER: _ClassVar[int]
+    FRAGMENT_COUNT_FIELD_NUMBER: _ClassVar[int]
+    BYTE_OFFSET_FIELD_NUMBER: _ClassVar[int]
+    PAYLOAD_FIELD_NUMBER: _ClassVar[int]
+    fragment_index: int
+    fragment_count: int
+    byte_offset: int
+    payload: bytes
+    def __init__(self, fragment_index: _Optional[int] = ..., fragment_count: _Optional[int] = ..., byte_offset: _Optional[int] = ..., payload: _Optional[bytes] = ...) -> None: ...
+
+class NamedTensor(_message.Message):
+    __slots__ = ("name", "shape", "dtype", "byte_order", "total_bytes", "compression", "checksum", "fragments")
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    SHAPE_FIELD_NUMBER: _ClassVar[int]
+    DTYPE_FIELD_NUMBER: _ClassVar[int]
+    BYTE_ORDER_FIELD_NUMBER: _ClassVar[int]
+    TOTAL_BYTES_FIELD_NUMBER: _ClassVar[int]
+    COMPRESSION_FIELD_NUMBER: _ClassVar[int]
+    CHECKSUM_FIELD_NUMBER: _ClassVar[int]
+    FRAGMENTS_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    shape: _containers.RepeatedScalarFieldContainer[int]
+    dtype: DType
+    byte_order: ByteOrder
+    total_bytes: int
+    compression: Compression
+    checksum: Checksum
+    fragments: _containers.RepeatedCompositeFieldContainer[TensorFragment]
+    def __init__(self, name: _Optional[str] = ..., shape: _Optional[_Iterable[int]] = ..., dtype: _Optional[_Union[DType, str]] = ..., byte_order: _Optional[_Union[ByteOrder, str]] = ..., total_bytes: _Optional[int] = ..., compression: _Optional[_Union[Compression, str]] = ..., checksum: _Optional[_Union[Checksum, _Mapping]] = ..., fragments: _Optional[_Iterable[_Union[TensorFragment, _Mapping]]] = ...) -> None: ...
+
+class TensorBundle(_message.Message):
+    __slots__ = ("bundle_version", "tensors")
+    BUNDLE_VERSION_FIELD_NUMBER: _ClassVar[int]
+    TENSORS_FIELD_NUMBER: _ClassVar[int]
+    bundle_version: int
+    tensors: _containers.RepeatedCompositeFieldContainer[NamedTensor]
+    def __init__(self, bundle_version: _Optional[int] = ..., tensors: _Optional[_Iterable[_Union[NamedTensor, _Mapping]]] = ...) -> None: ...
+
+class Fingerprint(_message.Message):
+    __slots__ = ("model_artifact_digest", "runtime_recipe_digest", "recipe_id", "recipe_version", "catalogue_version")
+    MODEL_ARTIFACT_DIGEST_FIELD_NUMBER: _ClassVar[int]
+    RUNTIME_RECIPE_DIGEST_FIELD_NUMBER: _ClassVar[int]
+    RECIPE_ID_FIELD_NUMBER: _ClassVar[int]
+    RECIPE_VERSION_FIELD_NUMBER: _ClassVar[int]
+    CATALOGUE_VERSION_FIELD_NUMBER: _ClassVar[int]
+    model_artifact_digest: str
+    runtime_recipe_digest: str
+    recipe_id: str
+    recipe_version: str
+    catalogue_version: str
+    def __init__(self, model_artifact_digest: _Optional[str] = ..., runtime_recipe_digest: _Optional[str] = ..., recipe_id: _Optional[str] = ..., recipe_version: _Optional[str] = ..., catalogue_version: _Optional[str] = ...) -> None: ...
+
+class ShardRange(_message.Message):
+    __slots__ = ("start_layer", "end_layer", "effective_start_layer")
+    START_LAYER_FIELD_NUMBER: _ClassVar[int]
+    END_LAYER_FIELD_NUMBER: _ClassVar[int]
+    EFFECTIVE_START_LAYER_FIELD_NUMBER: _ClassVar[int]
+    start_layer: int
+    end_layer: int
+    effective_start_layer: int
+    def __init__(self, start_layer: _Optional[int] = ..., end_layer: _Optional[int] = ..., effective_start_layer: _Optional[int] = ...) -> None: ...
+
+class PositionSpan(_message.Message):
+    __slots__ = ("first_position", "token_count")
+    FIRST_POSITION_FIELD_NUMBER: _ClassVar[int]
+    TOKEN_COUNT_FIELD_NUMBER: _ClassVar[int]
+    first_position: int
+    token_count: int
+    def __init__(self, first_position: _Optional[int] = ..., token_count: _Optional[int] = ...) -> None: ...
+
+class ChunkInfo(_message.Message):
+    __slots__ = ("chunk_index", "chunk_count", "final_chunk")
+    CHUNK_INDEX_FIELD_NUMBER: _ClassVar[int]
+    CHUNK_COUNT_FIELD_NUMBER: _ClassVar[int]
+    FINAL_CHUNK_FIELD_NUMBER: _ClassVar[int]
+    chunk_index: int
+    chunk_count: int
+    final_chunk: bool
+    def __init__(self, chunk_index: _Optional[int] = ..., chunk_count: _Optional[int] = ..., final_chunk: _Optional[bool] = ...) -> None: ...
+
+class CacheExpectation(_message.Message):
+    __slots__ = ("mode", "expected_past_len")
+    MODE_FIELD_NUMBER: _ClassVar[int]
+    EXPECTED_PAST_LEN_FIELD_NUMBER: _ClassVar[int]
+    mode: CacheMode
+    expected_past_len: int
+    def __init__(self, mode: _Optional[_Union[CacheMode, str]] = ..., expected_past_len: _Optional[int] = ...) -> None: ...
+
+class CacheResult(_message.Message):
+    __slots__ = ("mode", "past_len", "cache_hit")
+    MODE_FIELD_NUMBER: _ClassVar[int]
+    PAST_LEN_FIELD_NUMBER: _ClassVar[int]
+    CACHE_HIT_FIELD_NUMBER: _ClassVar[int]
+    mode: CacheMode
+    past_len: int
+    cache_hit: bool
+    def __init__(self, mode: _Optional[_Union[CacheMode, str]] = ..., past_len: _Optional[int] = ..., cache_hit: _Optional[bool] = ...) -> None: ...
+
+class Envelope(_message.Message):
+    __slots__ = ("schema_version", "work_id", "route_session_id", "route_epoch", "fingerprint", "shard_range", "phase", "position", "idempotency_step", "cache_expectation", "deadline_unix_nanos", "chunk")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    FINGERPRINT_FIELD_NUMBER: _ClassVar[int]
+    SHARD_RANGE_FIELD_NUMBER: _ClassVar[int]
+    PHASE_FIELD_NUMBER: _ClassVar[int]
+    POSITION_FIELD_NUMBER: _ClassVar[int]
+    IDEMPOTENCY_STEP_FIELD_NUMBER: _ClassVar[int]
+    CACHE_EXPECTATION_FIELD_NUMBER: _ClassVar[int]
+    DEADLINE_UNIX_NANOS_FIELD_NUMBER: _ClassVar[int]
+    CHUNK_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    work_id: str
+    route_session_id: str
+    route_epoch: int
+    fingerprint: Fingerprint
+    shard_range: ShardRange
+    phase: Phase
+    position: PositionSpan
+    idempotency_step: int
+    cache_expectation: CacheExpectation
+    deadline_unix_nanos: int
+    chunk: ChunkInfo
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., work_id: _Optional[str] = ..., route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ..., fingerprint: _Optional[_Union[Fingerprint, _Mapping]] = ..., shard_range: _Optional[_Union[ShardRange, _Mapping]] = ..., phase: _Optional[_Union[Phase, str]] = ..., position: _Optional[_Union[PositionSpan, _Mapping]] = ..., idempotency_step: _Optional[int] = ..., cache_expectation: _Optional[_Union[CacheExpectation, _Mapping]] = ..., deadline_unix_nanos: _Optional[int] = ..., chunk: _Optional[_Union[ChunkInfo, _Mapping]] = ...) -> None: ...
+
+class ShardError(_message.Message):
+    __slots__ = ("code", "detail", "retryable", "actual_past_len")
+    CODE_FIELD_NUMBER: _ClassVar[int]
+    DETAIL_FIELD_NUMBER: _ClassVar[int]
+    RETRYABLE_FIELD_NUMBER: _ClassVar[int]
+    ACTUAL_PAST_LEN_FIELD_NUMBER: _ClassVar[int]
+    code: ErrorCode
+    detail: str
+    retryable: bool
+    actual_past_len: int
+    def __init__(self, code: _Optional[_Union[ErrorCode, str]] = ..., detail: _Optional[str] = ..., retryable: _Optional[bool] = ..., actual_past_len: _Optional[int] = ...) -> None: ...
+
+class FlowControl(_message.Message):
+    __slots__ = ("credits_granted", "max_inflight_chunks", "max_chunk_bytes", "max_prefill_chunk_tokens")
+    CREDITS_GRANTED_FIELD_NUMBER: _ClassVar[int]
+    MAX_INFLIGHT_CHUNKS_FIELD_NUMBER: _ClassVar[int]
+    MAX_CHUNK_BYTES_FIELD_NUMBER: _ClassVar[int]
+    MAX_PREFILL_CHUNK_TOKENS_FIELD_NUMBER: _ClassVar[int]
+    credits_granted: int
+    max_inflight_chunks: int
+    max_chunk_bytes: int
+    max_prefill_chunk_tokens: int
+    def __init__(self, credits_granted: _Optional[int] = ..., max_inflight_chunks: _Optional[int] = ..., max_chunk_bytes: _Optional[int] = ..., max_prefill_chunk_tokens: _Optional[int] = ...) -> None: ...
+
+class SessionOpen(_message.Message):
+    __slots__ = ("schema_version", "route_session_id", "route_epoch", "fingerprint", "shard_range", "proposed_flow_control", "accepted_compression")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    FINGERPRINT_FIELD_NUMBER: _ClassVar[int]
+    SHARD_RANGE_FIELD_NUMBER: _ClassVar[int]
+    PROPOSED_FLOW_CONTROL_FIELD_NUMBER: _ClassVar[int]
+    ACCEPTED_COMPRESSION_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    route_session_id: str
+    route_epoch: int
+    fingerprint: Fingerprint
+    shard_range: ShardRange
+    proposed_flow_control: FlowControl
+    accepted_compression: _containers.RepeatedScalarFieldContainer[Compression]
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ..., fingerprint: _Optional[_Union[Fingerprint, _Mapping]] = ..., shard_range: _Optional[_Union[ShardRange, _Mapping]] = ..., proposed_flow_control: _Optional[_Union[FlowControl, _Mapping]] = ..., accepted_compression: _Optional[_Iterable[_Union[Compression, str]]] = ...) -> None: ...
+
+class SessionAccepted(_message.Message):
+    __slots__ = ("schema_version", "route_session_id", "route_epoch", "flow_control", "accepted_compression", "fingerprint")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    FLOW_CONTROL_FIELD_NUMBER: _ClassVar[int]
+    ACCEPTED_COMPRESSION_FIELD_NUMBER: _ClassVar[int]
+    FINGERPRINT_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    route_session_id: str
+    route_epoch: int
+    flow_control: FlowControl
+    accepted_compression: _containers.RepeatedScalarFieldContainer[Compression]
+    fingerprint: Fingerprint
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ..., flow_control: _Optional[_Union[FlowControl, _Mapping]] = ..., accepted_compression: _Optional[_Iterable[_Union[Compression, str]]] = ..., fingerprint: _Optional[_Union[Fingerprint, _Mapping]] = ...) -> None: ...
+
+class ActivationChunk(_message.Message):
+    __slots__ = ("envelope", "bundle")
+    ENVELOPE_FIELD_NUMBER: _ClassVar[int]
+    BUNDLE_FIELD_NUMBER: _ClassVar[int]
+    envelope: Envelope
+    bundle: TensorBundle
+    def __init__(self, envelope: _Optional[_Union[Envelope, _Mapping]] = ..., bundle: _Optional[_Union[TensorBundle, _Mapping]] = ...) -> None: ...
+
+class DecodeStep(_message.Message):
+    __slots__ = ("idempotency_step", "position", "expected_past_len", "tensor", "work_id", "deadline_unix_nanos")
+    IDEMPOTENCY_STEP_FIELD_NUMBER: _ClassVar[int]
+    POSITION_FIELD_NUMBER: _ClassVar[int]
+    EXPECTED_PAST_LEN_FIELD_NUMBER: _ClassVar[int]
+    TENSOR_FIELD_NUMBER: _ClassVar[int]
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    DEADLINE_UNIX_NANOS_FIELD_NUMBER: _ClassVar[int]
+    idempotency_step: int
+    position: int
+    expected_past_len: int
+    tensor: NamedTensor
+    work_id: str
+    deadline_unix_nanos: int
+    def __init__(self, idempotency_step: _Optional[int] = ..., position: _Optional[int] = ..., expected_past_len: _Optional[int] = ..., tensor: _Optional[_Union[NamedTensor, _Mapping]] = ..., work_id: _Optional[str] = ..., deadline_unix_nanos: _Optional[int] = ...) -> None: ...
+
+class ReleaseSignal(_message.Message):
+    __slots__ = ("route_session_id", "route_epoch", "work_id")
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    route_session_id: str
+    route_epoch: int
+    work_id: str
+    def __init__(self, route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ..., work_id: _Optional[str] = ...) -> None: ...
+
+class CancelSignal(_message.Message):
+    __slots__ = ("route_session_id", "route_epoch", "work_id", "reason")
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    REASON_FIELD_NUMBER: _ClassVar[int]
+    route_session_id: str
+    route_epoch: int
+    work_id: str
+    reason: str
+    def __init__(self, route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ..., work_id: _Optional[str] = ..., reason: _Optional[str] = ...) -> None: ...
+
+class Ack(_message.Message):
+    __slots__ = ("work_id", "idempotency_step", "cache_result", "duplicate", "execution_nanos")
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    IDEMPOTENCY_STEP_FIELD_NUMBER: _ClassVar[int]
+    CACHE_RESULT_FIELD_NUMBER: _ClassVar[int]
+    DUPLICATE_FIELD_NUMBER: _ClassVar[int]
+    EXECUTION_NANOS_FIELD_NUMBER: _ClassVar[int]
+    work_id: str
+    idempotency_step: int
+    cache_result: CacheResult
+    duplicate: bool
+    execution_nanos: int
+    def __init__(self, work_id: _Optional[str] = ..., idempotency_step: _Optional[int] = ..., cache_result: _Optional[_Union[CacheResult, _Mapping]] = ..., duplicate: _Optional[bool] = ..., execution_nanos: _Optional[int] = ...) -> None: ...
+
+class ShardStatus(_message.Message):
+    __slots__ = ("work_id", "route_session_id", "idempotency_step", "error", "terminal")
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    IDEMPOTENCY_STEP_FIELD_NUMBER: _ClassVar[int]
+    ERROR_FIELD_NUMBER: _ClassVar[int]
+    TERMINAL_FIELD_NUMBER: _ClassVar[int]
+    work_id: str
+    route_session_id: str
+    idempotency_step: int
+    error: ShardError
+    terminal: bool
+    def __init__(self, work_id: _Optional[str] = ..., route_session_id: _Optional[str] = ..., idempotency_step: _Optional[int] = ..., error: _Optional[_Union[ShardError, _Mapping]] = ..., terminal: _Optional[bool] = ...) -> None: ...
+
+class SessionRequest(_message.Message):
+    __slots__ = ("open", "chunk", "decode", "flow_control", "release", "cancel")
+    OPEN_FIELD_NUMBER: _ClassVar[int]
+    CHUNK_FIELD_NUMBER: _ClassVar[int]
+    DECODE_FIELD_NUMBER: _ClassVar[int]
+    FLOW_CONTROL_FIELD_NUMBER: _ClassVar[int]
+    RELEASE_FIELD_NUMBER: _ClassVar[int]
+    CANCEL_FIELD_NUMBER: _ClassVar[int]
+    open: SessionOpen
+    chunk: ActivationChunk
+    decode: DecodeStep
+    flow_control: FlowControl
+    release: ReleaseSignal
+    cancel: CancelSignal
+    def __init__(self, open: _Optional[_Union[SessionOpen, _Mapping]] = ..., chunk: _Optional[_Union[ActivationChunk, _Mapping]] = ..., decode: _Optional[_Union[DecodeStep, _Mapping]] = ..., flow_control: _Optional[_Union[FlowControl, _Mapping]] = ..., release: _Optional[_Union[ReleaseSignal, _Mapping]] = ..., cancel: _Optional[_Union[CancelSignal, _Mapping]] = ...) -> None: ...
+
+class SessionResponse(_message.Message):
+    __slots__ = ("accepted", "chunk", "ack", "flow_control", "status")
+    ACCEPTED_FIELD_NUMBER: _ClassVar[int]
+    CHUNK_FIELD_NUMBER: _ClassVar[int]
+    ACK_FIELD_NUMBER: _ClassVar[int]
+    FLOW_CONTROL_FIELD_NUMBER: _ClassVar[int]
+    STATUS_FIELD_NUMBER: _ClassVar[int]
+    accepted: SessionAccepted
+    chunk: ActivationChunk
+    ack: Ack
+    flow_control: FlowControl
+    status: ShardStatus
+    def __init__(self, accepted: _Optional[_Union[SessionAccepted, _Mapping]] = ..., chunk: _Optional[_Union[ActivationChunk, _Mapping]] = ..., ack: _Optional[_Union[Ack, _Mapping]] = ..., flow_control: _Optional[_Union[FlowControl, _Mapping]] = ..., status: _Optional[_Union[ShardStatus, _Mapping]] = ...) -> None: ...
+
+class CapabilityRequest(_message.Message):
+    __slots__ = ("schema_version",)
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ...) -> None: ...
+
+class CapabilityReport(_message.Message):
+    __slots__ = ("schema_version", "fingerprint", "shard_range", "backend", "device", "validated", "detail", "max_concurrent_sessions", "max_context_tokens", "flow_control", "accepted_compression", "supported_schema_versions", "validated_at_unix_nanos")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    FINGERPRINT_FIELD_NUMBER: _ClassVar[int]
+    SHARD_RANGE_FIELD_NUMBER: _ClassVar[int]
+    BACKEND_FIELD_NUMBER: _ClassVar[int]
+    DEVICE_FIELD_NUMBER: _ClassVar[int]
+    VALIDATED_FIELD_NUMBER: _ClassVar[int]
+    DETAIL_FIELD_NUMBER: _ClassVar[int]
+    MAX_CONCURRENT_SESSIONS_FIELD_NUMBER: _ClassVar[int]
+    MAX_CONTEXT_TOKENS_FIELD_NUMBER: _ClassVar[int]
+    FLOW_CONTROL_FIELD_NUMBER: _ClassVar[int]
+    ACCEPTED_COMPRESSION_FIELD_NUMBER: _ClassVar[int]
+    SUPPORTED_SCHEMA_VERSIONS_FIELD_NUMBER: _ClassVar[int]
+    VALIDATED_AT_UNIX_NANOS_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    fingerprint: Fingerprint
+    shard_range: ShardRange
+    backend: str
+    device: str
+    validated: bool
+    detail: str
+    max_concurrent_sessions: int
+    max_context_tokens: int
+    flow_control: FlowControl
+    accepted_compression: _containers.RepeatedScalarFieldContainer[Compression]
+    supported_schema_versions: _containers.RepeatedScalarFieldContainer[SchemaVersion]
+    validated_at_unix_nanos: int
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., fingerprint: _Optional[_Union[Fingerprint, _Mapping]] = ..., shard_range: _Optional[_Union[ShardRange, _Mapping]] = ..., backend: _Optional[str] = ..., device: _Optional[str] = ..., validated: _Optional[bool] = ..., detail: _Optional[str] = ..., max_concurrent_sessions: _Optional[int] = ..., max_context_tokens: _Optional[int] = ..., flow_control: _Optional[_Union[FlowControl, _Mapping]] = ..., accepted_compression: _Optional[_Iterable[_Union[Compression, str]]] = ..., supported_schema_versions: _Optional[_Iterable[_Union[SchemaVersion, str]]] = ..., validated_at_unix_nanos: _Optional[int] = ...) -> None: ...
+
+class HealthRequest(_message.Message):
+    __slots__ = ("schema_version",)
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ...) -> None: ...
+
+class HealthReport(_message.Message):
+    __slots__ = ("schema_version", "state", "active_sessions", "queued_chunks", "batch_occupancy", "kv_pressure", "resident_bytes", "detail")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    STATE_FIELD_NUMBER: _ClassVar[int]
+    ACTIVE_SESSIONS_FIELD_NUMBER: _ClassVar[int]
+    QUEUED_CHUNKS_FIELD_NUMBER: _ClassVar[int]
+    BATCH_OCCUPANCY_FIELD_NUMBER: _ClassVar[int]
+    KV_PRESSURE_FIELD_NUMBER: _ClassVar[int]
+    RESIDENT_BYTES_FIELD_NUMBER: _ClassVar[int]
+    DETAIL_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    state: ServingState
+    active_sessions: int
+    queued_chunks: int
+    batch_occupancy: int
+    kv_pressure: float
+    resident_bytes: int
+    detail: str
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., state: _Optional[_Union[ServingState, str]] = ..., active_sessions: _Optional[int] = ..., queued_chunks: _Optional[int] = ..., batch_occupancy: _Optional[int] = ..., kv_pressure: _Optional[float] = ..., resident_bytes: _Optional[int] = ..., detail: _Optional[str] = ...) -> None: ...
+
+class ReleaseRequest(_message.Message):
+    __slots__ = ("schema_version", "route_session_id", "route_epoch")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    route_session_id: str
+    route_epoch: int
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ...) -> None: ...
+
+class ReleaseResponse(_message.Message):
+    __slots__ = ("released", "error")
+    RELEASED_FIELD_NUMBER: _ClassVar[int]
+    ERROR_FIELD_NUMBER: _ClassVar[int]
+    released: bool
+    error: ShardError
+    def __init__(self, released: _Optional[bool] = ..., error: _Optional[_Union[ShardError, _Mapping]] = ...) -> None: ...
+
+class CancelRequest(_message.Message):
+    __slots__ = ("schema_version", "route_session_id", "route_epoch", "work_id", "reason")
+    SCHEMA_VERSION_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_SESSION_ID_FIELD_NUMBER: _ClassVar[int]
+    ROUTE_EPOCH_FIELD_NUMBER: _ClassVar[int]
+    WORK_ID_FIELD_NUMBER: _ClassVar[int]
+    REASON_FIELD_NUMBER: _ClassVar[int]
+    schema_version: SchemaVersion
+    route_session_id: str
+    route_epoch: int
+    work_id: str
+    reason: str
+    def __init__(self, schema_version: _Optional[_Union[SchemaVersion, str]] = ..., route_session_id: _Optional[str] = ..., route_epoch: _Optional[int] = ..., work_id: _Optional[str] = ..., reason: _Optional[str] = ...) -> None: ...
+
+class CancelResponse(_message.Message):
+    __slots__ = ("cancelled_work_items", "error")
+    CANCELLED_WORK_ITEMS_FIELD_NUMBER: _ClassVar[int]
+    ERROR_FIELD_NUMBER: _ClassVar[int]
+    cancelled_work_items: int
+    error: ShardError
+    def __init__(self, cancelled_work_items: _Optional[int] = ..., error: _Optional[_Union[ShardError, _Mapping]] = ...) -> None: ...

packages/node/meshnet_node/native_protocol/generated/shard_runtime_pb2_grpc.py

@@ -0,0 +1,295 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+import warnings
+
+from . import shard_runtime_pb2 as shard__runtime__pb2
+
+GRPC_GENERATED_VERSION = '1.82.1'
+GRPC_VERSION = grpc.__version__
+_version_not_supported = False
+
+try:
+    from grpc._utilities import first_version_is_lower
+    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
+except ImportError:
+    _version_not_supported = True
+
+if _version_not_supported:
+    raise RuntimeError(
+        f'The grpc package installed is at version {GRPC_VERSION},'
+        + ' but the generated code in shard_runtime_pb2_grpc.py depends on'
+        + f' grpcio>={GRPC_GENERATED_VERSION}.'
+        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
+    )
+
+
+class ShardRuntimeStub:
+    """---------------------------------------------------------------------------
+    Service
+    ---------------------------------------------------------------------------
+
+    """
+
+    def __init__(self, channel):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.GetCapability = channel.unary_unary(
+                '/meshnet.shard.v1.ShardRuntime/GetCapability',
+                request_serializer=shard__runtime__pb2.CapabilityRequest.SerializeToString,
+                response_deserializer=shard__runtime__pb2.CapabilityReport.FromString,
+                _registered_method=True)
+        self.Health = channel.unary_unary(
+                '/meshnet.shard.v1.ShardRuntime/Health',
+                request_serializer=shard__runtime__pb2.HealthRequest.SerializeToString,
+                response_deserializer=shard__runtime__pb2.HealthReport.FromString,
+                _registered_method=True)
+        self.Session = channel.stream_stream(
+                '/meshnet.shard.v1.ShardRuntime/Session',
+                request_serializer=shard__runtime__pb2.SessionRequest.SerializeToString,
+                response_deserializer=shard__runtime__pb2.SessionResponse.FromString,
+                _registered_method=True)
+        self.Release = channel.unary_unary(
+                '/meshnet.shard.v1.ShardRuntime/Release',
+                request_serializer=shard__runtime__pb2.ReleaseRequest.SerializeToString,
+                response_deserializer=shard__runtime__pb2.ReleaseResponse.FromString,
+                _registered_method=True)
+        self.Cancel = channel.unary_unary(
+                '/meshnet.shard.v1.ShardRuntime/Cancel',
+                request_serializer=shard__runtime__pb2.CancelRequest.SerializeToString,
+                response_deserializer=shard__runtime__pb2.CancelResponse.FromString,
+                _registered_method=True)
+
+
+class ShardRuntimeServicer:
+    """---------------------------------------------------------------------------
+    Service
+    ---------------------------------------------------------------------------
+
+    """
+
+    def GetCapability(self, request, context):
+        """What this worker can execute. Read before a route is built.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Health(self, request, context):
+        """Live load and serving state.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Session(self, request_iterator, context):
+        """One long-lived bidirectional stream per Route Session Activation Seam.
+
+        The stream opens with SessionOpen/SessionAccepted, then carries bounded
+        prefill chunks and decode steps in both directions for the life of the
+        session. Per-token channel creation is a non-goal: the handshake cost is
+        paid once and the hot path carries only what changes.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Release(self, request, context):
+        """Drop session state out of band. Idempotent.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Cancel(self, request, context):
+        """Cancel out of band, on a fresh call.
+
+        In-band CancelSignal is preferred, but a sender that is blocked on flow
+        control cannot write one — a cancel that can only travel down a wedged
+        stream is not a cancel. This RPC always has a path to the worker.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+
+def add_ShardRuntimeServicer_to_server(servicer, server):
+    rpc_method_handlers = {
+            'GetCapability': grpc.unary_unary_rpc_method_handler(
+                    servicer.GetCapability,
+                    request_deserializer=shard__runtime__pb2.CapabilityRequest.FromString,
+                    response_serializer=shard__runtime__pb2.CapabilityReport.SerializeToString,
+            ),
+            'Health': grpc.unary_unary_rpc_method_handler(
+                    servicer.Health,
+                    request_deserializer=shard__runtime__pb2.HealthRequest.FromString,
+                    response_serializer=shard__runtime__pb2.HealthReport.SerializeToString,
+            ),
+            'Session': grpc.stream_stream_rpc_method_handler(
+                    servicer.Session,
+                    request_deserializer=shard__runtime__pb2.SessionRequest.FromString,
+                    response_serializer=shard__runtime__pb2.SessionResponse.SerializeToString,
+            ),
+            'Release': grpc.unary_unary_rpc_method_handler(
+                    servicer.Release,
+                    request_deserializer=shard__runtime__pb2.ReleaseRequest.FromString,
+                    response_serializer=shard__runtime__pb2.ReleaseResponse.SerializeToString,
+            ),
+            'Cancel': grpc.unary_unary_rpc_method_handler(
+                    servicer.Cancel,
+                    request_deserializer=shard__runtime__pb2.CancelRequest.FromString,
+                    response_serializer=shard__runtime__pb2.CancelResponse.SerializeToString,
+            ),
+    }
+    generic_handler = grpc.method_handlers_generic_handler(
+            'meshnet.shard.v1.ShardRuntime', rpc_method_handlers)
+    server.add_generic_rpc_handlers((generic_handler,))
+    server.add_registered_method_handlers('meshnet.shard.v1.ShardRuntime', rpc_method_handlers)
+
+
+ # This class is part of an EXPERIMENTAL API.
+class ShardRuntime:
+    """---------------------------------------------------------------------------
+    Service
+    ---------------------------------------------------------------------------
+
+    """
+
+    @staticmethod
+    def GetCapability(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/meshnet.shard.v1.ShardRuntime/GetCapability',
+            shard__runtime__pb2.CapabilityRequest.SerializeToString,
+            shard__runtime__pb2.CapabilityReport.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def Health(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/meshnet.shard.v1.ShardRuntime/Health',
+            shard__runtime__pb2.HealthRequest.SerializeToString,
+            shard__runtime__pb2.HealthReport.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def Session(request_iterator,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.stream_stream(
+            request_iterator,
+            target,
+            '/meshnet.shard.v1.ShardRuntime/Session',
+            shard__runtime__pb2.SessionRequest.SerializeToString,
+            shard__runtime__pb2.SessionResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def Release(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/meshnet.shard.v1.ShardRuntime/Release',
+            shard__runtime__pb2.ReleaseRequest.SerializeToString,
+            shard__runtime__pb2.ReleaseResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def Cancel(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/meshnet.shard.v1.ShardRuntime/Cancel',
+            shard__runtime__pb2.CancelRequest.SerializeToString,
+            shard__runtime__pb2.CancelResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)

packages/node/native/CMakeLists.txt

@@ -0,0 +1,74 @@
+# Native Shard protocol: C++ schema generation, build wiring and conformance test.
+#
+# The C++ stubs are generated into the build tree at configure time — they are
+# never committed. A C++ consumer already needs a toolchain, so committing
+# generated C++ would only create a second copy of the schema that can rot.
+#
+# gRPC C++ is optional here on purpose. The conformance test only needs message
+# types, so the schema can be verified on a machine that has protobuf but not
+# the gRPC C++ stack. When gRPC *is* found, the service stubs are generated too
+# and exported as `shard_runtime_grpc` for the worker (DGR-008) to link.
+#
+# Build:
+#   cmake -S packages/node/native -B build/native -DCMAKE_PREFIX_PATH=<protobuf-install>
+#   cmake --build build/native -j
+#   ctest --test-dir build/native --output-on-failure
+#
+# `scripts/bootstrap_native_toolchain.sh` builds a protobuf install from source
+# when the system has none.
+
+cmake_minimum_required(VERSION 3.24)
+project(meshnet_shard_protocol CXX)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+find_package(protobuf CONFIG REQUIRED)
+find_package(gRPC CONFIG QUIET)
+
+set(SHARD_PROTO "${CMAKE_CURRENT_SOURCE_DIR}/proto/shard_runtime.proto")
+
+# Message types: always available.
+add_library(shard_runtime_proto STATIC "${SHARD_PROTO}")
+target_link_libraries(shard_runtime_proto PUBLIC protobuf::libprotobuf)
+target_include_directories(shard_runtime_proto PUBLIC "${CMAKE_CURRENT_BINARY_DIR}")
+protobuf_generate(
+  TARGET shard_runtime_proto
+  LANGUAGE cpp
+  IMPORT_DIRS "${CMAKE_CURRENT_SOURCE_DIR}/proto"
+  PROTOC_OUT_DIR "${CMAKE_CURRENT_BINARY_DIR}"
+)
+
+# Service stubs: only when the gRPC C++ stack is present.
+if(gRPC_FOUND)
+  add_library(shard_runtime_grpc STATIC "${SHARD_PROTO}")
+  target_link_libraries(shard_runtime_grpc PUBLIC shard_runtime_proto gRPC::grpc++)
+  target_include_directories(shard_runtime_grpc PUBLIC "${CMAKE_CURRENT_BINARY_DIR}")
+  protobuf_generate(
+    TARGET shard_runtime_grpc
+    LANGUAGE grpc
+    GENERATE_EXTENSIONS .grpc.pb.h .grpc.pb.cc
+    PLUGIN "protoc-gen-grpc=$<TARGET_FILE:gRPC::grpc_cpp_plugin>"
+    IMPORT_DIRS "${CMAKE_CURRENT_SOURCE_DIR}/proto"
+    PROTOC_OUT_DIR "${CMAKE_CURRENT_BINARY_DIR}"
+  )
+  message(STATUS "gRPC C++ found: building ShardRuntime service stubs")
+else()
+  message(STATUS "gRPC C++ not found: building message types only "
+                 "(sufficient for the conformance test)")
+endif()
+
+enable_testing()
+
+add_executable(shard_protocol_conformance tests/test_shard_protocol_conformance.cpp)
+target_link_libraries(shard_protocol_conformance PRIVATE shard_runtime_proto)
+
+# The test asserts against the same committed vectors the Python test uses, and
+# writes its own re-serialization back out so Python can prove the two languages
+# agree byte-for-byte.
+add_test(
+  NAME shard_protocol_conformance
+  COMMAND shard_protocol_conformance
+          "${CMAKE_CURRENT_SOURCE_DIR}/testdata"
+          "${CMAKE_CURRENT_BINARY_DIR}"
+)

packages/node/native/README.md

@@ -0,0 +1,69 @@
+# Native Shard protocol
+
+`proto/shard_runtime.proto` is the semantic contract between a Meshnet node and
+a Shard worker: Protocol Buffers over gRPC/HTTP2 (ADR-0020). It is the source of
+truth. The Python and C++ types are generated from it; neither is the contract.
+
+## What lives here
+
+| Path | Purpose |
+|---|---|
+| `proto/shard_runtime.proto` | The schema: capability, health, session stream, release, cancel |
+| `testdata/*.binpb` | Committed conformance vectors both languages assert against |
+| `tests/test_shard_protocol_conformance.cpp` | C++ conformance test |
+| `CMakeLists.txt` | C++ generation, build wiring, and `ctest` registration |
+
+The Python stubs are generated into
+`packages/node/meshnet_node/native_protocol/generated/` and are committed, so
+installing a node needs no protoc. The C++ stubs are generated into the build
+tree and are never committed — a C++ consumer already has a toolchain, and a
+committed copy could only rot.
+
+## Regenerating
+
+```bash
+pip install grpcio-tools==1.82.1          # bundles protoc; no system protoc needed
+python scripts/generate_native_protocol.py           # rewrite the Python stubs
+python scripts/generate_native_protocol.py --check   # fail if they drifted
+python scripts/generate_protocol_goldens.py --check  # fail if the vectors drifted
+```
+
+Both `--check` modes run in CI via `tests/test_native_shard_protocol.py`, so a
+schema edit that is not accompanied by regenerated output fails the suite rather
+than shipping stubs that disagree with the schema they claim to implement.
+
+## Building and running the C++ conformance test
+
+If the machine has no protobuf C++ toolchain:
+
+```bash
+scripts/bootstrap_native_toolchain.sh build/native-toolchain
+```
+
+Then:
+
+```bash
+cmake -S packages/node/native -B build/native \
+      -DCMAKE_PREFIX_PATH="$PWD/build/native-toolchain"
+cmake --build build/native -j
+ctest --test-dir build/native --output-on-failure
+```
+
+gRPC C++ is optional: without it, CMake builds the message types only, which is
+all the conformance test needs. When gRPC C++ *is* found, the `ShardRuntime`
+service stubs are built too and exported as `shard_runtime_grpc` for the worker
+(DGR-008) to link.
+
+## How the cross-language check actually proves something
+
+Two codecs that each round-trip their own output prove only that each is
+self-consistent. Instead:
+
+1. Python builds the canonical message and commits its bytes to `testdata/`.
+2. The C++ test parses *those* bytes, asserts every field, independently
+   recomputes the CRC32C from the polynomial, and re-serializes to
+   `cpp_roundtrip.binpb` in the build tree.
+3. `test_cpp_and_python_agree_byte_for_byte` compares that file to the golden.
+
+Byte equality across the two implementations is the claim; anything less is two
+parallel test suites that can drift apart.

packages/node/native/proto/shard_runtime.proto

@@ -0,0 +1,592 @@
+// The native Shard data plane: Protocol Buffers over gRPC/HTTP2 (ADR-0020).
+//
+// This schema — not any Python or C++ type — is the semantic contract between a
+// Meshnet node and a Shard worker. Direct hops speak gRPC. When direct
+// connectivity is unavailable the existing relay carries these exact serialized
+// frames as opaque binary, which is why every deadline, cancellation and
+// identity field is carried *in the schema* rather than left to gRPC metadata:
+// a relayed frame has no HTTP/2 context to inherit them from.
+//
+// Meshnet remains the only control plane. Nothing here selects routes, prices
+// work, or authenticates peers; the worker executes a layer range it has been
+// admitted for and reports what it did.
+
+syntax = "proto3";
+
+package meshnet.shard.v1;
+
+option cc_enable_arenas = true;
+option go_package = "github.com/meshnet/shard/v1;shardv1";
+
+// ---------------------------------------------------------------------------
+// Versioning
+// ---------------------------------------------------------------------------
+
+// Wire-compatibility generation of this schema. A worker rejects a peer whose
+// SCHEMA_VERSION it cannot satisfy instead of guessing at field meaning.
+//
+// This is deliberately NOT the `X-Meshnet-Wire` version of the legacy HTTP
+// activation format (currently "2", see ADR-0008). The native protocol is a
+// separate contract with its own generation counter and starts at 1.
+enum SchemaVersion {
+  SCHEMA_VERSION_UNSPECIFIED = 0;
+  SCHEMA_VERSION_1 = 1;
+}
+
+// ---------------------------------------------------------------------------
+// Tensor bundle — the public activation boundary
+// ---------------------------------------------------------------------------
+
+// Element type of a tensor payload.
+//
+// bfloat16 is the canonical activation dtype at every Shard boundary regardless
+// of the weight quantization a node uses internally (ADR-0008): weights may be
+// NF4 or INT8, compute upcasts, and the boundary stays bfloat16. The other
+// values exist for auxiliary tensors (position ids, attention masks) and for
+// architectures whose boundary is not a plain hidden state.
+enum DType {
+  DTYPE_UNSPECIFIED = 0;
+  DTYPE_BFLOAT16 = 1;
+  DTYPE_FLOAT16 = 2;
+  DTYPE_FLOAT32 = 3;
+  DTYPE_INT32 = 4;
+  DTYPE_INT64 = 5;
+  DTYPE_UINT8 = 6;
+  DTYPE_INT8 = 7;
+  DTYPE_BOOL = 8;
+}
+
+// Byte order of a tensor payload. Little-endian is canonical on the wire; the
+// field is explicit so a big-endian peer is rejected loudly rather than
+// silently reading byte-swapped activations.
+enum ByteOrder {
+  BYTE_ORDER_UNSPECIFIED = 0;
+  BYTE_ORDER_LITTLE_ENDIAN = 1;
+  BYTE_ORDER_BIG_ENDIAN = 2;
+}
+
+// Payload compression. Mirrors the policy-driven zstd decision the existing
+// activation seam already makes (`activation_compression.py`): compression is a
+// transport optimisation and NONE is always a legal choice for any payload.
+enum Compression {
+  COMPRESSION_UNSPECIFIED = 0;
+  COMPRESSION_NONE = 1;
+  COMPRESSION_ZSTD = 2;
+}
+
+// Integrity algorithm covering a payload.
+enum ChecksumAlgorithm {
+  CHECKSUM_ALGORITHM_UNSPECIFIED = 0;
+  CHECKSUM_ALGORITHM_NONE = 1;
+  CHECKSUM_ALGORITHM_CRC32C = 2;
+}
+
+// An integrity check over the *uncompressed* canonical payload bytes.
+//
+// Checksumming the uncompressed bytes (not the compressed frame) means the
+// value is stable whether a hop compressed the payload or not, so a relay may
+// re-frame or a peer may decline compression without invalidating it.
+message Checksum {
+  ChecksumAlgorithm algorithm = 1;
+  // Big-endian encoding of the checksum value.
+  bytes value = 2;
+}
+
+// One slice of a tensor's wire payload.
+//
+// Fragments bound individual payload pieces and make coverage independently
+// verifiable. The negotiated `FlowControl.max_chunk_bytes` bounds the enclosing
+// serialized stream message; `byte_offset` proves the pieces tile the tensor
+// exactly — no hole, no overlap — instead of trusting arrival order.
+//
+// Offsets and payloads describe the *wire* body, which is the compressed frame
+// when the parent tensor declares a compression. A zstd frame is not decodable
+// per fragment, so a receiver reassembles first and decompresses once. The
+// uncompressed size is not repeated here: `NamedTensor.total_bytes` is the
+// single source of truth, and a second copy could only ever disagree with it.
+message TensorFragment {
+  uint32 fragment_index = 1;
+  uint32 fragment_count = 2;
+  // Offset of this fragment within the tensor's wire body.
+  uint64 byte_offset = 3;
+  // Fragment bytes, compressed iff the parent tensor declares a compression.
+  bytes payload = 4;
+
+  reserved 5;
+  reserved "uncompressed_size";
+}
+
+// One named tensor at an architecture boundary.
+message NamedTensor {
+  // Architecture-defined boundary name, e.g. "hidden_states", "position_ids".
+  // A boundary may need more than one tensor, which is why the payload is a
+  // named bundle rather than a bare buffer (ADR-0020).
+  string name = 1;
+  repeated int64 shape = 2;
+  DType dtype = 3;
+  ByteOrder byte_order = 4;
+  // Total uncompressed payload size. A receiver sizes its buffer from this
+  // before reading fragments and refuses a bundle that exceeds its budget.
+  uint64 total_bytes = 5;
+  Compression compression = 6;
+  // Over the uncompressed payload; see Checksum.
+  Checksum checksum = 7;
+  // Ordered fragments. A tensor small enough for one message carries exactly
+  // one fragment with fragment_count == 1.
+  repeated TensorFragment fragments = 8;
+}
+
+// A versioned named-tensor bundle: the public activation boundary payload.
+message TensorBundle {
+  // Generation of the bundle layout itself, independent of SchemaVersion so a
+  // boundary payload can evolve without a whole-protocol version bump.
+  uint32 bundle_version = 1;
+  repeated NamedTensor tensors = 2;
+}
+
+// ---------------------------------------------------------------------------
+// Identity: what work this is, for which route, against which artifact
+// ---------------------------------------------------------------------------
+
+// Exact identity of the executable thing a Shard runs.
+//
+// Both fingerprints must match end to end across a route. Two nodes agreeing on
+// a model name but not on quantization, kernel or tail-norm placement would
+// produce silently wrong tokens, so identity is a digest, not a label.
+message Fingerprint {
+  // Digest of the exact Model Artifact (weights + config), e.g. the GGUF hash.
+  string model_artifact_digest = 1;
+  // Digest of the exact runtime recipe (backend, quantization, kernels, dtype).
+  string runtime_recipe_digest = 2;
+  // Human-readable recipe identity, mirroring the capability report an admitted
+  // node already registers with (ADR-0023). Labels are for diagnosis; the
+  // digests above are what a peer actually compares.
+  string recipe_id = 3;
+  string recipe_version = 4;
+  string catalogue_version = 5;
+}
+
+// The contiguous transformer layer range a Shard owns.
+message ShardRange {
+  // Registered range, inclusive start, exclusive end.
+  uint32 start_layer = 1;
+  uint32 end_layer = 2;
+  // The layer this Shard must actually begin at for this route (ADR-0012).
+  //
+  // Shard ranges may overlap; the Tracker resolves the overlap when it builds
+  // the route and every hop is told where the previous hop stopped. Executing
+  // from `start_layer` when `effective_start_layer` is higher would re-apply
+  // layers already applied to the incoming activation and silently corrupt the
+  // result. A worker executes [effective_start_layer, end_layer).
+  uint32 effective_start_layer = 3;
+}
+
+// Which part of generation a message belongs to.
+enum Phase {
+  PHASE_UNSPECIFIED = 0;
+  PHASE_PREFILL = 1;
+  PHASE_DECODE = 2;
+  PHASE_RELEASE = 3;
+  PHASE_CANCEL = 4;
+}
+
+// Token span carried by one chunk.
+message PositionSpan {
+  // Absolute position of the first token in this chunk within the sequence.
+  uint64 first_position = 1;
+  // Number of token positions in this chunk. A decode step carries 1.
+  uint32 token_count = 2;
+}
+
+// Bounded chunking for a prefill.
+//
+// A long prompt is split into token-aligned chunks before the first forward
+// pass (ADR-0008) so peak transfer per boundary stays bounded regardless of
+// prompt length. Splits never fall mid-token.
+message ChunkInfo {
+  uint32 chunk_index = 1;
+  uint32 chunk_count = 2;
+  // True on the final chunk of a prefill. A receiver that has not seen a final
+  // chunk knows the prefill is still incomplete.
+  bool final_chunk = 3;
+}
+
+// How the sender expects the receiver's Hot KV State to be positioned.
+enum CacheMode {
+  CACHE_MODE_UNSPECIFIED = 0;
+  // No session state; the payload carries everything needed. Always a legal
+  // fallback and the recovery path after a miss.
+  CACHE_MODE_STATELESS = 1;
+  // Establish fresh session state for this Shard's layer range.
+  CACHE_MODE_PREFILL = 2;
+  // Continue existing session state. `expected_past_len` must match exactly.
+  CACHE_MODE_DECODE = 3;
+}
+
+// The sender's expectation about receiver-side cache state (ADR-0022).
+//
+// A mismatch is always a declared cache miss, never a silent stateless forward:
+// running a single-token decode payload without the matching cache would emit
+// plausible garbage. The receiver answers a miss with ShardError.CACHE_MISS and
+// the head re-prefills the whole sequence under the same Route Session ID.
+message CacheExpectation {
+  CacheMode mode = 1;
+  // Decode only: the number of tokens the receiver's session cache must already
+  // hold for this Shard's layer range.
+  uint64 expected_past_len = 2;
+}
+
+// What the receiver's cache actually did.
+message CacheResult {
+  CacheMode mode = 1;
+  // Tokens held for this Shard's range after the step completed.
+  uint64 past_len = 2;
+  // True when session state was reused rather than rebuilt.
+  bool cache_hit = 3;
+}
+
+// Everything required to interpret one unit of work, independent of transport.
+//
+// Carried on every request and response because a relayed frame arrives as
+// opaque binary with no gRPC metadata, no deadline and no channel identity.
+message Envelope {
+  SchemaVersion schema_version = 1;
+  // Unique id for this unit of work; also the unit of billing attribution.
+  string work_id = 2;
+  // The Route Session this work belongs to. Together with `route_epoch` it maps
+  // to exactly one isolated llama sequence / bounded context (ADR-0020).
+  string route_session_id = 3;
+  // Bumped by the control plane whenever the route changes. A worker refuses
+  // work from a stale epoch rather than mixing it into live session state.
+  uint64 route_epoch = 4;
+  Fingerprint fingerprint = 5;
+  ShardRange shard_range = 6;
+  Phase phase = 7;
+  PositionSpan position = 8;
+  // Monotonic step within (route_session_id, route_epoch).
+  //
+  // Idempotency key: a retried or duplicated step carries the same value and
+  // must be acknowledged, not re-applied. Re-applying a decode step would
+  // advance the KV cache twice and desynchronise the route.
+  uint64 idempotency_step = 9;
+  CacheExpectation cache_expectation = 10;
+  // Absolute deadline. Carried in-band so a relayed frame keeps its deadline;
+  // on a direct hop it is set consistently with the gRPC deadline.
+  int64 deadline_unix_nanos = 11;
+  // Chunking, when this message is part of a bounded prefill.
+  ChunkInfo chunk = 12;
+}
+
+// ---------------------------------------------------------------------------
+// Structured errors
+// ---------------------------------------------------------------------------
+
+// Why a unit of work could not be executed as asked.
+//
+// These are semantic outcomes, distinct from transport failure. They travel as
+// a ShardStatus on the stream (and as google.rpc.Status details on unary RPCs)
+// so a relayed frame carries the same diagnosis a direct gRPC hop would.
+enum ErrorCode {
+  ERROR_CODE_UNSPECIFIED = 0;
+  // Peer cannot satisfy the requested SchemaVersion.
+  ERROR_CODE_SCHEMA_UNSUPPORTED = 1;
+  // Model artifact or runtime recipe digest differs from this worker's.
+  ERROR_CODE_FINGERPRINT_MISMATCH = 2;
+  // Work arrived for a route epoch the worker has already moved past.
+  ERROR_CODE_EPOCH_STALE = 3;
+  // Requested layer range is not the range this worker serves.
+  ERROR_CODE_SHARD_RANGE_MISMATCH = 4;
+  // Session state absent or positioned differently than expected (ADR-0022).
+  // The head recovers with one full re-prefill; this is not a fatal error.
+  ERROR_CODE_CACHE_MISS = 5;
+  // Admission budget (weights, KV, scratch, queue depth) would be exceeded.
+  ERROR_CODE_RESOURCE_EXHAUSTED = 6;
+  // Payload failed its checksum, or fragments did not cover the tensor.
+  ERROR_CODE_PAYLOAD_CORRUPT = 7;
+  // Work was cancelled by the control plane or the peer.
+  ERROR_CODE_CANCELLED = 8;
+  // Deadline in the envelope had already passed when work was dequeued.
+  ERROR_CODE_DEADLINE_EXCEEDED = 9;
+  // Sender exceeded its granted flow-control credit.
+  ERROR_CODE_FLOW_CONTROL_VIOLATION = 10;
+  // The worker failed while executing; detail is sanitized.
+  ERROR_CODE_INTERNAL = 11;
+}
+
+message ShardError {
+  ErrorCode code = 1;
+  // Sanitized, operator-facing explanation. Never a raw exception, file path or
+  // credential (ADR-0023 sanitization rule).
+  string detail = 2;
+  // True when the same work may be retried unchanged. A CACHE_MISS is not
+  // retryable unchanged — the head must re-prefill first.
+  bool retryable = 3;
+  // Set on CACHE_MISS so the head knows where the receiver actually is.
+  uint64 actual_past_len = 4;
+}
+
+// ---------------------------------------------------------------------------
+// Flow control
+// ---------------------------------------------------------------------------
+
+// Application-level credit, layered on top of HTTP/2 flow control.
+//
+// HTTP/2 bounds bytes in flight; it does not bound how much *work* a worker has
+// queued, and a relayed frame gets no HTTP/2 window at all. Credits bound the
+// number of un-acked chunks a sender may have outstanding, which is what keeps
+// worker queues and KV pressure bounded and lets prefill avoid starving decode.
+message FlowControl {
+  // Additional chunks the receiver is willing to accept beyond those already
+  // granted. Purely additive: a grant never reduces outstanding credit.
+  uint32 credits_granted = 1;
+  // Hard ceiling on un-acked chunks, independent of granted credit.
+  uint32 max_inflight_chunks = 2;
+  // Hard ceiling on the serialized size of one chunk message.
+  uint64 max_chunk_bytes = 3;
+  // Hard ceiling on token positions per prefill chunk.
+  uint32 max_prefill_chunk_tokens = 4;
+}
+
+// ---------------------------------------------------------------------------
+// Session stream messages
+// ---------------------------------------------------------------------------
+
+// Opens one long-lived bidirectional stream for one Route Session Activation
+// Seam. The handshake settles version, identity and the initial flow-control
+// window before any activation is sent, so an incompatible peer fails at open
+// rather than mid-generation.
+message SessionOpen {
+  // Highest schema version the initiator supports; the acceptor replies with
+  // the version actually negotiated.
+  SchemaVersion schema_version = 1;
+  string route_session_id = 2;
+  uint64 route_epoch = 3;
+  Fingerprint fingerprint = 4;
+  ShardRange shard_range = 5;
+  // Flow-control limits the initiator can honour. The acceptor replies with the
+  // limits that actually apply.
+  FlowControl proposed_flow_control = 6;
+  // Compressions the initiator can decode. NONE is implicitly always supported.
+  repeated Compression accepted_compression = 7;
+}
+
+message SessionAccepted {
+  // The version both peers will use for the life of this stream.
+  SchemaVersion schema_version = 1;
+  string route_session_id = 2;
+  uint64 route_epoch = 3;
+  // Authoritative limits. The initiator must not exceed these.
+  FlowControl flow_control = 4;
+  repeated Compression accepted_compression = 5;
+  // Fingerprint the worker actually serves, so a mismatch is visible at open.
+  Fingerprint fingerprint = 6;
+}
+
+// One unit of activation work: a bounded prefill chunk or a decode step.
+message ActivationChunk {
+  Envelope envelope = 1;
+  TensorBundle bundle = 2;
+}
+
+// The small decode fast path.
+//
+// A decode step is one token: the envelope's identity fields are already fixed
+// for the life of the stream, so repeating them per token is pure overhead on
+// the hottest path. A DecodeStep carries only what changes — the step, the
+// position, and one tensor — and inherits the rest from the SessionOpen
+// handshake. A peer may always fall back to ActivationChunk with PHASE_DECODE.
+message DecodeStep {
+  // Idempotency step within the session; also orders the stream.
+  uint64 idempotency_step = 1;
+  // Absolute position of this token.
+  uint64 position = 2;
+  // Tokens the receiver's cache must already hold. A mismatch is a CACHE_MISS.
+  uint64 expected_past_len = 3;
+  // The single boundary tensor for this token, typically [1, 1, hidden].
+  NamedTensor tensor = 4;
+  // Work id for attribution; the session/epoch/fingerprint/range are inherited.
+  string work_id = 5;
+  int64 deadline_unix_nanos = 6;
+}
+
+// Drop session state for a Route Session. Bounded memory does not depend on
+// this arriving — workers also evict by TTL and LRU (ADR-0022) — but an
+// explicit release returns KV immediately instead of holding it for the TTL.
+message ReleaseSignal {
+  string route_session_id = 1;
+  uint64 route_epoch = 2;
+  string work_id = 3;
+}
+
+// Abandon in-flight work.
+//
+// Cancellation must remain possible when the stream is wedged behind flow
+// control, which is why Cancel also exists as a unary RPC on a fresh call.
+message CancelSignal {
+  string route_session_id = 1;
+  uint64 route_epoch = 2;
+  // Cancel one unit of work; empty cancels every in-flight unit for the session.
+  string work_id = 3;
+  string reason = 4;
+}
+
+// Acknowledges a chunk or decode step, releasing its flow-control credit and
+// recording where the receiver's cache now stands.
+message Ack {
+  string work_id = 1;
+  uint64 idempotency_step = 2;
+  CacheResult cache_result = 3;
+  // True when the step was recognised as an already-applied duplicate and was
+  // therefore acknowledged without being re-applied.
+  bool duplicate = 4;
+  // Observed execution time, for Generation Telemetry.
+  uint64 execution_nanos = 5;
+}
+
+// A terminal outcome for one unit of work, or for the session.
+message ShardStatus {
+  string work_id = 1;
+  string route_session_id = 2;
+  uint64 idempotency_step = 3;
+  ShardError error = 4;
+  // True when the stream itself is finished and no further work will be served.
+  bool terminal = 5;
+}
+
+message SessionRequest {
+  oneof kind {
+    SessionOpen open = 1;
+    ActivationChunk chunk = 2;
+    DecodeStep decode = 3;
+    FlowControl flow_control = 4;
+    ReleaseSignal release = 5;
+    CancelSignal cancel = 6;
+  }
+}
+
+message SessionResponse {
+  oneof kind {
+    SessionAccepted accepted = 1;
+    // Result activations forwarded toward the next Shard, or to the head.
+    ActivationChunk chunk = 2;
+    Ack ack = 3;
+    FlowControl flow_control = 4;
+    ShardStatus status = 5;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Capability and health
+// ---------------------------------------------------------------------------
+
+message CapabilityRequest {
+  SchemaVersion schema_version = 1;
+}
+
+// What this worker can actually execute, proven by a bounded real forward
+// (ADR-0023). Detected hardware is not a capability; only a validated recipe is.
+message CapabilityReport {
+  SchemaVersion schema_version = 1;
+  Fingerprint fingerprint = 2;
+  ShardRange shard_range = 3;
+  // Backend/device identity, e.g. "rocm:gfx1151", "cpu:avx512".
+  string backend = 4;
+  string device = 5;
+  // True only when a bounded real forward has passed for exactly this
+  // (artifact, range, recipe, device). Never set from hardware detection alone.
+  bool validated = 6;
+  // Sanitized reason when `validated` is false.
+  string detail = 7;
+  // Admission budgets this worker will enforce.
+  uint64 max_concurrent_sessions = 8;
+  uint64 max_context_tokens = 9;
+  FlowControl flow_control = 10;
+  repeated Compression accepted_compression = 11;
+  repeated SchemaVersion supported_schema_versions = 12;
+  int64 validated_at_unix_nanos = 13;
+}
+
+message HealthRequest {
+  SchemaVersion schema_version = 1;
+}
+
+enum ServingState {
+  SERVING_STATE_UNSPECIFIED = 0;
+  // Accepting new Route Sessions.
+  SERVING_STATE_SERVING = 1;
+  // Alive but refusing new sessions; existing sessions continue draining.
+  SERVING_STATE_DRAINING = 2;
+  // Alive but not routable — e.g. registered-but-dark pending certification.
+  SERVING_STATE_NOT_SERVING = 3;
+}
+
+// Live load, for backpressure and Generation Telemetry.
+message HealthReport {
+  SchemaVersion schema_version = 1;
+  ServingState state = 2;
+  uint32 active_sessions = 3;
+  uint32 queued_chunks = 4;
+  // Occupancy of the continuous decode batch.
+  uint32 batch_occupancy = 5;
+  // Fraction of the KV budget in use, 0..1.
+  float kv_pressure = 6;
+  uint64 resident_bytes = 7;
+  string detail = 8;
+}
+
+message ReleaseRequest {
+  SchemaVersion schema_version = 1;
+  string route_session_id = 2;
+  uint64 route_epoch = 3;
+}
+
+message ReleaseResponse {
+  // True when session state existed and was dropped; false when there was
+  // nothing to drop, which is a success, not an error — release is idempotent.
+  bool released = 1;
+  ShardError error = 2;
+}
+
+message CancelRequest {
+  SchemaVersion schema_version = 1;
+  string route_session_id = 2;
+  uint64 route_epoch = 3;
+  // Empty cancels every in-flight unit for the session.
+  string work_id = 4;
+  string reason = 5;
+}
+
+message CancelResponse {
+  uint32 cancelled_work_items = 1;
+  ShardError error = 2;
+}
+
+// ---------------------------------------------------------------------------
+// Service
+// ---------------------------------------------------------------------------
+
+service ShardRuntime {
+  // What this worker can execute. Read before a route is built.
+  rpc GetCapability(CapabilityRequest) returns (CapabilityReport);
+
+  // Live load and serving state.
+  rpc Health(HealthRequest) returns (HealthReport);
+
+  // One long-lived bidirectional stream per Route Session Activation Seam.
+  //
+  // The stream opens with SessionOpen/SessionAccepted, then carries bounded
+  // prefill chunks and decode steps in both directions for the life of the
+  // session. Per-token channel creation is a non-goal: the handshake cost is
+  // paid once and the hot path carries only what changes.
+  rpc Session(stream SessionRequest) returns (stream SessionResponse);
+
+  // Drop session state out of band. Idempotent.
+  rpc Release(ReleaseRequest) returns (ReleaseResponse);
+
+  // Cancel out of band, on a fresh call.
+  //
+  // In-band CancelSignal is preferred, but a sender that is blocked on flow
+  // control cannot write one — a cancel that can only travel down a wedged
+  // stream is not a cancel. This RPC always has a path to the worker.
+  rpc Cancel(CancelRequest) returns (CancelResponse);
+}

packages/node/native/testdata/capability_report_golden.binpb

@@ -0,0 +1,2 @@
+
F
+sha256:1f0c9d2esha256:ab77e410llama-gguf-q4km-rocm"
3*	2026.07.1"rocm*gfx11510
@H€@R€€€ €
Z
b

h€€Ð<E282AC>éθý

packages/node/native/tests/test_shard_protocol_conformance.cpp

@@ -0,0 +1,337 @@
+// C++ conformance test for the native Shard protocol.
+//
+// This test does not check that C++ can round-trip its own output — that would
+// only prove C++ is self-consistent. It parses the *Python-produced* committed
+// vectors, asserts every field the protocol promises to carry, independently
+// recomputes the CRC32C over the reassembled tensor, and re-serializes the
+// message back out for the Python test to compare byte-for-byte. Only that
+// closes the loop: both languages agree on the same bytes and the same meaning.
+//
+// Run via ctest; see packages/node/native/CMakeLists.txt.
+
+#include "shard_runtime.pb.h"
+
+#include <cstdint>
+#include <cstdio>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <sstream>
+#include <string>
+#include <vector>
+
+// `pb` is already taken by protobuf's own generated headers.
+namespace sp = meshnet::shard::v1;
+
+namespace {
+
+int g_failures = 0;
+
+#define CHECK(cond)                                                            \
+  do {                                                                         \
+    if (!(cond)) {                                                             \
+      std::cerr << "FAIL " << __FILE__ << ":" << __LINE__ << ": " << #cond     \
+                << "\n";                                                       \
+      ++g_failures;                                                            \
+    }                                                                          \
+  } while (0)
+
+#define CHECK_EQ(actual, expected)                                             \
+  do {                                                                         \
+    auto &&a_ = (actual);                                                      \
+    auto &&e_ = (expected);                                                    \
+    if (!(a_ == e_)) {                                                         \
+      std::cerr << "FAIL " << __FILE__ << ":" << __LINE__ << ": " << #actual   \
+                << " == " << #expected << "\n  actual:   " << a_               \
+                << "\n  expected: " << e_ << "\n";                             \
+      ++g_failures;                                                            \
+    }                                                                          \
+  } while (0)
+
+// Independent CRC32C (Castagnoli). Written from the polynomial rather than
+// shared with the Python side on purpose: a checksum that both languages
+// compute with the *same* code proves nothing about interoperability.
+uint32_t Crc32c(const std::string &data) {
+  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) ^ 0x82F63B78u : (c >> 1);
+      }
+      table[i] = c;
+    }
+    built = true;
+  }
+  uint32_t crc = 0xFFFFFFFFu;
+  for (unsigned char byte : data) {
+    crc = (crc >> 8) ^ table[(crc ^ byte) & 0xFF];
+  }
+  return crc ^ 0xFFFFFFFFu;
+}
+
+std::string ReadFile(const std::filesystem::path &path) {
+  std::ifstream in(path, std::ios::binary);
+  if (!in) {
+    std::cerr << "FAIL cannot read " << path << "\n";
+    ++g_failures;
+    return {};
+  }
+  std::ostringstream buffer;
+  buffer << in.rdbuf();
+  return buffer.str();
+}
+
+// Reassemble a tensor's fragments and validate coverage, exactly as a worker
+// must before it feeds the bytes to a forward pass.
+std::string ReassembleUncompressed(const sp::NamedTensor &tensor) {
+  std::string body;
+  std::vector<const sp::TensorFragment *> fragments;
+  for (const auto &fragment : tensor.fragments()) {
+    fragments.push_back(&fragment);
+  }
+  CHECK(!fragments.empty());
+  for (uint32_t index = 0; index < fragments.size(); ++index) {
+    const sp::TensorFragment *found = nullptr;
+    for (const auto *fragment : fragments) {
+      if (fragment->fragment_index() == index) {
+        found = fragment;
+        break;
+      }
+    }
+    CHECK(found != nullptr);
+    if (found == nullptr) {
+      return {};
+    }
+    CHECK_EQ(found->fragment_count(),
+             static_cast<uint32_t>(fragments.size()));
+    // Offsets must tile the wire body exactly: no hole, no overlap.
+    CHECK_EQ(found->byte_offset(), static_cast<uint64_t>(body.size()));
+    body.append(found->payload());
+  }
+  return body;
+}
+
+void CheckFingerprint(const sp::Fingerprint &fingerprint) {
+  CHECK_EQ(fingerprint.model_artifact_digest(), std::string("sha256:1f0c9d2e"));
+  CHECK_EQ(fingerprint.runtime_recipe_digest(), std::string("sha256:ab77e410"));
+  CHECK_EQ(fingerprint.recipe_id(), std::string("llama-gguf-q4km-rocm"));
+  CHECK_EQ(fingerprint.recipe_version(), std::string("3"));
+  CHECK_EQ(fingerprint.catalogue_version(), std::string("2026.07.1"));
+}
+
+// The canonical session request, as produced by Python.
+void TestSessionRequestVector(const std::string &bytes) {
+  sp::SessionRequest request;
+  CHECK(request.ParseFromString(bytes));
+  CHECK(request.kind_case() == sp::SessionRequest::kChunk);
+
+  const sp::ActivationChunk &chunk = request.chunk();
+  const sp::Envelope &envelope = chunk.envelope();
+
+  // Every field the protocol promises to carry (acceptance criterion 4).
+  CHECK_EQ(envelope.schema_version(), sp::SCHEMA_VERSION_1);
+  CHECK_EQ(envelope.work_id(), std::string("work-7f3a"));
+  CHECK_EQ(envelope.route_session_id(), std::string("rs-2b91"));
+  CHECK_EQ(envelope.route_epoch(), 7u);
+  CheckFingerprint(envelope.fingerprint());
+  CHECK_EQ(envelope.shard_range().start_layer(), 12u);
+  CHECK_EQ(envelope.shard_range().end_layer(), 24u);
+  // Overlap-safe effective start (ADR-0012): a worker that begins at
+  // start_layer instead would re-apply layers 12..15.
+  CHECK_EQ(envelope.shard_range().effective_start_layer(), 16u);
+  CHECK_EQ(envelope.phase(), sp::PHASE_PREFILL);
+  CHECK_EQ(envelope.position().first_position(), 256u);
+  CHECK_EQ(envelope.position().token_count(), 128u);
+  CHECK_EQ(envelope.idempotency_step(), 42u);
+  CHECK_EQ(envelope.cache_expectation().mode(), sp::CACHE_MODE_PREFILL);
+  CHECK_EQ(envelope.cache_expectation().expected_past_len(), 256u);
+  CHECK_EQ(envelope.deadline_unix_nanos(), 1800000000000000000LL);
+  CHECK_EQ(envelope.chunk().chunk_index(), 1u);
+  CHECK_EQ(envelope.chunk().chunk_count(), 3u);
+  CHECK_EQ(envelope.chunk().final_chunk(), false);
+
+  // The versioned named-tensor bundle (acceptance criterion 5).
+  const sp::TensorBundle &bundle = chunk.bundle();
+  CHECK_EQ(bundle.bundle_version(), 1u);
+  CHECK_EQ(bundle.tensors_size(), 1);
+
+  const sp::NamedTensor &tensor = bundle.tensors(0);
+  CHECK_EQ(tensor.name(), std::string("hidden_states"));
+  CHECK_EQ(tensor.shape_size(), 3);
+  CHECK_EQ(tensor.shape(0), 1);
+  CHECK_EQ(tensor.shape(1), 128);
+  CHECK_EQ(tensor.shape(2), 8);
+  CHECK_EQ(tensor.dtype(), sp::DTYPE_BFLOAT16);
+  CHECK_EQ(tensor.byte_order(), sp::BYTE_ORDER_LITTLE_ENDIAN);
+  CHECK_EQ(tensor.total_bytes(), 1u * 128u * 8u * 2u);
+  CHECK_EQ(tensor.compression(), sp::COMPRESSION_NONE);
+
+  // Multi-fragment on purpose: reassembly is what a worker actually does.
+  CHECK(tensor.fragments_size() > 1);
+  const std::string payload = ReassembleUncompressed(tensor);
+  CHECK_EQ(payload.size(), static_cast<size_t>(tensor.total_bytes()));
+
+  // The payload Python generated, recomputed here from its rule.
+  std::string expected;
+  expected.reserve(payload.size());
+  for (size_t i = 0; i < payload.size(); ++i) {
+    expected.push_back(static_cast<char>((i * 7 + 11) % 256));
+  }
+  CHECK(payload == expected);
+
+  // Checksum: computed by Python, verified by an independent C++ CRC32C.
+  CHECK_EQ(tensor.checksum().algorithm(), sp::CHECKSUM_ALGORITHM_CRC32C);
+  const std::string &checksum = tensor.checksum().value();
+  CHECK_EQ(checksum.size(), 4u);
+  if (checksum.size() == 4) {
+    const uint32_t actual = Crc32c(payload);
+    const uint32_t declared =
+        (static_cast<uint32_t>(static_cast<unsigned char>(checksum[0])) << 24) |
+        (static_cast<uint32_t>(static_cast<unsigned char>(checksum[1])) << 16) |
+        (static_cast<uint32_t>(static_cast<unsigned char>(checksum[2])) << 8) |
+        static_cast<uint32_t>(static_cast<unsigned char>(checksum[3]));
+    CHECK_EQ(actual, declared);
+  }
+}
+
+void TestCapabilityReportVector(const std::string &bytes) {
+  sp::CapabilityReport report;
+  CHECK(report.ParseFromString(bytes));
+  CHECK_EQ(report.schema_version(), sp::SCHEMA_VERSION_1);
+  CheckFingerprint(report.fingerprint());
+  CHECK_EQ(report.backend(), std::string("rocm"));
+  CHECK_EQ(report.device(), std::string("gfx1151"));
+  CHECK_EQ(report.validated(), true);
+  CHECK_EQ(report.max_concurrent_sessions(), 4u);
+  CHECK_EQ(report.max_context_tokens(), 8192u);
+  CHECK_EQ(report.flow_control().max_prefill_chunk_tokens(), 128u);
+  CHECK_EQ(report.flow_control().max_chunk_bytes(), 4u * 1024u * 1024u);
+  CHECK_EQ(report.accepted_compression_size(), 2);
+  CHECK_EQ(report.supported_schema_versions_size(), 1);
+}
+
+// Forward compatibility: a field this build has never heard of must survive a
+// parse/serialize cycle untouched. Without this, an old Shard silently strips
+// fields a newer peer depends on as it forwards activations down the route.
+void TestUnknownFieldsArePreserved(const std::string &bytes) {
+  // Field 9999, wire type 0 (varint), value 12345 — not in this schema.
+  std::string forward = bytes;
+  forward.push_back(static_cast<char>(0xB8));  // tag: (9999 << 3) | 0
+  forward.push_back(static_cast<char>(0xE0));
+  forward.push_back(static_cast<char>(0x04));
+  forward.push_back(static_cast<char>(0xB9));  // value: 12345
+  forward.push_back(static_cast<char>(0x60));
+
+  sp::SessionRequest request;
+  CHECK(request.ParseFromString(forward));
+  // Known fields still parse.
+  CHECK_EQ(request.chunk().envelope().work_id(), std::string("work-7f3a"));
+  // And the unknown field is retained rather than dropped.
+  CHECK(!request.GetReflection()->GetUnknownFields(request).empty());
+
+  std::string reserialized;
+  CHECK(request.SerializeToString(&reserialized));
+  CHECK_EQ(reserialized.size(), forward.size());
+
+  sp::SessionRequest reparsed;
+  CHECK(reparsed.ParseFromString(reserialized));
+  CHECK(!reparsed.GetReflection()->GetUnknownFields(reparsed).empty());
+}
+
+// Backward compatibility: a message from a peer that sets none of the optional
+// groups must still parse, yielding proto3 defaults rather than an error.
+void TestSparseMessageParses() {
+  sp::SessionRequest minimal;
+  minimal.mutable_chunk()->mutable_envelope()->set_work_id("w");
+  std::string bytes;
+  CHECK(minimal.SerializeToString(&bytes));
+
+  sp::SessionRequest parsed;
+  CHECK(parsed.ParseFromString(bytes));
+  CHECK_EQ(parsed.chunk().envelope().work_id(), std::string("w"));
+  CHECK_EQ(parsed.chunk().envelope().route_epoch(), 0u);
+  CHECK_EQ(parsed.chunk().envelope().phase(), sp::PHASE_UNSPECIFIED);
+  CHECK_EQ(parsed.chunk().bundle().tensors_size(), 0);
+}
+
+// The decode fast path must stay small: repeating the full envelope per token
+// is pure overhead on the hottest path in the system.
+void TestDecodeFastPathIsSmall() {
+  sp::SessionRequest decode;
+  sp::DecodeStep *step = decode.mutable_decode();
+  step->set_idempotency_step(9);
+  step->set_position(1024);
+  step->set_expected_past_len(1024);
+  step->set_work_id("work-7f3a");
+  sp::NamedTensor *tensor = step->mutable_tensor();
+  tensor->set_name("hidden_states");
+  tensor->add_shape(1);
+  tensor->add_shape(1);
+  tensor->add_shape(8);
+  tensor->set_dtype(sp::DTYPE_BFLOAT16);
+  tensor->set_byte_order(sp::BYTE_ORDER_LITTLE_ENDIAN);
+  tensor->set_total_bytes(16);
+  tensor->set_compression(sp::COMPRESSION_NONE);
+  sp::TensorFragment *fragment = tensor->add_fragments();
+  fragment->set_fragment_index(0);
+  fragment->set_fragment_count(1);
+  fragment->set_byte_offset(0);
+  fragment->set_payload(std::string(16, '\x01'));
+
+  std::string bytes;
+  CHECK(decode.SerializeToString(&bytes));
+  // Payload is 16 bytes; the framing around it must stay well under the
+  // envelope-carrying prefill path.
+  CHECK(bytes.size() < 96);
+
+  sp::SessionRequest parsed;
+  CHECK(parsed.ParseFromString(bytes));
+  CHECK(parsed.kind_case() == sp::SessionRequest::kDecode);
+  CHECK_EQ(parsed.decode().position(), 1024u);
+  CHECK_EQ(parsed.decode().tensor().total_bytes(), 16u);
+}
+
+}  // namespace
+
+int main(int argc, char **argv) {
+  GOOGLE_PROTOBUF_VERIFY_VERSION;
+  if (argc < 3) {
+    std::cerr << "usage: " << argv[0] << " <testdata-dir> <output-dir>\n";
+    return 2;
+  }
+  const std::filesystem::path testdata(argv[1]);
+  const std::filesystem::path output(argv[2]);
+
+  const std::string session_bytes =
+      ReadFile(testdata / "session_request_golden.binpb");
+  const std::string capability_bytes =
+      ReadFile(testdata / "capability_report_golden.binpb");
+
+  TestSessionRequestVector(session_bytes);
+  TestCapabilityReportVector(capability_bytes);
+  TestUnknownFieldsArePreserved(session_bytes);
+  TestSparseMessageParses();
+  TestDecodeFastPathIsSmall();
+
+  // Re-serialize the canonical message from the C++ object model and hand it
+  // back for Python to compare against the golden bytes. If the two languages
+  // disagreed about any field's encoding, this file would differ.
+  sp::SessionRequest request;
+  if (request.ParseFromString(session_bytes)) {
+    std::string reserialized;
+    if (request.SerializeToString(&reserialized)) {
+      std::ofstream out(output / "cpp_roundtrip.binpb", std::ios::binary);
+      out.write(reserialized.data(),
+                static_cast<std::streamsize>(reserialized.size()));
+    }
+  }
+
+  if (g_failures != 0) {
+    std::cerr << g_failures << " check(s) failed\n";
+    return 1;
+  }
+  std::cout << "all C++ conformance checks passed\n";
+  return 0;
+}

packages/node/pyproject.toml

@@ -13,6 +13,11 @@ dependencies = [
     "huggingface-hub>=0.20",
     "accelerate>=0.28",
     "bitsandbytes>=0.43",
+    # Keep these floors aligned with the committed grpcio-tools 1.82.1 output.
+    # Older runtimes satisfy broad constraints but fail while importing the
+    # generated modules before a node can report a useful protocol error.
+    "grpcio>=1.82.1",
+    "protobuf>=7.35.0",
     "rich>=13",
     "safetensors>=0.4",
     "torch>=2.1",
@@ -23,6 +28,11 @@ dependencies = [
     "kernels>=0.11.1,<0.16",
 ]
 
+[project.optional-dependencies]
+# Regenerating the native Shard protocol stubs. Not needed to run a node: the
+# generated modules are committed, and `grpcio-tools` bundles its own protoc.
+proto = ["grpcio-tools==1.82.1"]
+
 [project.scripts]
 meshnet-node = "meshnet_node.cli:main"
 

scripts/bootstrap_native_toolchain.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+# Build a protobuf C++ toolchain for the native Shard protocol.
+#
+# The Python side needs nothing beyond `pip install grpcio-tools` — it bundles
+# protoc. The C++ side needs libprotobuf headers and a protoc binary, and a
+# machine that has neither (no protobuf-devel, no cmake, no system protoc) can
+# still get a working one from source with this script. It is the exact recipe
+# DGR-002 used to build and run the C++ conformance test.
+#
+# gRPC C++ is deliberately NOT built here. The conformance test only needs
+# message types, so verifying the schema does not require the whole gRPC stack.
+# The worker (DGR-008) will need gRPC C++ and should extend this script then.
+#
+# Usage:
+#   scripts/bootstrap_native_toolchain.sh [install-prefix]
+#
+# Then:
+#   cmake -S packages/node/native -B build/native -DCMAKE_PREFIX_PATH=<prefix>
+#   cmake --build build/native -j
+#   ctest --test-dir build/native --output-on-failure
+
+set -euo pipefail
+
+PREFIX="${1:-${PWD}/build/native-toolchain}"
+WORK="$(mktemp -d)"
+trap 'rm -rf "${WORK}"' EXIT
+
+# Pinned: the C++ runtime a generated stub is compiled against must be a version
+# that stub is allowed to use, so these are exact, not floating.
+PROTOBUF_VERSION="33.1"
+ABSEIL_VERSION="20250814.1"
+
+command -v cmake >/dev/null || {
+  echo "cmake is required (pip install cmake==4.4.0)" >&2
+  exit 1
+}
+
+echo "--- fetching protobuf ${PROTOBUF_VERSION} and abseil ${ABSEIL_VERSION}"
+cd "${WORK}"
+curl -sfL -o protobuf.tar.gz \
+  "https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOBUF_VERSION}/protobuf-${PROTOBUF_VERSION}.tar.gz"
+tar xzf protobuf.tar.gz
+
+# The protobuf release tarball ships utf8_range but not abseil, and its default
+# CMake provider expects abseil as a submodule, so vendor it into place.
+curl -sfL -o abseil.tar.gz \
+  "https://github.com/abseil/abseil-cpp/releases/download/${ABSEIL_VERSION}/abseil-cpp-${ABSEIL_VERSION}.tar.gz"
+tar xzf abseil.tar.gz
+rm -rf "protobuf-${PROTOBUF_VERSION}/third_party/abseil-cpp"
+mv "abseil-cpp-${ABSEIL_VERSION}" "protobuf-${PROTOBUF_VERSION}/third_party/abseil-cpp"
+
+echo "--- building protobuf into ${PREFIX}"
+cmake -S "protobuf-${PROTOBUF_VERSION}" -B build \
+  -DCMAKE_BUILD_TYPE=Release \
+  -DCMAKE_INSTALL_PREFIX="${PREFIX}" \
+  -DCMAKE_POSITION_INDEPENDENT_CODE=ON \
+  -Dprotobuf_ABSL_PROVIDER=module \
+  -Dprotobuf_BUILD_TESTS=OFF \
+  -Dprotobuf_BUILD_SHARED_LIBS=OFF \
+  -DABSL_PROPAGATE_CXX_STD=ON
+cmake --build build -j"$(nproc)"
+cmake --install build
+
+echo "--- done"
+"${PREFIX}/bin/protoc" --version
+echo "configure the protocol build with: -DCMAKE_PREFIX_PATH=${PREFIX}"

scripts/generate_native_protocol.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Generate the Python Shard-protocol stubs from `shard_runtime.proto`.
+
+The `.proto` file is the contract; the Python modules under
+`meshnet_node/native_protocol/generated/` are build output that happens to be
+committed. They are committed so that installing the node does not require a
+protoc toolchain, and `--check` exists so a committed stub can never silently
+drift from the schema it claims to implement.
+
+Usage::
+
+    python scripts/generate_native_protocol.py           # regenerate in place
+    python scripts/generate_native_protocol.py --check   # fail if out of date
+
+C++ stubs are *not* generated here. They are build artifacts produced by CMake
+(`packages/node/native/CMakeLists.txt`) into the build tree, because a C++ build
+already requires a toolchain and nothing is gained by committing them.
+"""
+
+from __future__ import annotations
+
+import argparse
+import pathlib
+import shutil
+import subprocess
+import sys
+import tempfile
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
+PROTO_DIR = REPO_ROOT / "packages/node/native/proto"
+PROTO_FILE = PROTO_DIR / "shard_runtime.proto"
+OUT_DIR = REPO_ROOT / "packages/node/meshnet_node/native_protocol/generated"
+
+# Regenerating with a different protoc emits different gencode headers, so the
+# generator version is part of the contract and `--check` would catch a drift.
+REQUIRED_GRPCIO_TOOLS = "1.82.1"
+
+_HEADER = "# Generated by scripts/generate_native_protocol.py. Do not edit.\n"
+
+
+def _generate(into: pathlib.Path) -> None:
+    """Run protoc, writing generated modules into `into`."""
+    try:
+        from grpc_tools import protoc
+    except ImportError:  # pragma: no cover - exercised only without the toolchain
+        sys.exit(
+            "grpc_tools is required to generate stubs:\n"
+            f"    pip install grpcio-tools=={REQUIRED_GRPCIO_TOOLS}"
+        )
+
+    into.mkdir(parents=True, exist_ok=True)
+    # grpc_tools bundles protoc and the well-known types, so generation needs no
+    # system protoc and produces identical output on any machine.
+    well_known = pathlib.Path(protoc.__file__).parent / "_proto"
+    args = [
+        "protoc",
+        f"--proto_path={PROTO_DIR}",
+        f"--proto_path={well_known}",
+        f"--python_out={into}",
+        f"--pyi_out={into}",
+        f"--grpc_python_out={into}",
+        str(PROTO_FILE),
+    ]
+    if protoc.main(args) != 0:
+        sys.exit("protoc failed")
+
+    # protoc emits `import shard_runtime_pb2` — a bare top-level import that only
+    # resolves if the generated directory happens to be on sys.path. Rewrite it
+    # to a relative import so the package works as an installed package.
+    grpc_module = into / "shard_runtime_pb2_grpc.py"
+    text = grpc_module.read_text()
+    text = text.replace(
+        "import shard_runtime_pb2 as shard__runtime__pb2",
+        "from . import shard_runtime_pb2 as shard__runtime__pb2",
+    )
+    grpc_module.write_text(text)
+
+    (into / "__init__.py").write_text(
+        _HEADER + '"""Generated protobuf/gRPC stubs for the native Shard protocol."""\n'
+    )
+
+
+def _tracked_files(directory: pathlib.Path) -> dict[str, bytes]:
+    return {
+        path.name: path.read_bytes()
+        for path in sorted(directory.iterdir())
+        if path.is_file() and path.suffix in {".py", ".pyi"}
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="verify the committed stubs match the .proto instead of rewriting them",
+    )
+    args = parser.parse_args()
+
+    if args.check:
+        with tempfile.TemporaryDirectory() as tmp:
+            fresh = pathlib.Path(tmp) / "generated"
+            _generate(fresh)
+            if not OUT_DIR.is_dir():
+                print(f"generated stubs are missing: {OUT_DIR}", file=sys.stderr)
+                return 1
+            if _tracked_files(fresh) != _tracked_files(OUT_DIR):
+                print(
+                    "generated stubs are out of date with shard_runtime.proto.\n"
+                    "Run: python scripts/generate_native_protocol.py",
+                    file=sys.stderr,
+                )
+                return 1
+        print("generated stubs are up to date")
+        return 0
+
+    if OUT_DIR.exists():
+        shutil.rmtree(OUT_DIR)
+    _generate(OUT_DIR)
+    print(f"wrote {OUT_DIR.relative_to(REPO_ROOT)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/generate_protocol_goldens.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Write the committed cross-language conformance vectors.
+
+The bytes under `packages/node/native/testdata/` are the reference both the
+Python and the C++ conformance tests assert against. They are committed so the
+C++ test can run without a Python step, and `--check` exists so they can never
+drift from the schema unnoticed: if a schema edit changes the canonical
+message's encoding, `--check` fails and the change has to be acknowledged.
+
+Usage::
+
+    python scripts/generate_protocol_goldens.py           # rewrite vectors
+    python scripts/generate_protocol_goldens.py --check    # fail if stale
+"""
+
+from __future__ import annotations
+
+import argparse
+import pathlib
+import sys
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT / "packages/node"))
+
+from meshnet_node.native_protocol import conformance  # noqa: E402
+
+
+def _vectors() -> dict[str, bytes]:
+    return {
+        conformance.GOLDEN_SESSION_REQUEST: conformance.serialize(
+            conformance.canonical_session_request()
+        ),
+        conformance.GOLDEN_CAPABILITY_REPORT: conformance.serialize(
+            conformance.canonical_capability_report()
+        ),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--check", action="store_true")
+    args = parser.parse_args()
+
+    out_dir = conformance.TESTDATA_DIR
+    out_dir.mkdir(parents=True, exist_ok=True)
+    stale = []
+
+    for name, payload in _vectors().items():
+        path = out_dir / name
+        if args.check:
+            if not path.is_file() or path.read_bytes() != payload:
+                stale.append(name)
+            continue
+        path.write_bytes(payload)
+        print(f"wrote {path.relative_to(REPO_ROOT)} ({len(payload)} bytes)")
+
+    if stale:
+        print(
+            "conformance vectors are stale: " + ", ".join(stale) + "\n"
+            "The canonical message no longer encodes to the committed bytes. If "
+            "that is intended, run: python scripts/generate_protocol_goldens.py",
+            file=sys.stderr,
+        )
+        return 1
+    if args.check:
+        print("conformance vectors are up to date")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_native_shard_protocol.py

@@ -0,0 +1,577 @@
+"""Conformance tests for the native Shard protocol (ADR-0020, DGR-002).
+
+Three layers are tested, and they are not the same thing:
+
+1. The *schema* — asserted against the descriptor, not against the Python
+   helpers. If a field the protocol promises to carry were dropped from the
+   `.proto`, a test that only exercised the codec would still pass.
+2. The *codec* — that a payload which is corrupt, short, holed, or byte-swapped
+   is rejected rather than fed to a forward pass.
+3. *Compatibility* — that an old build preserves fields a newer peer added, and
+   that the committed cross-language vectors still encode as promised.
+"""
+
+from __future__ import annotations
+
+import pathlib
+import subprocess
+import sys
+
+import pytest
+
+pytest.importorskip("google.protobuf", reason="protobuf runtime is required")
+
+from google.protobuf import descriptor_pb2
+
+from meshnet_node.activation_compression import CompressionPolicy
+from meshnet_node.native_protocol import (
+    DEFAULT_MAX_CHUNK_BYTES,
+    DEFAULT_MAX_FRAGMENTS_PER_TENSOR,
+    DEFAULT_MAX_PREFILL_CHUNK_TOKENS,
+    DEFAULT_MAX_TENSORS_PER_BUNDLE,
+    HIDDEN_STATES,
+    PayloadCorrupt,
+    ProtocolError,
+    checksum_of,
+    decode_bundle,
+    decode_tensor,
+    default_flow_control,
+    encode_bundle,
+    encode_tensor,
+    negotiate_flow_control,
+    pb,
+    plan_prefill_chunks,
+    validate_session_message_size,
+)
+from meshnet_node.native_protocol import conformance
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
+CPP_ROUNDTRIP = REPO_ROOT / "build/native" / conformance.CPP_ROUNDTRIP
+
+
+# --- The schema itself ------------------------------------------------------
+
+
+def test_service_exposes_capability_health_session_release_and_cancel():
+    service = pb.DESCRIPTOR.services_by_name["ShardRuntime"]
+    assert set(service.methods_by_name) == {
+        "GetCapability",
+        "Health",
+        "Session",
+        "Release",
+        "Cancel",
+    }
+
+
+def test_session_is_one_long_lived_bidirectional_stream():
+    session = pb.DESCRIPTOR.services_by_name["ShardRuntime"].methods_by_name["Session"]
+    assert session.client_streaming, "the seam must stream requests"
+    assert session.server_streaming, "the seam must stream responses"
+
+    # Cancellation must not have to travel down a stream that flow control has
+    # wedged, so it also exists as its own unary call.
+    cancel = pb.DESCRIPTOR.services_by_name["ShardRuntime"].methods_by_name["Cancel"]
+    assert not cancel.client_streaming and not cancel.server_streaming
+
+
+def test_envelope_carries_every_field_the_protocol_promises():
+    # Asserted against the descriptor: this is the acceptance criterion, and it
+    # must fail if the .proto drops a field, not merely if the codec stops
+    # setting one.
+    fields = set(pb.Envelope.DESCRIPTOR.fields_by_name)
+    assert {
+        "schema_version",
+        "work_id",
+        "route_session_id",
+        "route_epoch",
+        "fingerprint",
+        "shard_range",
+        "phase",
+        "position",
+        "idempotency_step",
+        "cache_expectation",
+        "deadline_unix_nanos",
+        "chunk",
+    } <= fields
+
+    assert {"model_artifact_digest", "runtime_recipe_digest"} <= set(
+        pb.Fingerprint.DESCRIPTOR.fields_by_name
+    )
+    # Overlap-safe start (ADR-0012) is a distinct field from the registered one.
+    assert {"start_layer", "end_layer", "effective_start_layer"} <= set(
+        pb.ShardRange.DESCRIPTOR.fields_by_name
+    )
+    assert {"mode", "expected_past_len"} <= set(
+        pb.CacheExpectation.DESCRIPTOR.fields_by_name
+    )
+
+
+def test_named_tensor_bundle_is_versioned_and_fully_described():
+    assert "bundle_version" in pb.TensorBundle.DESCRIPTOR.fields_by_name
+    assert {"name", "shape", "dtype", "byte_order", "compression", "checksum", "fragments"} <= set(
+        pb.NamedTensor.DESCRIPTOR.fields_by_name
+    )
+
+
+def test_phases_and_error_codes_cover_the_lifecycle():
+    assert {"PHASE_PREFILL", "PHASE_DECODE", "PHASE_RELEASE", "PHASE_CANCEL"} <= set(
+        pb.Phase.keys()
+    )
+    # A cache miss is a first-class, recoverable outcome (ADR-0022), not a crash.
+    assert {
+        "ERROR_CODE_CACHE_MISS",
+        "ERROR_CODE_FINGERPRINT_MISMATCH",
+        "ERROR_CODE_EPOCH_STALE",
+        "ERROR_CODE_SCHEMA_UNSUPPORTED",
+        "ERROR_CODE_DEADLINE_EXCEEDED",
+        "ERROR_CODE_FLOW_CONTROL_VIOLATION",
+    } <= set(pb.ErrorCode.keys())
+
+
+# --- Tensor bundle round-trip ----------------------------------------------
+
+
+def test_tensor_round_trips_through_fragments():
+    payload = bytes(range(256)) * 4
+    tensor = encode_tensor(
+        HIDDEN_STATES, payload, [1, 64, 8], pb.DTYPE_BFLOAT16, max_fragment_bytes=100
+    )
+
+    assert len(tensor.fragments) > 1, "the bound must actually split the payload"
+    assert tensor.total_bytes == len(payload)
+    assert decode_tensor(tensor) == payload
+
+
+def test_bundle_round_trips_multiple_named_tensors():
+    # An architecture boundary may need more than one tensor; that is why the
+    # payload is a named bundle rather than a bare buffer.
+    hidden = encode_tensor(HIDDEN_STATES, b"\x01\x02" * 8, [1, 8, 1], pb.DTYPE_BFLOAT16)
+    positions = encode_tensor(
+        "position_ids", (7).to_bytes(4, "little") * 8, [8], pb.DTYPE_INT32
+    )
+    bundle = encode_bundle([hidden, positions])
+
+    restored = decode_bundle(pb.TensorBundle.FromString(bundle.SerializeToString()))
+    assert restored == {
+        HIDDEN_STATES: b"\x01\x02" * 8,
+        "position_ids": (7).to_bytes(4, "little") * 8,
+    }
+
+
+def test_compressed_tensor_round_trips_and_keeps_its_uncompressed_checksum():
+    pytest.importorskip("zstandard")
+    # Highly compressible, and over the policy's minimum input size.
+    payload = b"\x00" * 65536
+    always = CompressionPolicy(min_input_bytes=0, min_savings_bytes=0, min_savings_ratio=0.0)
+    tensor = encode_tensor(
+        HIDDEN_STATES, payload, [1, 4096, 8], pb.DTYPE_BFLOAT16, policy=always
+    )
+
+    assert tensor.compression == pb.COMPRESSION_ZSTD
+    assert sum(len(f.payload) for f in tensor.fragments) < len(payload)
+    # The checksum covers the uncompressed bytes, so it stays valid whether or
+    # not a hop chose to compress.
+    assert tensor.checksum == checksum_of(payload)
+    assert decode_tensor(tensor) == payload
+
+
+# --- The codec refuses what it cannot account for --------------------------
+
+
+def test_corrupt_payload_is_rejected_by_checksum():
+    tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
+    # Flip one byte, as a lossy relay or a bad NIC would.
+    tensor.fragments[0].payload = b"\xab" + tensor.fragments[0].payload[1:]
+
+    with pytest.raises(PayloadCorrupt, match="CRC32C"):
+        decode_tensor(tensor)
+
+
+def test_missing_fragment_is_rejected_rather_than_silently_truncated():
+    tensor = encode_tensor(
+        HIDDEN_STATES, b"\xaa" * 64, [1, 32, 1], pb.DTYPE_BFLOAT16, max_fragment_bytes=16
+    )
+    del tensor.fragments[1]
+
+    with pytest.raises(PayloadCorrupt):
+        decode_tensor(tensor)
+
+
+def test_fragment_hole_is_rejected():
+    tensor = encode_tensor(
+        HIDDEN_STATES, b"\xaa" * 64, [1, 32, 1], pb.DTYPE_BFLOAT16, max_fragment_bytes=16
+    )
+    # A gap in coverage: offsets no longer tile the body.
+    tensor.fragments[2].byte_offset += 4
+
+    with pytest.raises(PayloadCorrupt, match="expected"):
+        decode_tensor(tensor)
+
+
+def test_shape_that_disagrees_with_payload_is_rejected_at_encode():
+    with pytest.raises(ProtocolError, match="carries"):
+        encode_tensor(HIDDEN_STATES, b"\x01\x02", [1, 8, 8], pb.DTYPE_BFLOAT16)
+
+
+def test_shape_that_disagrees_with_declared_bytes_is_rejected_at_decode():
+    tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
+    # A peer claiming a larger tensor than its bytes describe.
+    tensor.shape[1] = 32
+
+    with pytest.raises(PayloadCorrupt, match="implies"):
+        decode_tensor(tensor)
+
+
+def test_big_endian_tensor_is_rejected_loudly():
+    tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
+    tensor.byte_order = pb.BYTE_ORDER_BIG_ENDIAN
+
+    # Byte-swapped activations would be plausible-looking garbage, so this is an
+    # error rather than a best-effort read.
+    with pytest.raises(ProtocolError, match="big-endian"):
+        decode_tensor(tensor)
+
+
+def test_bundle_from_a_newer_layout_is_refused():
+    bundle = encode_bundle([])
+    bundle.bundle_version = 99
+
+    with pytest.raises(ProtocolError, match="not supported"):
+        decode_bundle(bundle)
+
+
+def test_bundle_with_unspecified_version_is_refused():
+    with pytest.raises(ProtocolError, match="version 0"):
+        decode_bundle(pb.TensorBundle())
+
+
+def test_tensor_with_unspecified_compression_is_refused():
+    tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
+    tensor.compression = pb.COMPRESSION_UNSPECIFIED
+
+    with pytest.raises(ProtocolError, match="unspecified"):
+        decode_tensor(tensor)
+
+
+def test_tensor_with_unspecified_checksum_is_refused():
+    tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
+    tensor.checksum.algorithm = pb.CHECKSUM_ALGORITHM_UNSPECIFIED
+
+    with pytest.raises(ProtocolError, match="unspecified"):
+        decode_tensor(tensor)
+
+
+def test_declared_tensor_size_cannot_exceed_negotiated_chunk_bound():
+    tensor = pb.NamedTensor(
+        name=HIDDEN_STATES,
+        shape=[1],
+        dtype=pb.DTYPE_UINT8,
+        byte_order=pb.BYTE_ORDER_LITTLE_ENDIAN,
+        total_bytes=DEFAULT_MAX_CHUNK_BYTES + 1,
+        compression=pb.COMPRESSION_NONE,
+    )
+
+    with pytest.raises(ProtocolError, match="negotiated chunk bound"):
+        decode_tensor(tensor)
+
+
+def test_stricter_session_negotiation_overrides_global_defaults():
+    tensor = encode_tensor(
+        HIDDEN_STATES,
+        b"\xaa" * 32,
+        [32],
+        pb.DTYPE_UINT8,
+        max_fragment_bytes=32,
+    )
+
+    with pytest.raises(ProtocolError, match="16-byte negotiated"):
+        decode_tensor(tensor, max_chunk_bytes=16, max_fragment_bytes=16)
+    with pytest.raises(ProtocolError, match="16-byte"):
+        decode_bundle(
+            encode_bundle([tensor]),
+            max_chunk_bytes=16,
+            max_fragment_bytes=16,
+        )
+
+
+def test_fragment_cannot_exceed_protocol_fragment_bound():
+    tensor = encode_tensor(
+        HIDDEN_STATES,
+        b"\xaa" * (1024 * 1024 + 1),
+        [1024 * 1024 + 1],
+        pb.DTYPE_UINT8,
+        max_fragment_bytes=1024 * 1024 + 1,
+    )
+
+    with pytest.raises(PayloadCorrupt, match="fragment larger"):
+        decode_tensor(tensor)
+
+
+def test_fragment_count_is_bounded_before_sorting_or_allocation():
+    tensor = encode_tensor(
+        HIDDEN_STATES,
+        b"\xaa" * (DEFAULT_MAX_FRAGMENTS_PER_TENSOR + 1),
+        [DEFAULT_MAX_FRAGMENTS_PER_TENSOR + 1],
+        pb.DTYPE_UINT8,
+        max_fragment_bytes=1,
+        max_fragments=DEFAULT_MAX_FRAGMENTS_PER_TENSOR + 1,
+    )
+
+    with pytest.raises(PayloadCorrupt, match="fragments.*exceeding"):
+        decode_tensor(tensor)
+
+
+def test_tensor_count_is_bounded_before_bundle_decoding():
+    tensors = [
+        pb.NamedTensor(name=f"t-{index}")
+        for index in range(DEFAULT_MAX_TENSORS_PER_BUNDLE + 1)
+    ]
+
+    with pytest.raises(ProtocolError, match="tensors.*exceeding"):
+        encode_bundle(tensors)
+
+
+@pytest.mark.parametrize(
+    ("shape", "match"),
+    [
+        ([1] * 9, "rank"),
+        ([-1], "dimension outside"),
+        ([1 << 31], "dimension outside"),
+    ],
+)
+def test_shape_rank_and_dimensions_are_bounded_before_payload_work(shape, match):
+    with pytest.raises(ProtocolError, match=match):
+        encode_tensor(HIDDEN_STATES, b"\x00", shape, pb.DTYPE_UINT8)
+
+
+def test_complete_session_message_includes_envelope_in_byte_bound():
+    small = pb.SessionRequest(chunk=pb.ActivationChunk(bundle=encode_bundle([])))
+    assert validate_session_message_size(small, max_chunk_bytes=64) == small.ByteSize()
+
+    oversized = pb.SessionRequest(
+        chunk=pb.ActivationChunk(
+            envelope=pb.Envelope(work_id="w" * 128),
+            bundle=encode_bundle([]),
+        )
+    )
+    with pytest.raises(ProtocolError, match="serialized session message"):
+        validate_session_message_size(oversized, max_chunk_bytes=64)
+
+
+def test_bundle_serialized_size_is_bounded():
+    payload = b"\xaa" * (DEFAULT_MAX_CHUNK_BYTES // 2)
+    first = encode_tensor("first", payload, [len(payload)], pb.DTYPE_UINT8)
+    second = encode_tensor("second", payload, [len(payload)], pb.DTYPE_UINT8)
+
+    with pytest.raises(ProtocolError, match="serialized tensor bundle"):
+        encode_bundle([first, second])
+
+
+def test_compressed_tensor_cannot_expand_past_declared_size():
+    pytest.importorskip("zstandard")
+    payload = b"\x00" * 65536
+    always = CompressionPolicy(min_input_bytes=0, min_savings_bytes=0, min_savings_ratio=0.0)
+    tensor = encode_tensor(
+        HIDDEN_STATES, payload, [len(payload)], pb.DTYPE_UINT8, policy=always
+    )
+    assert tensor.compression == pb.COMPRESSION_ZSTD
+    # Simulate a hostile frame that advertises a small output while expanding to
+    # the original 64 KiB. Shape and total_bytes agree, so only the bounded
+    # decompressor protects the receiver.
+    tensor.shape[:] = [1024]
+    tensor.total_bytes = 1024
+
+    with pytest.raises(PayloadCorrupt, match="bounded zstd"):
+        decode_tensor(tensor)
+
+
+def test_generated_runtime_floors_match_committed_stubs():
+    metadata = (REPO_ROOT / "packages/node/pyproject.toml").read_text()
+    assert '"grpcio>=1.82.1"' in metadata
+    assert '"protobuf>=7.35.0"' in metadata
+
+
+# --- Bounded prefill chunking and the decode fast path ----------------------
+
+
+def test_prefill_is_split_into_bounded_token_aligned_chunks():
+    chunks = plan_prefill_chunks(2048)
+
+    assert all(c.token_count <= DEFAULT_MAX_PREFILL_CHUNK_TOKENS for c in chunks)
+    assert sum(c.token_count for c in chunks) == 2048
+    # Contiguous, token-aligned, and no split falls mid-token.
+    assert [c.first_position for c in chunks] == [
+        i * DEFAULT_MAX_PREFILL_CHUNK_TOKENS for i in range(len(chunks))
+    ]
+    assert [c.final_chunk for c in chunks] == [False] * (len(chunks) - 1) + [True]
+
+
+def test_final_prefill_chunk_carries_the_remainder():
+    chunks = plan_prefill_chunks(300, max_tokens=128)
+
+    assert [c.token_count for c in chunks] == [128, 128, 44]
+    assert chunks[-1].chunk_info().final_chunk
+    assert chunks[0].position().first_position == 0
+    assert chunks[-1].position().first_position == 256
+
+
+def test_empty_prefill_is_refused():
+    with pytest.raises(ProtocolError):
+        plan_prefill_chunks(0)
+
+
+def test_decode_fast_path_is_much_smaller_than_a_full_envelope_chunk():
+    hidden = b"\x01\x02" * 8  # one token, hidden=8, bfloat16
+    tensor = encode_tensor(HIDDEN_STATES, hidden, [1, 1, 8], pb.DTYPE_BFLOAT16)
+
+    fast = pb.SessionRequest(
+        decode=pb.DecodeStep(
+            idempotency_step=9,
+            position=1024,
+            expected_past_len=1024,
+            tensor=tensor,
+            work_id="work-7f3a",
+        )
+    )
+    # The same single token carried the long way, repeating identity that the
+    # handshake already fixed for the life of the stream.
+    full = pb.SessionRequest(
+        chunk=pb.ActivationChunk(
+            envelope=conformance.canonical_session_request().chunk.envelope,
+            bundle=encode_bundle([tensor]),
+        )
+    )
+
+    assert len(fast.SerializeToString()) * 2 < len(full.SerializeToString())
+    assert decode_tensor(fast.decode.tensor) == hidden
+
+
+def test_flow_control_defaults_bound_the_queue_and_the_message():
+    limits = default_flow_control()
+
+    assert limits.max_prefill_chunk_tokens == DEFAULT_MAX_PREFILL_CHUNK_TOKENS
+    assert limits.max_chunk_bytes == DEFAULT_MAX_CHUNK_BYTES
+    assert limits.max_inflight_chunks > 0
+
+
+def test_flow_control_negotiation_takes_the_strictest_bound():
+    proposed = pb.FlowControl(
+        credits_granted=64, max_inflight_chunks=64, max_chunk_bytes=64 << 20,
+        max_prefill_chunk_tokens=1024,
+    )
+    settled = negotiate_flow_control(proposed, default_flow_control())
+
+    # A sender cannot talk a worker into unbounded queues by proposing a large
+    # window: neither peer can raise the other's ceiling.
+    assert settled.max_inflight_chunks == default_flow_control().max_inflight_chunks
+    assert settled.credits_granted <= settled.max_inflight_chunks
+    assert settled.max_chunk_bytes == DEFAULT_MAX_CHUNK_BYTES
+    assert settled.max_prefill_chunk_tokens == DEFAULT_MAX_PREFILL_CHUNK_TOKENS
+
+
+def test_initial_credits_never_exceed_negotiated_inflight_limit():
+    proposed = pb.FlowControl(credits_granted=64, max_inflight_chunks=64)
+    limits = pb.FlowControl(credits_granted=64, max_inflight_chunks=8)
+
+    settled = negotiate_flow_control(proposed, limits)
+
+    assert settled.credits_granted == 8
+    assert settled.max_inflight_chunks == 8
+
+
+# --- Compatibility ----------------------------------------------------------
+
+
+def test_committed_vectors_still_encode_as_promised():
+    # The C++ test asserts against these exact bytes. If a schema change alters
+    # the canonical encoding, it must be acknowledged by regenerating them.
+    golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
+    assert conformance.serialize(conformance.canonical_session_request()) == golden
+
+    report = (conformance.TESTDATA_DIR / conformance.GOLDEN_CAPABILITY_REPORT).read_bytes()
+    assert conformance.serialize(conformance.canonical_capability_report()) == report
+
+
+def test_golden_session_request_round_trips_with_every_field_intact():
+    golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
+    request = pb.SessionRequest.FromString(golden)
+    envelope = request.chunk.envelope
+
+    assert envelope.work_id == conformance.WORK_ID
+    assert envelope.route_session_id == conformance.ROUTE_SESSION_ID
+    assert envelope.route_epoch == conformance.ROUTE_EPOCH
+    assert envelope.idempotency_step == conformance.IDEMPOTENCY_STEP
+    assert envelope.shard_range.effective_start_layer == conformance.EFFECTIVE_START_LAYER
+    assert envelope.phase == pb.PHASE_PREFILL
+    assert envelope.cache_expectation.mode == pb.CACHE_MODE_PREFILL
+    assert envelope.deadline_unix_nanos == conformance.DEADLINE_UNIX_NANOS
+    assert decode_bundle(request.chunk.bundle) == {
+        HIDDEN_STATES: conformance.canonical_payload()
+    }
+    assert request.SerializeToString(deterministic=True) == golden
+
+
+def test_unknown_fields_from_a_newer_peer_survive_a_forwarding_hop():
+    # A Shard forwards activations onward. If it silently dropped fields a newer
+    # peer added, it would corrupt a route it is merely a waypoint on.
+    golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
+    future_field = b"\xb8\xe0\x04\xb9\x60"  # field 9999, varint 12345
+    request = pb.SessionRequest.FromString(golden + future_field)
+
+    assert request.chunk.envelope.work_id == conformance.WORK_ID
+    assert request.SerializeToString() == golden + future_field
+
+
+def test_a_message_missing_newer_field_groups_still_parses():
+    sparse = pb.SessionRequest(chunk=pb.ActivationChunk(envelope=pb.Envelope(work_id="w")))
+    parsed = pb.SessionRequest.FromString(sparse.SerializeToString())
+
+    assert parsed.chunk.envelope.work_id == "w"
+    assert parsed.chunk.envelope.route_epoch == 0
+    assert parsed.chunk.envelope.phase == pb.PHASE_UNSPECIFIED
+
+
+def test_retired_fragment_field_stays_reserved():
+    # `uncompressed_size` (field 5) was removed because NamedTensor.total_bytes
+    # is the single source of truth. The number stays reserved so it can never
+    # be recycled for a different meaning — a recycled number is the one schema
+    # change that old and new peers cannot detect, because the bytes still parse.
+    descriptor = descriptor_pb2.DescriptorProto()
+    pb.TensorFragment.DESCRIPTOR.CopyToProto(descriptor)
+
+    assert 5 not in {field.number for field in descriptor.field}
+    assert any(r.start <= 5 < r.end for r in descriptor.reserved_range)
+    assert "uncompressed_size" in descriptor.reserved_name
+
+
+def test_a_peer_still_sending_the_retired_field_does_not_corrupt_the_tensor():
+    # An older peer that still sets field 5 must be parsed, not rejected: the
+    # value lands in unknown fields and the payload is unaffected.
+    tensor = encode_tensor(HIDDEN_STATES, b"\xaa" * 32, [1, 16, 1], pb.DTYPE_BFLOAT16)
+    wire = tensor.SerializeToString()
+    fragment = pb.TensorFragment.FromString(tensor.fragments[0].SerializeToString() + b"\x28\x20")
+
+    assert fragment.payload == tensor.fragments[0].payload
+    assert decode_tensor(pb.NamedTensor.FromString(wire)) == b"\xaa" * 32
+
+
+def test_generated_python_stubs_match_the_proto():
+    pytest.importorskip("grpc_tools", reason="protoc toolchain is required to verify")
+    result = subprocess.run(
+        [sys.executable, str(REPO_ROOT / "scripts/generate_native_protocol.py"), "--check"],
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
+@pytest.mark.skipif(
+    not CPP_ROUNDTRIP.is_file(),
+    reason="build the C++ conformance test to check cross-language agreement",
+)
+def test_cpp_and_python_agree_byte_for_byte():
+    # Written by the C++ conformance test: it parsed the golden bytes into its
+    # own object model and serialized them back. Byte equality means both
+    # languages encode every field of this schema identically.
+    golden = (conformance.TESTDATA_DIR / conformance.GOLDEN_SESSION_REQUEST).read_bytes()
+    assert CPP_ROUNDTRIP.read_bytes() == golden