fix(rpc): align SSZ witness response decoding with current draft

packages/testing/src/execution_testing/rpc/rpc_types.py

@@ -13,7 +13,6 @@
 from remerkleable.byte_arrays import ByteList, ByteVector
 from remerkleable.complex import Container
 from remerkleable.complex import List as SszList
-from remerkleable.union import Union as SszUnion
 
 from execution_testing.base_types import (
     Address,
@@ -386,10 +385,9 @@ class EthConfigResponse(CamelModel):
     last: ForkConfig | None = None
 
 
-# SSZ schema for the REST POST /new-payload-with-witness response.
+# SSZ schema for the REST witness payload response.
 
-VALIDATION_ERROR_MAX = 8192
-MAX_WITNESS_BYTES = 2**30  # 1 GiB
+MAX_ERROR_BYTES = 1024
 MAX_WITNESS_ITEMS = 2**20
 MAX_WITNESS_ITEM_BYTES = 2**20
 
@@ -400,11 +398,15 @@ class _SszExecutionWitness(Container):
     headers: SszList[ByteList[MAX_WITNESS_ITEM_BYTES], MAX_WITNESS_ITEMS]
 
 
-class _SszNewPayloadWithWitnessResponse(Container):
+class _SszPayloadStatus(Container):
     status: uint8
-    latest_valid_hash: SszUnion[None, ByteVector[32]]
-    validation_error: SszUnion[None, ByteList[VALIDATION_ERROR_MAX]]
-    witness: ByteList[MAX_WITNESS_BYTES]
+    latest_valid_hash: SszList[ByteVector[32], 1]
+    validation_error: SszList[ByteList[MAX_ERROR_BYTES], 1]
+
+
+class _SszPayloadStatusWithWitness(Container):
+    payload_status: _SszPayloadStatus
+    witness: SszList[_SszExecutionWitness, 1]
 
 
 class _NewPayloadWithWitnessJSONRPCResult(CamelModel):
@@ -423,7 +425,6 @@ class _NewPayloadWithWitnessJSONRPCResult(CamelModel):
     1: PayloadStatusEnum.INVALID,
     2: PayloadStatusEnum.SYNCING,
     3: PayloadStatusEnum.ACCEPTED,
-    4: PayloadStatusEnum.INVALID_BLOCK_HASH,
 }
 
 
@@ -531,32 +532,34 @@ class NewPayloadWithWitnessResponse:
 
     @classmethod
     def from_ssz_bytes(cls, data: bytes) -> Self:
-        """Decode an SSZ-encoded NewPayloadWithWitnessResponseV1 body."""
-        resp = _SszNewPayloadWithWitnessResponse.decode_bytes(data)
+        """Decode an SSZ-encoded PayloadStatusWithWitness body."""
+        resp = _SszPayloadStatusWithWitness.decode_bytes(data)
+        payload_status = resp.payload_status
 
-        status_int = int(resp.status)
+        status_int = int(payload_status.status)
         try:
             status = _SSZ_STATUS_TO_ENUM[status_int]
         except KeyError as e:
             raise ValueError(f"Unknown SSZ status byte: {status_int}") from e
 
         latest_valid_hash: Hash | None = None
-        if resp.latest_valid_hash.selector() == 1:
-            latest_valid_hash = Hash(bytes(resp.latest_valid_hash.value()))
+        if payload_status.latest_valid_hash.length() > 0:
+            latest_valid_hash = Hash(
+                bytes(payload_status.latest_valid_hash[0])
+            )
 
         validation_error: str | None = None
-        if resp.validation_error.selector() == 1:
-            raw = bytes(resp.validation_error.value())
+        if payload_status.validation_error.length() > 0:
+            raw = bytes(payload_status.validation_error[0])
             validation_error = raw.decode("utf-8", errors="replace")
 
         witness: ExecutionWitness | None = None
-        witness_bytes = bytes(resp.witness)
-        if witness_bytes:
+        if resp.witness.length() > 0:
             if status != PayloadStatusEnum.VALID:
                 raise ValueError(
                     f"{status.value} SSZ response must not contain a witness"
                 )
