Ethereum Execution Layer Specifications (Updated up to June 13 2026)

contract Storage {
    uint256 a; // p = 0
    uint256 b; // p = 1
    mapping(uint256 => uint256) c; // p = 2
}

struct Player {
    uint256 score;      // stored at slot p
    uint128 health;     // stored at slot p + 1 (bytes 0-15)
    uint128 magic;      // stored at slot p + 1 (bytes 16-31, packed with health)
    address guild;      // stored at slot p + 2
}
Player myPlayer; // If p = 0, this struct occupies slots 0, 1, and 2.

// To find the slot for dynamicArray[i] where p is the base slot:
uint256 dataStart = uint256(keccak256(abi.encode(p)));
uint256 elementSlot = dataStart + i; 

// To find the slot for myMapping[key] where p is the base slot:
uint256 elementSlot = uint256(keccak256(abi.encode(key, p)));

from __future__ import annotations

import copy
from dataclasses import dataclass, field
from typing import (
    TYPE_CHECKING,
    Callable,
    Dict,
    Generic,
    List,
    Mapping,
    MutableMapping,
    Optional,
    Sequence,
    Tuple,
    TypeVar,
    cast,
)

from ethereum_rlp import Extended, rlp
from ethereum_types.bytes import Bytes
from ethereum_types.frozen import slotted_freezable
from ethereum_types.numeric import Uint
from typing_extensions import assert_type

from ethereum.crypto.hash import Hash32, keccak256
from ethereum.utils.hexadecimal import hex_to_bytes

if TYPE_CHECKING:
    from ethereum.state import Account, Address, Root

# note: an empty trie (regardless of whether it is secured) has root:
#   keccak256(RLP(b''))
#       ==
#   56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421 # noqa: E501
EMPTY_TRIE_ROOT = Hash32(
    hex_to_bytes(
        "56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
    )
)


@slotted_freezable
@dataclass
class LeafNode:
    rest_of_key: Bytes
    value: Extended


@slotted_freezable
@dataclass
class ExtensionNode:
    key_segment: Bytes
    subnode: Extended


BranchSubnodes = Tuple[
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
    Extended,
]


@slotted_freezable
@dataclass
class BranchNode:
    subnodes: BranchSubnodes
    value: Extended


InternalNode = LeafNode | ExtensionNode | BranchNode


K = TypeVar("K", bound=Bytes)
V = TypeVar("V", bound=Extended | None)


def encode_account(raw_account_data: Account, storage_root: Bytes) -> Bytes:
    """
    Encode `Account` dataclass.

    Storage is not stored in the `Account` dataclass, so `Accounts` cannot be
    encoded without providing a storage root.
    """
    return rlp.encode(
        (
            raw_account_data.nonce,
            raw_account_data.balance,
            storage_root,
            raw_account_data.code_hash,
        )
    )


def encode_internal_node(node: Optional[InternalNode]) -> Extended:
    """
    Encodes a Merkle Trie node into its RLP form. The RLP will then be
    serialized into a `Bytes` object and hashed unless it is less than 32 bytes
    when serialized.

    This function also accepts `None`, representing the absence of a node,
    which is encoded to `b""`.

    Parameters
    ----------
    node : Optional[InternalNode]
        The node to encode.

    Returns
    -------
    encoded : `Extended`
        The node encoded as RLP.

    """
    unencoded: Extended
    if node is None:
        unencoded = b""
    elif isinstance(node, LeafNode):
        unencoded = (
            nibble_list_to_compact(node.rest_of_key, True),
            node.value,
        )
    elif isinstance(node, ExtensionNode):
        unencoded = (
            nibble_list_to_compact(node.key_segment, False),
            node.subnode,
        )
    elif isinstance(node, BranchNode):
        unencoded = list(node.subnodes) + [node.value]
    else:
        raise AssertionError(f"Invalid internal node type {type(node)}!")

    encoded = rlp.encode(unencoded)
    if len(encoded) < 32:
        return unencoded
    else:
        return keccak256(encoded)


def encode_node(node: Extended, storage_root: Bytes | None = None) -> Bytes:
    """
    Encode a Node for storage in the Merkle Trie.
    """
    from ethereum.state import Account

    if isinstance(node, Account):
        assert storage_root is not None
        return encode_account(node, storage_root)
    elif isinstance(node, Bytes):
        return node
    else:
        return rlp.encode(node)


@dataclass
class Trie(Generic[K, V]):
    """
    The Merkle Trie.
    """

    secured: bool
    default: V
    _data: Dict[K, V] = field(default_factory=dict)


def copy_trie(trie: Trie[K, V]) -> Trie[K, V]:
    """
    Create a copy of `trie`. Since only frozen objects may be stored in tries,
    the contents are reused.

    Parameters
    ----------
    trie: `Trie`
        Trie to copy.

    Returns
    -------
    new_trie : `Trie[K, V]`
        A copy of the trie.

    """
    return Trie(trie.secured, trie.default, copy.copy(trie._data))


def trie_set(trie: Trie[K, V], key: K, value: V) -> None:
    """
    Stores an item in a Merkle Trie.

    This method deletes the key if `value == trie.default`, because the Merkle
    Trie represents the default value by omitting it from the trie.

    Parameters
    ----------
    trie: `Trie`
        Trie to store in.
    key : `Bytes`
        Key to lookup.
    value : `V`
        Node to insert at `key`.

    """
    if value == trie.default:
        if key in trie._data:
            del trie._data[key]
    else:
        trie._data[key] = value


def trie_get(trie: Trie[K, V], key: K) -> V:
    """
    Gets an item from the Merkle Trie.

    This method returns `trie.default` if the key is missing.

    Parameters
    ----------
    trie:
        Trie to lookup in.
    key :
        Key to lookup.

    Returns
    -------
    node : `V`
        Node at `key` in the trie.

    """
    return trie._data.get(key, trie.default)


def common_prefix_length(a: Sequence, b: Sequence) -> int:
    """
    Find the longest common prefix of two sequences.
    """
    for i in range(len(a)):
        if i >= len(b) or a[i] != b[i]:
            return i
    return len(a)


def nibble_list_to_compact(x: Bytes, is_leaf: bool) -> Bytes:
    """
    Compresses nibble-list into a standard byte array with a flag.

    A nibble-list is a list of byte values no greater than `15`. The flag is
    encoded in high nibble of the highest byte. The flag nibble can be broken
    down into two two-bit flags.

    Highest nibble::

        +---+---+----------+--------+
        | _ | _ | is_leaf | parity |
        +---+---+----------+--------+
          3   2      1         0


    The lowest bit of the nibble encodes the parity of the length of the
    remaining nibbles -- `0` when even and `1` when odd. The second lowest bit
    is used to distinguish leaf and extension nodes. The other two bits are not
    used.

    Parameters
    ----------
    x :
        Array of nibbles.
    is_leaf :
        True if this is part of a leaf node, or false if it is an extension
        node.

    Returns
    -------
    compressed : `bytearray`
        Compact byte array.

    """
    compact = bytearray()

    if len(x) % 2 == 0:  # ie even length
        compact.append(16 * (2 * is_leaf))
        for i in range(0, len(x), 2):
            compact.append(16 * x[i] + x[i + 1])
    else:
        compact.append(16 * ((2 * is_leaf) + 1) + x[0])
        for i in range(1, len(x), 2):
            compact.append(16 * x[i] + x[i + 1])

    return Bytes(compact)


def bytes_to_nibble_list(bytes_: Bytes) -> Bytes:
    """
    Converts a `Bytes` into to a sequence of nibbles (bytes with value < 16).

    Parameters
    ----------
    bytes_:
        The `Bytes` to convert.

    Returns
    -------
    nibble_list : `Bytes`
        The `Bytes` in nibble-list format.

    """
    nibble_list = bytearray(2 * len(bytes_))
    for byte_index, byte in enumerate(bytes_):
        nibble_list[byte_index * 2] = (byte & 0xF0) >> 4
        nibble_list[byte_index * 2 + 1] = byte & 0x0F
    return Bytes(nibble_list)


def _prepare_trie(
    trie: Trie[K, V],
    get_storage_root: Optional[Callable[[Address], Root]] = None,
) -> Mapping[Bytes, Bytes]:
    """
    Prepares the trie for root calculation. Removes values that are empty,
    hashes the keys (if `secured == True`) and encodes all the nodes.

    Parameters
    ----------
    trie :
        The `Trie` to prepare.
    get_storage_root :
        Function to get the storage root of an account. Needed to encode
        `Account` objects.

    Returns
    -------
    out : `Mapping[ethereum.base_types.Bytes, Node]`
        Object with keys mapped to nibble-byte form.

    """
    from ethereum.state import Account, Address

    mapped: MutableMapping[Bytes, Bytes] = {}

    for preimage, value in trie._data.items():
        if isinstance(value, Account):
            assert get_storage_root is not None
            address = Address(preimage)
            encoded_value = encode_node(value, get_storage_root(address))
        elif value is None:
            raise AssertionError("cannot encode `None`")
        else:
            encoded_value = encode_node(value)
        if encoded_value == b"":
            raise AssertionError
        key: Bytes
        if trie.secured:
            # "secure" tries hash keys once before construction
            key = keccak256(preimage)
        else:
            key = preimage
        mapped[bytes_to_nibble_list(key)] = encoded_value

    return mapped


