feat(genvm): vendor leader public data codec

backend/node/base.py

@@ -990,7 +990,7 @@ async def _run_genvm(
         self.timing_callback("GENVM_PREPARATION_START")
 
         leader_res: None | dict[int, bytes]
-        if self.leader_receipt is None or not self.leader_receipt.eq_outputs:
+        if self.leader_receipt is None or self.leader_receipt.eq_outputs is None:
             leader_res = None
         else:
             leader_res = {

backend/node/genvm/base.py

@@ -51,6 +51,7 @@
 # subset the SDK exposes, so import it last to win over the star import.
 from .origin.host_fns import ResultCode
 from .origin import logger as genvm_logger
+from .origin.leader_public_data import LeaderPublicData
 from .error_codes import (
     extract_error_code,
     extract_error_code_from_timeout,
@@ -656,8 +657,10 @@ def provide_result(
                         )
                     )
 
-        # Extract eq_outputs from result_nondet_results
-        eq_outputs = {i: data for i, data in enumerate(res.result_nondet_results)}
+        leader_public_data = LeaderPublicData.decode(res.result_leader_public_data)
+        eq_outputs = {
+            i: data for i, data in enumerate(leader_public_data.nondet_block_outputs)
+        }
 
         execution_stats = dict(ctx.stats)
         llm_token_metrics = _extract_llm_token_metrics(res.metrics)
@@ -909,16 +912,16 @@ def _create_timeout_result(
     )
 
 
-def _leader_results_to_list(
+def _encode_leader_public_data(
     leader_results: dict[int, bytes] | None,
-) -> list[bytes] | None:
-    """Convert dict[int, bytes] keyed by call_no to ordered list[bytes]."""
+) -> bytes | None:
     if leader_results is None:
         return None
-    if not leader_results:
-        return []
-    max_key = max(leader_results.keys())
-    return [leader_results.get(i, b"") for i in range(max_key + 1)]
+    outputs = []
+    if leader_results:
+        max_key = max(leader_results.keys())
+        outputs = [leader_results.get(i, b"") for i in range(max_key + 1)]
+    return LeaderPublicData(outputs).encode()
 
 
 async def run_genvm_host(
@@ -1035,7 +1038,7 @@ async def run_genvm_host(
                 leader_results = fresh_args.get(
                     "leader_results", host_args.get("leader_results")
                 )
-                leader_nondet_results = _leader_results_to_list(leader_results)
+                leader_public_data = _encode_leader_public_data(leader_results)
 
                 try:
                     # Fresh manager websocket per attempt: run_genvm never owns
@@ -1065,7 +1068,7 @@ async def run_genvm_host(
                             calldata=fresh_args.get(
                                 "calldata_bytes", host_args.get("calldata_bytes", b"")
                             ),
-                            leader_nondet_results=leader_nondet_results,
+                            leader_public_data=leader_public_data,
                             unsafe_overrides=base_host.UnsafeOverrides(
                                 reroute_to=genvm_executor_selector or ""
                             ),

backend/node/genvm/origin/base_host.py

@@ -506,7 +506,7 @@ class ConsumedResult:
     result_fingerprint: ResultFingerprint | None = None
     result_storage_deltas: list[tuple[bytes, bytes]] = field(default_factory=list)
     result_emissions: list[ResultEmission] = field(default_factory=list)
-    result_nondet_results: list[bytes] = field(default_factory=list)
+    result_leader_public_data: bytes = b""
     data_fees_remaining: list[int] = field(default_factory=list)
 
     @classmethod
@@ -559,7 +559,7 @@ def decode(cls, raw: typing.Any) -> "ConsumedResult":
             result_fingerprint=decoded.get("fingerprint"),
             result_storage_deltas=decoded.get("storage_deltas", []),
             result_emissions=decoded.get("emissions", []),
-            result_nondet_results=decoded.get("nondet_results", []),
+            result_leader_public_data=decoded.get("leader_public_data", b""),
             data_fees_remaining=decoded.get("data_fees_remaining", []),
         )
 
@@ -579,7 +579,7 @@ class RunHostAndProgramRes:
     result_fingerprint: ResultFingerprint | None
     result_storage_deltas: list[tuple[bytes, bytes]]
     result_emissions: list[ResultEmission]
-    result_nondet_results: list[bytes]
+    result_leader_public_data: bytes
     data_fees_remaining: list[int]
     metrics: dict[str, typing.Any] | None = None
     vm_error_description: str | None = None
@@ -1177,7 +1177,7 @@ async def run_genvm(
     bucket_totals: list[int],
     code: bytes | None = None,
     calldata: bytes,
-    leader_nondet_results: list[bytes] | None = None,
+    leader_public_data: bytes | None = None,
     message_fee_allocation: collections.abc.Sequence[fees.MessageAllocationNode] = (),
     unsafe_overrides: UnsafeOverrides | None = None,
     request_extra: collections.abc.Mapping[
@@ -1228,7 +1228,7 @@ async def run_genvm(
             "extra_args": list(extra_args),
             "code": code,
             "calldata": calldata,
-            "leader_nondet_results": leader_nondet_results,
+            "leader_public_data": leader_public_data,
             "bucket_totals": bucket_totals,
             "gas_data": effective_gas_data,
             "message_fee_allocation": list(message_fee_allocation),
@@ -1403,7 +1403,7 @@ async def cancel_on_shutdown():
             result_fingerprint=consumed.result_fingerprint,
             result_storage_deltas=consumed.result_storage_deltas,
             result_emissions=consumed.result_emissions,
-            result_nondet_results=consumed.result_nondet_results,
+            result_leader_public_data=consumed.result_leader_public_data,
             data_fees_remaining=consumed.data_fees_remaining,
             vm_error_description=vm_error_description,
             execution_time=time.time() - started_at,

backend/node/genvm/origin/leader_public_data.py

@@ -0,0 +1,98 @@
+import collections.abc
+from dataclasses import dataclass
+from typing import Self
+
+_PADDING = b"padded"
+
+
+@dataclass
+class LeaderPublicData:
+    nondet_block_outputs: list[bytes]
+
+    def encode(self) -> bytes:
+        payload = b"".join(
+            _encode_bytes(value) for value in (*self.nondet_block_outputs, _PADDING)
+        )
+        return _encode_length(len(payload), 0xC0, 0xF7) + payload
+
+    @classmethod
+    def decode(cls, encoded: collections.abc.Buffer) -> Self:
+        data = bytes(encoded)
+        if not data:
+            return cls([])
+
+        payload_start, payload_len = _decode_length(data, 0, list_=True)
+        payload_end = payload_start + payload_len
+        if payload_end != len(data):
+            raise ValueError("trailing leader public data")
+
+        outputs: list[bytes] = []
+        cursor = payload_start
+        while cursor < payload_end:
+            value_start, value_len = _decode_length(data, cursor, list_=False)
+            value_end = value_start + value_len
+            if value_end > payload_end:
+                raise ValueError("leader public data item exceeds list")
+            outputs.append(data[value_start:value_end])
+            cursor = value_end
+
+        if not outputs or outputs[-1] != _PADDING:
+            raise ValueError("leader public data padding is missing")
+        return cls(outputs[:-1])
+
+
+def encode(value: LeaderPublicData) -> bytes:
+    return value.encode()
+
+
+def decode(encoded: collections.abc.Buffer) -> LeaderPublicData:
+    return LeaderPublicData.decode(encoded)
+
+
+def _encode_bytes(value: bytes) -> bytes:
+    if len(value) == 1 and value[0] < 0x80:
+        return value
+    return _encode_length(len(value), 0x80, 0xB7) + value
+
+
+def _encode_length(length: int, short_base: int, long_base: int) -> bytes:
+    if length <= 55:
+        return bytes([short_base + length])
+    length_bytes = length.to_bytes((length.bit_length() + 7) // 8, "big")
+    return bytes([long_base + len(length_bytes)]) + length_bytes
+
+
+def _decode_length(
+    encoded: bytes,
+    offset: int,
+    *,
+    list_: bool,
+) -> tuple[int, int]:
+    if offset >= len(encoded):
+        raise ValueError("truncated leader public data")
+
+    prefix = encoded[offset]
+    short_base = 0xC0 if list_ else 0x80
+    long_base = 0xF7 if list_ else 0xB7
+    if not list_ and prefix < 0x80:
+        return offset, 1
+    if prefix < short_base or prefix > long_base + 8:
+        raise ValueError("invalid RLP prefix")
+    if prefix <= long_base:
+        if not list_ and prefix == 0x81 and offset + 1 < len(encoded):
+            if encoded[offset + 1] < 0x80:
+                raise ValueError("non-canonical RLP string")
+        return offset + 1, prefix - short_base
+
+    length_len = prefix - long_base
+    length_start = offset + 1
+    length_end = length_start + length_len
+    if length_end > len(encoded):
+        raise ValueError("truncated RLP length")
+    length_bytes = encoded[length_start:length_end]
+    if length_bytes[0] == 0:
+        raise ValueError("non-canonical RLP length")
+    length = int.from_bytes(length_bytes, "big")
+    if length <= 55:
+        raise ValueError("non-canonical long RLP value")
+    return length_end, length

tests/unit/test_node_state_proxy_metrics.py

@@ -13,6 +13,7 @@
 import backend.node.genvm.origin.calldata as gvm_calldata
 from backend.node.genvm.origin.base_host import RunHostAndProgramRes
 from backend.node.genvm.origin.host_fns import ResultCode
+from backend.node.genvm.origin.leader_public_data import LeaderPublicData
 from backend.node.types import Address, ExecutionMode, ExecutionResultStatus
 from backend.protocol_rpc.fees import (
     GENVM_UNMETERED_DATA_FEE_BUCKET,
@@ -186,7 +187,7 @@ def test_host_provide_result_preserves_fee_metadata_from_genvm_emissions():
                 "gasUsed": 123,
             },
         ],
-        result_nondet_results=[],
+        result_leader_public_data=LeaderPublicData([]).encode(),
         data_fees_remaining=[100, 90, 80],
     )
 

tests/unit/test_provide_result_readonly.py

@@ -3,6 +3,7 @@
 from backend.node.genvm.base import Context, Host
 from backend.node.genvm.origin.base_host import RunHostAndProgramRes
 from backend.node.genvm.origin.host_fns import ResultCode
+from backend.node.genvm.origin.leader_public_data import LeaderPublicData
 
 
 def _result_with_storage_change():
@@ -17,7 +18,7 @@ def _result_with_storage_change():
         result_fingerprint=None,
         result_storage_deltas=[(b"\x11" * 32 + (0).to_bytes(4, "big"), b"\xaa")],
         result_emissions=[],
-        result_nondet_results=[],
+        result_leader_public_data=LeaderPublicData([]).encode(),
         data_fees_remaining=[],
     )
 

tests/unit/test_usage_metrics_service.py

@@ -5,6 +5,7 @@
 
 from backend.node.genvm import base as genvm_base
 from backend.node.genvm.origin import host_fns
+from backend.node.genvm.origin.leader_public_data import LeaderPublicData
 from backend.services.usage_metrics_service import UsageMetricsService
 
 
@@ -169,7 +170,7 @@ def test_provide_result_preserves_llm_token_metrics():
         result_data={"ok": True},
         result_storage_deltas=[],
         result_emissions=[],
-        result_nondet_results=[],
+        result_leader_public_data=LeaderPublicData([]).encode(),
         stdout="",
         stderr="",
         genvm_log=[],

third_party/genvm/version

@@ -1 +1 @@
-fix/vm-fatal-errors:8a4816ebea2d2c1f210c6167778e94330fb3d46a
+feat/leader-data-to-genvm:201dc510c61c0fe86371cd40518f3e40fd3fb8c7