diff --git a/tests/core/pyspec/eth_consensus_specs/test/electra/fork_choice/test_deposit_with_reorg.py b/tests/core/pyspec/eth_consensus_specs/test/electra/fork_choice/test_deposit_with_reorg.py
index 2b465e0057..89bd6986d1 100644
--- a/tests/core/pyspec/eth_consensus_specs/test/electra/fork_choice/test_deposit_with_reorg.py
+++ b/tests/core/pyspec/eth_consensus_specs/test/electra/fork_choice/test_deposit_with_reorg.py
@@ -28,7 +28,6 @@
 )
 
 
-# TODO(jtraglia): In gloas, how do we set execution requests in the payload envelope?
 @with_all_phases_from_to(ELECTRA, GLOAS)
 @spec_state_test
 @with_presets([MINIMAL], reason="too slow")
diff --git a/tests/core/pyspec/eth_consensus_specs/test/gloas/block_processing/test_process_execution_payload.py b/tests/core/pyspec/eth_consensus_specs/test/gloas/block_processing/test_process_execution_payload.py
index ab91061fa2..7cc7036174 100644
--- a/tests/core/pyspec/eth_consensus_specs/test/gloas/block_processing/test_process_execution_payload.py
+++ b/tests/core/pyspec/eth_consensus_specs/test/gloas/block_processing/test_process_execution_payload.py
@@ -6,8 +6,8 @@
 )
 from eth_consensus_specs.test.helpers.execution_payload import (
     build_empty_execution_payload,
+    prepare_execution_payload_envelope,
 )