def root(
    trie: Trie[K, V],
    get_storage_root: Optional[Callable[[Address], Root]] = None,
) -> Root:
    """
    Computes the root of a modified merkle patricia trie (MPT).

    Parameters
    ----------
    trie :
        `Trie` to get the root of.
    get_storage_root :
        Function to get the storage root of an account. Needed to encode
        `Account` objects.


    Returns
    -------
    root : `.state.Root`
        MPT root of the underlying key-value pairs.

    """
    from ethereum.state import Root

    obj = _prepare_trie(trie, get_storage_root)

    root_node = encode_internal_node(patricialize(obj, Uint(0)))
    if len(rlp.encode(root_node)) < 32:
        return keccak256(rlp.encode(root_node))
    else:
        assert isinstance(root_node, Bytes)
        return Root(root_node)


def patricialize(
    obj: Mapping[Bytes, Bytes], level: Uint
) -> Optional[InternalNode]:
    """
    Structural composition function.

    Used to recursively patricialize and merkleize a dictionary. Includes
    memoization of the tree structure and hashes.

    Parameters
    ----------
    obj :
        Underlying trie key-value pairs, with keys in nibble-list format.
    level :
        Current trie level.

    Returns
    -------
    node : `ethereum.base_types.Bytes`
        Root node of `obj`.

    """
    if len(obj) == 0:
        return None

    arbitrary_key = next(iter(obj))

    # if leaf node
    if len(obj) == 1:
        leaf = LeafNode(arbitrary_key[level:], obj[arbitrary_key])
        return leaf

    # prepare for extension node check by finding max j such that all keys in
    # obj have the same key[i:j]
    substring = arbitrary_key[level:]
    prefix_length = len(substring)
    for key in obj:
        prefix_length = min(
            prefix_length, common_prefix_length(substring, key[level:])
        )

        # finished searching, found another key at the current level
        if prefix_length == 0:
            break

    # if extension node
    if prefix_length > 0:
        prefix = arbitrary_key[int(level) : int(level) + prefix_length]
        return ExtensionNode(
            prefix,
            encode_internal_node(
                patricialize(obj, level + Uint(prefix_length))
            ),
        )

    branches: List[MutableMapping[Bytes, Bytes]] = []
    for _ in range(16):
        branches.append({})
    value = b""
    for key in obj:
        if len(key) == level:
            value = obj[key]
        else:
            branches[key[level]][key] = obj[key]

    subnodes = tuple(
        encode_internal_node(patricialize(branches[k], level + Uint(1)))
        for k in range(16)
    )
    return BranchNode(
        cast(BranchSubnodes, assert_type(subnodes, Tuple[Extended, ...])),
        value,
    )

"""
The `State` class is the in-memory implementation of `PreState` (interface for State).
It consists of a main account trie and storage tries for each contract.

There is a distinction between an account that does not exist and
`EMPTY_ACCOUNT`.
"""

from dataclasses import dataclass, field
from typing import AbstractSet, Dict, List, Optional, Protocol, Set, Tuple

from ethereum_types.bytes import Bytes, Bytes20, Bytes32
from ethereum_types.frozen import slotted_freezable
from ethereum_types.numeric import U256, Uint

from ethereum.crypto.hash import Hash32, keccak256
from ethereum.merkle_patricia_trie import (
    EMPTY_TRIE_ROOT,
    InternalNode,
    Trie,
    copy_trie,
    root,
    trie_get,
    trie_set,
)

Address = Bytes20
Root = Hash32

EMPTY_CODE_HASH = keccak256(b"")


@slotted_freezable
@dataclass
class Account:
    """
    State associated with an address.
    """

    nonce: Uint
    balance: U256
    code_hash: Hash32


EMPTY_ACCOUNT = Account(
    nonce=Uint(0),
    balance=U256(0),
    code_hash=EMPTY_CODE_HASH,
)


@dataclass
class BlockDiff:
    """
    State changes produced by executing a block.
    """

    account_changes: Dict[Address, Optional[Account]]
    """Per-address account diffs produced by execution."""

    storage_changes: Dict[Address, Dict[Bytes32, U256]]
    """Per-address storage diffs produced by execution."""

    code_changes: Dict[Hash32, Bytes]
    """New bytecodes (keyed by code hash) introduced by execution."""

    storage_clears: Set[Address] = field(default_factory=set)
    """
    Addresses whose pre-existing storage was wiped during block
    execution (via a pre-EIP-6780 `SELFDESTRUCT`). Their storage
    tries are dropped before [`storage_changes`][sc] is applied, so any
    post-wipe writes begin from empty storage.

    [sc]: ref:ethereum.state.BlockDiff.storage_changes
    """


@dataclass
class State:
    """
    Contains all information that is preserved between transactions.
    """

    _main_trie: Trie[Address, Optional[Account]] = field(
        default_factory=lambda: Trie(secured=True, default=None)
    )
    _storage_tries: Dict[Address, Trie[Bytes32, U256]] = field(
        default_factory=dict
    )
    _code_store: Dict[Hash32, Bytes] = field(
        default_factory=dict, compare=False
    )

    def get_code(self, code_hash: Hash32) -> Bytes:
        """
        Get the bytecode for a given code hash.

        Return ``b""`` for ``EMPTY_CODE_HASH``.
        """
        if code_hash == EMPTY_CODE_HASH:
            return b""
        return self._code_store[code_hash]

    def get_account_optional(self, address: Address) -> Optional[Account]:
        """
        Get the account at an address.

        Return ``None`` if there is no account at the address.
        """
        return trie_get(self._main_trie, address)

    def get_storage(self, address: Address, key: Bytes32) -> U256:
        """
        Get a storage value.

        Return ``U256(0)`` if the key has not been set.
        """
        trie = self._storage_tries.get(address)
        if trie is None:
            return U256(0)

        value = trie_get(trie, key)

        assert isinstance(value, U256)
        return value

    def account_has_storage(self, address: Address) -> bool:
        """
        Check whether an account has any storage.

        Only needed for EIP-7610.
        """
        return address in self._storage_tries

    def compute_state_root_and_trie_changes(
        self,
        account_changes: Dict[Address, Optional[Account]],
        storage_changes: Dict[Address, Dict[Bytes32, U256]],
        storage_clears: AbstractSet[Address] = frozenset(),
    ) -> Tuple[Root, List["InternalNode"]]:
        """
        Compute the state root after applying changes to the pre-state.

        ``storage_clears`` lists addresses whose pre-existing storage
        tries are dropped before ``storage_changes`` is applied, so any
        post-wipe writes begin from empty storage.

        Return the new state root together with the internal trie nodes
        that were created or modified.
        """
        main_trie = copy_trie(self._main_trie)
        storage_tries = {
            k: copy_trie(v)
            for k, v in self._storage_tries.items()
            if k not in storage_clears
        }

        for address, account in account_changes.items():
            trie_set(main_trie, address, account)

        for address, slots in storage_changes.items():
            trie = storage_tries.get(address)
            if trie is None:
                trie = Trie(secured=True, default=U256(0))
                storage_tries[address] = trie
            for key, value in slots.items():
                trie_set(trie, key, value)
            if trie._data == {}:
                del storage_tries[address]

        def get_storage_root(addr: Address) -> Root:
            if addr in storage_tries:
                return root(storage_tries[addr])
            return EMPTY_TRIE_ROOT

        state_root_value = root(main_trie, get_storage_root=get_storage_root)

        return state_root_value, []


def apply_changes_to_state(state: State, diff: BlockDiff) -> None:
    """
    Apply block-level diff to the ``State`` for the next block.

    Parameters
    ----------
    state :
        The state to update.
    diff :
        Account, storage, and code changes to apply.

    """
    for address in diff.storage_clears:
        state._storage_tries.pop(address, None)

    for address, account in diff.account_changes.items():
        trie_set(state._main_trie, address, account)

    for address, slots in diff.storage_changes.items():
        trie = state._storage_tries.get(address)
        if trie is None:
            trie = Trie(secured=True, default=U256(0))
            state._storage_tries[address] = trie
        for key, value in slots.items():
            trie_set(trie, key, value)
        if trie._data == {}:
            del state._storage_tries[address]

    state._code_store.update(diff.code_changes)

class LegacyTransaction:
    """
    Atomic operation performed on the block chain.
    """

    nonce: U256
    """
    A scalar value equal to the number of transactions sent by the sender.
    """

    gas_price: Uint
    """
    The price of gas for this transaction, in wei.
    """

    gas: Uint
    """
    The maximum amount of gas that can be used by this transaction.
    """

    to: Bytes0 | Address
    """
    The address of the recipient. If empty, the transaction is a contract
    creation.
    """

    value: U256
    """
    The amount of ether (in wei) to send with this transaction.
    """

    data: Bytes
    """
    The data payload of the transaction, which can be used to call functions
    on contracts or to create new contracts.
    """

    v: U256
    """
    The recovery id of the signature.
    """

    r: U256
    """
    The first part of the signature.
    """

    s: U256
    """
    The second part of the signature.
    """

