feat(consume): add engine-witness simulator for witness verification

Adds `consume engine-witness`, a Hive simulator that drives Amsterdam
blockchain-engine fixtures through a witness-emitting endpoint and diffs
the client-generated `ExecutionWitness` against the fixture's expected
witness (set-equality on state/codes/headers).

Supports two transports:

- Default: JSON-RPC `engine_newPayloadWithWitnessVX` with RLP witness
  (geth PR #30069, already implemented in go-ethereum and forks).
- `--ssz`: REST `POST /new-payload-with-witness` with SSZ witness
  (execution-apis PR #773, implemented on Ethrex `feat/zkengine-http`).

Both paths converge on a shared `NewPayloadWithWitnessResponse` dataclass
and the same assertion helper. Clients that do not implement the chosen
transport skip cleanly (HTTP 404/405 for REST, `-32601 Method not found`
for JSON-RPC).

Author: spencer-tb

packages/testing/src/execution_testing/cli/pytest_commands/consume.py

@@ -51,6 +51,13 @@ def get_command_logic_test_paths(command_name: str) -> List[Path]:
             / "simulator_logic"
             / f"test_via_{test_command}.py"
         ]
+    elif command_name == "engine_witness":
+        command_logic_test_paths = [
+            base_path
+            / "simulators"
+            / "simulator_logic"
+            / "test_via_engine_witness.py"
+        ]
     elif command_name == "sync":
         command_logic_test_paths = [
             base_path / "simulators" / "simulator_logic" / "test_via_sync.py"
@@ -79,9 +86,10 @@ def decorator(func: Callable[..., Any]) -> click.Command:
         command_name = func.__name__
         command_help = func.__doc__
         command_logic_test_paths = get_command_logic_test_paths(command_name)
+        cli_name = command_name.replace("_", "-")
 
         @consume.command(
-            name=command_name,
+            name=cli_name,
             help=command_help,
             context_settings={"ignore_unknown_options": True},
         )
@@ -120,6 +128,18 @@ def engine() -> None:
     pass
 
 
+@consume_command(is_hive=True)
+def engine_witness() -> None:
+    """
+    Verify client-emitted execution witnesses against the fixture.
+
+    Default transport: engine_newPayloadWithWitnessVX JSON-RPC (geth).
+    Pass --ssz to use the REST POST /new-payload-with-witness endpoint
+    (execution-apis PR #773).
+    """
+    pass
+
+
 @consume_command(is_hive=True)
 def enginex() -> None:
     """Consume via Engine API with pre-alloc optimization."""

packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/engine_witness/__init__.py

@@ -0,0 +1,102 @@
+"""
+Pytest fixtures for the `consume engine-witness` simulator.
+
+Drives the Hive back-end and EL clients through the REST
+`POST /new-payload-with-witness` endpoint (execution-apis PR #773),
+asserting the client-generated execution witness matches the fixture.
+"""
+
+import io
+from typing import Mapping
+
+import pytest
+from hive.client import Client
+
+from execution_testing.exceptions import ExceptionMapper
+from execution_testing.fixtures import BlockchainEngineFixture
+from execution_testing.fixtures.blockchain import FixtureHeader
+from execution_testing.rpc import EngineWitnessRPC
+
+pytest_plugins = (
+    "execution_testing.cli.pytest_commands.plugins.pytest_hive.pytest_hive",
+    "execution_testing.cli.pytest_commands.plugins.consume.simulators.base",
+    "execution_testing.cli.pytest_commands.plugins.consume.simulators.single_test_client",
+    "execution_testing.cli.pytest_commands.plugins.consume.simulators.test_case_description",
+    "execution_testing.cli.pytest_commands.plugins.consume.simulators.timing_data",
+    "execution_testing.cli.pytest_commands.plugins.consume.simulators.exceptions",
+    "execution_testing.cli.pytest_commands.plugins.consume.simulators.engine_api",
+)
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """Register the `--ssz` transport flag for the engine-witness simulator."""
+    parser.addoption(
+        "--ssz",
+        action="store_true",
+        default=False,
+        help=(
+            "Use the REST POST /new-payload-with-witness endpoint with "
+            "SSZ-encoded response (execution-apis PR #773) instead of the "
+            "default JSON-RPC engine_newPayloadWithWitnessVX with "
+            "RLP-encoded witness (geth-style)."
+        ),
+    )
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Set the supported fixture formats for the engine-witness simulator."""
+    config.supported_fixture_formats = [BlockchainEngineFixture]  # type: ignore[attr-defined]
+
+
+@pytest.fixture(scope="session")
+def use_ssz_transport(request: pytest.FixtureRequest) -> bool:
+    """Return True when `--ssz` was passed on the CLI."""
+    return bool(request.config.getoption("--ssz"))
+
+
+@pytest.fixture(scope="module")
+def test_suite_name() -> str:
+    """The name of the hive test suite used in this simulator."""
+    return "eels/consume-engine-witness"
+
+
+@pytest.fixture(scope="module")
+def test_suite_description() -> str:
+    """The description of the hive test suite used in this simulator."""
+    return (
+        "Execute blockchain-engine fixtures via the REST "
+        "POST /new-payload-with-witness endpoint (execution-apis PR #773), "
+        "verifying the client-generated execution witness against the "
+        "fixture witness."
+    )
+
+
+@pytest.fixture(scope="function")
+def client_files(
+    buffered_genesis: io.BufferedReader,
+) -> Mapping[str, io.BufferedReader]:
+    """Define the files that hive will start the client with."""
+    files = {}
+    files["/genesis.json"] = buffered_genesis
+    return files
+
+
+@pytest.fixture(scope="function")
+def genesis_header(fixture: BlockchainEngineFixture) -> "FixtureHeader":
+    """Provide the genesis header from the fixture."""
+    return fixture.genesis
+
+
+@pytest.fixture(scope="function")
+def engine_witness_rpc(
+    client: Client, client_exception_mapper: ExceptionMapper | None
+) -> EngineWitnessRPC:
+    """Provide a REST client for POST /new-payload-with-witness."""
+    if client_exception_mapper:
+        return EngineWitnessRPC(
+            f"http://{client.ip}:8551",
+            response_validation_context={
+                "exception_mapper": client_exception_mapper,
+            },
+        )
+    return EngineWitnessRPC(f"http://{client.ip}:8551")

packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/engine_witness/conftest.py

@@ -0,0 +1,70 @@
+"""Tests for set-based witness comparison helper."""
+
+import pytest
+
+from execution_testing.base_types import Bytes
+from execution_testing.cli.pytest_commands.plugins.consume.simulators.helpers.witness_diff import (  # noqa: E501
+    WitnessMismatchError,
+    assert_witness_matches,
+)
+from execution_testing.test_types.execution_witness import ExecutionWitness
+
+
+def _w(
+    state: list[bytes] | None = None,
+    codes: list[bytes] | None = None,
+    headers: list[bytes] | None = None,
+) -> ExecutionWitness:
+    return ExecutionWitness(
+        state=[Bytes(b) for b in state or []],
+        codes=[Bytes(b) for b in codes or []],
+        headers=[Bytes(b) for b in headers or []],
+    )
+
+
+def test_matching_witnesses_pass() -> None:
+    """Byte-equal witnesses match."""
+    w = _w(state=[b"\xaa", b"\xbb"], codes=[b"\x60"], headers=[b"\xf9"])
+    assert_witness_matches(expected=w, actual=w)
+
+
+def test_reordered_witness_matches() -> None:
+    """Set-equality ignores ordering — PR #773 does not mandate it."""
+    expected = _w(state=[b"\xaa", b"\xbb"], codes=[b"\x60", b"\x70"])
+    actual = _w(state=[b"\xbb", b"\xaa"], codes=[b"\x70", b"\x60"])
+    assert_witness_matches(expected=expected, actual=actual)
+
+
+def test_duplicates_reduced_to_set() -> None:
+    """Duplicate items on either side collapse to a single set element."""
+    expected = _w(state=[b"\xaa"])
+    actual = _w(state=[b"\xaa", b"\xaa"])
+    assert_witness_matches(expected=expected, actual=actual)
+
+
+def test_missing_state_node_fails() -> None:
+    """Client missing a state node gives a 'missing' diff line."""
+    expected = _w(state=[b"\xaa", b"\xbb"])
+    actual = _w(state=[b"\xaa"])
+    with pytest.raises(WitnessMismatchError, match="state: 1 missing"):
+        assert_witness_matches(expected=expected, actual=actual)
+
+
+def test_extra_code_fails() -> None:
+    """Client over-collecting a code gives an 'extra' diff line."""
+    expected = _w(codes=[b"\x60"])
+    actual = _w(codes=[b"\x60", b"\x70"])
+    with pytest.raises(WitnessMismatchError, match=r"codes: 1 extra"):
+        assert_witness_matches(expected=expected, actual=actual)
+
+
+def test_multi_field_mismatch_reports_all() -> None:
+    """All mismatching fields are reported in one exception."""
+    expected = _w(state=[b"\xaa"], codes=[b"\x60"], headers=[b"\xf9"])
+    actual = _w(state=[b"\xbb"], codes=[b"\x61"], headers=[])
+    with pytest.raises(WitnessMismatchError) as excinfo:
+        assert_witness_matches(expected=expected, actual=actual)
+    msg = str(excinfo.value)
+    assert "state:" in msg
+    assert "codes:" in msg
+    assert "headers:" in msg

packages/testing/src/execution_testing/cli/pytest_commands/plugins/consume/simulators/helpers/tests/__init__.py

@@ -0,0 +1,55 @@
+"""Set-based witness comparison helper for the engine-witness simulator."""
+
+from typing import Iterable, List
+
+from execution_testing.base_types import Bytes
+from execution_testing.test_types.execution_witness import ExecutionWitness
+
+
+def _diff_sets(
+    field: str, expected: Iterable[Bytes], actual: Iterable[Bytes]
+) -> List[str]:
+    """Return human-readable diff messages for a single witness field."""
+    exp = {bytes(x) for x in expected}
+    act = {bytes(x) for x in actual}
+    missing = exp - act
+    extra = act - exp
+    messages: List[str] = []
+    if missing:
+        preview = ", ".join(sorted("0x" + m.hex()[:16] for m in missing)[:5])
+        messages.append(
+            f"{field}: {len(missing)} missing (not emitted by client): {preview}"
+        )
+    if extra:
+        preview = ", ".join(sorted("0x" + e.hex()[:16] for e in extra)[:5])
+        messages.append(
+            f"{field}: {len(extra)} extra (over-collected by client): {preview}"
+        )
+    return messages
+
+
+class WitnessMismatchError(AssertionError):
+    """Raised when a client-emitted witness does not match the fixture's."""
+
+
+def assert_witness_matches(
+    expected: ExecutionWitness, actual: ExecutionWitness
+) -> None:
+    """
+    Assert the client-emitted `actual` witness matches the fixture `expected`
+    witness under set-equality on each of `state`, `codes`, `headers`.
+
+    Ordering is not mandated by execution-apis PR #773, so any permutation
+    the client produces is acceptable. Duplicate items on either side are
+    reduced to a single set element.
+    """
+    messages: List[str] = []
+    messages += _diff_sets("state", expected.state, actual.state)
+    messages += _diff_sets("codes", expected.codes, actual.codes)
+    messages += _diff_sets("headers", expected.headers, actual.headers)
+
+    if messages:
+        raise WitnessMismatchError(
+            "client witness does not match fixture witness:\n  "
+            + "\n  ".join(messages)
+        )