From 0019ecea980cba612078b474e733f5928be5f0a3 Mon Sep 17 00:00:00 2001
From: spencer-tb <spencer.tb@ethereum.org>
Date: Wed, 12 Aug 2026 11:39:47 +0200
Subject: [PATCH 1/5] feat(spec-specs): add consensus-layer execution engine
 interface to amsterdam

---
 .../amsterdam/execution_engine/__init__.py    |  52 +++
 .../execution_engine/forkchoice_update.py     |  26 ++
 .../amsterdam/execution_engine/get_payload.py |  16 +
 .../amsterdam/execution_engine/new_payload.py | 124 +++++++
 .../amsterdam/execution_engine/requests.py    | 314 ++++++++++++++++++
 .../forks/amsterdam/execution_engine/types.py | 143 ++++++++
 .../execution_engine/validation_helpers.py    | 122 +++++++
 vulture_whitelist.py                          |  19 ++
 8 files changed, 816 insertions(+)
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/__init__.py
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/forkchoice_update.py
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/get_payload.py
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/new_payload.py
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/requests.py
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/types.py
 create mode 100644 src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py

diff --git a/src/ethereum/forks/amsterdam/execution_engine/__init__.py b/src/ethereum/forks/amsterdam/execution_engine/__init__.py
new file mode 100644
index 00000000000..e70905eec8d
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/__init__.py
@@ -0,0 +1,52 @@
+"""
+Interface between the consensus layer and the execution layer.
+
+The consensus layer drives the execution layer through a small set of
+methods defined as the `ExecutionEngine` abstraction in the
+consensus-specs, carried between clients by the [Engine API]. Each new
+beacon block carries an [`ExecutionPayload`] that the execution layer
+validates and applies to its state with
+[`verify_and_notify_new_payload`].
+
+[Engine API]: https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md
+[`ExecutionPayload`]:
+    ref:ethereum.forks.amsterdam.execution_engine.types.ExecutionPayload
+[`verify_and_notify_new_payload`]:
+    ref:ethereum.forks.amsterdam.execution_engine.new_payload.verify_and_notify_new_payload
+"""  # noqa: E501
+
+from .forkchoice_update import notify_forkchoice_updated
+from .get_payload import get_payload
+from .new_payload import (
+    is_valid_block_hash,
+    is_valid_versioned_hashes,
+    notify_new_payload,
+    verify_and_notify_new_payload,
+)
+from .types import (
+    BlobsBundle,
+    ExecutionEngine,
+    ExecutionPayload,
+    ExecutionRequests,
+    GetPayloadResponse,
+    NewPayloadRequest,
+    PayloadAttributes,
+    PayloadId,
+)
+
+__all__ = [
+    "BlobsBundle",
+    "ExecutionEngine",
+    "ExecutionPayload",
+    "ExecutionRequests",
+    "GetPayloadResponse",
+    "NewPayloadRequest",
+    "PayloadAttributes",
+    "PayloadId",
+    "get_payload",
+    "is_valid_block_hash",
+    "is_valid_versioned_hashes",
+    "notify_forkchoice_updated",
+    "notify_new_payload",
+    "verify_and_notify_new_payload",
+]
diff --git a/src/ethereum/forks/amsterdam/execution_engine/forkchoice_update.py b/src/ethereum/forks/amsterdam/execution_engine/forkchoice_update.py
new file mode 100644
index 00000000000..2f96bf4e9a3
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/forkchoice_update.py
@@ -0,0 +1,26 @@
+"""
+Forkchoice update and payload build signal.
+"""
+
+from typing import Optional
+
+from ethereum.crypto.hash import Hash32
+
+from .types import (
+    ExecutionEngine,
+    PayloadAttributes,
+    PayloadId,
+)
+
+
+def notify_forkchoice_updated(
+    _chain: ExecutionEngine,
+    _head_block_hash: Hash32,
+    _safe_block_hash: Hash32,
+    _finalized_block_hash: Hash32,
+    _payload_attributes: Optional[PayloadAttributes],
+) -> Optional[PayloadId]:
+    """
+    Notify the execution engine about the latest fork-choice state.
+    """
+    raise NotImplementedError
diff --git a/src/ethereum/forks/amsterdam/execution_engine/get_payload.py b/src/ethereum/forks/amsterdam/execution_engine/get_payload.py
new file mode 100644
index 00000000000..832d5643d82
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/get_payload.py
@@ -0,0 +1,16 @@
+"""
+Payload build storage and retrieval helpers.
+"""
+
+from .types import (
+    GetPayloadResponse,
+    PayloadId,
+)
+
+
+def get_payload(_payload_id: PayloadId) -> GetPayloadResponse:
+    """
+    Return a prepared payload response for a previously returned
+    ``PayloadId``.
+    """
+    raise NotImplementedError
diff --git a/src/ethereum/forks/amsterdam/execution_engine/new_payload.py b/src/ethereum/forks/amsterdam/execution_engine/new_payload.py
new file mode 100644
index 00000000000..1bb8c73188b
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/new_payload.py
@@ -0,0 +1,124 @@
+"""
+Payload verification and execution.
+"""
+
+from ethereum_rlp import rlp
+
+from ethereum.crypto.hash import keccak256
+from ethereum.exceptions import EthereumException
+from ethereum.state import Root
+
+from ..fork import state_transition
+from ..fork_types import VersionedHash
+from ..transactions import BlobTransaction, decode_transaction
+from .requests import ExecutionRequests
+from .types import ExecutionEngine, ExecutionPayload, NewPayloadRequest
+from .validation_helpers import _payload_block, _payload_header
+
+
+def is_valid_block_hash(
+    execution_payload: ExecutionPayload,
+    parent_beacon_block_root: Root,
+    execution_requests: ExecutionRequests,
+) -> bool:
+    """
+    Return `True` if and only if `execution_payload.block_hash` is
+    computed correctly.
+    """
+    try:
+        header = _payload_header(
+            execution_payload,
+            parent_beacon_block_root,
+            execution_requests,
+        )
+    except Exception:
+        # Any decoding or conversion failure means the payload
+        # cannot produce a valid header.
+        return False
+    return keccak256(rlp.encode(header)) == execution_payload.block_hash
+
+
+def is_valid_versioned_hashes(
+    new_payload_request: NewPayloadRequest,
+) -> bool:
+    """
+    Return `True` if and only if the versioned hashes computed by blob
+    transactions in `new_payload_request.execution_payload` match
+    `new_payload_request.versioned_hashes`.
+    """
+    computed_versioned_hashes: list[VersionedHash] = []
+
+    try:
+        for encoded_tx in new_payload_request.execution_payload.transactions:
+            tx = decode_transaction(encoded_tx)
+            if isinstance(tx, BlobTransaction):
+                computed_versioned_hashes.extend(tx.blob_versioned_hashes)
+    except Exception:
+        # Any decoding failure means versioned hashes cannot be
+        # verified.
+        return False
+
+    return tuple(computed_versioned_hashes) == (
+        new_payload_request.versioned_hashes
+    )
+
+
+def notify_new_payload(
+    chain: ExecutionEngine,
+    new_payload_request: NewPayloadRequest,
+) -> bool:
+    """
+    Execute the payload against the chain head and return `True` if and
+    only if it forms a valid block.
+
+    The payload is converted into a [`Block`] and applied with
+    [`state_transition`], which appends it to the chain on success.
+
+    [`Block`]: ref:ethereum.forks.amsterdam.blocks.Block
+    [`state_transition`]: ref:ethereum.forks.amsterdam.fork.state_transition
+    """
+    block = _payload_block(
+        new_payload_request.execution_payload,
+        new_payload_request.parent_beacon_block_root,
+        new_payload_request.execution_requests,
+    )
+
+    try:
+        state_transition(chain, block)
+    except EthereumException:
+        return False
+
+    return True
+
+
+def verify_and_notify_new_payload(
+    chain: ExecutionEngine,
+    new_payload_request: NewPayloadRequest,
+) -> bool:
+    """
+    Validate the payload and, if valid, apply it to the chain.
+
+    Mirrors the consensus-layer `verify_and_notify_new_payload` method
+    of the `ExecutionEngine`: the payload must carry a correctly
+    computed `block_hash` and matching blob versioned hashes before it
+    is executed by [`notify_new_payload`].
+
+    [`notify_new_payload`]:
+        ref:ethereum.forks.amsterdam.execution_engine.new_payload.notify_new_payload
+    """  # noqa: E501
+    payload = new_payload_request.execution_payload
+
+    if b"" in payload.transactions:
+        return False
+
+    if not is_valid_block_hash(
+        payload,
+        new_payload_request.parent_beacon_block_root,
+        new_payload_request.execution_requests,
+    ):
+        return False
+
+    if not is_valid_versioned_hashes(new_payload_request):
+        return False
+
+    return notify_new_payload(chain, new_payload_request)
diff --git a/src/ethereum/forks/amsterdam/execution_engine/requests.py b/src/ethereum/forks/amsterdam/execution_engine/requests.py
new file mode 100644
index 00000000000..77c2f749014
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/requests.py
@@ -0,0 +1,314 @@
+"""
+Typed execution-layer requests and engine-API wire-form codecs.
+
+The consensus layer defines ``ExecutionRequests`` as a typed Container
+holding deposit, withdrawal, consolidation, builder deposit, and builder exit
+lists.
+"""
+
+from dataclasses import dataclass
+from typing import Sequence, Tuple, final
+
+from ethereum_types.bytes import Bytes, Bytes32, Bytes48, Bytes96
+from ethereum_types.frozen import slotted_freezable
+from ethereum_types.numeric import U64
+
+from ethereum.exceptions import InvalidBlock
+from ethereum.state import Address
+
+from ..requests import (
+    BUILDER_DEPOSIT_REQUEST_TYPE,
+    BUILDER_EXIT_REQUEST_TYPE,
+    CONSOLIDATION_REQUEST_TYPE,
+    DEPOSIT_REQUEST_TYPE,
+    WITHDRAWAL_REQUEST_TYPE,
+)
+
+DEPOSIT_REQUEST_SIZE = 48 + 32 + 8 + 96 + 8
+WITHDRAWAL_REQUEST_SIZE = 20 + 48 + 8
+CONSOLIDATION_REQUEST_SIZE = 20 + 48 + 48
+BUILDER_DEPOSIT_REQUEST_SIZE = 184
+BUILDER_EXIT_REQUEST_SIZE = 68
+
+
+@final
+@slotted_freezable
+@dataclass
+class DepositRequest:
+    """A single EIP-6110 deposit request."""
+
+    pubkey: Bytes48
+    withdrawal_credentials: Bytes32
+    amount: U64
+    signature: Bytes96
+    index: U64
+
+
+@final
+@slotted_freezable
+@dataclass
+class WithdrawalRequest:
+    """A single EIP-7002 withdrawal request."""
+
+    source_address: Address
+    validator_pubkey: Bytes48
+    amount: U64
+
+
+@final
+@slotted_freezable
+@dataclass
+class ConsolidationRequest:
+    """A single EIP-7251 consolidation request."""
+
+    source_address: Address
+    source_pubkey: Bytes48
+    target_pubkey: Bytes48
+
+
+@final
+@slotted_freezable
+@dataclass
+class BuilderDepositRequest:
+    """A single EIP-8282 builder deposit request."""
+
+    pubkey: Bytes48
+    withdrawal_credentials: Bytes32
+    amount: U64
+    signature: Bytes96
+
+
+@final
+@slotted_freezable
+@dataclass
+class BuilderExitRequest:
+    """A single EIP-8282 builder exit request."""
+
+    source_address: Address
+    pubkey: Bytes48
+
+
+@final
+@slotted_freezable
+@dataclass
+class ExecutionRequests:
+    """
+    Typed engine-API container of execution-layer triggered requests.
+
+    Mirrors the consensus-layer ``ExecutionRequests`` Container.
+    """
+
+    deposits: Tuple[DepositRequest, ...]
+    withdrawals: Tuple[WithdrawalRequest, ...]
+    consolidations: Tuple[ConsolidationRequest, ...]
+    builder_deposits: Tuple[BuilderDepositRequest, ...]
+    builder_exits: Tuple[BuilderExitRequest, ...]
+
+
+def _encode_deposit(d: DepositRequest) -> Bytes:
+    return Bytes(
+        bytes(d.pubkey)
+        + bytes(d.withdrawal_credentials)
+        + bytes(d.amount.to_le_bytes8())
+        + bytes(d.signature)
+        + bytes(d.index.to_le_bytes8())
+    )
+
+
+def _encode_withdrawal(w: WithdrawalRequest) -> Bytes:
+    return Bytes(
+        bytes(w.source_address)
+        + bytes(w.validator_pubkey)
+        + bytes(w.amount.to_le_bytes8())
+    )
+
+
+def _encode_consolidation(c: ConsolidationRequest) -> Bytes:
+    return Bytes(
+        bytes(c.source_address)
+        + bytes(c.source_pubkey)
+        + bytes(c.target_pubkey)
+    )
+
+
+def _encode_builder_deposit(b: BuilderDepositRequest) -> Bytes:
+    return Bytes(
+        bytes(b.pubkey)
+        + bytes(b.withdrawal_credentials)
+        + bytes(b.amount.to_le_bytes8())
+        + bytes(b.signature)
+    )
+
+
+def _encode_builder_exit(b: BuilderExitRequest) -> Bytes:
+    return Bytes(bytes(b.source_address) + bytes(b.pubkey))
+
+
+def _decode_deposit(payload: Bytes) -> DepositRequest:
+    return DepositRequest(
+        pubkey=Bytes48(payload[0:48]),
+        withdrawal_credentials=Bytes32(payload[48:80]),
+        amount=U64.from_le_bytes(payload[80:88]),
+        signature=Bytes96(payload[88:184]),
+        index=U64.from_le_bytes(payload[184:192]),
+    )
+
+
+def _decode_withdrawal(payload: Bytes) -> WithdrawalRequest:
+    return WithdrawalRequest(
+        source_address=Address(payload[0:20]),
+        validator_pubkey=Bytes48(payload[20:68]),
+        amount=U64.from_le_bytes(payload[68:76]),
+    )
+
+
+def _decode_consolidation(payload: Bytes) -> ConsolidationRequest:
+    return ConsolidationRequest(
+        source_address=Address(payload[0:20]),
+        source_pubkey=Bytes48(payload[20:68]),
+        target_pubkey=Bytes48(payload[68:116]),
+    )
+
+
+def _decode_builder_deposit(payload: Bytes) -> BuilderDepositRequest:
+    return BuilderDepositRequest(
+        pubkey=Bytes48(payload[0:48]),
+        withdrawal_credentials=Bytes32(payload[48:80]),
+        amount=U64.from_le_bytes(payload[80:88]),
+        signature=Bytes96(payload[88:184]),
+    )
+
+
+def _decode_builder_exit(payload: Bytes) -> BuilderExitRequest:
+    return BuilderExitRequest(
+        source_address=Address(payload[0:20]),
+        pubkey=Bytes48(payload[20:68]),
+    )
+
+
+def encode_execution_requests(
+    requests: ExecutionRequests,
+) -> Tuple[Bytes, ...]:
+    """
+    Flatten a typed ``ExecutionRequests`` into the engine-API wire form.
+
+    Each non-empty list is emitted as a single blob
+    ``TYPE_BYTE || concat(serialize(item) for item)``, in ascending
+    type order. Empty lists are omitted. Mirrors CL's
+    ``get_execution_requests_list()``.
+    """
+    output: list[Bytes] = []
+    if requests.deposits:
+        body = b"".join(_encode_deposit(d) for d in requests.deposits)
+        output.append(Bytes(DEPOSIT_REQUEST_TYPE + body))
+    if requests.withdrawals:
+        body = b"".join(_encode_withdrawal(w) for w in requests.withdrawals)
+        output.append(Bytes(WITHDRAWAL_REQUEST_TYPE + body))
+    if requests.consolidations:
+        body = b"".join(
+            _encode_consolidation(c) for c in requests.consolidations
+        )
+        output.append(Bytes(CONSOLIDATION_REQUEST_TYPE + body))
+    if requests.builder_deposits:
+        body = b"".join(
+            _encode_builder_deposit(b) for b in requests.builder_deposits
+        )
+        output.append(Bytes(BUILDER_DEPOSIT_REQUEST_TYPE + body))
+    if requests.builder_exits:
+        body = b"".join(
+            _encode_builder_exit(b) for b in requests.builder_exits
+        )
+        output.append(Bytes(BUILDER_EXIT_REQUEST_TYPE + body))
+    return tuple(output)
+
+
+def decode_execution_requests(
+    wire: Sequence[Bytes],
+) -> ExecutionRequests:
+    """
+    Parse the engine-API wire form into a typed ``ExecutionRequests``.
+
+    Validates strict ascending type order, no duplicate type bytes, no
+    unknown type bytes, and that each payload's length is a multiple of
+    the per-type item size.
+    """
+    deposits: Tuple[DepositRequest, ...] = ()
+    withdrawals: Tuple[WithdrawalRequest, ...] = ()
+    consolidations: Tuple[ConsolidationRequest, ...] = ()
+    builder_deposits: Tuple[BuilderDepositRequest, ...] = ()
+    builder_exits: Tuple[BuilderExitRequest, ...] = ()
+
+    last_type = -1
+    for blob in wire:
+        if len(blob) < 1:
+            raise InvalidBlock("Empty execution request blob")
+        type_byte = bytes(blob[0:1])
+        body = bytes(blob[1:])
+        type_int = type_byte[0]
+        if type_int <= last_type:
+            raise InvalidBlock(
+                "Execution requests must be in strict ascending type order"
+            )
+        last_type = type_int
+
+        if type_byte == DEPOSIT_REQUEST_TYPE:
+            if len(body) % DEPOSIT_REQUEST_SIZE != 0:
+                raise InvalidBlock("Invalid deposit request payload length")
+            deposits = tuple(
+                _decode_deposit(Bytes(body[i : i + DEPOSIT_REQUEST_SIZE]))
+                for i in range(0, len(body), DEPOSIT_REQUEST_SIZE)
+            )
+        elif type_byte == WITHDRAWAL_REQUEST_TYPE:
+            if len(body) % WITHDRAWAL_REQUEST_SIZE != 0:
+                raise InvalidBlock("Invalid withdrawal request payload length")
+            withdrawals = tuple(
+                _decode_withdrawal(
+                    Bytes(body[i : i + WITHDRAWAL_REQUEST_SIZE])
+                )
+                for i in range(0, len(body), WITHDRAWAL_REQUEST_SIZE)
+            )
+        elif type_byte == CONSOLIDATION_REQUEST_TYPE:
+            if len(body) % CONSOLIDATION_REQUEST_SIZE != 0:
+                raise InvalidBlock(
+                    "Invalid consolidation request payload length"
+                )
+            consolidations = tuple(
+                _decode_consolidation(
+                    Bytes(body[i : i + CONSOLIDATION_REQUEST_SIZE])
+                )
+                for i in range(0, len(body), CONSOLIDATION_REQUEST_SIZE)
+            )
+        elif type_byte == BUILDER_DEPOSIT_REQUEST_TYPE:
+            if len(body) % BUILDER_DEPOSIT_REQUEST_SIZE != 0:
+                raise InvalidBlock(
+                    "Invalid builder deposit request payload length"
+                )
+            builder_deposits = tuple(
+                _decode_builder_deposit(
+                    Bytes(body[i : i + BUILDER_DEPOSIT_REQUEST_SIZE])
+                )
+                for i in range(0, len(body), BUILDER_DEPOSIT_REQUEST_SIZE)
+            )
+        elif type_byte == BUILDER_EXIT_REQUEST_TYPE:
+            if len(body) % BUILDER_EXIT_REQUEST_SIZE != 0:
+                raise InvalidBlock(
+                    "Invalid builder exit request payload length"
+                )
+            builder_exits = tuple(
+                _decode_builder_exit(
+                    Bytes(body[i : i + BUILDER_EXIT_REQUEST_SIZE])
+                )
+                for i in range(0, len(body), BUILDER_EXIT_REQUEST_SIZE)
+            )
+        else:
+            raise InvalidBlock(
+                f"Unknown execution request type byte {type_byte!r}"
+            )
+
+    return ExecutionRequests(
+        deposits=deposits,
+        withdrawals=withdrawals,
+        consolidations=consolidations,
+        builder_deposits=builder_deposits,
+        builder_exits=builder_exits,
+    )
diff --git a/src/ethereum/forks/amsterdam/execution_engine/types.py b/src/ethereum/forks/amsterdam/execution_engine/types.py
new file mode 100644
index 00000000000..d39990bbd43
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/types.py
@@ -0,0 +1,143 @@
+"""
+Execution engine data structures and aliases.
+"""
+
+from dataclasses import dataclass
+from typing import Tuple, final
+
+from ethereum_types.bytes import Bytes, Bytes8, Bytes32
+from ethereum_types.frozen import slotted_freezable
+from ethereum_types.numeric import U64, U256, Uint
+
+from ethereum.crypto.hash import Hash32
+from ethereum.state import Address, Root
+
+from ..blocks import Withdrawal
+from ..fork import BlockChain
+from ..fork_types import Bloom, VersionedHash
+from .requests import ExecutionRequests
+
+ExecutionEngine = BlockChain
+"""
+Chain and state container that the execution engine methods operate on.
+"""
+
+PayloadId = Bytes8
+"""
+Identifier of a payload build process, returned by
+[`notify_forkchoice_updated`] and consumed by [`get_payload`].
+
+[`notify_forkchoice_updated`]:
+    ref:ethereum.forks.amsterdam.execution_engine.forkchoice_update.notify_forkchoice_updated
+[`get_payload`]:
+    ref:ethereum.forks.amsterdam.execution_engine.get_payload.get_payload
+"""  # noqa: E501
+
+
+@final
+@slotted_freezable
+@dataclass
+class ExecutionPayload:
+    """
+    Represent a new block to be processed by the execution layer.
+
+    The consensus layer constructs this from a beacon block body and
+    passes it to the execution engine for validation. Mirrors the
+    [`ExecutionPayloadV4`] structure of the Engine API.
+
+    The execution requests are not a direct field in the payload but are
+    indirectly committed to via `block_hash`, since `requests_hash` is
+    part of the execution-layer block header.
+
+    [`ExecutionPayloadV4`]: https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md
+    """  # noqa: E501
+
+    parent_hash: Hash32
+    fee_recipient: Address
+    state_root: Root
+    receipts_root: Root
+    logs_bloom: Bloom
+    prev_randao: Bytes32
+    block_number: Uint
+    gas_limit: Uint
+    gas_used: Uint
+    timestamp: U256
+    extra_data: Bytes
+    base_fee_per_gas: Uint
+    block_hash: Hash32
+    transactions: Tuple[Bytes, ...]
+    withdrawals: Tuple[Withdrawal, ...]
+    blob_gas_used: U64
+    excess_blob_gas: U64
+    block_access_list: Bytes
+    slot_number: U64
+
+
+@final
+@slotted_freezable
+@dataclass
+class NewPayloadRequest:
+    """
+    Contain an execution payload along with versioned hashes, the parent
+    beacon block root, and execution requests for the
+    [`verify_and_notify_new_payload`] entry point.
+
+    Corresponds to the consensus-layer [`NewPayloadRequest`] container
+    and carries the parameters of the Engine API `engine_newPayloadV5`
+    method.
+
+    [`verify_and_notify_new_payload`]:
+        ref:ethereum.forks.amsterdam.execution_engine.new_payload.verify_and_notify_new_payload
+    [`NewPayloadRequest`]: https://ethereum.github.io/consensus-specs/specs/electra/beacon-chain/#modified-newpayloadrequest
+    """  # noqa: E501
+
+    execution_payload: ExecutionPayload
+    versioned_hashes: Tuple[VersionedHash, ...]
+    parent_beacon_block_root: Root
+    execution_requests: ExecutionRequests
+
+
+@final
+@slotted_freezable
+@dataclass
+class PayloadAttributes:
+    """
+    Carry the parameters that the consensus layer supplies when it
+    requests the execution layer to build a new block.
+    """
+
+    timestamp: U256
+    prev_randao: Bytes32
+    suggested_fee_recipient: Address
+    withdrawals: Tuple[Withdrawal, ...]
+    parent_beacon_block_root: Root
+
+
+@final
+@slotted_freezable
+@dataclass
+class BlobsBundle:
+    """
+    Bundle of blobs data associated with a built payload.
+    """
+
+    commitments: Tuple[Bytes, ...]
+    proofs: Tuple[Bytes, ...]
+    blobs: Tuple[Bytes, ...]
+
+
+@final
+@slotted_freezable
+@dataclass
+class GetPayloadResponse:
+    """
+    Response returned by [`get_payload`] for a prepared payload build.
+
+    [`get_payload`]:
+        ref:ethereum.forks.amsterdam.execution_engine.get_payload.get_payload
+    """
+
+    execution_payload: ExecutionPayload
+    block_value: U256
+    blobs_bundle: BlobsBundle
+    execution_requests: ExecutionRequests
diff --git a/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py b/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py
new file mode 100644
index 00000000000..f76af3d1ffa
--- /dev/null
+++ b/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py
@@ -0,0 +1,122 @@
+"""
+Shared execution-engine conversion helpers.
+"""
+
+from typing import Optional
+
+from ethereum_rlp import rlp
+from ethereum_types.bytes import Bytes, Bytes8
+from ethereum_types.numeric import Uint
+
+from ethereum.crypto.hash import Hash32, keccak256
+from ethereum.merkle_patricia_trie import Trie, root, trie_set
+from ethereum.state import Root
+
+from ..blocks import Block, Header
+from ..fork import EMPTY_OMMER_HASH
+from ..requests import compute_requests_hash
+from ..transactions import LegacyTransaction, decode_transaction
+from .requests import ExecutionRequests, encode_execution_requests
+from .types import ExecutionPayload
+
+
+def _payload_header(
+    execution_payload: ExecutionPayload,
+    parent_beacon_block_root: Root,
+    execution_requests: ExecutionRequests,
+) -> Header:
+    """
+    Build the execution header implied by a payload request.
+    """
+    transactions_trie: Trie[Bytes, Optional[Bytes]] = Trie(
+        secured=False, default=None
+    )
+    for i, encoded_tx in enumerate(execution_payload.transactions):
+        trie_set(
+            transactions_trie,
+            rlp.encode(Uint(i)),
+            encoded_tx,
+        )
+    transactions_root = root(transactions_trie)
+
+    withdrawals_trie: Trie[Bytes, Optional[Bytes]] = Trie(
+        secured=False, default=None
+    )
+    for i, withdrawal in enumerate(execution_payload.withdrawals):
+        trie_set(
+            withdrawals_trie,
+            rlp.encode(Uint(i)),
+            rlp.encode(withdrawal),
+        )
+    withdrawals_root = root(withdrawals_trie)
+
+    requests_hash = Hash32(
+        compute_requests_hash(
+            list(encode_execution_requests(execution_requests))
+        )
+    )
+
+    return Header(
+        parent_hash=execution_payload.parent_hash,
+        ommers_hash=EMPTY_OMMER_HASH,
+        coinbase=execution_payload.fee_recipient,
+        state_root=execution_payload.state_root,
+        transactions_root=transactions_root,
+        receipt_root=execution_payload.receipts_root,
+        bloom=execution_payload.logs_bloom,
+        difficulty=Uint(0),
+        number=execution_payload.block_number,
+        gas_limit=execution_payload.gas_limit,
+        gas_used=execution_payload.gas_used,
+        timestamp=execution_payload.timestamp,
+        extra_data=execution_payload.extra_data,
+        prev_randao=execution_payload.prev_randao,
+        nonce=Bytes8(b"\x00\x00\x00\x00\x00\x00\x00\x00"),
+        base_fee_per_gas=execution_payload.base_fee_per_gas,
+        withdrawals_root=withdrawals_root,
+        blob_gas_used=execution_payload.blob_gas_used,
+        excess_blob_gas=execution_payload.excess_blob_gas,
+        parent_beacon_block_root=parent_beacon_block_root,
+        requests_hash=requests_hash,
+        block_access_list_hash=Hash32(
+            keccak256(execution_payload.block_access_list)
+        ),
+        slot_number=execution_payload.slot_number,
+    )
+
+
+def _payload_transaction_to_block_transaction(
+    encoded_transaction: Bytes,
+) -> LegacyTransaction | Bytes:
+    """Return the canonical block representation of a payload transaction."""
+    if not encoded_transaction or encoded_transaction[0] < 0xC0:
+        return encoded_transaction
+
+    transaction = decode_transaction(encoded_transaction)
+    assert isinstance(transaction, LegacyTransaction)
+    return transaction
+
+
+def _payload_block(
+    execution_payload: ExecutionPayload,
+    parent_beacon_block_root: Root,
+    execution_requests: ExecutionRequests,
+) -> Block:
+    """
+    Convert an execution payload request into an execution-layer block.
+    """
+    header = _payload_header(
+        execution_payload,
+        parent_beacon_block_root,
+        execution_requests,
+    )
+
+    return Block(
+        header=header,
+        transactions=tuple(
+            _payload_transaction_to_block_transaction(encoded_transaction)
+            for encoded_transaction in execution_payload.transactions
+        ),
+        ommers=(),
+        withdrawals=execution_payload.withdrawals,
+    )
diff --git a/vulture_whitelist.py b/vulture_whitelist.py
index b68fbccedab..64649b51a0f 100644
--- a/vulture_whitelist.py
+++ b/vulture_whitelist.py
@@ -203,3 +203,22 @@
 _configure_client_manager  # autouse fixture
 test_suite_name  # hive test suite name fixture
 genesis_header  # genesis header fixture