class Access:
    """
    A mapping from account address to storage slots that are pre-warmed as part
    of a transaction.
    """

    account: Address
    """
    The address of the account that is accessed.
    """

    slots: Tuple[Bytes32, ...]
    """
    A tuple of storage slots that are accessed in the account.
    """

class AccessListTransaction:
    """
    The transaction type added in [EIP-2930] to support access lists.

    This transaction type extends the legacy transaction with an access list
    and chain ID. The access list specifies which addresses and storage slots
    the transaction will access.

    [EIP-2930]: https://eips.ethereum.org/EIPS/eip-2930
    """

    chain_id: U64
    """
    The ID of the chain on which this transaction is executed.
    """

    nonce: U256
    """
    A scalar value equal to the number of transactions sent by the sender.
    """

    gas_price: Uint
    """
    The price of gas for this transaction.
    """

    gas: Uint
    """
    The maximum amount of gas that can be used by this transaction.
    """

    to: Bytes0 | Address
    """
    The address of the recipient. If empty, the transaction is a contract
    creation.
    """

    value: U256
    """
    The amount of ether (in wei) to send with this transaction.
    """

    data: Bytes
    """
    The data payload of the transaction, which can be used to call functions
    on contracts or to create new contracts.
    """

    access_list: Tuple[Access, ...]
    """
    A tuple of `Access` objects that specify which addresses and storage slots
    are accessed in the transaction.
    """

    y_parity: U256
    """
    The recovery id of the signature.
    """

    r: U256
    """
    The first part of the signature.
    """

    s: U256
    """
    The second part of the signature.
    """

class FeeMarketTransaction:
    """
    The transaction type added in [EIP-1559].

    This transaction type introduces a new fee market mechanism with two gas
    price parameters: max_priority_fee_per_gas and max_fee_per_gas.

    [EIP-1559]: https://eips.ethereum.org/EIPS/eip-1559
    """

    chain_id: U64
    """
    The ID of the chain on which this transaction is executed.
    """

    nonce: U256
    """
    A scalar value equal to the number of transactions sent by the sender.
    """

    max_priority_fee_per_gas: Uint
    """
    The maximum priority fee per gas that the sender is willing to pay.
    """

    max_fee_per_gas: Uint
    """
    The maximum fee per gas that the sender is willing to pay, including the
    base fee and priority fee.
    """

    gas: Uint
    """
    The maximum amount of gas that can be used by this transaction.
    """

    to: Bytes0 | Address
    """
    The address of the recipient. If empty, the transaction is a contract
    creation.
    """

    value: U256
    """
    The amount of ether (in wei) to send with this transaction.
    """

    data: Bytes
    """
    The data payload of the transaction, which can be used to call functions
    on contracts or to create new contracts.
    """

    access_list: Tuple[Access, ...]
    """
    A tuple of `Access` objects that specify which addresses and storage slots
    are accessed in the transaction.
    """

    y_parity: U256
    """
    The recovery id of the signature.
    """

    r: U256
    """
    The first part of the signature.
    """

    s: U256
    """
    The second part of the signature.
    """

ELASTICITY_MULTIPLIER=2 # Means that the target is half of the block gas limit
BASE_FEE_MAX_CHANGE_DENOMINATOR=8 # Means that the change step is 12.5 percent


def calculate_base_fee_per_gas(
    block_gas_limit: Uint,
    parent_gas_limit: Uint,
    parent_gas_used: Uint,
    parent_base_fee_per_gas: Uint,
) -> Uint:
    """
    Calculates the base fee per gas for the block.

    Parameters
    ----------
    block_gas_limit :
        Gas limit of the block for which the base fee is being calculated.
    parent_gas_limit :
        Gas limit of the parent block.
    parent_gas_used :
        Gas used in the parent block.
    parent_base_fee_per_gas :
        Base fee per gas of the parent block.

    Returns
    -------
    base_fee_per_gas : `Uint`
        Base fee per gas for the block.

    """
    parent_gas_target = parent_gas_limit // ELASTICITY_MULTIPLIER
    if not check_gas_limit(block_gas_limit, parent_gas_limit):
        raise InvalidBlock

    if parent_gas_used == parent_gas_target:
        expected_base_fee_per_gas = parent_base_fee_per_gas
    elif parent_gas_used > parent_gas_target:
        gas_used_delta = parent_gas_used - parent_gas_target

        parent_fee_gas_delta = parent_base_fee_per_gas * gas_used_delta
        target_fee_gas_delta = parent_fee_gas_delta // parent_gas_target

        base_fee_per_gas_delta = max(
            target_fee_gas_delta // BASE_FEE_MAX_CHANGE_DENOMINATOR,
            Uint(1),
        )

        expected_base_fee_per_gas = (
            parent_base_fee_per_gas + base_fee_per_gas_delta
        )
    else:
        gas_used_delta = parent_gas_target - parent_gas_used

        parent_fee_gas_delta = parent_base_fee_per_gas * gas_used_delta
        target_fee_gas_delta = parent_fee_gas_delta // parent_gas_target

        base_fee_per_gas_delta = (
            target_fee_gas_delta // BASE_FEE_MAX_CHANGE_DENOMINATOR
        )

        expected_base_fee_per_gas = (
            parent_base_fee_per_gas - base_fee_per_gas_delta
        )

    return Uint(expected_base_fee_per_gas)

# GasCosts.LIMIT_ADJUSTMENT_FACTOR = 1024
# GasCosts.LIMIT_MINIMUM = 5000

def check_gas_limit(gas_limit: Uint, parent_gas_limit: Uint) -> bool:  
    """
    Validates the gas limit for a block.

    The bounds of the gas limit, ``max_adjustment_delta``, is set as the
    quotient of the parent block's gas limit and the``LIMIT_ADJUSTMENT_FACTOR``. 

    Parameters
    ----------
    gas_limit :
        Gas limit to validate.

    parent_gas_limit :
        Gas limit of the parent block.

    Returns
    -------
    check : `bool`
        True if gas limit constraints are satisfied, False otherwise.

    """
    max_adjustment_delta = parent_gas_limit // GasCosts.LIMIT_ADJUSTMENT_FACTOR
    if gas_limit >= parent_gas_limit + max_adjustment_delta:
        return False
    if gas_limit <= parent_gas_limit - max_adjustment_delta:
        return False
    if gas_limit < GasCosts.LIMIT_MINIMUM:
        return False

    return True

class BlobTransaction:
    """
    The transaction type added in [EIP-4844].

    This transaction type extends the fee market transaction to support
    blob-carrying transactions.

    [EIP-4844]: https://eips.ethereum.org/EIPS/eip-4844
    """

    chain_id: U64
    """
    The ID of the chain on which this transaction is executed.
    """

    nonce: U256
    """
    A scalar value equal to the number of transactions sent by the sender.
    """

    max_priority_fee_per_gas: Uint
    """
    The maximum priority fee per gas that the sender is willing to pay.
    """

    max_fee_per_gas: Uint
    """
    The maximum fee per gas that the sender is willing to pay, including the
    base fee and priority fee.
    """

    gas: Uint
    """
    The maximum amount of gas that can be used by this transaction.
    """

    to: Address
    """
    The address of the recipient. If empty, the transaction is a contract
    creation.
    """

    value: U256
    """
    The amount of ether (in wei) to send with this transaction.
    """

    data: Bytes
    """
    The data payload of the transaction, which can be used to call functions
    on contracts or to create new contracts.
    """

    access_list: Tuple[Access, ...]
    """
    A tuple of `Access` objects that specify which addresses and storage slots
    are accessed in the transaction.
    """

    max_fee_per_blob_gas: U256
    """
    The maximum fee per blob gas that the sender is willing to pay.
    """

    blob_versioned_hashes: Tuple[VersionedHash, ...]
    """
    A tuple of objects that represent the versioned hashes of the blobs
    included in the transaction.
    """

    y_parity: U256
    """
    The recovery id of the signature.
    """

    r: U256
    """
    The first part of the signature.
    """

    s: U256
    """
    The second part of the signature.
    """

PER_BLOB = 2**17
BLOB_SCHEDULE_TARGET = 14
BLOB_TARGET_GAS_PER_BLOCK = PER_BLOB * BLOB_SCHEDULE_TARGET
BLOB_BASE_COST = 2**13
BLOB_SCHEDULE_MAX = 21
BLOB_MIN_GASPRICE = 1
BLOB_BASE_FEE_UPDATE_FRACTION = 11684671

