diff --git a/setup.py b/setup.py
index 3e94ac8fb..0bdf04b55 100644
--- a/setup.py
+++ b/setup.py
@@ -5,6 +5,7 @@ from distutils.util import convert_path
 import os
 import re
 import string
+import textwrap
 from typing import Dict, NamedTuple, List, Sequence, Optional
 from abc import ABC, abstractmethod
 import ast
@@ -48,8 +49,14 @@ def floorlog2(x: int) -> uint64:
 '''
 
 
+class ProtocolDefinition(NamedTuple):
+    # just function definitions currently. May expand with configuration vars in future.
+    functions: Dict[str, str]
+
+
 class SpecObject(NamedTuple):
     functions: Dict[str, str]
+    protocols: Dict[str, ProtocolDefinition]
     custom_types: Dict[str, str]
     constants: Dict[str, str]
     ssz_dep_constants: Dict[str, str]  # the constants that depend on ssz_objects
@@ -73,6 +80,18 @@ def _get_function_name_from_source(source: str) -> str:
     return fn.name
 
 
+def _get_self_type_from_source(source: str) -> Optional[str]:
+    fn = ast.parse(source).body[0]
+    args = fn.args.args
+    if len(args) == 0:
+        return None
+    if args[0].arg != 'self':
+        return None
+    if args[0].annotation is None:
+        return None
+    return args[0].annotation.id
+
+
 def _get_class_info_from_source(source: str) -> (str, Optional[str]):
     class_def = ast.parse(source).body[0]
     base = class_def.bases[0]
@@ -107,6 +126,7 @@ def _get_eth2_spec_comment(child: LinkRefDef) -> Optional[str]:
 
 def get_spec(file_name: str) -> SpecObject:
     functions: Dict[str, str] = {}
+    protocols: Dict[str, ProtocolDefinition] = {}
     constants: Dict[str, str] = {}
     ssz_dep_constants: Dict[str, str] = {}
     ssz_objects: Dict[str, str] = {}
@@ -132,7 +152,14 @@ def get_spec(file_name: str) -> SpecObject:
             source = _get_source_from_code_block(child)
             if source.startswith("def"):
                 current_name = _get_function_name_from_source(source)
-                functions[current_name] = "\n".join(line.rstrip() for line in source.splitlines())
+                self_type_name = _get_self_type_from_source(source)
+                function_def = "\n".join(line.rstrip() for line in source.splitlines())
+                if self_type_name is None:
+                    functions[current_name] = function_def
+                else:
+                    if self_type_name not in protocols:
+                        protocols[self_type_name] = ProtocolDefinition(functions={})
+                    protocols[self_type_name].functions[current_name] = function_def
             elif source.startswith("@dataclass"):
                 dataclasses[current_name] = "\n".join(line.rstrip() for line in source.splitlines())
             elif source.startswith("class"):
@@ -170,6 +197,7 @@ def get_spec(file_name: str) -> SpecObject:
 
     return SpecObject(
         functions=functions,
+        protocols=protocols,
         custom_types=custom_types,
         constants=constants,
         ssz_dep_constants=ssz_dep_constants,
@@ -422,7 +450,8 @@ class MergeSpecBuilder(Phase0SpecBuilder):
 
     @classmethod
     def imports(cls):
-        return super().imports() + '\n' + '''
+        return super().imports() + '''
+from typing import Protocol
 from eth2spec.phase0 import spec as phase0
 from eth2spec.utils.ssz.ssz_typing import Bytes20, ByteList, ByteVector, uint256
 from importlib import reload
@@ -452,12 +481,22 @@ def get_pow_chain_head() -> PowBlock:
     pass
 
 
-def verify_execution_state_transition(execution_payload: ExecutionPayload) -> bool:
-    return True
+class NoopExecutionEngine(ExecutionEngine):
+
+    def new_block(self, execution_payload: ExecutionPayload) -> bool:
+        return True
+
+    def set_head(self, block_hash: Hash32) -> bool:
+        return True
+
+    def finalize_block(self, block_hash: Hash32) -> bool:
+        return True
+
+    def assemble_block(self, block_hash: Hash32, timestamp: uint64) -> ExecutionPayload:
+        raise NotImplementedError("no default block production")
 
 
-def produce_execution_payload(parent_hash: Hash32, timestamp: uint64) -> ExecutionPayload:
-    pass"""
+EXECUTION_ENGINE = NoopExecutionEngine()"""
 
 
     @classmethod
@@ -495,6 +534,15 @@ def objects_to_spec(spec_object: SpecObject, builder: SpecBuilder, ordered_class
             ]
         )
     )
+
+    def format_protocol(protocol_name: str, protocol_def: ProtocolDefinition) -> str:
+        protocol = f"class {protocol_name}(Protocol):"
+        for fn_source in protocol_def.functions.values():
+            fn_source = fn_source.replace("self: "+protocol_name, "self")
+            protocol += "\n\n" + textwrap.indent(fn_source, "    ")
+        return protocol
+
+    protocols_spec = '\n\n\n'.join(format_protocol(k, v) for k, v in spec_object.protocols.items())
     for k in list(spec_object.functions):
         if "ceillog2" in k or "floorlog2" in k:
             del spec_object.functions[k]
@@ -520,6 +568,7 @@ def objects_to_spec(spec_object: SpecObject, builder: SpecBuilder, ordered_class
             + '\n\n' + constants_spec
             + '\n\n' + CONFIG_LOADER
             + '\n\n' + ordered_class_objects_spec
+            + ('\n\n\n' + protocols_spec if protocols_spec != '' else '')
             + '\n\n\n' + functions_spec
             + '\n\n' + builder.sundry_functions()
             # Since some constants are hardcoded in setup.py, the following assertions verify that the hardcoded constants are
@@ -531,6 +580,17 @@ def objects_to_spec(spec_object: SpecObject, builder: SpecBuilder, ordered_class
     return spec
 
 
+def combine_protocols(old_protocols: Dict[str, ProtocolDefinition],
+                      new_protocols: Dict[str, ProtocolDefinition]) -> Dict[str, ProtocolDefinition]:
+    for key, value in new_protocols.items():
+        if key not in old_protocols:
+            old_protocols[key] = value
+        else:
+            functions = combine_functions(old_protocols[key].functions, value.functions)
+            old_protocols[key] = ProtocolDefinition(functions=functions)
+    return old_protocols
+
+
 def combine_functions(old_functions: Dict[str, str], new_functions: Dict[str, str]) -> Dict[str, str]:
     for key, value in new_functions.items():
         old_functions[key] = value
@@ -589,8 +649,9 @@ def combine_spec_objects(spec0: SpecObject, spec1: SpecObject) -> SpecObject:
     """
     Takes in two spec variants (as tuples of their objects) and combines them using the appropriate combiner function.
     """
-    functions0, custom_types0, constants0, ssz_dep_constants0, ssz_objects0, dataclasses0 = spec0
-    functions1, custom_types1, constants1, ssz_dep_constants1, ssz_objects1, dataclasses1 = spec1
+    functions0, protocols0, custom_types0, constants0, ssz_dep_constants0, ssz_objects0, dataclasses0 = spec0
+    functions1, protocols1, custom_types1, constants1, ssz_dep_constants1, ssz_objects1, dataclasses1 = spec1
+    protocols = combine_protocols(protocols0, protocols1)
     functions = combine_functions(functions0, functions1)
     custom_types = combine_constants(custom_types0, custom_types1)
     constants = combine_constants(constants0, constants1)
@@ -599,6 +660,7 @@ def combine_spec_objects(spec0: SpecObject, spec1: SpecObject) -> SpecObject:
     dataclasses = combine_functions(dataclasses0, dataclasses1)
     return SpecObject(
         functions=functions,
+        protocols=protocols,
         custom_types=custom_types,
         constants=constants,
         ssz_dep_constants=ssz_dep_constants,
diff --git a/specs/merge/beacon-chain.md b/specs/merge/beacon-chain.md
index 626a86724..85953a547 100644
--- a/specs/merge/beacon-chain.md
+++ b/specs/merge/beacon-chain.md
@@ -22,6 +22,9 @@
   - [New containers](#new-containers)
     - [`ExecutionPayload`](#executionpayload)
     - [`ExecutionPayloadHeader`](#executionpayloadheader)
+- [Protocols](#protocols)
+  - [`ExecutionEngine`](#executionengine)
+    - [`new_block`](#new_block)
 - [Helper functions](#helper-functions)
   - [Misc](#misc)
     - [`is_execution_enabled`](#is_execution_enabled)
@@ -30,7 +33,6 @@
     - [`compute_time_at_slot`](#compute_time_at_slot)
   - [Block processing](#block-processing)
     - [Execution payload processing](#execution-payload-processing)
-      - [`verify_execution_state_transition`](#verify_execution_state_transition)
       - [`process_execution_payload`](#process_execution_payload)
 
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
@@ -137,6 +139,30 @@ class ExecutionPayloadHeader(Container):
     transactions_root: Root
 ```
 
+## Protocols
+
+### `ExecutionEngine`
+
+The `ExecutionEngine` protocol separates the consensus and execution sub-systems.
+The consensus implementation references an instance of this sub-system with `EXECUTION_ENGINE`. 
+
+The following methods are added to the `ExecutionEngine` protocol for use in the state transition:
+
+#### `new_block`
+
+Verifies the given `execution_payload` with respect to execution state transition, and persists changes if valid.
+
+The body of this function is implementation dependent.
+The Consensus API may be used to implement this with an external execution engine.
+
+```python
+def new_block(self: ExecutionEngine, execution_payload: ExecutionPayload) -> bool:
+    """
+    Returns True if the ``execution_payload`` was verified and processed successfully, False otherwise. 
+    """
+    ...
+```
+
 ## Helper functions
 
 ### Misc
@@ -182,20 +208,17 @@ def process_block(state: BeaconState, block: BeaconBlock) -> None:
     process_operations(state, block.body)
     # Pre-merge, skip execution payload processing
     if is_execution_enabled(state, block):
-        process_execution_payload(state, block.body.execution_payload)  # [New in Merge]
+        process_execution_payload(state, block.body.execution_payload, EXECUTION_ENGINE)  # [New in Merge]
 ```
 
 #### Execution payload processing
 
-##### `verify_execution_state_transition`
-
-Let `verify_execution_state_transition(execution_payload: ExecutionPayload) -> bool` be the function that verifies given `ExecutionPayload` with respect to execution state transition.
-The body of the function is implementation dependent.
-
 ##### `process_execution_payload`
 
 ```python
-def process_execution_payload(state: BeaconState, execution_payload: ExecutionPayload) -> None:
+def process_execution_payload(state: BeaconState,
+                              execution_payload: ExecutionPayload,
+                              execution_engine: ExecutionEngine) -> None:
     """
     Note: This function is designed to be able to be run in parallel with the other `process_block` sub-functions
     """
@@ -205,7 +228,7 @@ def process_execution_payload(state: BeaconState, execution_payload: ExecutionPa
 
     assert execution_payload.timestamp == compute_time_at_slot(state, state.slot)
 
-    assert verify_execution_state_transition(execution_payload)
+    assert execution_engine.new_block(execution_payload)
 
     state.latest_execution_payload_header = ExecutionPayloadHeader(
         block_hash=execution_payload.block_hash,
diff --git a/specs/merge/fork-choice.md b/specs/merge/fork-choice.md
index f478dd7e6..9e6c341bc 100644
--- a/specs/merge/fork-choice.md
+++ b/specs/merge/fork-choice.md
@@ -8,11 +8,16 @@
 <!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
 
 - [Introduction](#introduction)
-  - [Helpers](#helpers)
+- [Protocols](#protocols)
+  - [`ExecutionEngine`](#executionengine)
+    - [`set_head`](#set_head)
+    - [`finalize_block`](#finalize_block)
+- [Containers](#containers)
     - [`PowBlock`](#powblock)
+- [Helper functions](#helper-functions)
     - [`get_pow_block`](#get_pow_block)
     - [`is_valid_transition_block`](#is_valid_transition_block)
-  - [Updated fork-choice handlers](#updated-fork-choice-handlers)
+- [Updated fork-choice handlers](#updated-fork-choice-handlers)
     - [`on_block`](#on_block)
 
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
@@ -24,7 +29,44 @@ This is the modification of the fork choice according to the executable beacon c
 
 *Note*: It introduces the process of transition from the last PoW block to the first PoS block.
 
-### Helpers
+## Protocols
+
+### `ExecutionEngine`
+
+The following methods are added to the `ExecutionEngine` protocol for use in the fork choice:
+
+#### `set_head`
+
+Re-organizes the execution payload chain and corresponding state to make `block_hash` the head.
+
+The body of this function is implementation dependent.
+The Consensus API may be used to implement this with an external execution engine.
+
+```python
+def set_head(self: ExecutionEngine, block_hash: Hash32) -> bool:
+    """
+    Returns True if the ``block_hash`` was successfully set as head of the execution payload chain.
+    """
+    ...
+```
+
+#### `finalize_block`
+
+Applies finality to the execution state: it irreversibly persists the chain of all execution payloads
+and corresponding state, up to and including `block_hash`.
+
+The body of this function is implementation dependent.
+The Consensus API may be used to implement this with an external execution engine.
+
+```python
+def finalize_block(self: ExecutionEngine, block_hash: Hash32) -> bool:
+    """
+    Returns True if the data up to and including ``block_hash`` was successfully finalized.
+    """
+    ...
+```
+
+## Containers
 
 #### `PowBlock`
 
@@ -36,6 +78,8 @@ class PowBlock(Container):
     total_difficulty: uint256
 ```
 
+## Helper functions
+
 #### `get_pow_block`
 
 Let `get_pow_block(block_hash: Hash32) -> PowBlock` be the function that given the hash of the PoW block returns its data.
@@ -52,7 +96,7 @@ def is_valid_transition_block(block: PowBlock) -> bool:
     return block.is_valid and is_total_difficulty_reached
 ```
 
-### Updated fork-choice handlers
+## Updated fork-choice handlers
 
 #### `on_block`
 
diff --git a/specs/merge/validator.md b/specs/merge/validator.md
index dccc5727b..c4c396059 100644
--- a/specs/merge/validator.md
+++ b/specs/merge/validator.md
@@ -12,6 +12,9 @@
 
 - [Introduction](#introduction)
 - [Prerequisites](#prerequisites)
+- [Protocols](#protocols)
+  - [`ExecutionEngine`](#executionengine)
+    - [`assemble_block`](#assemble_block)
 - [Beacon chain responsibilities](#beacon-chain-responsibilities)
   - [Block proposal](#block-proposal)
     - [Constructing the `BeaconBlockBody`](#constructing-the-beaconblockbody)
@@ -32,6 +35,25 @@ This document is an extension of the [Phase 0 -- Validator](../phase0/validator.
 
 All terminology, constants, functions, and protocol mechanics defined in the updated Beacon Chain doc of [The Merge](./beacon-chain.md) are requisite for this document and used throughout. Please see related Beacon Chain doc before continuing and use them as a reference throughout.
 
+## Protocols
+
+### `ExecutionEngine`
+
+The following methods are added to the `ExecutionEngine` protocol for use as a validator:
+
+#### `assemble_block`
+
+Produces a new instance of an execution payload, with the specified `timestamp`,
+on top of the execution payload chain tip identified by `block_hash`.
+
+The body of this function is implementation dependent.
+The Consensus API may be used to implement this with an external execution engine.
+
+```python
+def assemble_block(self: ExecutionEngine, block_hash: Hash32, timestamp: uint64) -> ExecutionPayload:
+    ...
+```
+
 ## Beacon chain responsibilities
 
 All validator responsibilities remain unchanged other than those noted below. Namely, the transition block handling and the addition of `ExecutionPayload`.
@@ -49,12 +71,12 @@ Let `get_pow_chain_head() -> PowBlock` be the function that returns the head of
 ###### `produce_execution_payload`
 
 Let `produce_execution_payload(parent_hash: Hash32, timestamp: uint64) -> ExecutionPayload` be the function that produces new instance of execution payload.
-The body of this function is implementation dependent.
+The `ExecutionEngine` protocol is used for the implementation specific part of execution payload proposals.
 
 * Set `block.body.execution_payload = get_execution_payload(state)` where:
 
 ```python
-def get_execution_payload(state: BeaconState) -> ExecutionPayload:
+def get_execution_payload(state: BeaconState, execution_engine: ExecutionEngine) -> ExecutionPayload:
     if not is_transition_completed(state):
         pow_block = get_pow_chain_head()
         if not is_valid_transition_block(pow_block):
@@ -63,10 +85,10 @@ def get_execution_payload(state: BeaconState) -> ExecutionPayload:
         else:
             # Signify merge via producing on top of the last PoW block
             timestamp = compute_time_at_slot(state, state.slot)
-            return produce_execution_payload(pow_block.block_hash, timestamp)
+            return execution_engine.assemble_block(pow_block.block_hash, timestamp)
 
     # Post-merge, normal payload
     execution_parent_hash = state.latest_execution_payload_header.block_hash
     timestamp = compute_time_at_slot(state, state.slot)
-    return produce_execution_payload(execution_parent_hash, timestamp)
+    return execution_engine.assemble_block(execution_parent_hash, timestamp)
 ```
diff --git a/tests/core/pyspec/eth2spec/test/helpers/execution_payload.py b/tests/core/pyspec/eth2spec/test/helpers/execution_payload.py
index 093b7cf2e..7774aa4d9 100644
--- a/tests/core/pyspec/eth2spec/test/helpers/execution_payload.py
+++ b/tests/core/pyspec/eth2spec/test/helpers/execution_payload.py
@@ -1,4 +1,3 @@
-
 def build_empty_execution_payload(spec, state):
     """
     Assuming a pre-state of the same slot, build a valid ExecutionPayload without any transactions.
@@ -24,3 +23,37 @@ def build_empty_execution_payload(spec, state):
     payload.block_hash = spec.Hash32(spec.hash(payload.hash_tree_root() + b"FAKE RLP HASH"))
 
     return payload
+
+
+def get_execution_payload_header(spec, execution_payload):
+    return spec.ExecutionPayloadHeader(
+        block_hash=execution_payload.block_hash,
+        parent_hash=execution_payload.parent_hash,
+        coinbase=execution_payload.coinbase,
+        state_root=execution_payload.state_root,
+        number=execution_payload.number,
+        gas_limit=execution_payload.gas_limit,
+        gas_used=execution_payload.gas_used,
+        timestamp=execution_payload.timestamp,
+        receipt_root=execution_payload.receipt_root,
+        logs_bloom=execution_payload.logs_bloom,
+        transactions_root=spec.hash_tree_root(execution_payload.transactions)
+    )
+
+
+def build_state_with_incomplete_transition(spec, state):
+    return build_state_with_execution_payload_header(spec, state, spec.ExecutionPayloadHeader())
+
+
+def build_state_with_complete_transition(spec, state):
+    pre_state_payload = build_empty_execution_payload(spec, state)
+    payload_header = get_execution_payload_header(spec, pre_state_payload)
+
+    return build_state_with_execution_payload_header(spec, state, payload_header)
+
+
+def build_state_with_execution_payload_header(spec, state, execution_payload_header):
+    pre_state = state.copy()
+    pre_state.latest_execution_payload_header = execution_payload_header
+
+    return pre_state
diff --git a/tests/core/pyspec/eth2spec/test/merge/block_processing/test_process_execution_payload.py b/tests/core/pyspec/eth2spec/test/merge/block_processing/test_process_execution_payload.py
index fb1da8758..5edd31960 100644
--- a/tests/core/pyspec/eth2spec/test/merge/block_processing/test_process_execution_payload.py
+++ b/tests/core/pyspec/eth2spec/test/merge/block_processing/test_process_execution_payload.py
@@ -1,4 +1,9 @@
-from eth2spec.test.helpers.execution_payload import build_empty_execution_payload
+from eth2spec.test.helpers.execution_payload import (
+    build_empty_execution_payload,
+    get_execution_payload_header,
+    build_state_with_incomplete_transition,
+    build_state_with_complete_transition,
+)
 from eth2spec.test.context import spec_state_test, expect_assertion_error, with_merge_and_later
 from eth2spec.test.helpers.state import next_slot
 
@@ -13,31 +18,184 @@ def run_execution_payload_processing(spec, state, execution_payload, valid=True,
     If ``valid == False``, run expecting ``AssertionError``
     """
 
-    pre_exec_header = state.latest_execution_payload_header.copy()
-
     yield 'pre', state
     yield 'execution', {'execution_valid': execution_valid}
     yield 'execution_payload', execution_payload
 
+    called_new_block = False
+
+    class TestEngine(spec.NoopExecutionEngine):
+        def new_block(self, payload) -> bool:
+            nonlocal called_new_block, execution_valid
+            called_new_block = True
+            assert payload == execution_payload
+            return execution_valid
+
     if not valid:
-        expect_assertion_error(lambda: spec.process_execution_payload(state, execution_payload))
+        expect_assertion_error(lambda: spec.process_execution_payload(state, execution_payload, TestEngine()))
         yield 'post', None
         return
 
-    spec.process_execution_payload(state, execution_payload)
+    spec.process_execution_payload(state, execution_payload, TestEngine())
+
+    # Make sure we called the engine
+    assert called_new_block
 
     yield 'post', state
 
-    assert pre_exec_header != state.latest_execution_payload_header
-    # TODO: any more assertions to make?
+    assert state.latest_execution_payload_header == get_execution_payload_header(spec, execution_payload)
 
 
 @with_merge_and_later
 @spec_state_test
 def test_success_first_payload(spec, state):
+    # pre-state
+    state = build_state_with_incomplete_transition(spec, state)
     next_slot(spec, state)
-    assert not spec.is_transition_completed(state)
 
+    # execution payload
     execution_payload = build_empty_execution_payload(spec, state)
 
     yield from run_execution_payload_processing(spec, state, execution_payload)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_success_regular_payload(spec, state):
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+
+    yield from run_execution_payload_processing(spec, state, execution_payload)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_success_first_payload_with_gap_slot(spec, state):
+    # pre-state
+    state = build_state_with_incomplete_transition(spec, state)
+    next_slot(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+
+    yield from run_execution_payload_processing(spec, state, execution_payload)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_success_regular_payload_with_gap_slot(spec, state):
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+
+    yield from run_execution_payload_processing(spec, state, execution_payload)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_execution_first_payload(spec, state):
+    # completely valid payload, but execution itself fails (e.g. block exceeds gas limit)
+
+    # pre-state
+    state = build_state_with_incomplete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False, execution_valid=False)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_execution_regular_payload(spec, state):
+    # completely valid payload, but execution itself fails (e.g. block exceeds gas limit)
+
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False, execution_valid=False)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_parent_hash_regular_payload(spec, state):
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+    execution_payload.parent_hash = spec.Hash32()
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_number_regular_payload(spec, state):
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+    execution_payload.number = execution_payload.number + 1
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_everything_regular_payload(spec, state):
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+    execution_payload.parent_hash = spec.Hash32()
+    execution_payload.number = execution_payload.number + 1
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_timestamp_first_payload(spec, state):
+    # pre-state
+    state = build_state_with_incomplete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+    execution_payload.timestamp = execution_payload.timestamp + 1
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False)
+
+
+@with_merge_and_later
+@spec_state_test
+def test_bad_timestamp_regular_payload(spec, state):
+    # pre-state
+    state = build_state_with_complete_transition(spec, state)
+    next_slot(spec, state)
+
+    # execution payload
+    execution_payload = build_empty_execution_payload(spec, state)
+    execution_payload.timestamp = execution_payload.timestamp + 1
+
+    yield from run_execution_payload_processing(spec, state, execution_payload, valid=False)