+
+# src/ethereum/forks/amsterdam/execution_engine - consensus-layer interface
+# surface; consumed by external engine callers rather than the spec itself
+from ethereum.forks.amsterdam.execution_engine.requests import (
+    decode_execution_requests,
+)
+from ethereum.forks.amsterdam.execution_engine.types import (
+    BlobsBundle,
+    GetPayloadResponse,
+    PayloadAttributes,
+)
+
+decode_execution_requests
+PayloadAttributes.suggested_fee_recipient
+BlobsBundle.commitments
+BlobsBundle.proofs
+BlobsBundle.blobs
+GetPayloadResponse.block_value
+GetPayloadResponse.blobs_bundle

From 044a44922ddee6369b1d8e83a1dbb528de2f2077 Mon Sep 17 00:00:00 2001
From: spencer-tb <spencer.tb@ethereum.org>
Date: Wed, 12 Aug 2026 15:02:20 +0200
Subject: [PATCH 2/5] feat(tooling): add ethereum-spec-engine Engine API server

---
 pyproject.toml                                |   1 +
 .../engine_server/__init__.py                 |  98 +++
 .../engine_server/genesis.py                  | 148 ++++
 .../engine_server/server.py                   | 649 ++++++++++++++++++
 vulture_whitelist.py                          |   5 +
 5 files changed, 901 insertions(+)
 create mode 100644 src/ethereum_spec_tools/engine_server/__init__.py
 create mode 100644 src/ethereum_spec_tools/engine_server/genesis.py
 create mode 100644 src/ethereum_spec_tools/engine_server/server.py