def calculate_excess_blob_gas(
    parent_header: Header | PreviousHeader,
) -> U64:
    """
    Calculates the excess blob gas for the current block based
    on the gas used in the parent block.

    Parameters
    ----------
    parent_header :
        The parent block of the current block.

    Returns
    -------
    excess_blob_gas: `ethereum.base_types.U64`
        The excess blob gas for the current block.

    """
    excess_blob_gas = parent_header.excess_blob_gas
    blob_gas_used = parent_header.blob_gas_used
    base_fee_per_gas = parent_header.base_fee_per_gas

    parent_blob_gas = excess_blob_gas + blob_gas_used
    if parent_blob_gas < GasCosts.BLOB_TARGET_GAS_PER_BLOCK:
        return U64(0)

    target_blob_gas_price = Uint(GasCosts.PER_BLOB)
    target_blob_gas_price *= calculate_blob_gas_price(excess_blob_gas)

    base_blob_tx_price = GasCosts.BLOB_BASE_COST * base_fee_per_gas
    if base_blob_tx_price > target_blob_gas_price:
        blob_schedule_delta = (
            GasCosts.BLOB_SCHEDULE_MAX - GasCosts.BLOB_SCHEDULE_TARGET
        )
        return (
            excess_blob_gas
            + blob_gas_used * blob_schedule_delta // GasCosts.BLOB_SCHEDULE_MAX
        )

    return parent_blob_gas - GasCosts.BLOB_TARGET_GAS_PER_BLOCK

def calculate_blob_gas_price(excess_blob_gas: U64) -> Uint:
    """
    Calculate the blob gasprice for a block.

    Parameters
    ----------
    excess_blob_gas :
        The excess blob gas for the block.

    Returns
    -------
    blob_gasprice: `Uint`
        The blob gasprice.

    """
    return taylor_exponential(
        GasCosts.BLOB_MIN_GASPRICE,
        Uint(excess_blob_gas),
        GasCosts.BLOB_BASE_FEE_UPDATE_FRACTION,
    )

class Authorization:
    """
    The authorization for a set code transaction.
    """

    chain_id: U256
    address: Address
    nonce: U64
    y_parity: U8
    r: U256
    s: U256

class SetCodeTransaction:
    """
    The transaction type added in [EIP-7702].

    This transaction type allows Ethereum Externally Owned Accounts (EOAs)
    to set code on their account, enabling them to act as smart contracts.

    [EIP-7702]: https://eips.ethereum.org/EIPS/eip-7702
    """

    chain_id: U64
    """
    The ID of the chain on which this transaction is executed.
    """

    nonce: U64
    """
    A scalar value equal to the number of transactions sent by the sender.
    """

    max_priority_fee_per_gas: Uint
    """
    The maximum priority fee per gas that the sender is willing to pay.
    """

    max_fee_per_gas: Uint
    """
    The maximum fee per gas that the sender is willing to pay, including the
    base fee and priority fee.
    """

    gas: Uint
    """
    The maximum amount of gas that can be used by this transaction.
    """

    to: Address
    """
    The address of the recipient. If empty, the transaction is a contract
    creation.
    """

    value: U256
    """
    The amount of ether (in wei) to send with this transaction.
    """

    data: Bytes
    """
    The data payload of the transaction, which can be used to call functions
    on contracts or to create new contracts.
    """

    access_list: Tuple[Access, ...]
    """
    A tuple of `Access` objects that specify which addresses and storage slots
    are accessed in the transaction.
    """

    authorizations: Tuple[Authorization, ...]
    """
    A tuple of `Authorization` objects that specify what code the signer
    desires to execute in the context of their EOA.
    """

    y_parity: U256
    """
    The recovery id of the signature.
    """

    r: U256
    """
    The first part of the signature.
    """

    s: U256
    """
    The second part of the signature.
    """

SET_CODE_TX_MAGIC = b"\x05"
EOA_DELEGATION_MARKER = b"\xef\x01\x00"
EOA_DELEGATION_MARKER_LENGTH = len(EOA_DELEGATION_MARKER)
EOA_DELEGATED_CODE_LENGTH = 23
REFUND_AUTH_PER_EXISTING_ACCOUNT = 12500
NULL_ADDRESS = hex_to_address("0x0000000000000000000000000000000000000000")


def is_valid_delegation(code: bytes) -> bool:
    """
    Whether the code is a valid delegation designation.

    Parameters
    ----------
    code: `bytes`
        The code to check.

    Returns
    -------
    valid : `bool`
        True if the code is a valid delegation designation,
        False otherwise.

    """
    if (
        len(code) == EOA_DELEGATED_CODE_LENGTH
        and code[:EOA_DELEGATION_MARKER_LENGTH] == EOA_DELEGATION_MARKER
    ):
        return True
    return False


def set_delegation(message: Message) -> U256:
    """
    Set the delegation code for the authorities in the message.

    Parameters
    ----------
    message :
        Transaction specific items.

    Returns
    -------
    refund_counter: `U256`
        Refund from authority which already exists in state.

    """
    tx_state = message.tx_env.state
    refund_counter = U256(0)
    for auth in message.tx_env.authorizations:
        if auth.chain_id not in (message.block_env.chain_id, U256(0)):
            continue

        if auth.nonce >= U64.MAX_VALUE:
            continue

        try:
            authority = recover_authority(auth)
        except InvalidSignatureError:
            continue

        message.accessed_addresses.add(authority)

        authority_account = get_account(tx_state, authority)
        authority_code = get_code(tx_state, authority_account.code_hash)

        if authority_code and not is_valid_delegation(authority_code):
            continue

        authority_nonce = authority_account.nonce
        if authority_nonce != auth.nonce:
            continue

        if account_exists(tx_state, authority):
            refund_counter += U256(
                GasCosts.AUTH_PER_EMPTY_ACCOUNT
                - REFUND_AUTH_PER_EXISTING_ACCOUNT
            )

        if auth.address == NULL_ADDRESS:
            code_to_set = b""
        else:
            code_to_set = EOA_DELEGATION_MARKER + auth.address

        set_code(tx_state, authority, code_to_set)
        increment_nonce(tx_state, authority)

    if message.code_address is None:
        raise InvalidBlock("Invalid type 4 transaction: no target")

    message.code = get_code(
        tx_state,
        get_account(tx_state, message.code_address).code_hash,
    )

    return refund_counter

"""
State Tracking for Block Execution.

Track state changes on top of a read-only ``PreState``.  At block end,
accumulated diffs feed into
``PreState.compute_state_root_and_trie_changes()``.

.. contents:: Table of Contents
    :backlinks: none
    :local:

Introduction
------------

Replace the mutable ``State`` class with lightweight state trackers that
record diffs.  ``BlockState`` accumulates committed transaction
changes across a block.  ``TransactionState`` tracks in-flight changes
within a single transaction and supports copy-on-write rollback.
"""

from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Callable, Dict, Optional, Set, Tuple

from ethereum_types.bytes import Bytes, Bytes32
from ethereum_types.frozen import modify
from ethereum_types.numeric import U256, Uint

from ethereum.crypto.hash import Hash32, keccak256
from ethereum.state import (
    EMPTY_ACCOUNT,
    EMPTY_CODE_HASH,
    Account,
    Address,
    BlockDiff,
    PreState,
)

if TYPE_CHECKING:
    from .block_access_lists import BlockAccessListBuilder


@dataclass
class BlockState:
    """
    Accumulate committed transaction-level changes across a block.

    Read chain: block writes -> pre_state.

    ``account_reads`` and ``storage_reads`` accumulate across all
    transactions for BAL generation.
    """

    pre_state: PreState
    account_reads: Set[Address] = field(default_factory=set)
    account_writes: Dict[Address, Optional[Account]] = field(
        default_factory=dict
    )
    storage_reads: Set[Tuple[Address, Bytes32]] = field(default_factory=set)
    storage_writes: Dict[Address, Dict[Bytes32, U256]] = field(
        default_factory=dict
    )
    code_writes: Dict[Hash32, Bytes] = field(default_factory=dict)


@dataclass
class TransactionState:
    """
    Track in-flight state changes within a single transaction.

    Read chain: tx writes -> block writes -> pre_state.

    ``storage_reads`` and ``account_reads`` are shared references
    that survive rollback (reads from failed calls still appear in the
    Block Access List).
    """

    parent: BlockState
    account_reads: Set[Address] = field(default_factory=set)
    account_writes: Dict[Address, Optional[Account]] = field(
        default_factory=dict
    )
    storage_reads: Set[Tuple[Address, Bytes32]] = field(default_factory=set)
    storage_writes: Dict[Address, Dict[Bytes32, U256]] = field(
        default_factory=dict
    )
    code_writes: Dict[Hash32, Bytes] = field(default_factory=dict)
    created_accounts: Set[Address] = field(default_factory=set)
    transient_storage: Dict[Tuple[Address, Bytes32], U256] = field(
        default_factory=dict
    )