-from eth_consensus_specs.test.helpers.keys import builder_privkeys, privkeys
 
 
 def run_execution_payload_processing(
@@ -52,112 +52,6 @@ def verify_and_notify_new_payload(self, new_payload_request) -> bool:
     yield "post", state
 
 
-def prepare_execution_payload_envelope(
-    spec,
-    state,
-    builder_index=None,
-    slot=None,
-    beacon_block_root=None,
-    state_root=None,
-    execution_payload=None,
-    execution_requests=None,
-    valid_signature=True,
-):
-    """
-    Helper to create a signed execution payload envelope with customizable parameters.
-    Note: This should be called AFTER setting up the state with the committed bid.
-    """
-    if builder_index is None:
-        builder_index = spec.BUILDER_INDEX_SELF_BUILD
-
-    if slot is None:
-        slot = state.slot
-
-    if beacon_block_root is None:
-        # Cache latest block header state root if not already set
-        if state.latest_block_header.state_root == spec.Root():
-            state.latest_block_header.state_root = state.hash_tree_root()
-        beacon_block_root = state.latest_block_header.hash_tree_root()
-
-    if execution_payload is None:
-        execution_payload = build_empty_execution_payload(spec, state)
-
-    if execution_requests is None:
-        execution_requests = spec.ExecutionRequests(
-            deposits=spec.List[spec.DepositRequest, spec.MAX_DEPOSIT_REQUESTS_PER_PAYLOAD](),
-            withdrawals=spec.List[
-                spec.WithdrawalRequest, spec.MAX_WITHDRAWAL_REQUESTS_PER_PAYLOAD
-            ](),
-            consolidations=spec.List[
-                spec.ConsolidationRequest, spec.MAX_CONSOLIDATION_REQUESTS_PER_PAYLOAD
-            ](),
-        )
-
-    # Create a copy of state for computing state_root after execution payload processing
-    if state_root is None:
-        post_state = state.copy()
-        # Simulate the state changes that process_execution_payload will make
-
-        # Cache latest block header state root if empty (matches process_execution_payload)
-        previous_state_root = post_state.hash_tree_root()
-        if post_state.latest_block_header.state_root == spec.Root():
-            post_state.latest_block_header.state_root = previous_state_root
-
-        # Process execution requests if any
-        if execution_requests is not None:
-            for deposit in execution_requests.deposits:
-                spec.process_deposit_request(post_state, deposit)
-            for withdrawal in execution_requests.withdrawals:
-                spec.process_withdrawal_request(post_state, withdrawal)
-            for consolidation in execution_requests.consolidations:
-                spec.process_consolidation_request(post_state, consolidation)
-
-        # Process builder payment (only if amount > 0)
-        payment = post_state.builder_pending_payments[
-            spec.SLOTS_PER_EPOCH + state.slot % spec.SLOTS_PER_EPOCH
-        ]
-        if payment.withdrawal.amount > 0:
-            post_state.builder_pending_withdrawals.append(payment.withdrawal)
-
-        # Clear the pending payment
-        post_state.builder_pending_payments[
-            spec.SLOTS_PER_EPOCH + state.slot % spec.SLOTS_PER_EPOCH
-        ] = spec.BuilderPendingPayment()
-
-        # Update execution payload availability and latest block hash
-        post_state.execution_payload_availability[state.slot % spec.SLOTS_PER_HISTORICAL_ROOT] = 0b1
-        post_state.latest_block_hash = execution_payload.block_hash
-        state_root = post_state.hash_tree_root()
-
-    envelope = spec.ExecutionPayloadEnvelope(
-        payload=execution_payload,
-        execution_requests=execution_requests,
-        builder_index=builder_index,
-        beacon_block_root=beacon_block_root,
-        slot=slot,
-        state_root=state_root,
-    )
-
-    if valid_signature:
-        if envelope.builder_index == spec.BUILDER_INDEX_SELF_BUILD:
-            privkey = privkeys[state.latest_block_header.proposer_index]
-        else:
-            privkey = builder_privkeys[envelope.builder_index]
-        signature = spec.get_execution_payload_envelope_signature(
-            state,
-            envelope,
-            privkey,
-        )
-    else:
-        # Invalid signature
-        signature = spec.BLSSignature()
-
-    return spec.SignedExecutionPayloadEnvelope(
-        message=envelope,
-        signature=signature,
-    )
-
-
 def setup_state_with_payload_bid(
     spec, state, builder_index=None, value=None, prev_randao=None, blob_kzg_commitments=None
 ):
diff --git a/tests/core/pyspec/eth_consensus_specs/test/gloas/fork_choice/__init__.py b/tests/core/pyspec/eth_consensus_specs/test/gloas/fork_choice/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/core/pyspec/eth_consensus_specs/test/gloas/fork_choice/test_deposit_with_reorg.py b/tests/core/pyspec/eth_consensus_specs/test/gloas/fork_choice/test_deposit_with_reorg.py
new file mode 100644
index 0000000000..c85da4b51a
--- /dev/null
+++ b/tests/core/pyspec/eth_consensus_specs/test/gloas/fork_choice/test_deposit_with_reorg.py
@@ -0,0 +1,115 @@
+from eth_consensus_specs.test.context import (
+    spec_state_test,
+    with_gloas_and_later,
+    with_presets,
+)
+from eth_consensus_specs.test.helpers.block import (
+    build_empty_block_for_next_slot,
+)
+from eth_consensus_specs.test.helpers.constants import MINIMAL
+from eth_consensus_specs.test.helpers.deposits import (
+    prepare_deposit_request,
+)
+from eth_consensus_specs.test.helpers.execution_payload import (
+    reveal_payload_to_state,
+)
+from eth_consensus_specs.test.helpers.fork_choice import (
+    apply_next_slots_with_attestations,
+    get_genesis_forkchoice_store_and_block,
+    tick_and_add_block,
+)
+from eth_consensus_specs.test.helpers.state import (
+    next_slot,
+    state_transition_and_sign_block,
+)
+
+
+@with_gloas_and_later
+@spec_state_test
+@with_presets([MINIMAL], reason="too slow")
+def test_new_validator_deposit_with_multiple_epoch_transitions(spec, state):
+    """
+    Test that deposit processing works correctly through fork choice with reorgs in Gloas.
+
+    In Gloas, execution_requests (including deposits) are delivered via ExecutionPayloadEnvelope
+    (on_execution_payload) rather than directly in the BeaconBlockBody.
+    """
+    # signify the eth1 bridge deprecation
+    state.deposit_requests_start_index = state.eth1_deposit_index
+
+    # yield anchor state and block
+    store, anchor_block = get_genesis_forkchoice_store_and_block(spec, state)
+    yield "anchor_state", state
+    yield "anchor_block", anchor_block
+
+    test_steps = []
+
+    # (1) create deposit request for a new validator via execution payload envelope
+    deposit_request = prepare_deposit_request(
+        spec, len(state.validators), spec.MIN_ACTIVATION_BALANCE, signed=True
+    )
+    execution_requests = spec.ExecutionRequests(
+        deposits=spec.List[spec.DepositRequest, spec.MAX_DEPOSIT_REQUESTS_PER_PAYLOAD](
+            [deposit_request]
+        ),
+        withdrawals=spec.List[spec.WithdrawalRequest, spec.MAX_WITHDRAWAL_REQUESTS_PER_PAYLOAD](),
+        consolidations=spec.List[
+            spec.ConsolidationRequest, spec.MAX_CONSOLIDATION_REQUESTS_PER_PAYLOAD
+        ](),
+    )
+
+    deposit_block = build_empty_block_for_next_slot(spec, state)
+    signed_deposit_block = state_transition_and_sign_block(spec, state, deposit_block)
+
+    deposit_envelope = reveal_payload_to_state(spec, state, execution_requests=execution_requests)
+
+    pending_deposit = spec.PendingDeposit(
+        pubkey=deposit_request.pubkey,
+        withdrawal_credentials=deposit_request.withdrawal_credentials,
+        amount=deposit_request.amount,
+        signature=deposit_request.signature,
+        slot=deposit_block.slot,
+    )
+
+    assert state.pending_deposits == [pending_deposit]
+
+    yield from tick_and_add_block(
+        spec, store, signed_deposit_block, test_steps, envelope=deposit_envelope
+    )
+
+    # (2) finalize and process pending deposit on one fork
+    slots = 4 * spec.SLOTS_PER_EPOCH - state.slot
+    post_state, _, latest_block = yield from apply_next_slots_with_attestations(
+        spec, state, store, slots, True, True, test_steps, with_payload_reveal=True
+    )
+
+    # check new validator has been created
+    assert post_state.pending_deposits == []
+    new_validator = post_state.validators[len(post_state.validators) - 1]
+    assert new_validator.pubkey == pending_deposit.pubkey
+    assert new_validator.withdrawal_credentials == pending_deposit.withdrawal_credentials
+
+    # (3) create a conflicting block that triggers deposit processing on another fork
+    prev_epoch_ancestor = store.blocks[latest_block.message.parent_root]
+    # important to skip last block of the epoch to make client do the epoch processing
+    # otherwise, client can read the post-epoch from cache
+    prev_epoch_ancestor = store.blocks[prev_epoch_ancestor.parent_root]
+    ancestor_root = prev_epoch_ancestor.hash_tree_root()
+    # Use post-on_execution_payload state for the reorg base
+    another_fork_state = store.payload_states[ancestor_root].copy()
+
+    assert another_fork_state.pending_deposits == [pending_deposit]
+
+    # skip a slot to create and process a fork block
+    next_slot(spec, another_fork_state)
+    post_state, _, _ = yield from apply_next_slots_with_attestations(
+        spec, another_fork_state, store, 1, True, True, test_steps, with_payload_reveal=True
+    )
+
+    # check new validator has been created on another fork
+    assert post_state.pending_deposits == []
+    new_validator = post_state.validators[len(post_state.validators) - 1]
+    assert new_validator.pubkey == pending_deposit.pubkey
+    assert new_validator.withdrawal_credentials == pending_deposit.withdrawal_credentials
+
+    yield "steps", test_steps
diff --git a/tests/core/pyspec/eth_consensus_specs/test/helpers/attestations.py b/tests/core/pyspec/eth_consensus_specs/test/helpers/attestations.py
index 5f4a43847e..2b76290fb5 100644
--- a/tests/core/pyspec/eth_consensus_specs/test/helpers/attestations.py
+++ b/tests/core/pyspec/eth_consensus_specs/test/helpers/attestations.py
@@ -2,10 +2,12 @@
 
 from eth_consensus_specs.test.context import expect_assertion_error
 from eth_consensus_specs.test.helpers.block import build_empty_block_for_next_slot
+from eth_consensus_specs.test.helpers.execution_payload import reveal_payload_to_state
 from eth_consensus_specs.test.helpers.forks import (
     is_post_altair,
     is_post_deneb,
     is_post_electra,
+    is_post_gloas,
 )
 from eth_consensus_specs.test.helpers.keys import privkeys
 from eth_consensus_specs.test.helpers.state import (
@@ -282,7 +284,7 @@ def get_valid_attestation_at_slot(
 
 
 def next_slots_with_attestations(
-    spec, state, slot_count, fill_cur_epoch, fill_prev_epoch, participation_fn=None
+    spec, state, slot_count, fill_cur_epoch, fill_prev_epoch, participation_fn=None, envelopes=None
 ):
     """
     participation_fn: (slot, committee_index, committee_indices_set) -> participants_indices_set
@@ -296,6 +298,7 @@ def next_slots_with_attestations(
             fill_cur_epoch,
             fill_prev_epoch,
             participation_fn,
+            envelopes=envelopes,
         )
         signed_blocks.append(signed_block)
 
@@ -323,7 +326,7 @@ def _add_valid_attestations(spec, state, block, slot_to_attest, participation_fn
 
 
 def next_epoch_with_attestations(
-    spec, state, fill_cur_epoch, fill_prev_epoch, participation_fn=None
+    spec, state, fill_cur_epoch, fill_prev_epoch, participation_fn=None, envelopes=None
 ):
     assert state.slot % spec.SLOTS_PER_EPOCH == 0
 
@@ -334,6 +337,7 @@ def next_epoch_with_attestations(
         fill_cur_epoch,
         fill_prev_epoch,
         participation_fn,
+        envelopes=envelopes,
     )
 
 
@@ -345,9 +349,13 @@ def state_transition_with_full_block(
     participation_fn=None,
     sync_aggregate=None,
     block=None,
+    envelopes=None,
 ):
     """
     Build and apply a block with attestations at the calculated `slot_to_attest` of current epoch and/or previous epoch.
+
+    For Gloas: when ``envelopes`` is provided, also applies ``process_execution_payload`` to the state
+    and appends the signed envelope to the list.
     """
     if block is None:
         block = build_empty_block_for_next_slot(spec, state)
@@ -366,6 +374,11 @@ def state_transition_with_full_block(
         block.body.sync_aggregate = sync_aggregate
 
     signed_block = state_transition_and_sign_block(spec, state, block)
+
+    if envelopes is not None and is_post_gloas(spec):
+        signed_envelope = reveal_payload_to_state(spec, state)
+        envelopes.append(signed_envelope)
+
     return signed_block
 
 
diff --git a/tests/core/pyspec/eth_consensus_specs/test/helpers/execution_payload.py b/tests/core/pyspec/eth_consensus_specs/test/helpers/execution_payload.py
index ca8e13c74c..4c8ee35621 100644
--- a/tests/core/pyspec/eth_consensus_specs/test/helpers/execution_payload.py
+++ b/tests/core/pyspec/eth_consensus_specs/test/helpers/execution_payload.py
@@ -13,7 +13,7 @@
     is_post_electra,
     is_post_gloas,
 )
-from eth_consensus_specs.test.helpers.keys import builder_privkeys
+from eth_consensus_specs.test.helpers.keys import builder_privkeys, privkeys
 from eth_consensus_specs.test.helpers.withdrawals import get_expected_withdrawals
 from eth_consensus_specs.utils.ssz.ssz_impl import hash_tree_root
 
@@ -478,3 +478,133 @@ def build_state_with_execution_payload_bid(spec, state, execution_payload_bid):
 
 def get_random_tx(rng):
     return get_random_bytes_list(rng, rng.randint(1, 1000))
+
+
+def prepare_execution_payload_envelope(
+    spec,
+    state,
+    builder_index=None,
+    slot=None,
+    beacon_block_root=None,
+    state_root=None,
+    execution_payload=None,
+    execution_requests=None,
+    valid_signature=True,
+):
+    """
+    Helper to create a signed execution payload envelope with customizable parameters.
+    Note: This should be called AFTER setting up the state with the committed bid.
+    """
+    if builder_index is None:
+        builder_index = spec.BUILDER_INDEX_SELF_BUILD
+
+    if slot is None:
+        slot = state.slot
+
+    if beacon_block_root is None:
+        # Cache latest block header state root if not already set
+        if state.latest_block_header.state_root == spec.Root():
+            state.latest_block_header.state_root = state.hash_tree_root()
+        beacon_block_root = state.latest_block_header.hash_tree_root()
+
+    if execution_payload is None:
+        execution_payload = build_empty_execution_payload(spec, state)
+
+    if execution_requests is None:
+        execution_requests = spec.ExecutionRequests(
+            deposits=spec.List[spec.DepositRequest, spec.MAX_DEPOSIT_REQUESTS_PER_PAYLOAD](),
+            withdrawals=spec.List[
+                spec.WithdrawalRequest, spec.MAX_WITHDRAWAL_REQUESTS_PER_PAYLOAD
+            ](),
+            consolidations=spec.List[
+                spec.ConsolidationRequest, spec.MAX_CONSOLIDATION_REQUESTS_PER_PAYLOAD
+            ](),
+        )
+
+    # Create a copy of state for computing state_root after execution payload processing
+    if state_root is None:
+        post_state = state.copy()
+        # Simulate the state changes that process_execution_payload will make
+
+        # Cache latest block header state root if empty (matches process_execution_payload)
+        previous_state_root = post_state.hash_tree_root()
+        if post_state.latest_block_header.state_root == spec.Root():
+            post_state.latest_block_header.state_root = previous_state_root
+
+        # Process execution requests if any
+        if execution_requests is not None:
+            for deposit in execution_requests.deposits:
+                spec.process_deposit_request(post_state, deposit)
+            for withdrawal in execution_requests.withdrawals:
+                spec.process_withdrawal_request(post_state, withdrawal)
+            for consolidation in execution_requests.consolidations:
+                spec.process_consolidation_request(post_state, consolidation)
+
+        # Process builder payment (only if amount > 0)
+        payment = post_state.builder_pending_payments[
+            spec.SLOTS_PER_EPOCH + state.slot % spec.SLOTS_PER_EPOCH
+        ]
+        if payment.withdrawal.amount > 0:
+            post_state.builder_pending_withdrawals.append(payment.withdrawal)
+
+        # Clear the pending payment
+        post_state.builder_pending_payments[
+            spec.SLOTS_PER_EPOCH + state.slot % spec.SLOTS_PER_EPOCH
+        ] = spec.BuilderPendingPayment()
+
+        # Update execution payload availability and latest block hash
+        post_state.execution_payload_availability[state.slot % spec.SLOTS_PER_HISTORICAL_ROOT] = 0b1
+        post_state.latest_block_hash = execution_payload.block_hash
+        state_root = post_state.hash_tree_root()
+
+    envelope = spec.ExecutionPayloadEnvelope(
+        payload=execution_payload,
+        execution_requests=execution_requests,
+        builder_index=builder_index,
+        beacon_block_root=beacon_block_root,
+        slot=slot,
+        state_root=state_root,
+    )
+
+    if valid_signature:
+        if envelope.builder_index == spec.BUILDER_INDEX_SELF_BUILD:
+            privkey = privkeys[state.latest_block_header.proposer_index]
+        else:
+            privkey = builder_privkeys[envelope.builder_index]
+        signature = spec.get_execution_payload_envelope_signature(
+            state,
+            envelope,
+            privkey,
+        )
+    else:
+        # Invalid signature
+        signature = spec.BLSSignature()
+
+    return spec.SignedExecutionPayloadEnvelope(
+        message=envelope,
+        signature=signature,
+    )
+
+
+def reveal_payload_to_state(spec, state, execution_requests=None):
+    """
+    For Gloas and later: create an execution payload envelope and apply
+    ``process_execution_payload`` to the state. Returns the signed envelope.
+
+    This should be called AFTER ``state_transition_and_sign_block``,
+    when the state has latest_execution_payload_bid set from the block's bid.
+    """
+    execution_payload = build_empty_execution_payload(spec, state)
+    # Override block_hash to match the committed bid
+    execution_payload.block_hash = state.latest_execution_payload_bid.block_hash
+
+    signed_envelope = prepare_execution_payload_envelope(
+        spec,
+        state,
+        execution_payload=execution_payload,
+        execution_requests=execution_requests,
+    )
+
+    spec.process_execution_payload(state, signed_envelope, spec.NoopExecutionEngine())
+
+    return signed_envelope
diff --git a/tests/core/pyspec/eth_consensus_specs/test/helpers/fork_choice.py b/tests/core/pyspec/eth_consensus_specs/test/helpers/fork_choice.py
index 367997ffd6..0683ffc224 100644
--- a/tests/core/pyspec/eth_consensus_specs/test/helpers/fork_choice.py
+++ b/tests/core/pyspec/eth_consensus_specs/test/helpers/fork_choice.py
@@ -169,6 +169,7 @@ def tick_and_add_block(
     block_not_found=False,
     is_optimistic=False,
     blob_data: BlobData | None = None,
+    envelope=None,
 ):
     pre_state = store.block_states[signed_block.message.parent_root]
     if merge_block:
@@ -195,6 +196,10 @@ def tick_and_add_block(
         blob_data=blob_data,
     )
 
+    # For Gloas: apply on_execution_payload to the store when an explicit envelope is provided.
+    if envelope is not None and valid and post_state is not None:
+        spec.on_execution_payload(store, envelope)
+
     return post_state
 
 
@@ -503,19 +508,38 @@ def apply_next_epoch_with_attestations(
 
 
 def apply_next_slots_with_attestations(
-    spec, state, store, slots, fill_cur_epoch, fill_prev_epoch, test_steps, participation_fn=None
+    spec,
+    state,
+    store,
+    slots,
+    fill_cur_epoch,
+    fill_prev_epoch,
+    test_steps,
+    participation_fn=None,
+    with_payload_reveal=False,
 ):
+    envelopes = [] if with_payload_reveal else None
     _, new_signed_blocks, post_state = next_slots_with_attestations(
-        spec, state, slots, fill_cur_epoch, fill_prev_epoch, participation_fn=participation_fn
+        spec,
+        state,
+        slots,
+        fill_cur_epoch,
+        fill_prev_epoch,
+        participation_fn=participation_fn,
+        envelopes=envelopes,
     )
-    for signed_block in new_signed_blocks:
+    for i, signed_block in enumerate(new_signed_blocks):
         block = signed_block.message
-        yield from tick_and_add_block(spec, store, signed_block, test_steps)
+        envelope = envelopes[i] if envelopes else None
+        yield from tick_and_add_block(spec, store, signed_block, test_steps, envelope=envelope)
         block_root = block.hash_tree_root()
         assert store.blocks[block_root] == block
         last_signed_block = signed_block
 
-    assert store.block_states[block_root].hash_tree_root() == post_state.hash_tree_root()
+    if with_payload_reveal:
+        assert store.payload_states[block_root].hash_tree_root() == post_state.hash_tree_root()
+    else:
+        assert store.block_states[block_root].hash_tree_root() == post_state.hash_tree_root()
 
     return post_state, store, last_signed_block