diff --git a/pyproject.toml b/pyproject.toml
index 23a1eff6527..35e74b5628f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -273,6 +273,7 @@ ethereum-spec-sync = "ethereum_spec_tools.sync:main"
 ethereum-spec-new-fork = "ethereum_spec_tools.new_fork.cli:main"
 ethereum-spec-patch = "ethereum_spec_tools.patch_tool:main"
 ethereum-spec-evm = "ethereum_spec_tools.evm_tools:main"
+ethereum-spec-engine = "ethereum_spec_tools.engine_server:main"
 whitelist = "ethereum_spec_tools.whitelist:main"
 
 [project.entry-points."docc.plugins"]
diff --git a/src/ethereum_spec_tools/engine_server/__init__.py b/src/ethereum_spec_tools/engine_server/__init__.py
new file mode 100644
index 00000000000..00b4e335400
--- /dev/null
+++ b/src/ethereum_spec_tools/engine_server/__init__.py
@@ -0,0 +1,98 @@
+"""
+Run the execution specification as an Engine API client.
+
+Loads a hive genesis file into an Amsterdam [`BlockChain`] and serves
+the Engine API and `eth` JSON-RPC methods needed by a consensus-layer
+driver, allowing tools like the hive `consume engine` simulator to feed
+blocks directly to the specification.
+
+Only the Amsterdam fork is supported.
+
+[`BlockChain`]: ref:ethereum.forks.amsterdam.fork.BlockChain
+"""
+
+import argparse
+import os
+import threading
+from pathlib import Path
+
+from ethereum_types.numeric import U64
+
+from .genesis import load_genesis_chain
+from .server import DEFAULT_JWT_SECRET, EngineBackend, serve
+
+
+def parse_arguments() -> argparse.Namespace:
+    """Parse the engine server's command line arguments."""
+    parser = argparse.ArgumentParser(
+        prog="ethereum-spec-engine",
+        description="Run the execution specs as an Engine API client.",
+    )
+    parser.add_argument(
+        "--genesis",
+        type=Path,
+        required=True,
+        help="Path to the hive genesis JSON file.",
+    )
+    parser.add_argument(
+        "--fork",
+        default="Amsterdam",
+        help="Fork to run (only Amsterdam is supported).",
+    )
+    parser.add_argument(
+        "--chain-id",
+        type=lambda value: int(value, 0),
+        default=int(os.environ.get("HIVE_CHAIN_ID", "1")),
+        help="Chain id (defaults to $HIVE_CHAIN_ID or 1).",
+    )
+    parser.add_argument(
+        "--address",
+        default="0.0.0.0",
+        help="Address to bind the HTTP listeners to.",
+    )
+    parser.add_argument(
+        "--rpc-port",
+        type=int,
+        default=8545,
+        help="Port for the unauthenticated eth namespace.",
+    )
+    parser.add_argument(
+        "--engine-port",
+        type=int,
+        default=8551,
+        help="Port for the JWT-authenticated engine namespace.",
+    )
+    parser.add_argument(
+        "--jwt-secret",
+        type=lambda value: bytes.fromhex(value.removeprefix("0x")),
+        default=DEFAULT_JWT_SECRET,
+        help="JWT secret as a hex string (defaults to the hive secret).",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    """Start the engine server and serve until interrupted."""
+    options = parse_arguments()
+
+    if options.fork != "Amsterdam":
+        raise SystemExit(
+            f"unsupported fork: {options.fork} (only Amsterdam is supported)"
+        )
+
+    chain = load_genesis_chain(options.genesis, U64(options.chain_id))
+    backend = EngineBackend(chain)
+    serve(
+        backend,
+        options.address,
+        options.rpc_port,
+        options.engine_port,
+        options.jwt_secret,
+    )
+    print(
+        f"engine server listening on {options.address}:"
+        f"{options.rpc_port} (eth) and {options.address}:"
+        f"{options.engine_port} (engine)",
+        flush=True,
+    )
+    threading.Event().wait()
diff --git a/src/ethereum_spec_tools/engine_server/genesis.py b/src/ethereum_spec_tools/engine_server/genesis.py
new file mode 100644
index 00000000000..a40fb83a14c
--- /dev/null
+++ b/src/ethereum_spec_tools/engine_server/genesis.py
@@ -0,0 +1,148 @@
+"""
+Load a hive client genesis file into an Amsterdam blockchain.
+
+Hive simulators hand clients a genesis file containing the fixture's
+genesis block header fields together with an `alloc` object describing
+the pre-state. This module parses that file into a [`BlockChain`] ready
+to accept payloads through the execution engine interface.
+
+[`BlockChain`]: ref:ethereum.forks.amsterdam.fork.BlockChain
+"""
+
+import json
+from pathlib import Path
+from typing import Any, Mapping
+
+from ethereum_rlp import rlp
+from ethereum_types.bytes import Bytes, Bytes8, Bytes32
+from ethereum_types.numeric import U64, U256, Uint
+
+from ethereum.crypto.hash import Hash32, keccak256
+from ethereum.forks.amsterdam.blocks import Block, Header
+from ethereum.forks.amsterdam.fork import BlockChain
+from ethereum.forks.amsterdam.fork_types import Bloom
+from ethereum.state import Account, Address, Root
+from ethereum.state_mpt import (
+    State,
+    set_account,
+    set_storage,
+    state_root,
+    store_code,
+)
+
+
+def _bytes(value: Any) -> bytes:
+    """Convert a `0x`-prefixed hex string to bytes."""
+    if not isinstance(value, str) or not value.startswith("0x"):
+        raise ValueError(f"expected hex string, got {value!r}")
+    return bytes.fromhex(value[2:])
+
+
+def _int(value: Any) -> int:
+    """Convert a hex string or plain integer to an integer."""
+    if isinstance(value, int):
+        return value
+    if isinstance(value, str):
+        return int(value, 16)
+    raise ValueError(f"expected integer or hex string, got {value!r}")
+
+
+def genesis_header_from_json(genesis: Mapping[str, Any]) -> Header:
+    """
+    Build the genesis block header from a hive genesis file.
+
+    The field names follow the blockchain test fixture header encoding,
+    which is what the consume simulators write for the client.
+    """
+    return Header(
+        parent_hash=Hash32(_bytes(genesis["parentHash"])),
+        ommers_hash=Hash32(_bytes(genesis["uncleHash"])),
+        coinbase=Address(_bytes(genesis["coinbase"])),
+        state_root=Root(_bytes(genesis["stateRoot"])),
+        transactions_root=Root(_bytes(genesis["transactionsTrie"])),
+        receipt_root=Root(_bytes(genesis["receiptTrie"])),
+        bloom=Bloom(_bytes(genesis["bloom"])),
+        difficulty=Uint(_int(genesis["difficulty"])),
+        number=Uint(_int(genesis["number"])),
+        gas_limit=Uint(_int(genesis["gasLimit"])),
+        gas_used=Uint(_int(genesis["gasUsed"])),
+        timestamp=U256(_int(genesis["timestamp"])),
+        extra_data=Bytes(_bytes(genesis["extraData"])),
+        prev_randao=Bytes32(_bytes(genesis["mixHash"])),
+        nonce=Bytes8(_bytes(genesis["nonce"])),
+        base_fee_per_gas=Uint(_int(genesis["baseFeePerGas"])),
+        withdrawals_root=Root(_bytes(genesis["withdrawalsRoot"])),
+        blob_gas_used=U64(_int(genesis["blobGasUsed"])),
+        excess_blob_gas=U64(_int(genesis["excessBlobGas"])),
+        parent_beacon_block_root=Root(
+            _bytes(genesis["parentBeaconBlockRoot"])
+        ),
+        requests_hash=Hash32(_bytes(genesis["requestsHash"])),
+        block_access_list_hash=Hash32(_bytes(genesis["blockAccessListHash"])),
+        slot_number=U64(_int(genesis["slotNumber"])),
+    )
+
+
+def state_from_alloc(alloc: Mapping[str, Any]) -> State:
+    """
+    Build the genesis state from a hive genesis `alloc` object.
+
+    Account keys may appear with or without a `0x` prefix.
+    """
+    state = State()
+    for address_hex, account in alloc.items():
+        address = Address(bytes.fromhex(address_hex.removeprefix("0x")))
+        code_hash = store_code(state, Bytes(_bytes(account.get("code", "0x"))))
+        set_account(
+            state,
+            address,
+            Account(
+                nonce=Uint(_int(account.get("nonce", 0))),
+                balance=U256(_int(account.get("balance", 0))),
+                code_hash=code_hash,
+            ),
+        )
+        for key, value in account.get("storage", {}).items():
+            set_storage(
+                state,
+                address,
+                Bytes32(_int(key).to_bytes(32, "big")),
+                U256(_int(value)),
+            )
+    return state
+
+
+def load_genesis_chain(genesis_path: Path, chain_id: U64) -> BlockChain:
+    """
+    Load a hive genesis file into a single-block chain.
+
+    The state built from `alloc` must produce the header's `state_root`,
+    and when the file carries the expected genesis `hash`, the header
+    must reproduce it.
+    """
+    genesis = json.loads(genesis_path.read_text())
+
+    header = genesis_header_from_json(genesis)
+    state = state_from_alloc(genesis.get("alloc", {}))
+
+    computed_state_root = state_root(state)
+    if computed_state_root != header.state_root:
+        raise ValueError(
+            f"alloc state root {computed_state_root.hex()} does not match "
+            f"genesis header state root {header.state_root.hex()}"
+        )
+
+    block_hash = keccak256(rlp.encode(header))
+    if "hash" in genesis and block_hash != Hash32(_bytes(genesis["hash"])):
+        raise ValueError(
+            f"computed genesis hash {block_hash.hex()} does not match "
+            f"declared genesis hash {genesis['hash']}"
+        )
+
+    return BlockChain(
+        blocks=[
+            Block(header=header, transactions=(), ommers=(), withdrawals=())
+        ],
+        state=state,
+        chain_id=chain_id,
+    )
diff --git a/src/ethereum_spec_tools/engine_server/server.py b/src/ethereum_spec_tools/engine_server/server.py
new file mode 100644
index 00000000000..b039401423b
--- /dev/null
+++ b/src/ethereum_spec_tools/engine_server/server.py
@@ -0,0 +1,649 @@
+"""
+JSON-RPC server exposing the Amsterdam execution engine interface.
+
+Serves the subset of the [Engine API] and `eth` namespace that a
+consensus-layer driver — such as the hive `consume engine` simulator —
+needs to feed blocks to the execution layer specification:
+
+- `engine_newPayloadV5` validates and executes a payload through
+  [`ethereum.forks.amsterdam.execution_engine`].
+- `engine_forkchoiceUpdatedV4` acknowledges the chain head.
+- `eth_getBlockByNumber` and friends answer basic chain queries.
+
+The engine namespace is authenticated with a JWT bearer token as
+described in the Engine API's authentication specification.
+
+[Engine API]: https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md
+"""  # noqa: E501
+
+import base64
+import hashlib
+import hmac
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Dict, List, Optional, Tuple
+
+from ethereum_rlp import rlp
+from ethereum_rlp.exceptions import DecodingError
+from ethereum_types.bytes import Bytes, Bytes32
+from ethereum_types.numeric import U64, U256, Uint
+from typing_extensions import override
+
+from ethereum.crypto.hash import Hash32, keccak256
+from ethereum.exceptions import EthereumException
+from ethereum.forks.amsterdam.block_access_lists import BlockAccessList
+from ethereum.forks.amsterdam.blocks import Block, Withdrawal
+from ethereum.forks.amsterdam.execution_engine import (
+    ExecutionPayload,
+    NewPayloadRequest,
+    is_valid_block_hash,
+    is_valid_versioned_hashes,
+)
+from ethereum.forks.amsterdam.execution_engine.requests import (
+    decode_execution_requests,
+)
+from ethereum.forks.amsterdam.execution_engine.validation_helpers import (
+    _payload_block,
+)
+from ethereum.forks.amsterdam.fork import BlockChain, state_transition
+from ethereum.forks.amsterdam.fork_types import Bloom, VersionedHash
+from ethereum.forks.amsterdam.transactions import LegacyTransaction
+from ethereum.state import Address, Root
+
+# JSON-RPC and Engine API error codes.
+PARSE_ERROR = -32700
+INVALID_REQUEST = -32600
+METHOD_NOT_FOUND = -32601
+INVALID_PARAMS = -32602
+INTERNAL_ERROR = -32603
+INVALID_PAYLOAD_ATTRIBUTES = -38003
+UNSUPPORTED_FORK = -38005
+
+DEFAULT_JWT_SECRET = b"secretsecretsecretsecretsecretse"
+"""Default JWT secret used by hive."""
+
+CLIENT_VERSION = "eels/execution-specs/amsterdam"
+"""Version string reported by `web3_clientVersion`."""
+
+
+class RpcError(Exception):
+    """JSON-RPC error raised by method handlers."""
+
+    def __init__(self, code: int, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+def _hex(data: bytes) -> str:
+    """Encode bytes as a `0x`-prefixed hex string."""
+    return "0x" + data.hex()
+
+
+def _hex_int(value: int) -> str:
+    """Encode an integer as a `0x`-prefixed hex quantity."""
+    return hex(int(value))
+
+
+def _decode_hex(value: Any, name: str, size: Optional[int] = None) -> bytes:
+    """Decode a `0x`-prefixed hex string, validating its byte length."""
+    if not isinstance(value, str) or not value.startswith("0x"):
+        raise RpcError(INVALID_PARAMS, f"{name}: expected hex string")
+    try:
+        data = bytes.fromhex(value[2:])
+    except ValueError as e:
+        raise RpcError(INVALID_PARAMS, f"{name}: invalid hex: {e}") from e
+    if size is not None and len(data) != size:
+        raise RpcError(
+            INVALID_PARAMS,
+            f"{name}: expected {size} bytes, got {len(data)}",
+        )
+    return data
+
+
+def _decode_quantity(value: Any, name: str) -> int:
+    """Decode a `0x`-prefixed hex quantity."""
+    if not isinstance(value, str) or not value.startswith("0x"):
+        raise RpcError(INVALID_PARAMS, f"{name}: expected hex quantity")
+    try:
+        return int(value, 16)
+    except ValueError as e:
+        raise RpcError(INVALID_PARAMS, f"{name}: invalid quantity") from e
+
+
+def _field(obj: Dict[str, Any], name: str) -> Any:
+    """Fetch a required field from a JSON object."""
+    if not isinstance(obj, dict) or name not in obj:
+        raise RpcError(INVALID_PARAMS, f"missing field: {name}")
+    return obj[name]
+
+
+def _withdrawal_from_json(obj: Dict[str, Any]) -> Withdrawal:
+    """Decode a withdrawal object."""
+    return Withdrawal(
+        index=U64(_decode_quantity(_field(obj, "index"), "index")),
+        validator_index=U64(
+            _decode_quantity(_field(obj, "validatorIndex"), "validatorIndex")
+        ),
+        address=Address(_decode_hex(_field(obj, "address"), "address", 20)),
+        amount=U64(_decode_quantity(_field(obj, "amount"), "amount")),
+    )
+
+
+def _payload_from_json(obj: Any) -> ExecutionPayload:
+    """
+    Decode an `ExecutionPayloadV4` structure.
+
+    Raise an invalid-params error for any missing or malformed field,
+    per the Engine API requirement that structural failures are
+    reported at the RPC layer rather than as an `INVALID` status.
+    """
+    if not isinstance(obj, dict):
+        raise RpcError(INVALID_PARAMS, "executionPayload: expected object")
+    withdrawals = _field(obj, "withdrawals")
+    transactions = _field(obj, "transactions")
+    if not isinstance(withdrawals, list) or not isinstance(transactions, list):
+        raise RpcError(INVALID_PARAMS, "expected array")
+    return ExecutionPayload(
+        parent_hash=Hash32(
+            _decode_hex(_field(obj, "parentHash"), "parentHash", 32)
+        ),
+        fee_recipient=Address(
+            _decode_hex(_field(obj, "feeRecipient"), "feeRecipient", 20)
+        ),
+        state_root=Root(
+            _decode_hex(_field(obj, "stateRoot"), "stateRoot", 32)
+        ),
+        receipts_root=Root(
+            _decode_hex(_field(obj, "receiptsRoot"), "receiptsRoot", 32)
+        ),
+        logs_bloom=Bloom(
+            _decode_hex(_field(obj, "logsBloom"), "logsBloom", 256)
+        ),
+        prev_randao=Bytes32(
+            _decode_hex(_field(obj, "prevRandao"), "prevRandao", 32)
+        ),
+        block_number=Uint(
+            _decode_quantity(_field(obj, "blockNumber"), "blockNumber")
+        ),
+        gas_limit=Uint(_decode_quantity(_field(obj, "gasLimit"), "gasLimit")),
+        gas_used=Uint(_decode_quantity(_field(obj, "gasUsed"), "gasUsed")),
+        timestamp=U256(
+            _decode_quantity(_field(obj, "timestamp"), "timestamp")
+        ),
+        extra_data=Bytes(_decode_hex(_field(obj, "extraData"), "extraData")),
+        base_fee_per_gas=Uint(
+            _decode_quantity(_field(obj, "baseFeePerGas"), "baseFeePerGas")
+        ),
+        block_hash=Hash32(
+            _decode_hex(_field(obj, "blockHash"), "blockHash", 32)
+        ),
+        transactions=tuple(
+            Bytes(_decode_hex(tx, "transaction")) for tx in transactions
+        ),
+        withdrawals=tuple(_withdrawal_from_json(w) for w in withdrawals),
+        blob_gas_used=U64(
+            _decode_quantity(_field(obj, "blobGasUsed"), "blobGasUsed")
+        ),
+        excess_blob_gas=U64(
+            _decode_quantity(_field(obj, "excessBlobGas"), "excessBlobGas")
+        ),
+        block_access_list=Bytes(
+            _decode_hex(_field(obj, "blockAccessList"), "blockAccessList")
+        ),
+        slot_number=U64(
+            _decode_quantity(_field(obj, "slotNumber"), "slotNumber")
+        ),
+    )
+
+
+def _block_hash(block: Block) -> Hash32:
+    """Compute the hash of a block's header."""
+    return keccak256(rlp.encode(block.header))
+
+
+def _transaction_hash(tx: Any) -> Hash32:
+    """Compute the hash of a block transaction."""
+    if isinstance(tx, LegacyTransaction):
+        return keccak256(rlp.encode(tx))
+    return keccak256(tx)
+
+
+def _block_to_json(block: Block) -> Dict[str, Any]:
+    """
+    Encode a block in the `eth_getBlockByNumber` response format.
+
+    Transactions are always returned as hashes.
+    """
+    header = block.header
+    return {
+        "number": _hex_int(int(header.number)),
+        "hash": _hex(_block_hash(block)),
+        "parentHash": _hex(header.parent_hash),
+        "nonce": _hex(header.nonce),
+        "sha3Uncles": _hex(header.ommers_hash),
+        "logsBloom": _hex(header.bloom),
+        "transactionsRoot": _hex(header.transactions_root),
+        "stateRoot": _hex(header.state_root),
+        "receiptsRoot": _hex(header.receipt_root),
+        "miner": _hex(header.coinbase),
+        "difficulty": _hex_int(int(header.difficulty)),
+        "extraData": _hex(header.extra_data),
+        "size": _hex_int(len(rlp.encode(block))),
+        "gasLimit": _hex_int(int(header.gas_limit)),
+        "gasUsed": _hex_int(int(header.gas_used)),
+        "timestamp": _hex_int(int(header.timestamp)),
+        "mixHash": _hex(header.prev_randao),
+        "baseFeePerGas": _hex_int(int(header.base_fee_per_gas)),
+        "withdrawalsRoot": _hex(header.withdrawals_root),
+        "blobGasUsed": _hex_int(int(header.blob_gas_used)),
+        "excessBlobGas": _hex_int(int(header.excess_blob_gas)),
+        "parentBeaconBlockRoot": _hex(header.parent_beacon_block_root),
+        "requestsHash": _hex(header.requests_hash),
+        "blockAccessListHash": _hex(header.block_access_list_hash),
+        "slotNumber": _hex_int(int(header.slot_number)),
+        "transactions": [
+            _hex(_transaction_hash(tx)) for tx in block.transactions
+        ],
+        "withdrawals": [],
+        "uncles": [],
+    }
+
+
+class EngineBackend:
+    """
+    Chain state and JSON-RPC method handlers.
+
+    Holds the [`BlockChain`] that payloads are applied to, guarded by a
+    lock so that concurrent HTTP requests observe a consistent chain.
+
+    [`BlockChain`]: ref:ethereum.forks.amsterdam.fork.BlockChain
+    """
+
+    def __init__(self, chain: BlockChain) -> None:
+        self.chain = chain
+        self.lock = threading.Lock()
+        self.genesis_block = chain.blocks[0]
+        # All block hashes ever applied, mapping to their block number.
+        # Survives the chain's 255-block trim so forkchoice updates can
+        # recognize any previously validated head.
+        self.known_blocks: Dict[Hash32, Uint] = {
+            _block_hash(self.genesis_block): self.genesis_block.header.number
+        }
+
+    def head_hash(self) -> Hash32:
+        """Return the hash of the current chain head."""
+        return _block_hash(self.chain.blocks[-1])
+
+    def handle(self, method: str, params: List[Any]) -> Any:
+        """Dispatch a JSON-RPC method call."""
+        handlers = {
+            "web3_clientVersion": self.client_version,
+            "eth_chainId": self.chain_id,
+            "eth_getBlockByNumber": self.get_block_by_number,
+            "eth_getBlockByHash": self.get_block_by_hash,
+            "engine_exchangeCapabilities": self.exchange_capabilities,
+            "engine_newPayloadV5": self.new_payload_v5,
+            "engine_forkchoiceUpdatedV4": self.forkchoice_updated_v4,
+        }
+        if method in handlers:
+            return handlers[method](params)
+        # Earlier versions of the supported engine methods exist, but
+        # target forks preceding Amsterdam.
+        if method.startswith("engine_newPayloadV") or method.startswith(
+            "engine_forkchoiceUpdatedV"
+        ):
+            raise RpcError(UNSUPPORTED_FORK, "Unsupported fork")
+        raise RpcError(METHOD_NOT_FOUND, f"the method {method} does not exist")
+
+    def client_version(self, _params: List[Any]) -> str:
+        """`web3_clientVersion`: identify this client."""
+        return CLIENT_VERSION
+
+    def chain_id(self, _params: List[Any]) -> str:
+        """`eth_chainId`: return the chain id of the loaded chain."""
+        return _hex_int(int(self.chain.chain_id))
+
+    def exchange_capabilities(self, _params: List[Any]) -> List[str]:
+        """`engine_exchangeCapabilities`: list supported engine methods."""
+        return [
+            "engine_exchangeCapabilities",
+            "engine_newPayloadV5",
+            "engine_forkchoiceUpdatedV4",
+        ]
+
+    def _find_block(self, tag: Any) -> Optional[Block]:
+        """Resolve a block-number tag to a block, if present."""
+        with self.lock:
+            if tag in ("latest", "safe", "finalized", "pending"):
+                return self.chain.blocks[-1]
+            if tag == "earliest":
+                return self.genesis_block
+            number = Uint(_decode_quantity(tag, "blockNumber"))
+            if number == self.genesis_block.header.number:
+                return self.genesis_block
+            for block in self.chain.blocks:
+                if block.header.number == number:
+                    return block
+        return None
+
+    def get_block_by_number(self, params: List[Any]) -> Any:
+        """`eth_getBlockByNumber`: return a block by number or tag."""
+        if len(params) != 2:
+            raise RpcError(INVALID_PARAMS, "expected 2 params")
+        block = self._find_block(params[0])
+        if block is None:
+            return None
+        return _block_to_json(block)
+
+    def get_block_by_hash(self, params: List[Any]) -> Any:
+        """`eth_getBlockByHash`: return a block by hash."""
+        if len(params) != 2:
+            raise RpcError(INVALID_PARAMS, "expected 2 params")
+        block_hash = Hash32(_decode_hex(params[0], "blockHash", 32))
+        with self.lock:
+            for block in self.chain.blocks:
+                if _block_hash(block) == block_hash:
+                    return _block_to_json(block)
+        return None
+
+    def new_payload_v5(self, params: List[Any]) -> Dict[str, Any]:
+        """
+        `engine_newPayloadV5`: validate and execute a payload.
+
+        Follows the consensus-layer `verify_and_notify_new_payload`
+        sequence, surfacing each failure as an `INVALID` payload status
+        with a validation error message. Malformed parameters and
+        execution-request violations are JSON-RPC errors instead.
+        """
+        if len(params) != 4:
+            raise RpcError(INVALID_PARAMS, "expected 4 params")
+        payload_json, hashes_json, beacon_root_json, requests_json = params
+
+        payload = _payload_from_json(payload_json)
+        try:
+            rlp.decode_to(BlockAccessList, payload.block_access_list)
+        except DecodingError as e:
+            # A structurally undecodable block access list is an
+            # invalid parameter, not an invalid block.
+            raise RpcError(INVALID_PARAMS, f"blockAccessList: {e}") from e
+        if not isinstance(hashes_json, list):
+            raise RpcError(
+                INVALID_PARAMS, "expectedBlobVersionedHashes: expected array"
+            )
+        versioned_hashes = tuple(
+            VersionedHash(_decode_hex(h, "versionedHash", 32))
+            for h in hashes_json
+        )
+        parent_beacon_block_root = Root(
+            _decode_hex(beacon_root_json, "parentBeaconBlockRoot", 32)
+        )
+        if not isinstance(requests_json, list):
+            raise RpcError(INVALID_PARAMS, "executionRequests: expected array")
+        try:
+            execution_requests = decode_execution_requests(
+                tuple(
+                    Bytes(_decode_hex(r, "executionRequest"))
+                    for r in requests_json
+                )
+            )
+        except EthereumException as e:
+            # Ordering, size, and unknown-type violations are invalid
+            # params per the Engine API.
+            raise RpcError(INVALID_PARAMS, str(e)) from e
+
+        request = NewPayloadRequest(
+            execution_payload=payload,
+            versioned_hashes=versioned_hashes,
+            parent_beacon_block_root=parent_beacon_block_root,
+            execution_requests=execution_requests,
+        )
+
+        with self.lock:
+            return self._execute_payload(request)
+
+    def _execute_payload(self, request: NewPayloadRequest) -> Dict[str, Any]:
+        """Run the new-payload validation sequence for one request."""
+        payload = request.execution_payload
+
+        if b"" in payload.transactions:
+            return _payload_status(
+                "INVALID", None, "empty transaction in payload"
+            )
+
+        if not is_valid_block_hash(
+            payload,
+            request.parent_beacon_block_root,
+            request.execution_requests,
+        ):
+            return _payload_status("INVALID", None, "invalid block hash")
+
+        if not is_valid_versioned_hashes(request):
+            return _payload_status(
+                "INVALID", None, "invalid blob versioned hashes"
+            )
+
+        parent_is_head = payload.parent_hash == self.head_hash()
+
+        try:
+            block = _payload_block(
+                payload,
+                request.parent_beacon_block_root,
+                request.execution_requests,
+            )
+            state_transition(self.chain, block)
+        except EthereumException as e:
+            latest_valid: Optional[Hash32] = (
+                Hash32(payload.parent_hash) if parent_is_head else None
+            )
+            return _payload_status(
+                "INVALID", latest_valid, f"{type(e).__name__}: {e}"
+            )
+
+        self.known_blocks[Hash32(payload.block_hash)] = payload.block_number
+        return _payload_status("VALID", Hash32(payload.block_hash), None)
+
+    def forkchoice_updated_v4(self, params: List[Any]) -> Dict[str, Any]:
+        """
+        `engine_forkchoiceUpdatedV4`: acknowledge a forkchoice state.
+
+        Payload building is not supported, so non-null payload
+        attributes are rejected. The optional third parameter (the
+        EIP-8070 custody-column bitmap) is accepted and ignored.
+        """
+        if len(params) not in (2, 3):
+            raise RpcError(INVALID_PARAMS, "expected 2 or 3 params")
+        forkchoice_state = params[0]
+        payload_attributes = params[1]
+
+        head = Hash32(
+            _decode_hex(
+                _field(forkchoice_state, "headBlockHash"),
+                "headBlockHash",
+                32,
+            )
+        )
+
+        if payload_attributes is not None:
+            raise RpcError(
+                INVALID_PAYLOAD_ATTRIBUTES,
+                "payload building is not supported",
+            )
+
+        with self.lock:
+            known = head in self.known_blocks
+        if not known:
+            return {
+                "payloadStatus": _payload_status("SYNCING", None, None),
+                "payloadId": None,
+            }
+        return {
+            "payloadStatus": _payload_status("VALID", head, None),
+            "payloadId": None,
+        }
+
+
+def _payload_status(
+    status: str, latest_valid_hash: Optional[Hash32], error: Optional[str]
+) -> Dict[str, Any]:
+    """Encode a `PayloadStatusV1` object."""
+    return {
+        "status": status,
+        "latestValidHash": (
+            _hex(latest_valid_hash) if latest_valid_hash is not None else None
+        ),
+        "validationError": error,
+    }
+
+
+def _b64url_decode(data: str) -> bytes:
+    """Decode unpadded base64url data."""
+    return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))
+
+
+def verify_jwt(token: str, secret: bytes) -> bool:
+    """
+    Verify an HS256 JWT signature.
+
+    Only the signature is checked; `iat` freshness is intentionally not
+    enforced.
+    """
+    parts = token.split(".")
+    if len(parts) != 3:
+        return False
+    signing_input = f"{parts[0]}.{parts[1]}".encode()
+    try:
+        header = json.loads(_b64url_decode(parts[0]))
+        signature = _b64url_decode(parts[2])
+    except (ValueError, UnicodeDecodeError):
+        return False
+    if header.get("alg") != "HS256":
+        return False
+    expected = hmac.new(secret, signing_input, hashlib.sha256).digest()
+    return hmac.compare_digest(signature, expected)
+
+
+class _RpcHandler(BaseHTTPRequestHandler):
+    """HTTP handler translating JSON-RPC requests to backend calls."""
+
+    backend: "EngineBackend"
+    jwt_secret: Optional[bytes] = None
+
+    @override
+    def log_message(self, format: str, *args: Any) -> None:
+        """Suppress default request logging."""
+
+    def _respond(self, status: int, body: Dict[str, Any] | List[Any]) -> None:
+        data = json.dumps(body).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def _authorized(self) -> bool:
+        if self.jwt_secret is None:
+            return True
+        authorization = self.headers.get("Authorization", "")
+        if not authorization.startswith("Bearer "):
+            return False
+        return verify_jwt(
+            authorization.removeprefix("Bearer "), self.jwt_secret
+        )
+
+    def _handle_single(self, request: Any) -> Dict[str, Any]:
+        request_id = request.get("id") if isinstance(request, dict) else None
+        response: Dict[str, Any] = {"jsonrpc": "2.0", "id": request_id}
+        if not isinstance(request, dict) or "method" not in request:
+            response["error"] = {
+                "code": INVALID_REQUEST,
+                "message": "invalid request",
+            }
+            return response
+        try:
+            response["result"] = self.backend.handle(
+                request["method"], request.get("params", [])
+            )
+        except RpcError as e:
+            response["error"] = {"code": e.code, "message": e.message}
+        except Exception as e:  # noqa: BLE001
+            response["error"] = {
+                "code": INTERNAL_ERROR,
+                "message": f"{type(e).__name__}: {e}",
+            }
+        return response
+
+    def do_POST(self) -> None:  # noqa: N802
+        """Serve one JSON-RPC request or batch."""
+        if not self._authorized():
+            self._respond(
+                401,
+                {
+                    "jsonrpc": "2.0",
+                    "id": None,
+                    "error": {
+                        "code": INVALID_REQUEST,
+                        "message": "missing or invalid JWT",
+                    },
+                },
+            )
+            return
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+            request = json.loads(self.rfile.read(length))
+        except (ValueError, json.JSONDecodeError):
+            self._respond(
+                200,
+                {
+                    "jsonrpc": "2.0",
+                    "id": None,
+                    "error": {
+                        "code": PARSE_ERROR,
+                        "message": "parse error",
+                    },
+                },
+            )
+            return
+        if isinstance(request, list):
+            self._respond(200, [self._handle_single(item) for item in request])
+        else:
+            self._respond(200, self._handle_single(request))
+
+
+def serve(
+    backend: EngineBackend,
+    address: str,
+    rpc_port: int,
+    engine_port: int,
+    jwt_secret: bytes,
+) -> Tuple[ThreadingHTTPServer, ThreadingHTTPServer]:
+    """
+    Start the `eth` and authenticated `engine` HTTP listeners.
+
+    Both listeners dispatch to the same backend; only the engine
+    listener requires JWT authentication. Returns the two servers with
+    their serving threads already running.
+    """
+
+    class EthHandler(_RpcHandler):
+        pass
+
+    class EngineHandler(_RpcHandler):
+        pass
+
+    EthHandler.backend = backend
+    EthHandler.jwt_secret = None
+    EngineHandler.backend = backend
+    EngineHandler.jwt_secret = jwt_secret
+
+    rpc_server = ThreadingHTTPServer((address, rpc_port), EthHandler)
+    engine_server = ThreadingHTTPServer((address, engine_port), EngineHandler)
+
+    for http_server in (rpc_server, engine_server):
+        thread = threading.Thread(
+            target=http_server.serve_forever, daemon=True
+        )
+        thread.start()
+
+    return rpc_server, engine_server
diff --git a/vulture_whitelist.py b/vulture_whitelist.py
index 64649b51a0f..57fdbd4d065 100644
--- a/vulture_whitelist.py
+++ b/vulture_whitelist.py
@@ -222,3 +222,8 @@
 BlobsBundle.blobs
 GetPayloadResponse.block_value
 GetPayloadResponse.blobs_bundle