def get_account_optional(
    tx_state: TransactionState, address: Address
) -> Optional[Account]:
    """
    Get the ``Account`` object at an address. Return ``None`` (rather than
    ``EMPTY_ACCOUNT``) if there is no account at the address.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address to look up.

    Returns
    -------
    account : ``Optional[Account]``
        Account at address.

    """
    tx_state.account_reads.add(address)
    if address in tx_state.account_writes:
        return tx_state.account_writes[address]
    if address in tx_state.parent.account_writes:
        return tx_state.parent.account_writes[address]
    return tx_state.parent.pre_state.get_account_optional(address)


def get_account(tx_state: TransactionState, address: Address) -> Account:
    """
    Get the ``Account`` object at an address. Return ``EMPTY_ACCOUNT``
    if there is no account at the address.

    Use ``get_account_optional()`` if you care about the difference
    between a non-existent account and ``EMPTY_ACCOUNT``.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address to look up.

    Returns
    -------
    account : ``Account``
        Account at address.

    """
    account = get_account_optional(tx_state, address)
    if account is None:
        return EMPTY_ACCOUNT
    else:
        return account


def get_code(tx_state: TransactionState, code_hash: Hash32) -> Bytes:
    """
    Get the bytecode for a given code hash.

    Read chain: tx code_writes -> block code_writes -> pre_state.

    Parameters
    ----------
    tx_state :
        The transaction state.
    code_hash :
        Hash of the code to look up.

    Returns
    -------
    code : ``Bytes``
        The bytecode.

    """
    if code_hash == EMPTY_CODE_HASH:
        return b""
    if code_hash in tx_state.code_writes:
        return tx_state.code_writes[code_hash]
    if code_hash in tx_state.parent.code_writes:
        return tx_state.parent.code_writes[code_hash]
    return tx_state.parent.pre_state.get_code(code_hash)


def get_storage(
    tx_state: TransactionState, address: Address, key: Bytes32
) -> U256:
    """
    Get a value at a storage key on an account. Return ``U256(0)`` if
    the storage key has not been set previously.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account.
    key :
        Key to look up.

    Returns
    -------
    value : ``U256``
        Value at the key.

    """
    tx_state.storage_reads.add((address, key))
    if address in tx_state.storage_writes:
        if key in tx_state.storage_writes[address]:
            return tx_state.storage_writes[address][key]
    if address in tx_state.parent.storage_writes:
        if key in tx_state.parent.storage_writes[address]:
            return tx_state.parent.storage_writes[address][key]
    return tx_state.parent.pre_state.get_storage(address, key)


def get_storage_original(
    tx_state: TransactionState, address: Address, key: Bytes32
) -> U256:
    """
    Get the original value in a storage slot i.e. the value before the
    current transaction began. Read from block-level writes, then
    pre_state. Return ``U256(0)`` for accounts created in the current
    transaction.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account to read the value from.
    key :
        Key of the storage slot.

    """
    if address in tx_state.created_accounts:
        return U256(0)
    if address in tx_state.parent.storage_writes:
        if key in tx_state.parent.storage_writes[address]:
            return tx_state.parent.storage_writes[address][key]
    return tx_state.parent.pre_state.get_storage(address, key)


def get_transient_storage(
    tx_state: TransactionState, address: Address, key: Bytes32
) -> U256:
    """
    Get a value at a storage key on an account from transient storage.
    Return ``U256(0)`` if the storage key has not been set previously.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account.
    key :
        Key to look up.

    Returns
    -------
    value : ``U256``
        Value at the key.

    """
    return tx_state.transient_storage.get((address, key), U256(0))


def account_exists(tx_state: TransactionState, address: Address) -> bool:
    """
    Check if an account exists in the state trie.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account that needs to be checked.

    Returns
    -------
    account_exists : ``bool``
        True if account exists in the state trie, False otherwise.

    """
    return get_account_optional(tx_state, address) is not None


def account_has_code_or_nonce(
    tx_state: TransactionState, address: Address
) -> bool:
    """
    Check if an account has non-zero nonce or non-empty code.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account that needs to be checked.

    Returns
    -------
    has_code_or_nonce : ``bool``
        True if the account has non-zero nonce or non-empty code,
        False otherwise.

    """
    account = get_account(tx_state, address)
    return account.nonce != Uint(0) or account.code_hash != EMPTY_CODE_HASH


def account_has_storage(tx_state: TransactionState, address: Address) -> bool:
    """
    Check if an account has storage.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account that needs to be checked.

    Returns
    -------
    has_storage : ``bool``
        True if the account has storage, False otherwise.

    """
    if tx_state.storage_writes.get(address):
        return True
    if tx_state.parent.storage_writes.get(address):
        return True
    return tx_state.parent.pre_state.account_has_storage(address)


def account_exists_and_is_empty(
    tx_state: TransactionState, address: Address
) -> bool:
    """
    Check if an account exists and has zero nonce, empty code and zero
    balance.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account that needs to be checked.

    Returns
    -------
    exists_and_is_empty : ``bool``
        True if an account exists and has zero nonce, empty code and
        zero balance, False otherwise.

    """
    account = get_account_optional(tx_state, address)
    return (
        account is not None
        and account.nonce == Uint(0)
        and account.code_hash == EMPTY_CODE_HASH
        and account.balance == 0
    )


def is_account_alive(tx_state: TransactionState, address: Address) -> bool:
    """
    Check whether an account is both in the state and non-empty.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account that needs to be checked.

    Returns
    -------
    is_alive : ``bool``
        True if the account is alive.

    """
    account = get_account_optional(tx_state, address)
    return account is not None and account != EMPTY_ACCOUNT


def set_account(
    tx_state: TransactionState,
    address: Address,
    account: Optional[Account],
) -> None:
    """
    Set the ``Account`` object at an address. Setting to ``None``
    deletes the account (but not its storage, see
    ``destroy_account()``).

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address to set.
    account :
        Account to set at address.

    """
    tx_state.account_writes[address] = account


def set_storage(
    tx_state: TransactionState,
    address: Address,
    key: Bytes32,
    value: U256,
) -> None:
    """
    Set a value at a storage key on an account.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account.
    key :
        Key to set.
    value :
        Value to set at the key.

    """
    assert get_account_optional(tx_state, address) is not None
    if address not in tx_state.storage_writes:
        tx_state.storage_writes[address] = {}
    tx_state.storage_writes[address][key] = value


def destroy_account(tx_state: TransactionState, address: Address) -> None:
    """
    Completely remove the account at ``address`` and all of its storage.

    This function is made available exclusively for the ``SELFDESTRUCT``
    opcode. It is expected that ``SELFDESTRUCT`` will be disabled in a
    future hardfork and this function will be removed. Only supports same
    transaction destruction.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of account to destroy.

    """
    destroy_storage(tx_state, address)
    set_account(tx_state, address, None)


def destroy_storage(tx_state: TransactionState, address: Address) -> None:
    """
    Completely remove the storage at ``address``.

    Convert storage writes to reads before deleting so that accesses
    from created-then-destroyed accounts appear in the Block Access
    List. Only supports same transaction destruction.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of account whose storage is to be deleted.

    """
    if address in tx_state.storage_writes:
        for key in tx_state.storage_writes[address]:
            tx_state.storage_reads.add((address, key))
        del tx_state.storage_writes[address]


def mark_account_created(tx_state: TransactionState, address: Address) -> None:
    """
    Mark an account as having been created in the current transaction.
    This information is used by ``get_storage_original()`` to handle an
    obscure edgecase, and to respect the constraints added to
    SELFDESTRUCT by EIP-6780.

    The marker is not removed even if the account creation reverts.
    Since the account cannot have had code prior to its creation and
    can't call ``get_storage_original()``, this is harmless.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account that has been created.

    """
    tx_state.created_accounts.add(address)


def set_transient_storage(
    tx_state: TransactionState,
    address: Address,
    key: Bytes32,
    value: U256,
) -> None:
    """
    Set a value at a storage key on an account in transient storage.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account.
    key :
        Key to set.
    value :
        Value to set at the key.

    """
    if value == U256(0):
        tx_state.transient_storage.pop((address, key), None)
    else:
        tx_state.transient_storage[(address, key)] = value


def modify_state(
    tx_state: TransactionState,
    address: Address,
    f: Callable[[Account], None],
) -> None:
    """
    Modify an ``Account`` in the state. If, after modification, the
    account exists and has zero nonce, empty code, and zero balance, it
    is destroyed.
    """
    set_account(tx_state, address, modify(get_account(tx_state, address), f))
    if account_exists_and_is_empty(tx_state, address):
        destroy_account(tx_state, address)


def move_ether(
    tx_state: TransactionState,
    sender_address: Address,
    recipient_address: Address,
    amount: U256,
) -> None:
    """
    Move funds between accounts.

    Parameters
    ----------
    tx_state :
        The transaction state.
    sender_address :
        Address of the sender.
    recipient_address :
        Address of the recipient.
    amount :
        The amount to transfer.

    """

    def reduce_sender_balance(sender: Account) -> None:
        if sender.balance < amount:
            raise AssertionError
        sender.balance -= amount

    def increase_recipient_balance(recipient: Account) -> None:
        recipient.balance += amount

    modify_state(tx_state, sender_address, reduce_sender_balance)
    modify_state(tx_state, recipient_address, increase_recipient_balance)