-            inner = _SszExecutionWitness.decode_bytes(witness_bytes)
+            inner = resp.witness[0]
             witness = ExecutionWitness(
                 state=[Bytes(bytes(x)) for x in inner.state],
                 codes=[Bytes(bytes(x)) for x in inner.codes],

packages/testing/src/execution_testing/rpc/tests/test_new_payload_with_witness.py

@@ -7,64 +7,57 @@
 from remerkleable.byte_arrays import ByteList, ByteVector
 
 from execution_testing.rpc.rpc_types import (
-    MAX_WITNESS_BYTES,
+    MAX_ERROR_BYTES,
     MAX_WITNESS_ITEM_BYTES,
-    VALIDATION_ERROR_MAX,
     NewPayloadWithWitnessResponse,
     PayloadStatusEnum,
     _SszExecutionWitness,
-    _SszNewPayloadWithWitnessResponse,
+    _SszPayloadStatus,
+    _SszPayloadStatusWithWitness,
 )
 
 
 def _build_inner_witness(
     state: list[bytes], codes: list[bytes], headers: list[bytes]
-) -> bytes:
-    inner = _SszExecutionWitness(
+) -> _SszExecutionWitness:
+    return _SszExecutionWitness(
         state=[ByteList[MAX_WITNESS_ITEM_BYTES](b) for b in state],
         codes=[ByteList[MAX_WITNESS_ITEM_BYTES](b) for b in codes],
         headers=[ByteList[MAX_WITNESS_ITEM_BYTES](b) for b in headers],
     )
-    return inner.encode_bytes()
 
 
 def _build_response(
     status: int,
     latest_valid_hash: bytes | None,
     validation_error: str | None,
-    witness_bytes: bytes,
+    witness: _SszExecutionWitness | None,
 ) -> bytes:
-    fields = _SszNewPayloadWithWitnessResponse.fields()
-    lvh_type = fields["latest_valid_hash"]
-    ve_type = fields["validation_error"]
-
     if latest_valid_hash is None:
-        lvh = lvh_type(selector=0, value=None)
+        lvh = []
     else:
-        lvh = lvh_type(selector=1, value=ByteVector[32](latest_valid_hash))
+        lvh = [ByteVector[32](latest_valid_hash)]
 
     if validation_error is None:
-        ve = ve_type(selector=0, value=None)
+        ve = []
     else:
-        ve = ve_type(
-            selector=1,
-            value=ByteList[VALIDATION_ERROR_MAX](
-                validation_error.encode("utf-8")
-            ),
-        )
-
-    resp = _SszNewPayloadWithWitnessResponse(
-        status=uint8(status),
-        latest_valid_hash=lvh,
-        validation_error=ve,
-        witness=ByteList[MAX_WITNESS_BYTES](witness_bytes),
+        ve = [
+            ByteList[MAX_ERROR_BYTES](validation_error.encode("utf-8")),
+        ]
+
+    payload_status = _SszPayloadStatus(
+        status=uint8(status), latest_valid_hash=lvh, validation_error=ve
+    )
+    resp = _SszPayloadStatusWithWitness(
+        payload_status=payload_status,
+        witness=[] if witness is None else [witness],
     )
     return resp.encode_bytes()
 
 
 def test_decode_valid_with_witness() -> None:
     """A VALID response carries latestValidHash and a non-empty witness."""
-    witness_bytes = _build_inner_witness(
+    witness = _build_inner_witness(
         state=[b"\xaa\xaa", b"\xbb\xbb\xbb"],
         codes=[b"\x60\x01"],
         headers=[b"\xf9\x02"],
@@ -73,7 +66,7 @@ def test_decode_valid_with_witness() -> None:
         status=0,
         latest_valid_hash=b"\x11" * 32,
         validation_error=None,
-        witness_bytes=witness_bytes,
+        witness=witness,
     )
 
     decoded = NewPayloadWithWitnessResponse.from_ssz_bytes(raw)
@@ -97,7 +90,7 @@ def test_decode_invalid_with_validation_error() -> None:
         status=1,
         latest_valid_hash=None,
         validation_error="invalid state root",
-        witness_bytes=b"",
+        witness=None,
     )
 
     decoded = NewPayloadWithWitnessResponse.from_ssz_bytes(raw)
@@ -114,7 +107,7 @@ def test_decode_syncing_empty_witness() -> None:
         status=2,
         latest_valid_hash=None,
         validation_error=None,
-        witness_bytes=b"",
+        witness=None,
     )
 
     decoded = NewPayloadWithWitnessResponse.from_ssz_bytes(raw)
@@ -131,20 +124,33 @@ def test_decode_unknown_status_byte_raises() -> None:
         status=99,
         latest_valid_hash=None,
         validation_error=None,
-        witness_bytes=b"",
+        witness=None,
     )
 
     with pytest.raises(ValueError, match="Unknown SSZ status byte: 99"):
         NewPayloadWithWitnessResponse.from_ssz_bytes(raw)
 
 
+def test_decode_invalid_block_hash_status_byte_raises() -> None:
+    """SSZ PayloadStatus no longer includes INVALID_BLOCK_HASH."""
+    raw = _build_response(
+        status=4,
+        latest_valid_hash=None,
+        validation_error=None,
+        witness=None,
+    )
+
+    with pytest.raises(ValueError, match="Unknown SSZ status byte: 4"):
+        NewPayloadWithWitnessResponse.from_ssz_bytes(raw)
+
+
 def test_decode_invalid_with_witness_raises() -> None:
     """A non-VALID SSZ response must not carry witness bytes."""
     raw = _build_response(
         status=1,
         latest_valid_hash=None,
         validation_error="invalid state root",
-        witness_bytes=_build_inner_witness(
+        witness=_build_inner_witness(
             state=[b"\xaa"],
             codes=[],
             headers=[],