+
+# src/ethereum_spec_tools/engine_server - http.server override hook
+from ethereum_spec_tools.engine_server.server import _RpcHandler
+
+_RpcHandler.log_message

From 5e50a520b096392158c8ea9944a3088ad9249461 Mon Sep 17 00:00:00 2001
From: spencer-tb <spencer.tb@ethereum.org>
Date: Wed, 12 Aug 2026 17:16:24 +0200
Subject: [PATCH 3/5] feat(tooling): add block-tree reorg support to engine
 server for enginex

---
 src/ethereum/state_mpt.py                     |  19 +++
 .../engine_server/server.py                   | 115 ++++++++++++++----
 2 files changed, 111 insertions(+), 23 deletions(-)

diff --git a/src/ethereum/state_mpt.py b/src/ethereum/state_mpt.py
index dbdd7a718e0..8c4cb559ae3 100644
--- a/src/ethereum/state_mpt.py
+++ b/src/ethereum/state_mpt.py
@@ -138,6 +138,25 @@ def close_state(state: State) -> None:
     del state._code_store
 
 
+def copy_state(state: State) -> State:
+    """
+    Return an independent copy of `state`.
+
+    Mutating the copy, including through [`apply_changes_to_state`],
+    leaves the original untouched.
+
+    [`apply_changes_to_state`]: ref:ethereum.state_mpt.apply_changes_to_state
+    """
+    return State(
+        _main_trie=copy_trie(state._main_trie),
+        _storage_tries={
+            address: copy_trie(storage_trie)
+            for address, storage_trie in state._storage_tries.items()
+        },
+        _code_store=dict(state._code_store),
+    )
+
+
 def apply_changes_to_state(state: State, diff: BlockDiff) -> None:
     """
     Apply block-level diff to the ``State`` for the next block.
diff --git a/src/ethereum_spec_tools/engine_server/server.py b/src/ethereum_spec_tools/engine_server/server.py
index b039401423b..6f3a7ad30cc 100644
--- a/src/ethereum_spec_tools/engine_server/server.py
+++ b/src/ethereum_spec_tools/engine_server/server.py
@@ -21,6 +21,7 @@
 import hmac
 import json
 import threading
+from dataclasses import dataclass
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -46,10 +47,16 @@
 from ethereum.forks.amsterdam.execution_engine.validation_helpers import (
     _payload_block,
 )
-from ethereum.forks.amsterdam.fork import BlockChain, state_transition
+from ethereum.forks.amsterdam.fork import (
+    BlockChain,
+    ChainContext,
+    execute_block,
+    get_last_256_block_hashes,
+)
 from ethereum.forks.amsterdam.fork_types import Bloom, VersionedHash
 from ethereum.forks.amsterdam.transactions import LegacyTransaction
-from ethereum.state import Address, Root
+from ethereum.state import Address, BlockDiff, Root
+from ethereum.state_mpt import State, apply_changes_to_state, copy_state
 
 # JSON-RPC and Engine API error codes.
 PARSE_ERROR = -32700
@@ -251,31 +258,73 @@ def _block_to_json(block: Block) -> Dict[str, Any]:
     }
 
 
+@dataclass
+class _BlockRecord:
+    """A validated block, its parent link, and the diff it produced."""
+
+    block: Block
+    parent_hash: Optional[Hash32]
+    diff: Optional[BlockDiff]
+
+
 class EngineBackend:
     """
     Chain state and JSON-RPC method handlers.
 