def create_ether(
    tx_state: TransactionState, address: Address, amount: U256
) -> None:
    """
    Add newly created ether to an account.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account to which ether is added.
    amount :
        The amount of ether to be added to the account of interest.

    """

    def increase_balance(account: Account) -> None:
        account.balance += amount

    modify_state(tx_state, address, increase_balance)


def set_account_balance(
    tx_state: TransactionState, address: Address, amount: U256
) -> None:
    """
    Set the balance of an account.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account whose balance needs to be set.
    amount :
        The amount that needs to be set in the balance.

    """

    def set_balance(account: Account) -> None:
        account.balance = amount

    modify_state(tx_state, address, set_balance)


def increment_nonce(tx_state: TransactionState, address: Address) -> None:
    """
    Increment the nonce of an account.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account whose nonce needs to be incremented.

    """

    def increase_nonce(sender: Account) -> None:
        sender.nonce += Uint(1)

    modify_state(tx_state, address, increase_nonce)


def set_code(
    tx_state: TransactionState, address: Address, code: Bytes
) -> None:
    """
    Set Account code.

    Parameters
    ----------
    tx_state :
        The transaction state.
    address :
        Address of the account whose code needs to be updated.
    code :
        The bytecode that needs to be set.

    """
    code_hash = keccak256(code)
    if code_hash != EMPTY_CODE_HASH:
        tx_state.code_writes[code_hash] = code

    def write_code_hash(sender: Account) -> None:
        sender.code_hash = code_hash

    modify_state(tx_state, address, write_code_hash)


# -- Snapshot / Rollback ---------------------------------------------------


def copy_tx_state(tx_state: TransactionState) -> TransactionState:
    """
    Create a snapshot of the transaction state for rollback.

    Deep-copy writes and transient storage.  The parent reference,
    ``created_accounts``, ``storage_reads``, and ``account_reads``
    are shared (not rolled back).

    Parameters
    ----------
    tx_state :
        The transaction state to snapshot.

    Returns
    -------
    snapshot : ``TransactionState``
        A copy of the transaction state.

    """
    return TransactionState(
        parent=tx_state.parent,
        account_writes=dict(tx_state.account_writes),
        storage_writes={
            addr: dict(slots)
            for addr, slots in tx_state.storage_writes.items()
        },
        code_writes=dict(tx_state.code_writes),
        created_accounts=tx_state.created_accounts,
        transient_storage=dict(tx_state.transient_storage),
        storage_reads=tx_state.storage_reads,
        account_reads=tx_state.account_reads,
    )


def restore_tx_state(
    tx_state: TransactionState, snapshot: TransactionState
) -> None:
    """
    Restore transaction state from a snapshot (rollback on failure).

    Parameters
    ----------
    tx_state :
        The transaction state to restore.
    snapshot :
        The snapshot to restore from.

    """
    tx_state.account_writes = snapshot.account_writes
    tx_state.storage_writes = snapshot.storage_writes
    tx_state.code_writes = snapshot.code_writes
    tx_state.transient_storage = snapshot.transient_storage


# -- Lifecycle --------------------------------------------------------------


def incorporate_tx_into_block(
    tx_state: TransactionState,
    builder: "BlockAccessListBuilder",
) -> None:
    """
    Merge transaction writes into the block state and clear for reuse.

    Update the BAL builder incrementally by diffing this transaction's
    writes against the block's cumulative state.  Merge reads and
    touches into block-level sets.

    Parameters
    ----------
    tx_state :
        The transaction state to commit.
    builder :
        The BAL builder for incremental updates.

    """
    from .block_access_lists import update_builder_from_tx

    block = tx_state.parent

    # Update BAL builder before merging writes into block state
    update_builder_from_tx(builder, tx_state)

    # Merge reads and touches into block-level sets
    block.storage_reads.update(tx_state.storage_reads)
    block.account_reads.update(tx_state.account_reads)

    # Merge cumulative writes
    for address, account in tx_state.account_writes.items():
        block.account_writes[address] = account

    for address, slots in tx_state.storage_writes.items():
        if address not in block.storage_writes:
            block.storage_writes[address] = {}
        block.storage_writes[address].update(slots)

    block.code_writes.update(tx_state.code_writes)

    tx_state.account_writes.clear()
    tx_state.storage_writes.clear()
    tx_state.code_writes.clear()
    tx_state.created_accounts.clear()
    tx_state.transient_storage.clear()
    tx_state.storage_reads = set()
    tx_state.account_reads = set()


def extract_block_diff(block_state: BlockState) -> BlockDiff:
    """
    Extract account, storage, and code diff from the block state.

    Parameters
    ----------
    block_state :
        The block state.

    Returns
    -------
    diff : `BlockDiff`
        Account, storage, and code changes accumulated during block execution.

    """
    return BlockDiff(
        account_changes=block_state.account_writes,
        storage_changes=block_state.storage_writes,
        code_changes=block_state.code_writes,
    )

"""
Block access lists (BALs), originally defined in [EIP-7928], record all
accounts and storage locations accessed during block execution along with their
post-execution values.

BALs enable parallel disk reads, parallel transaction validation, parallel
state root computation, and applying state updates without executing bytecode.

See [`BlockAccessList`][bal] for more detail.

[EIP-7928]: https://eips.ethereum.org/EIPS/eip-7928
[bal]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
"""

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Tuple, TypeAlias

from ethereum_rlp import rlp
from ethereum_types.bytes import Bytes, Bytes32
from ethereum_types.frozen import slotted_freezable
from ethereum_types.numeric import U64, U256, Uint, ulen

from ethereum.crypto.hash import Hash32, keccak256
from ethereum.state import EMPTY_CODE_HASH, Account, Address, PreState

from .exceptions import BlockAccessListGasLimitExceededError
from .fork_types import BlockAccessIndex
from .state_tracker import BlockState, TransactionState, get_code


@slotted_freezable
@dataclass
class StorageChange:
    """
    In a [`SlotChanges`][s], represents a single change in an [`Account`]'s
    storage slot.

    [s]: ref:ethereum.forks.amsterdam.block_access_lists.SlotChanges
    [`Account`]: ref:ethereum.state.Account
    """

    block_access_index: BlockAccessIndex
    """
    Position within the set of all changes in a [`Block`].

    [`Block`]: ref:ethereum.forks.amsterdam.blocks.Block
    """

    new_value: U256
    """
    Value of an [`Account`]'s storage slot after this change has been applied.

    [`Account`]: ref:ethereum.state.Account
    """


@slotted_freezable
@dataclass
class BalanceChange:
    """
    In a [`BlockAccessList`][bal], represents a change in an [`Account`]'s
    balance.

    [bal]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    [`Account`]: ref:ethereum.state.Account
    """  # noqa: E501

    block_access_index: BlockAccessIndex
    """
    Position within the set of all changes in a [`Block`].

    [`Block`]: ref:ethereum.forks.amsterdam.blocks.Block
    """

    post_balance: U256
    """
    Balance of an [`Account`] after this change has been applied.

    [`Account`]: ref:ethereum.state.Account
    """


@slotted_freezable
@dataclass
class NonceChange:
    """
    In a [`BlockAccessList`][bal], represents a change in an [`Account`]'s
    nonce.

    [bal]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    [`Account`]: ref:ethereum.state.Account
    """  # noqa: E501

    block_access_index: BlockAccessIndex
    """
    Position within the set of all changes in a [`Block`].

    [`Block`]: ref:ethereum.forks.amsterdam.blocks.Block
    """

    new_nonce: U64
    """
    Nonce of an [`Account`] after this change has been applied.

    [`Account`]: ref:ethereum.state.Account
    """


@slotted_freezable
@dataclass
class CodeChange:
    """
    In a [`BlockAccessList`][bal], represents a change in an [`Account`]'s
    code.

    [bal]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    [`Account`]: ref:ethereum.state.Account
    """  # noqa: E501

    block_access_index: BlockAccessIndex
    """
    Position within the set of all changes in a [`Block`].

    [`Block`]: ref:ethereum.forks.amsterdam.blocks.Block
    """

    new_code: Bytes
    """
    Code of an [`Account`] after this change has been applied.

    [`Account`]: ref:ethereum.state.Account
    """


@slotted_freezable
@dataclass
class SlotChanges:
    """
    In a [`BlockAccessList`][bal], represents a change in an [`Account`]'s
    storage.

    [bal]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    [`Account`]: ref:ethereum.state.Account
    """  # noqa: E501

    slot: U256
    """
    Location within an [`Account`]'s storage that has been modified.

    [`Account`]: ref:ethereum.state.Account
    """

    changes: Tuple[StorageChange, ...]
    """
    Sequence of changes that have been made to one particular storage slot.
    """