-    Holds the [`BlockChain`] that payloads are applied to, guarded by a
-    lock so that concurrent HTTP requests observe a consistent chain.
+    Holds the [`BlockChain`] of the active branch, guarded by a lock so
+    that concurrent HTTP requests observe a consistent chain. Every
+    validated block is remembered in a block tree together with the
+    [`BlockDiff`] it produced, so a payload building on any known block
+    (or a forkchoice update selecting one) reorgs by rebuilding the
+    branch state from the genesis snapshot.
 
     [`BlockChain`]: ref:ethereum.forks.amsterdam.fork.BlockChain
+    [`BlockDiff`]: ref:ethereum.state.BlockDiff
     """
 
     def __init__(self, chain: BlockChain) -> None:
         self.chain = chain
         self.lock = threading.Lock()
         self.genesis_block = chain.blocks[0]
-        # All block hashes ever applied, mapping to their block number.
-        # Survives the chain's 255-block trim so forkchoice updates can
-        # recognize any previously validated head.
-        self.known_blocks: Dict[Hash32, Uint] = {
-            _block_hash(self.genesis_block): self.genesis_block.header.number
+        self._genesis_state: State = copy_state(chain.state)
+        # Block tree of every block ever validated; survives the active
+        # branch's 255-block trim.
+        self.records: Dict[Hash32, _BlockRecord] = {
+            _block_hash(self.genesis_block): _BlockRecord(
+                block=self.genesis_block, parent_hash=None, diff=None
+            )
         }
 
     def head_hash(self) -> Hash32:
         """Return the hash of the current chain head."""
         return _block_hash(self.chain.blocks[-1])
 
+    def _rebuild_to(self, target: Hash32) -> None:
+        """
+        Make `target` the chain head by rebuilding its branch.
+
+        Collect the ancestry of `target` in the block tree, copy the
+        genesis state, and reapply each block's diff along the branch.
+        """
+        branch = []
+        cursor: Optional[Hash32] = target
+        while cursor is not None:
+            record = self.records[cursor]
+            branch.append(record)
+            cursor = record.parent_hash
+        branch.reverse()
+
+        state = copy_state(self._genesis_state)
+        for record in branch[1:]:
+            assert record.diff is not None
+            apply_changes_to_state(state, record.diff)
+
+        self.chain = BlockChain(
+            blocks=[record.block for record in branch][-255:],
+            state=state,
+            chain_id=self.chain.chain_id,
+        )
+
     def handle(self, method: str, params: List[Any]) -> Any:
         """Dispatch a JSON-RPC method call."""
         handlers = {
@@ -343,9 +392,9 @@ def get_block_by_hash(self, params: List[Any]) -> Any:
             raise RpcError(INVALID_PARAMS, "expected 2 params")
         block_hash = Hash32(_decode_hex(params[0], "blockHash", 32))
         with self.lock:
-            for block in self.chain.blocks:
-                if _block_hash(block) == block_hash:
-                    return _block_to_json(block)
+            record = self.records.get(block_hash)
+            if record is not None:
+                return _block_to_json(record.block)
         return None
 
     def new_payload_v5(self, params: List[Any]) -> Dict[str, Any]:
@@ -424,7 +473,12 @@ def _execute_payload(self, request: NewPayloadRequest) -> Dict[str, Any]:
                 "INVALID", None, "invalid blob versioned hashes"
             )
 
-        parent_is_head = payload.parent_hash == self.head_hash()
+        parent_hash = Hash32(payload.parent_hash)
+        parent_known = parent_hash in self.records
+        if parent_known and parent_hash != self.head_hash():
+            # The payload builds on a known non-head block: reorg the
+            # active branch onto its parent before executing.
+            self._rebuild_to(parent_hash)
 
         try:
             block = _payload_block(
@@ -432,17 +486,30 @@ def _execute_payload(self, request: NewPayloadRequest) -> Dict[str, Any]:
                 request.parent_beacon_block_root,
                 request.execution_requests,
             )
-            state_transition(self.chain, block)
+            chain_context = ChainContext(
+                chain_id=self.chain.chain_id,
+                block_hashes=get_last_256_block_hashes(self.chain),
+                parent_header=self.chain.blocks[-1].header,
+            )
+            diff = execute_block(block, self.chain.state, chain_context)
         except EthereumException as e:
             latest_valid: Optional[Hash32] = (
-                Hash32(payload.parent_hash) if parent_is_head else None
+                parent_hash if parent_known else None
             )
             return _payload_status(
                 "INVALID", latest_valid, f"{type(e).__name__}: {e}"
             )
 
-        self.known_blocks[Hash32(payload.block_hash)] = payload.block_number
-        return _payload_status("VALID", Hash32(payload.block_hash), None)
+        apply_changes_to_state(self.chain.state, diff)
+        self.chain.blocks.append(block)
+        if len(self.chain.blocks) > 255:
+            self.chain.blocks = self.chain.blocks[-255:]
+
+        block_hash = Hash32(payload.block_hash)
+        self.records[block_hash] = _BlockRecord(
+            block=block, parent_hash=parent_hash, diff=diff
+        )
+        return _payload_status("VALID", block_hash, None)
 
     def forkchoice_updated_v4(self, params: List[Any]) -> Dict[str, Any]:
         """
@@ -472,12 +539,14 @@ def forkchoice_updated_v4(self, params: List[Any]) -> Dict[str, Any]:
             )
 
         with self.lock:
-            known = head in self.known_blocks
-        if not known:
-            return {
-                "payloadStatus": _payload_status("SYNCING", None, None),
-                "payloadId": None,
-            }
+            if head not in self.records:
+                return {
+                    "payloadStatus": _payload_status("SYNCING", None, None),
+                    "payloadId": None,
+                }
+            if head != self.head_hash():
+                # Selecting a known non-head block as head is a reorg.
+                self._rebuild_to(head)
         return {
             "payloadStatus": _payload_status("VALID", head, None),
             "payloadId": None,

From 3c282945b6d8cbc0f07af05dc6cddc822137e3e8 Mon Sep 17 00:00:00 2001
From: spencer-tb <spencer.tb@ethereum.org>
Date: Wed, 12 Aug 2026 17:47:18 +0200
Subject: [PATCH 4/5] perf(tooling): prepare genesis state copies off the
 engine server request path

---
 .../engine_server/server.py                   | 23 +++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/src/ethereum_spec_tools/engine_server/server.py b/src/ethereum_spec_tools/engine_server/server.py
index 6f3a7ad30cc..cea950a73f5 100644
--- a/src/ethereum_spec_tools/engine_server/server.py
+++ b/src/ethereum_spec_tools/engine_server/server.py
@@ -20,6 +20,7 @@
 import hashlib
 import hmac
 import json
+import queue
 import threading
 from dataclasses import dataclass
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -294,6 +295,24 @@ def __init__(self, chain: BlockChain) -> None:
                 block=self.genesis_block, parent_hash=None, diff=None
             )
         }