@slotted_freezable
@dataclass
class AccountChanges:
    """
    All changes for a single [`Account`], grouped by field type.

    [`Account`]: ref:ethereum.state.Account
    """

    address: Address
    """
    Address of the account containing these changes.
    """

    storage_changes: Tuple[SlotChanges, ...]
    """
    Writes to the storage of the associated [`Account`].

    [`Account`]: ref:ethereum.state.Account
    """

    storage_reads: Tuple[U256, ...]
    """
    Storage slots of the associated [`Account`] that have been read but not
    changed.

    [`Account`]: ref:ethereum.state.Account
    """

    balance_changes: Tuple[BalanceChange, ...]
    """
    Writes to the balance of the associated [`Account`].

    [`Account`]: ref:ethereum.state.Account
    """

    nonce_changes: Tuple[NonceChange, ...]
    """
    Writes to the nonce of the associated [`Account`].

    [`Account`]: ref:ethereum.state.Account
    """

    code_changes: Tuple[CodeChange, ...]
    """
    Writes to the code of the associated [`Account`].

    [`Account`]: ref:ethereum.state.Account
    """


BlockAccessList: TypeAlias = List[AccountChanges]
"""
List of state changes recorded across a [`Block`].

The hash of a block's access list is included in its [`Header`], though the
access list itself is not included in the block body.

A `BlockAccessList` includes, for example, the targets of:

- [`BALANCE`], [`EXTCODESIZE`][ecs], [`EXTCODECOPY`][ecc],
  and [`EXTCODEHASH`][ech] instructions;
- the [call family][call] of instructions _even if they revert_;
- the [create family][create] of instructions if the target is accessed;
- etc.

[`Block`]: ref:ethereum.forks.amsterdam.blocks.Block
[`Header`]: ref:ethereum.forks.amsterdam.blocks.Header
[`BALANCE`]: ref:ethereum.forks.amsterdam.vm.instructions.environment.balance
[ecs]: ref:ethereum.forks.amsterdam.vm.instructions.environment.extcodesize
[ecc]: ref:ethereum.forks.amsterdam.vm.instructions.environment.extcodecopy
[ech]: ref:ethereum.forks.amsterdam.vm.instructions.environment.extcodehash
[call]: ref:ethereum.forks.amsterdam.vm.instructions.system.call
[create]: ref:ethereum.forks.amsterdam.vm.instructions.system.create
"""


@dataclass
class AccountData:
    """
    Account data stored in the builder during block execution.

    This dataclass tracks all changes made to a single account throughout
    the execution of a block, organized by the type of change and the
    transaction index where it occurred.
    """

    storage_changes: Dict[U256, List[StorageChange]] = field(
        default_factory=dict
    )
    """
    Mapping from storage slot to list of changes made to that slot.
    Each change includes the transaction index and new value.
    """

    storage_reads: Set[U256] = field(default_factory=set)
    """
    Set of storage slots that were read but not modified.
    """

    balance_changes: List[BalanceChange] = field(default_factory=list)
    """
    List of balance changes for this account, ordered by transaction index.
    """

    nonce_changes: List[NonceChange] = field(default_factory=list)
    """
    List of nonce changes for this account, ordered by transaction index.
    """

    code_changes: List[CodeChange] = field(default_factory=list)
    """
    List of code changes (contract deployments) for this account,
    ordered by transaction index.
    """


@dataclass
class BlockAccessListBuilder:
    """
    Builder for constructing [`BlockAccessList`] efficiently during transaction
    execution.

    The builder accumulates all account and storage accesses during block
    execution and constructs a deterministic access list. Changes are tracked
    by address, field type, and transaction index to enable efficient
    reconstruction of state changes.

    The builder follows a two-phase approach:

    1. **Collection Phase**: During transaction execution, all state accesses
       are recorded via the tracking functions.
    1. **Build Phase**: After block execution, the accumulated data is sorted
       and encoded into the final deterministic format.

    [`BlockAccessList`]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    """  # noqa: E501

    block_access_index: BlockAccessIndex = BlockAccessIndex(0)
    """
    Current block access index.  Set by the caller before each
    [`incorporate_tx_into_block`] call (0 for system txs, i+1 for the
    i-th user tx, N+1 for post-execution operations).

    [`incorporate_tx_into_block`]: ref:ethereum.forks.amsterdam.state_tracker.incorporate_tx_into_block
    """  # noqa: E501

    accounts: Dict[Address, AccountData] = field(default_factory=dict)
    """
    Mapping from account address to its tracked changes during block execution.
    """


def ensure_account(builder: BlockAccessListBuilder, address: Address) -> None:
    """
    Ensure an account exists in the builder's tracking structure.

    Creates an empty [`AccountData`][ad] entry for the given address if it
    doesn't already exist. This function is idempotent and safe to call
    multiple times for the same address.

    [ad]: ref:ethereum.forks.amsterdam.block_access_lists.AccountData
    """
    if address not in builder.accounts:
        builder.accounts[address] = AccountData()


def add_storage_write(
    builder: BlockAccessListBuilder,
    address: Address,
    slot: U256,
    block_access_index: BlockAccessIndex,
    new_value: U256,
) -> None:
    """
    Add a storage write operation to the block access list.

    Records a storage slot modification for a given address at a specific
    transaction index. If multiple writes occur to the same slot within the
    same transaction (same `block_access_index`), only the final value is kept.
    """
    ensure_account(builder, address)

    if slot not in builder.accounts[address].storage_changes:
        builder.accounts[address].storage_changes[slot] = []

    # Check if there's already an entry with the same block_access_index
    # If so, update it with the new value, keeping only the final write
    changes = builder.accounts[address].storage_changes[slot]
    for i, existing_change in enumerate(changes):
        if existing_change.block_access_index == block_access_index:
            # Update the existing entry with the new value
            changes[i] = StorageChange(
                block_access_index=block_access_index, new_value=new_value
            )
            return

    # No existing entry found, append new change
    change = StorageChange(
        block_access_index=block_access_index, new_value=new_value
    )
    builder.accounts[address].storage_changes[slot].append(change)


def add_storage_read(
    builder: BlockAccessListBuilder, address: Address, slot: U256
) -> None:
    """
    Add a storage read operation to the block access list.

    Records that a storage slot was read during execution. Storage slots
    that are both read and written will only appear in the storage changes
    list, not in the storage reads list, as per [EIP-7928].
    """
    ensure_account(builder, address)
    builder.accounts[address].storage_reads.add(slot)


def add_balance_change(
    builder: BlockAccessListBuilder,
    address: Address,
    block_access_index: BlockAccessIndex,
    post_balance: U256,
) -> None:
    """
    Add a balance change to the block access list.

    Records the post-transaction balance for an account after it has been
    modified. This includes changes from transfers, gas fees, block rewards,
    and any other balance-affecting operations.
    """
    ensure_account(builder, address)

    # Balance value is already U256
    balance_value = post_balance

    # Check if we already have a balance change for this tx_index and update it
    # This ensures we only track the final balance per transaction
    existing_changes = builder.accounts[address].balance_changes
    for i, existing in enumerate(existing_changes):
        if existing.block_access_index == block_access_index:
            # Update the existing balance change with the new balance
            existing_changes[i] = BalanceChange(
                block_access_index=block_access_index,
                post_balance=balance_value,
            )
            return

    # No existing change for this tx_index, add a new one
    change = BalanceChange(
        block_access_index=block_access_index, post_balance=balance_value
    )
    builder.accounts[address].balance_changes.append(change)


def add_nonce_change(
    builder: BlockAccessListBuilder,
    address: Address,
    block_access_index: BlockAccessIndex,
    new_nonce: U64,
) -> None:
    """
    Add a nonce change to the block access list.

    Records a nonce increment for an account. This occurs when an EOA sends
    a transaction or when a contract performs [`CREATE`] or [`CREATE2`]
    operations.

    [`CREATE`]: ref:ethereum.forks.amsterdam.vm.instructions.system.create
    [`CREATE2`]: ref:ethereum.forks.amsterdam.vm.instructions.system.create2
    """
    ensure_account(builder, address)

    # Check if we already have a nonce change for this tx_index and update it
    # This ensures we only track the final (highest) nonce per transaction
    existing_changes = builder.accounts[address].nonce_changes
    for i, existing in enumerate(existing_changes):
        if existing.block_access_index == block_access_index:
            # Keep the highest nonce value
            if new_nonce > existing.new_nonce:
                existing_changes[i] = NonceChange(
                    block_access_index=block_access_index, new_nonce=new_nonce
                )
            return

    # No existing change for this tx_index, add a new one
    change = NonceChange(
        block_access_index=block_access_index, new_nonce=new_nonce
    )
    builder.accounts[address].nonce_changes.append(change)