+        # Reorgs back towards genesis need a fresh copy of the genesis
+        # state; keep one prepared in the background so the copy is off
+        # the request path (`_genesis_state` is never mutated, so the
+        # copier thread needs no lock).
+        self._genesis_copies: "queue.Queue[State]" = queue.Queue(maxsize=1)
+        threading.Thread(target=self._prepare_copies, daemon=True).start()
+
+    def _prepare_copies(self) -> None:
+        """Keep one spare copy of the genesis state ready."""
+        while True:
+            self._genesis_copies.put(copy_state(self._genesis_state))
+
+    def _fresh_genesis_state(self) -> State:
+        """Take the prepared genesis state copy, or copy synchronously."""
+        try:
+            return self._genesis_copies.get_nowait()
+        except queue.Empty:
+            return copy_state(self._genesis_state)
 
     def head_hash(self) -> Hash32:
         """Return the hash of the current chain head."""
@@ -303,7 +322,7 @@ def _rebuild_to(self, target: Hash32) -> None:
         """
         Make `target` the chain head by rebuilding its branch.
 
-        Collect the ancestry of `target` in the block tree, copy the
+        Collect the ancestry of `target` in the block tree, take a fresh
         genesis state, and reapply each block's diff along the branch.
         """
         branch = []
@@ -314,7 +333,7 @@ def _rebuild_to(self, target: Hash32) -> None:
             cursor = record.parent_hash
         branch.reverse()
 