def add_code_change(
    builder: BlockAccessListBuilder,
    address: Address,
    block_access_index: BlockAccessIndex,
    new_code: Bytes,
) -> None:
    """
    Add a code change to the block access list.

    Records contract code deployment or modification. This typically occurs
    during contract creation via [`CREATE`], [`CREATE2`], or
    [`SetCodeTransaction`][sct] operations.

    [`CREATE`]: ref:ethereum.forks.amsterdam.vm.instructions.system.create
    [`CREATE2`]: ref:ethereum.forks.amsterdam.vm.instructions.system.create2
    [sct]: ref:ethereum.forks.amsterdam.transactions.SetCodeTransaction
    """
    ensure_account(builder, address)

    # Check if we already have a code change for this block_access_index
    # This handles the case of in-transaction selfdestructs where code is
    # first deployed and then cleared in the same transaction
    existing_changes = builder.accounts[address].code_changes
    for i, existing in enumerate(existing_changes):
        if existing.block_access_index == block_access_index:
            # Replace the existing code change with the new one
            # For selfdestructs, this ensures we only record the final
            # state (empty code)
            existing_changes[i] = CodeChange(
                block_access_index=block_access_index, new_code=new_code
            )
            return

    # No existing change for this block_access_index, add a new one
    change = CodeChange(
        block_access_index=block_access_index, new_code=new_code
    )
    builder.accounts[address].code_changes.append(change)


def add_touched_account(
    builder: BlockAccessListBuilder, address: Address
) -> None:
    """
    Add an account that was accessed but not modified.

    Records that an account was accessed during execution without any state
    changes. This is used for operations like [`EXTCODEHASH`], [`BALANCE`],
    [`EXTCODESIZE`], and [`EXTCODECOPY`] that read account data without
    modifying it.

    [`EXTCODEHASH`]: ref:ethereum.forks.amsterdam.vm.instructions.environment.extcodehash
    [`BALANCE`]: ref:ethereum.forks.amsterdam.vm.instructions.environment.balance
    [`EXTCODESIZE`]: ref:ethereum.forks.amsterdam.vm.instructions.environment.extcodesize
    [`EXTCODECOPY`]: ref:ethereum.forks.amsterdam.vm.instructions.environment.extcodecopy
    """  # noqa: E501
    ensure_account(builder, address)


def _build_from_builder(
    builder: BlockAccessListBuilder,
) -> BlockAccessList:
    """
    Build the final [`BlockAccessList`] from a builder (internal helper).

    Constructs a deterministic block access list by sorting all accumulated
    changes. The resulting list is ordered by:

    1. Account addresses (lexicographically)
    2. Within each account:
       - Storage slots (lexicographically)
       - Transaction indices (numerically) for each change type

    Addresses, storage slots, and block access indices are unique.
    Storage reads that also appear in storage changes are excluded.

    [`BlockAccessList`]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    """  # noqa: E501
    block_access_list: BlockAccessList = []

    for address, changes in builder.accounts.items():
        storage_changes = []
        for slot, slot_changes in changes.storage_changes.items():
            sorted_changes = tuple(
                sorted(slot_changes, key=lambda x: x.block_access_index)
            )
            storage_changes.append(
                SlotChanges(slot=slot, changes=sorted_changes)
            )

        storage_reads = []
        for slot in changes.storage_reads:
            if slot not in changes.storage_changes:
                storage_reads.append(slot)

        balance_changes = tuple(
            sorted(changes.balance_changes, key=lambda x: x.block_access_index)
        )
        nonce_changes = tuple(
            sorted(changes.nonce_changes, key=lambda x: x.block_access_index)
        )
        code_changes = tuple(
            sorted(changes.code_changes, key=lambda x: x.block_access_index)
        )

        storage_changes.sort(key=lambda x: x.slot)
        storage_reads.sort()

        account_change = AccountChanges(
            address=address,
            storage_changes=tuple(storage_changes),
            storage_reads=tuple(storage_reads),
            balance_changes=balance_changes,
            nonce_changes=nonce_changes,
            code_changes=code_changes,
        )

        block_access_list.append(account_change)

    block_access_list.sort(key=lambda x: x.address)

    return block_access_list


def _get_pre_tx_account(
    pre_tx_accounts: Dict[Address, Optional[Account]],
    pre_state: PreState,
    address: Address,
) -> Optional[Account]:
    """
    Look up an account in cumulative state, falling back to `pre_state`.

    The cumulative account state (`pre_tx_accounts`) should contain state up
    to (but not including) the current transaction.

    Returns `None` if the `address` does not exist.
    """
    if address in pre_tx_accounts:
        return pre_tx_accounts[address]
    return pre_state.get_account_optional(address)


def _get_pre_tx_storage(
    pre_tx_storage: Dict[Address, Dict[Bytes32, U256]],
    pre_state: PreState,
    address: Address,
    key: Bytes32,
) -> U256:
    """
    Look up a storage value in cumulative state, falling back to `pre_state`.

    Returns `0` if not set.
    """
    if address in pre_tx_storage and key in pre_tx_storage[address]:
        return pre_tx_storage[address][key]
    return pre_state.get_storage(address, key)


def update_builder_from_tx(
    builder: BlockAccessListBuilder,
    tx_state: TransactionState,
) -> None:
    """
    Update the BAL builder with changes from a single transaction.

    Compare the transaction's writes against the block's cumulative
    state (falling back to `pre_state`) to extract balance, nonce, code, and
    storage changes.  Net-zero filtering is automatic: if the pre-tx value
    equals the post-tx value, no change is recorded.

    Must be called **before** the transaction's writes are merged into
    the block state.
    """
    block_state = tx_state.parent
    pre_state = block_state.pre_state
    idx = builder.block_access_index

    # Compare account writes against block cumulative state
    for address, post_account in tx_state.account_writes.items():
        pre_account = _get_pre_tx_account(
            block_state.account_writes, pre_state, address
        )

        pre_balance = pre_account.balance if pre_account else U256(0)
        post_balance = post_account.balance if post_account else U256(0)
        if pre_balance != post_balance:
            add_balance_change(builder, address, idx, post_balance)

        pre_nonce = pre_account.nonce if pre_account else Uint(0)
        post_nonce = post_account.nonce if post_account else Uint(0)
        if pre_nonce != post_nonce:
            add_nonce_change(builder, address, idx, U64(post_nonce))

        pre_code_hash = (
            pre_account.code_hash if pre_account else EMPTY_CODE_HASH
        )
        post_code_hash = (
            post_account.code_hash if post_account else EMPTY_CODE_HASH
        )
        if pre_code_hash != post_code_hash:
            post_code = get_code(tx_state, post_code_hash)
            add_code_change(builder, address, idx, post_code)

    # Compare storage writes against block cumulative state
    for address, slots in tx_state.storage_writes.items():
        for key, post_value in slots.items():
            pre_value = _get_pre_tx_storage(
                block_state.storage_writes, pre_state, address, key
            )
            if pre_value != post_value:
                # Convert slot from internal Bytes32 format to U256 for BAL.
                # EIP-7928 uses U256 as it's more space-efficient in RLP.
                u256_slot = U256.from_be_bytes(key)
                add_storage_write(builder, address, u256_slot, idx, post_value)


def build_block_access_list(
    builder: BlockAccessListBuilder,
    block_state: BlockState,
) -> BlockAccessList:
    """
    Build a [`BlockAccessList`] from the builder and block state.

    Feed accumulated reads from the block state into the builder, then produce
    the final sorted and encoded block access list.

    [`BlockAccessList`]: ref:ethereum.forks.amsterdam.block_access_lists.BlockAccessList
    """  # noqa: E501
    # Add storage reads (convert Bytes32 to U256 for BAL encoding)
    for address, slot in block_state.storage_reads:
        add_storage_read(builder, address, U256.from_be_bytes(slot))

    # Add touched addresses
    for address in block_state.account_reads:
        add_touched_account(builder, address)

    return _build_from_builder(builder)


def hash_block_access_list(
    block_access_list: BlockAccessList,
) -> Hash32:
    """
    Compute the hash of a Block Access List.
    """
    return keccak256(rlp.encode(block_access_list))


def validate_block_access_list_gas_limit(
    block_access_list: BlockAccessList,
    block_gas_limit: Uint,
) -> None:
    """
    Validate that the block access list does not exceed the gas limit.

    The total number of items (addresses + unique storage keys) must not
    exceed ``block_gas_limit // GAS_BLOCK_ACCESS_LIST_ITEM``.
    """
    from .vm.gas import GasCosts

    bal_items = Uint(0)
    for account in block_access_list:
        # Count each address as one item
        bal_items += Uint(1)

        # Collect unique storage keys across both
        # reads and writes
        unique_slots: Set[U256] = set()
        for slot_change in account.storage_changes:
            unique_slots.add(slot_change.slot)
        for slot in account.storage_reads:
            unique_slots.add(slot)

        # Count each unique storage key as one item
        bal_items += ulen(unique_slots)

    if bal_items > block_gas_limit // GasCosts.BLOCK_ACCESS_LIST_ITEM:
        raise BlockAccessListGasLimitExceededError(
            f"Block access list exceeds gas limit, {bal_items} items "
            f"exceeds limit of "
            f"{block_gas_limit // GasCosts.BLOCK_ACCESS_LIST_ITEM}."
        )