-        state = copy_state(self._genesis_state)
+        state = self._fresh_genesis_state()
         for record in branch[1:]:
             assert record.diff is not None
             apply_changes_to_state(state, record.diff)

From d5f7979ef5e30107c9f4ff5b6275b92e39ab6eec Mon Sep 17 00:00:00 2001
From: spencer-tb <spencer.tb@ethereum.org>
Date: Wed, 12 Aug 2026 20:53:15 +0200
Subject: [PATCH 5/5] fix(spec-specs,tooling): treat execution requests as
 opaque wire bytes per engine API

---
 .../amsterdam/execution_engine/__init__.py    |  2 +-
 .../amsterdam/execution_engine/new_payload.py |  6 ++-
 .../forks/amsterdam/execution_engine/types.py | 12 ++++--
 .../execution_engine/validation_helpers.py    | 15 +++-----
 .../engine_server/server.py                   | 38 ++++++++++++-------
 vulture_whitelist.py                          |  2 +
 6 files changed, 46 insertions(+), 29 deletions(-)

diff --git a/src/ethereum/forks/amsterdam/execution_engine/__init__.py b/src/ethereum/forks/amsterdam/execution_engine/__init__.py
index e70905eec8d..cdb95a323cc 100644
--- a/src/ethereum/forks/amsterdam/execution_engine/__init__.py
+++ b/src/ethereum/forks/amsterdam/execution_engine/__init__.py
@@ -23,11 +23,11 @@
     notify_new_payload,
     verify_and_notify_new_payload,
 )
+from .requests import ExecutionRequests
 from .types import (
     BlobsBundle,
     ExecutionEngine,
     ExecutionPayload,
-    ExecutionRequests,
     GetPayloadResponse,
     NewPayloadRequest,
     PayloadAttributes,
diff --git a/src/ethereum/forks/amsterdam/execution_engine/new_payload.py b/src/ethereum/forks/amsterdam/execution_engine/new_payload.py
index 1bb8c73188b..f6943acfabf 100644
--- a/src/ethereum/forks/amsterdam/execution_engine/new_payload.py
+++ b/src/ethereum/forks/amsterdam/execution_engine/new_payload.py
@@ -2,7 +2,10 @@
 Payload verification and execution.
 """
 
+from typing import Tuple
+
 from ethereum_rlp import rlp
+from ethereum_types.bytes import Bytes
 
 from ethereum.crypto.hash import keccak256
 from ethereum.exceptions import EthereumException
@@ -11,7 +14,6 @@
 from ..fork import state_transition
 from ..fork_types import VersionedHash
 from ..transactions import BlobTransaction, decode_transaction
-from .requests import ExecutionRequests
 from .types import ExecutionEngine, ExecutionPayload, NewPayloadRequest
 from .validation_helpers import _payload_block, _payload_header
 
@@ -19,7 +21,7 @@
 def is_valid_block_hash(
     execution_payload: ExecutionPayload,
     parent_beacon_block_root: Root,
-    execution_requests: ExecutionRequests,
+    execution_requests: Tuple[Bytes, ...],
 ) -> bool:
     """
     Return `True` if and only if `execution_payload.block_hash` is
diff --git a/src/ethereum/forks/amsterdam/execution_engine/types.py b/src/ethereum/forks/amsterdam/execution_engine/types.py
index d39990bbd43..3d69962b810 100644
--- a/src/ethereum/forks/amsterdam/execution_engine/types.py
+++ b/src/ethereum/forks/amsterdam/execution_engine/types.py
@@ -15,7 +15,6 @@
 from ..blocks import Withdrawal
 from ..fork import BlockChain
 from ..fork_types import Bloom, VersionedHash
-from .requests import ExecutionRequests
 
 ExecutionEngine = BlockChain
 """
@@ -84,7 +83,12 @@ class NewPayloadRequest:
 
     Corresponds to the consensus-layer [`NewPayloadRequest`] container
     and carries the parameters of the Engine API `engine_newPayloadV5`
-    method.
+    method. The execution requests are carried in their opaque wire
+    form; [`decode_execution_requests`] parses them when typed access
+    is needed.
+
+    [`decode_execution_requests`]:
+        ref:ethereum.forks.amsterdam.execution_engine.requests.decode_execution_requests
 
     [`verify_and_notify_new_payload`]:
         ref:ethereum.forks.amsterdam.execution_engine.new_payload.verify_and_notify_new_payload
@@ -94,7 +98,7 @@ class NewPayloadRequest:
     execution_payload: ExecutionPayload
     versioned_hashes: Tuple[VersionedHash, ...]
     parent_beacon_block_root: Root
-    execution_requests: ExecutionRequests
+    execution_requests: Tuple[Bytes, ...]
 
 
 @final
@@ -140,4 +144,4 @@ class GetPayloadResponse:
     execution_payload: ExecutionPayload
     block_value: U256
     blobs_bundle: BlobsBundle
-    execution_requests: ExecutionRequests
+    execution_requests: Tuple[Bytes, ...]
diff --git a/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py b/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py
index f76af3d1ffa..717f35978f9 100644
--- a/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py
+++ b/src/ethereum/forks/amsterdam/execution_engine/validation_helpers.py
@@ -2,7 +2,7 @@
 Shared execution-engine conversion helpers.
 """
 
-from typing import Optional
+from typing import Optional, Tuple
 
 from ethereum_rlp import rlp
 from ethereum_types.bytes import Bytes, Bytes8
@@ -16,14 +16,13 @@
 from ..fork import EMPTY_OMMER_HASH
 from ..requests import compute_requests_hash
 from ..transactions import LegacyTransaction, decode_transaction
-from .requests import ExecutionRequests, encode_execution_requests
 from .types import ExecutionPayload
 
 
 def _payload_header(
     execution_payload: ExecutionPayload,
     parent_beacon_block_root: Root,
-    execution_requests: ExecutionRequests,
+    execution_requests: Tuple[Bytes, ...],
 ) -> Header:
     """
     Build the execution header implied by a payload request.
@@ -50,11 +49,9 @@ def _payload_header(
         )
     withdrawals_root = root(withdrawals_trie)
 
-    requests_hash = Hash32(
-        compute_requests_hash(
-            list(encode_execution_requests(execution_requests))
-        )
-    )
+    # The wire-form requests are hashed as opaque items; their
+    # contents play no part in the block hash.
+    requests_hash = Hash32(compute_requests_hash(list(execution_requests)))
 
     return Header(
         parent_hash=execution_payload.parent_hash,
@@ -100,7 +97,7 @@ def _payload_transaction_to_block_transaction(
 def _payload_block(
     execution_payload: ExecutionPayload,
     parent_beacon_block_root: Root,
-    execution_requests: ExecutionRequests,
+    execution_requests: Tuple[Bytes, ...],
 ) -> Block:
     """
     Convert an execution payload request into an execution-layer block.
diff --git a/src/ethereum_spec_tools/engine_server/server.py b/src/ethereum_spec_tools/engine_server/server.py
index cea950a73f5..3a7f1fc3be8 100644
--- a/src/ethereum_spec_tools/engine_server/server.py
+++ b/src/ethereum_spec_tools/engine_server/server.py
@@ -42,9 +42,6 @@
     is_valid_block_hash,
     is_valid_versioned_hashes,
 )
-from ethereum.forks.amsterdam.execution_engine.requests import (
-    decode_execution_requests,
-)
 from ethereum.forks.amsterdam.execution_engine.validation_helpers import (
     _payload_block,
 )
@@ -449,17 +446,32 @@ def new_payload_v5(self, params: List[Any]) -> Dict[str, Any]:
         )
         if not isinstance(requests_json, list):
             raise RpcError(INVALID_PARAMS, "executionRequests: expected array")
-        try:
-            execution_requests = decode_execution_requests(
-                tuple(
-                    Bytes(_decode_hex(r, "executionRequest"))
-                    for r in requests_json
+        execution_requests = tuple(
+            Bytes(_decode_hex(r, "executionRequest")) for r in requests_json
+        )
+        # Per the Engine API, only structural violations of the requests
+        # list are parameter errors: empty items, items with no request
+        # data after the type byte, and type bytes out of strictly
+        # ascending order. Any other malformed content is hashed as
+        # opaque bytes and surfaces as an INVALID payload.
+        last_type = -1
+        for item in execution_requests:
+            if len(item) == 0:
+                raise RpcError(
+                    INVALID_PARAMS, "executionRequests: empty request item"
                 )
-            )
-        except EthereumException as e:
-            # Ordering, size, and unknown-type violations are invalid
-            # params per the Engine API.
-            raise RpcError(INVALID_PARAMS, str(e)) from e
+            if len(item) == 1:
+                raise RpcError(
+                    INVALID_PARAMS,
+                    "executionRequests: request item without data",
+                )
+            if item[0] <= last_type:
+                raise RpcError(
+                    INVALID_PARAMS,
+                    "executionRequests: request types not in strictly "
+                    "ascending order",
+                )
+            last_type = item[0]
 
         request = NewPayloadRequest(
             execution_payload=payload,
diff --git a/vulture_whitelist.py b/vulture_whitelist.py
index 57fdbd4d065..6647149f83c 100644
--- a/vulture_whitelist.py
+++ b/vulture_whitelist.py
@@ -208,6 +208,7 @@
 # surface; consumed by external engine callers rather than the spec itself
 from ethereum.forks.amsterdam.execution_engine.requests import (
     decode_execution_requests,
+    encode_execution_requests,
 )
 from ethereum.forks.amsterdam.execution_engine.types import (
     BlobsBundle,
@@ -216,6 +217,7 @@
 )
 
 decode_execution_requests
+encode_execution_requests
 PayloadAttributes.suggested_fee_recipient
 BlobsBundle.commitments
 BlobsBundle.proofs