diff --git a/engine/common/rpc/convert/execution_results_test.go b/engine/common/rpc/convert/execution_results_test.go
index cce0bd175e6..9529fc7769e 100644
--- a/engine/common/rpc/convert/execution_results_test.go
+++ b/engine/common/rpc/convert/execution_results_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
+// TODO: fails with input non-nil ChunkBody.ServiceEventCount
 func TestConvertExecutionResult(t *testing.T) {
 	t.Parallel()
 
@@ -25,6 +26,7 @@ func TestConvertExecutionResult(t *testing.T) {
 	assert.Equal(t, er, converted)
 }
 
+// TODO: fails with input non-nil ChunkBody.ServiceEventCount
 func TestConvertExecutionResults(t *testing.T) {
 	t.Parallel()
 
@@ -43,6 +45,7 @@ func TestConvertExecutionResults(t *testing.T) {
 	assert.Equal(t, results, converted)
 }
 
+// TODO: fails with input non-nil ChunkBody.ServiceEventCount
 func TestConvertExecutionResultMetaList(t *testing.T) {
 	t.Parallel()
 
diff --git a/engine/execution/block_result.go b/engine/execution/block_result.go
index 1cfbb9bc0d4..905ea58eb2d 100644
--- a/engine/execution/block_result.go
+++ b/engine/execution/block_result.go
@@ -2,6 +2,7 @@ package execution
 
 import (
 	"fmt"
+	"math"
 
 	"github.com/onflow/flow-go/fvm/storage/snapshot"
 	"github.com/onflow/flow-go/model/flow"
@@ -49,6 +50,18 @@ func (er *BlockExecutionResult) AllEvents() flow.EventsList {
 	return res
 }
 
+// ServiceEventCountForChunk returns the number of service events emitted in the given chunk.
+func (er *BlockExecutionResult) ServiceEventCountForChunk(chunkIndex int) uint16 {
+	serviceEventCount := len(er.collectionExecutionResults[chunkIndex].serviceEvents)
+	if serviceEventCount > math.MaxUint16 {
+		// The current protocol demands that the ServiceEventCount does not exceed 65535.
+		// For defensive programming, we explicitly enforce this limit as 65k could be produced by a bug.
+		// Execution nodes would be first to realize that this bound is violated, and crash (fail early).
+		panic(fmt.Sprintf("service event count (%d) exceeds maximum value of 65535", serviceEventCount))
+	}
+	return uint16(serviceEventCount)
+}
+
 func (er *BlockExecutionResult) AllServiceEvents() flow.EventsList {
 	res := make(flow.EventsList, 0)
 	for _, ce := range er.collectionExecutionResults {
@@ -199,6 +212,7 @@ func (ar *BlockAttestationResult) ChunkAt(index int) *flow.Chunk {
 		attestRes.startStateCommit,
 		len(execRes.TransactionResults()),
 		attestRes.eventCommit,
+		ar.ServiceEventCountForChunk(index),
 		attestRes.endStateCommit,
 		execRes.executionSnapshot.TotalComputationUsed(),
 	)
diff --git a/engine/execution/block_result_test.go b/engine/execution/block_result_test.go
new file mode 100644
index 00000000000..a96e3576728
--- /dev/null
+++ b/engine/execution/block_result_test.go
@@ -0,0 +1,81 @@
+package execution
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/onflow/flow-go/utils/slices"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// makeBlockExecutionResultFixture makes a BlockExecutionResult fixture
+// with the specified allocation of service events to chunks.
+func makeBlockExecutionResultFixture(serviceEventsPerChunk []int) *BlockExecutionResult {
+	fixture := new(BlockExecutionResult)
+	for _, nServiceEvents := range serviceEventsPerChunk {
+		fixture.collectionExecutionResults = append(fixture.collectionExecutionResults,
+			CollectionExecutionResult{
+				serviceEvents:          unittest.EventsFixture(nServiceEvents),
+				convertedServiceEvents: unittest.ServiceEventsFixture(nServiceEvents),
+			})
+	}
+	return fixture
+}
+
+// Tests that ServiceEventCountForChunk method works as expected under various circumstances:
+func TestBlockExecutionResult_ServiceEventCountForChunk(t *testing.T) {
+	t.Run("no service events", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 1 // always contains at least system chunk
+		blockResult := makeBlockExecutionResultFixture(make([]int, nChunks))
+		// all chunks should have 0 service event count
+		for chunkIndex := 0; chunkIndex < nChunks; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(0), count)
+		}
+	})
+	t.Run("service events only in system chunk", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 2 // at least 2 chunks
+		// add between 1 and 10 service events, all in the system chunk
+		serviceEventAllocation := make([]int, nChunks)
+		nServiceEvents := rand.Intn(10) + 1
+		serviceEventAllocation[nChunks-1] = nServiceEvents
+
+		blockResult := makeBlockExecutionResultFixture(serviceEventAllocation)
+		// all non-system chunks should have zero service event count
+		for chunkIndex := 0; chunkIndex < nChunks-1; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(0), count)
+		}
+		// the system chunk should contain all service events
+		assert.Equal(t, uint16(nServiceEvents), blockResult.ServiceEventCountForChunk(nChunks-1))
+	})
+	t.Run("service events only outside system chunk", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 2 // at least 2 chunks
+		// add 1 service event to all non-system chunks
+		serviceEventAllocation := slices.Fill(1, nChunks)
+		serviceEventAllocation[nChunks-1] = 0
+
+		blockResult := makeBlockExecutionResultFixture(serviceEventAllocation)
+		// all non-system chunks should have 1 service event
+		for chunkIndex := 0; chunkIndex < nChunks-1; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(1), count)
+		}
+		// the system chunk service event count should include all service events
+		assert.Equal(t, uint16(0), blockResult.ServiceEventCountForChunk(nChunks-1))
+	})
+	t.Run("service events in both system chunk and other chunks", func(t *testing.T) {
+		nChunks := rand.Intn(10) + 2 // at least 2 chunks
+		// add 1 service event to all chunks (including system chunk)
+		serviceEventAllocation := slices.Fill(1, nChunks)
+
+		blockResult := makeBlockExecutionResultFixture(serviceEventAllocation)
+		// all chunks should have service event count of 1
+		for chunkIndex := 0; chunkIndex < nChunks; chunkIndex++ {
+			count := blockResult.ServiceEventCountForChunk(chunkIndex)
+			assert.Equal(t, uint16(1), count)
+		}
+	})
+}
diff --git a/model/encoding/rlp/rlp_test.go b/model/encoding/rlp/rlp_test.go
new file mode 100644
index 00000000000..4b07e5d8a71
--- /dev/null
+++ b/model/encoding/rlp/rlp_test.go
@@ -0,0 +1,29 @@
+package rlp_test
+
+import (
+	"testing"
+
+	"github.com/onflow/go-ethereum/rlp"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRLPStructFieldOrder tests the field ordering property of RLP encoding.
+// It provides evidence that RLP encoding depends on struct field ordering.
+func TestRLPStructFieldOrder(t *testing.T) {
+	a := struct {
+		A uint32 // A first
+		B uint32
+	}{A: 2, B: 3}
+
+	b := struct {
+		B uint32 // B first
+		A uint32
+	}{A: 2, B: 3}
+
+	abin, err := rlp.EncodeToBytes(a)
+	require.NoError(t, err)
+	bbin, err := rlp.EncodeToBytes(b)
+	require.NoError(t, err)
+	assert.NotEqual(t, abin, bbin)
+}
diff --git a/model/flow/chunk.go b/model/flow/chunk.go
index 83eabde4b1e..d7ebfe4f102 100644
--- a/model/flow/chunk.go
+++ b/model/flow/chunk.go
@@ -2,9 +2,11 @@ package flow
 
 import (
 	"fmt"
+	"io"
 	"log"
 
 	"github.com/ipfs/go-cid"
+	"github.com/onflow/go-ethereum/rlp"
 	"github.com/vmihailenco/msgpack/v4"
 )
 
@@ -20,19 +22,111 @@ func init() {
 	}
 }
 
+// ChunkBodyV0 is the prior version of ChunkBody, used for computing backward-compatible IDs and tests.
+// Compared to ChunkBody, ChunkBodyV0 does not have the ServiceEventCount field.
+// Deprecated: to be removed in Mainnet27
+// TODO(mainnet27): Remove this data structure https://github.com/onflow/flow-go/issues/6773
+type ChunkBodyV0 struct {
+	CollectionIndex      uint
+	StartState           StateCommitment
+	EventCollection      Identifier
+	BlockID              Identifier
+	TotalComputationUsed uint64
+	NumberOfTransactions uint64
+}
+
 type ChunkBody struct {
 	CollectionIndex uint
 
 	// execution info
 	StartState      StateCommitment // start state when starting executing this chunk
 	EventCollection Identifier      // Events generated by executing results
-	BlockID         Identifier      // Block id of the execution result this chunk belongs to
+	// ServiceEventCount defines how many service events were emitted in this chunk.
+	// By reading these fields from the prior chunks in the same ExecutionResult, we can
+	// compute exactly what service events were emitted in this chunk.
+	//
+	// Let C be this chunk, K be the set of chunks in the ExecutionResult containing C.
+	// Then the service event indices for C are given by:
+	//    StartIndex = ∑Ci.ServiceEventCount : Ci ∈ K, Ci.Index < C.Index
+	//    EndIndex = StartIndex + C.ServiceEventCount
+	// The service events for C are given by:
+	//    ExecutionResult.ServiceEvents[StartIndex:EndIndex]
+	//
+	// BACKWARD COMPATIBILITY:
+	// (1) If ServiceEventCount is nil, this indicates that this chunk was created by an older software version
+	//     which did not support specifying a mapping between chunks and service events.
+	//     In this case, all service events are assumed to have been emitted in the system chunk (last chunk).
+	//     This was the implicit behaviour prior to the introduction of this field.
+	// (2) Otherwise, ServiceEventCount must be non-nil.
+	// Within an ExecutionResult, all chunks must use either representation (1) or (2), not both.
+	// TODO(mainnet27): make this field non-pointer https://github.com/onflow/flow-go/issues/6773
+	ServiceEventCount *uint16
+	BlockID           Identifier // Block id of the execution result this chunk belongs to
 
 	// Computation consumption info
 	TotalComputationUsed uint64 // total amount of computation used by running all txs in this chunk
 	NumberOfTransactions uint64 // number of transactions inside the collection
 }
 
+// We TEMPORARILY implement the [rlp.Encoder] interface to implement backwards-compatible ID computation.
+// TODO(mainnet27): remove EncodeRLP methods on Chunk and ChunkBody https://github.com/onflow/flow-go/issues/6773
+var _ rlp.Encoder = &ChunkBody{}
+
+// EncodeRLP defines custom encoding logic for the ChunkBody type.
+// NOTE: For correct operation when encoding a larger structure containing ChunkBody,
+// this method depends on Chunk also overriding EncodeRLP. Otherwise, since ChunkBody
+// is an embedded field, the RLP encoder will skip Chunk fields besides those in ChunkBody.
+//
+// The encoding is defined for backward compatibility with prior data model version (ChunkBodyV0):
+//   - All new ChunkBody instances must have non-nil ServiceEventCount field
+//   - A nil ServiceEventCount field indicates a v0 version of ChunkBody
+//   - when computing the ID of such a ChunkBody, the ServiceEventCount field is omitted from the fingerprint
+//
+// No errors expected during normal operations.
+// TODO(mainnet27): remove this method https://github.com/onflow/flow-go/issues/6773
+func (ch ChunkBody) EncodeRLP(w io.Writer) error {
+	var err error
+	if ch.ServiceEventCount == nil {
+		err = rlp.Encode(w, struct {
+			CollectionIndex      uint
+			StartState           StateCommitment
+			EventCollection      Identifier
+			BlockID              Identifier
+			TotalComputationUsed uint64
+			NumberOfTransactions uint64
+		}{
+			CollectionIndex:      ch.CollectionIndex,
+			StartState:           ch.StartState,
+			EventCollection:      ch.EventCollection,
+			BlockID:              ch.BlockID,
+			TotalComputationUsed: ch.TotalComputationUsed,
+			NumberOfTransactions: ch.NumberOfTransactions,
+		})
+	} else {
+		err = rlp.Encode(w, struct {
+			CollectionIndex      uint
+			StartState           StateCommitment
+			EventCollection      Identifier
+			ServiceEventCount    *uint16
+			BlockID              Identifier
+			TotalComputationUsed uint64
+			NumberOfTransactions uint64
+		}{
+			CollectionIndex:      ch.CollectionIndex,
+			StartState:           ch.StartState,
+			EventCollection:      ch.EventCollection,
+			ServiceEventCount:    ch.ServiceEventCount,
+			BlockID:              ch.BlockID,
+			TotalComputationUsed: ch.TotalComputationUsed,
+			NumberOfTransactions: ch.NumberOfTransactions,
+		})
+	}
+	if err != nil {
+		return fmt.Errorf("failed to rlp encode ChunkBody: %w", err)
+	}
+	return nil
+}
+
 type Chunk struct {
 	ChunkBody
 
@@ -41,12 +135,34 @@ type Chunk struct {
 	EndState StateCommitment
 }
 
+// We TEMPORARILY implement the [rlp.Encoder] interface to implement backwards-compatible ID computation.
+// TODO(mainnet27): remove EncodeRLP methods on Chunk and ChunkBody https://github.com/onflow/flow-go/issues/6773
+var _ rlp.Encoder = &Chunk{}
+
+// EncodeRLP defines custom encoding logic for the Chunk type.
+// This method exists only so that the embedded ChunkBody's EncodeRLP method is
+// not interpreted as the RLP encoding for the entire Chunk.
+// No errors expected during normal operation.
+// TODO(mainnet27): remove this method https://github.com/onflow/flow-go/issues/6773
+func (ch Chunk) EncodeRLP(w io.Writer) error {
+	return rlp.Encode(w, struct {
+		ChunkBody ChunkBody
+		Index     uint64
+		EndState  StateCommitment
+	}{
+		ChunkBody: ch.ChunkBody,
+		Index:     ch.Index,
+		EndState:  ch.EndState,
+	})
+}
+
 func NewChunk(
 	blockID Identifier,
 	collectionIndex int,
 	startState StateCommitment,
 	numberOfTransactions int,
 	eventCollection Identifier,
+	serviceEventCount uint16,
 	endState StateCommitment,
 	totalComputationUsed uint64,
 ) *Chunk {
@@ -57,6 +173,7 @@ func NewChunk(
 			StartState:           startState,
 			NumberOfTransactions: uint64(numberOfTransactions),
 			EventCollection:      eventCollection,
+			ServiceEventCount:    &serviceEventCount,
 			TotalComputationUsed: totalComputationUsed,
 		},
 		Index:    uint64(collectionIndex),
diff --git a/model/flow/chunk_test.go b/model/flow/chunk_test.go
index 9da330dcaaa..1af4fd4db2c 100644
--- a/model/flow/chunk_test.go
+++ b/model/flow/chunk_test.go
@@ -1,11 +1,14 @@
 package flow_test
 
 import (
+	"encoding/json"
 	"testing"
 
+	"github.com/fxamacker/cbor/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/onflow/flow-go/model/fingerprint"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/utils/rand"
 	"github.com/onflow/flow-go/utils/unittest"
@@ -117,6 +120,7 @@ func TestChunkIndexIsSet(t *testing.T) {
 		unittest.StateCommitmentFixture(),
 		21,
 		unittest.IdentifierFixture(),
+		0,
 		unittest.StateCommitmentFixture(),
 		17995,
 	)
@@ -135,6 +139,7 @@ func TestChunkNumberOfTxsIsSet(t *testing.T) {
 		unittest.StateCommitmentFixture(),
 		int(i),
 		unittest.IdentifierFixture(),
+		0,
 		unittest.StateCommitmentFixture(),
 		17995,
 	)
@@ -152,9 +157,152 @@ func TestChunkTotalComputationUsedIsSet(t *testing.T) {
 		unittest.StateCommitmentFixture(),
 		21,
 		unittest.IdentifierFixture(),
+		0,
 		unittest.StateCommitmentFixture(),
 		i,
 	)
 
 	assert.Equal(t, i, chunk.TotalComputationUsed)
 }
+
+// TestChunkEncodeDecode test encoding and decoding properties.
+// In particular, we confirm that `nil` values of the ServiceEventCount field are preserved (and
+// not conflated with 0) by the encoding schemes we use, because this difference is meaningful and
+// important for backward compatibility (see [ChunkBody.ServiceEventCount] for details).
+func TestChunkEncodeDecode(t *testing.T) {
+	chunk := unittest.ChunkFixture(unittest.IdentifierFixture(), 0)
+
+	t.Run("encode/decode preserves nil ServiceEventCount", func(t *testing.T) {
+		chunk.ServiceEventCount = nil
+		t.Run("json", func(t *testing.T) {
+			bz, err := json.Marshal(chunk)
+			require.NoError(t, err)
+			unmarshaled := new(flow.Chunk)
+			err = json.Unmarshal(bz, unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunk, unmarshaled)
+			assert.Nil(t, unmarshaled.ServiceEventCount)
+		})
+		t.Run("cbor", func(t *testing.T) {
+			bz, err := cbor.Marshal(chunk)
+			require.NoError(t, err)
+			unmarshaled := new(flow.Chunk)
+			err = cbor.Unmarshal(bz, unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunk, unmarshaled)
+			assert.Nil(t, unmarshaled.ServiceEventCount)
+		})
+	})
+	t.Run("encode/decode preserves empty but non-nil ServiceEventCount", func(t *testing.T) {
+		chunk.ServiceEventCount = unittest.PtrTo[uint16](0)
+		t.Run("json", func(t *testing.T) {
+			bz, err := json.Marshal(chunk)
+			require.NoError(t, err)
+			unmarshaled := new(flow.Chunk)
+			err = json.Unmarshal(bz, unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunk, unmarshaled)
+			assert.NotNil(t, unmarshaled.ServiceEventCount)
+		})
+		t.Run("cbor", func(t *testing.T) {
+			bz, err := cbor.Marshal(chunk)
+			require.NoError(t, err)
+			unmarshaled := new(flow.Chunk)
+			err = cbor.Unmarshal(bz, unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunk, unmarshaled)
+			assert.NotNil(t, unmarshaled.ServiceEventCount)
+		})
+	})
+}
+
+// TestChunk_ModelVersions_EncodeDecode tests that encoding and decoding between
+// supported versions works as expected.
+func TestChunk_ModelVersions_EncodeDecode(t *testing.T) {
+	chunkFixture := unittest.ChunkFixture(unittest.IdentifierFixture(), 1)
+	chunkFixture.ServiceEventCount = unittest.PtrTo[uint16](0) // non-nil extra field
+
+	t.Run("encoding v0 and decoding it into v1 should yield nil for ServiceEventCount", func(t *testing.T) {
+		var chunkv0 flow.ChunkBodyV0
+		unittest.EncodeDecodeDifferentVersions(t, chunkFixture.ChunkBody, &chunkv0)
+
+		t.Run("json", func(t *testing.T) {
+			bz, err := json.Marshal(chunkv0)
+			require.NoError(t, err)
+
+			var unmarshaled flow.ChunkBody
+			err = json.Unmarshal(bz, &unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunkv0.EventCollection, unmarshaled.EventCollection)
+			assert.Equal(t, chunkv0.BlockID, unmarshaled.BlockID)
+			assert.Nil(t, unmarshaled.ServiceEventCount)
+		})
+
+		t.Run("cbor", func(t *testing.T) {
+			bz, err := cbor.Marshal(chunkv0)
+			require.NoError(t, err)
+
+			var unmarshaled flow.ChunkBody
+			err = cbor.Unmarshal(bz, &unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunkv0.EventCollection, unmarshaled.EventCollection)
+			assert.Equal(t, chunkv0.BlockID, unmarshaled.BlockID)
+			assert.Nil(t, unmarshaled.ServiceEventCount)
+		})
+	})
+	t.Run("encoding v1 and decoding it into v0 should not error", func(t *testing.T) {
+		chunkv1 := chunkFixture.ChunkBody
+		chunkv1.ServiceEventCount = unittest.PtrTo[uint16](0) // ensure non-nil ServiceEventCount field
+
+		t.Run("json", func(t *testing.T) {
+			bz, err := json.Marshal(chunkv1)
+			require.NoError(t, err)
+
+			var unmarshaled flow.ChunkBodyV0
+			err = json.Unmarshal(bz, &unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunkv1.EventCollection, unmarshaled.EventCollection)
+			assert.Equal(t, chunkv1.BlockID, unmarshaled.BlockID)
+		})
+		t.Run("cbor", func(t *testing.T) {
+			bz, err := cbor.Marshal(chunkv1)
+			require.NoError(t, err)
+
+			var unmarshaled flow.ChunkBodyV0
+			err = cbor.Unmarshal(bz, &unmarshaled)
+			require.NoError(t, err)
+			assert.Equal(t, chunkv1.EventCollection, unmarshaled.EventCollection)
+			assert.Equal(t, chunkv1.BlockID, unmarshaled.BlockID)
+		})
+	})
+}
+
+// FingerprintBackwardCompatibility ensures that the Fingerprint and ID functions
+// are backward compatible with old data model versions. We emulate the
+// case where a peer running an older software version receives a `ChunkBody` that
+// was encoded in the new version. Specifically, if the new ServiceEventCount field
+// is nil, then the new model should produce IDs consistent with the old model.
+//
+// Backward compatibility is implemented by providing a custom EncodeRLP method.
+func TestChunk_FingerprintBackwardCompatibility(t *testing.T) {
+	chunk := unittest.ChunkFixture(unittest.IdentifierFixture(), 1)
+	chunk.ServiceEventCount = nil
+	chunkBody := chunk.ChunkBody
+	var chunkBodyV0 flow.ChunkBodyV0
+	unittest.EncodeDecodeDifferentVersions(t, chunkBody, &chunkBodyV0)
+
+	// A nil ServiceEventCount fields indicates a prior model version.
+	// The ID calculation for the old and new model version should be the same.
+	t.Run("nil ServiceEventCount fields", func(t *testing.T) {
+		chunkBody.ServiceEventCount = nil
+		assert.Equal(t, flow.MakeID(chunkBodyV0), flow.MakeID(chunkBody))
+		assert.Equal(t, fingerprint.Fingerprint(chunkBodyV0), fingerprint.Fingerprint(chunkBody))
+	})
+	// A non-nil ServiceEventCount fields indicates an up-to-date model version.
+	// The ID calculation for the old and new model version should be different,
+	// because the new model should include the ServiceEventCount field value.
+	t.Run("non-nil ServiceEventCount fields", func(t *testing.T) {
+		chunkBody.ServiceEventCount = unittest.PtrTo[uint16](0)
+		assert.NotEqual(t, flow.MakeID(chunkBodyV0), flow.MakeID(chunkBody))
+	})
+}
diff --git a/model/flow/execution_result.go b/model/flow/execution_result.go
index a24af0be53c..5a4fd9b91a4 100644
--- a/model/flow/execution_result.go
+++ b/model/flow/execution_result.go
@@ -43,7 +43,7 @@ func (er ExecutionResult) Checksum() Identifier {
 	return MakeID(er)
 }
 
-// ValidateChunksLength checks whether the number of chuncks is zero.
+// ValidateChunksLength checks whether the number of chunks is zero.
 //
 // It returns false if the number of chunks is zero (invalid).
 // By protocol definition, each ExecutionReceipt must contain at least one
@@ -76,6 +76,38 @@ func (er ExecutionResult) InitialStateCommit() (StateCommitment, error) {
 	return er.Chunks[0].StartState, nil
 }
 
+// SystemChunk is a system-generated chunk added to every block.
+// It is always the final chunk in an execution result.
+func (er ExecutionResult) SystemChunk() *Chunk {
+	return er.Chunks[len(er.Chunks)-1]
+}
+
+// ServiceEventsByChunk returns the list of service events emitted during the given chunk.
+func (er ExecutionResult) ServiceEventsByChunk(chunkIndex uint64) ServiceEventList {
+	serviceEventCount := er.Chunks[chunkIndex].ServiceEventCount
+	// CASE 1: Service event count is specified (non-nil)
+	if serviceEventCount != nil {
+		if *serviceEventCount == 0 {
+			return nil
+		}
+
+		startIndex := 0
+		for i := uint64(0); i < chunkIndex; i++ {
+			startIndex += int(*er.Chunks[i].ServiceEventCount)
+		}
+		return er.ServiceEvents[startIndex : startIndex+int(*serviceEventCount)]
+	}
+	// CASE 2: Service event count omitted (nil)
+	// This indicates the chunk was generated in an older data model version.
+	// In this case, all service events associated with the result are assumed
+	// to have been emitted within the system chunk (last chunk)
+	// TODO(mainnet27): remove this path https://github.com/onflow/flow-go/issues/6773
+	if chunkIndex == er.SystemChunk().Index {
+		return er.ServiceEvents
+	}
+	return nil
+}
+
 func (er ExecutionResult) MarshalJSON() ([]byte, error) {
 	type Alias ExecutionResult
 	return json.Marshal(struct {
diff --git a/model/flow/execution_result_test.go b/model/flow/execution_result_test.go
index b697488b86d..6e5a86228de 100644
--- a/model/flow/execution_result_test.go
+++ b/model/flow/execution_result_test.go
@@ -1,10 +1,12 @@
 package flow_test
 
 import (
+	"math/rand"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/onflow/flow-go/model/fingerprint"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/utils/unittest"
 )
@@ -41,3 +43,140 @@ func TestExecutionResultGroupBy(t *testing.T) {
 	unknown := groups.GetGroup(unittest.IdentifierFixture())
 	assert.Equal(t, 0, unknown.Size())
 }
+
+// FingerprintBackwardCompatibility ensures that the Fingerprint and ID functions
+// are backward compatible with old data model versions. Specifically, if the new
+// ServiceEventCount field is nil, then the new model should produce IDs consistent
+// with the old model.
+//
+// Backward compatibility is implemented by providing a custom EncodeRLP method.
+func TestExecutionResult_FingerprintBackwardCompatibility(t *testing.T) {
+	// Define a series of types which use flow.ChunkBodyV0
+	type ChunkV0 struct {
+		flow.ChunkBodyV0
+		Index    uint64
+		EndState flow.StateCommitment
+	}
+	type ChunkListV0 []*ChunkV0
+	type ExecutionResultV0 struct {
+		PreviousResultID flow.Identifier
+		BlockID          flow.Identifier
+		Chunks           ChunkListV0
+		ServiceEvents    flow.ServiceEventList
+		ExecutionDataID  flow.Identifier
+	}
+
+	// Construct an ExecutionResult with nil ServiceEventCount fields
+	result := unittest.ExecutionResultFixture()
+	for i := range result.Chunks {
+		result.Chunks[i].ServiceEventCount = nil
+	}
+
+	// Copy all fields to the prior-version model
+	var resultv0 ExecutionResultV0
+	unittest.EncodeDecodeDifferentVersions(t, result, &resultv0)
+
+	assert.Equal(t, result.ID(), flow.MakeID(resultv0))
+	assert.Equal(t, fingerprint.Fingerprint(result), fingerprint.Fingerprint(resultv0))
+}
+
+// Tests that [ExecutionResult.ServiceEventsByChunk] method works in a variety of circumstances.
+// It also tests the method against an ExecutionResult instance backed by both the
+// current and old data model version (with and with ServiceEventCount field)
+func TestExecutionResult_ServiceEventsByChunk(t *testing.T) {
+	t.Run("no service events", func(t *testing.T) {
+		t.Run("nil ServiceEventCount field (old model)", func(t *testing.T) {
+			result := unittest.ExecutionResultFixture()
+			for _, chunk := range result.Chunks {
+				chunk.ServiceEventCount = nil
+			}
+			// should return empty list for all chunks
+			for chunkIndex := 0; chunkIndex < result.Chunks.Len(); chunkIndex++ {
+				serviceEvents := result.ServiceEventsByChunk(uint64(chunkIndex))
+				assert.Len(t, serviceEvents, 0)
+			}
+		})
+		t.Run("populated ServiceEventCount field", func(t *testing.T) {
+			result := unittest.ExecutionResultFixture()
+			for _, chunk := range result.Chunks {
+				chunk.ServiceEventCount = unittest.PtrTo[uint16](0)
+			}
+			// should return empty list for all chunks
+			for chunkIndex := 0; chunkIndex < result.Chunks.Len(); chunkIndex++ {
+				serviceEvents := result.ServiceEventsByChunk(uint64(chunkIndex))
+				assert.Len(t, serviceEvents, 0)
+			}
+		})
+	})
+
+	t.Run("service events only in system chunk", func(t *testing.T) {
+		t.Run("nil ServiceEventCount field (old model)", func(t *testing.T) {
+			nServiceEvents := rand.Intn(10) + 1
+			result := unittest.ExecutionResultFixture(unittest.WithServiceEvents(nServiceEvents))
+			for _, chunk := range result.Chunks {
+				chunk.ServiceEventCount = nil
+			}
+
+			// should return empty list for all chunks
+			for chunkIndex := 0; chunkIndex < result.Chunks.Len()-1; chunkIndex++ {
+				serviceEvents := result.ServiceEventsByChunk(uint64(chunkIndex))
+				assert.Len(t, serviceEvents, 0)
+			}
+			// should return list of service events for system chunk
+			assert.Equal(t, result.ServiceEvents, result.ServiceEventsByChunk(result.SystemChunk().Index))
+		})
+		t.Run("populated ServiceEventCount field", func(t *testing.T) {
+			nServiceEvents := rand.Intn(10) + 1
+			result := unittest.ExecutionResultFixture(unittest.WithServiceEvents(nServiceEvents))
+			for _, chunk := range result.Chunks[:result.Chunks.Len()-1] {
+				chunk.ServiceEventCount = unittest.PtrTo[uint16](0)
+			}
+			result.SystemChunk().ServiceEventCount = unittest.PtrTo(uint16(nServiceEvents))
+
+			// should return empty list for all non-system chunks
+			for chunkIndex := 0; chunkIndex < result.Chunks.Len()-1; chunkIndex++ {
+				serviceEvents := result.ServiceEventsByChunk(uint64(chunkIndex))
+				assert.Len(t, serviceEvents, 0)
+			}
+			// should return list of service events for system chunk
+			assert.Equal(t, result.ServiceEvents, result.ServiceEventsByChunk(result.SystemChunk().Index))
+		})
+	})
+
+	// NOTE: service events in non-system chunks is unsupported by the old data model
+	t.Run("service only in non-system chunks", func(t *testing.T) {
+		result := unittest.ExecutionResultFixture()
+		unittest.WithServiceEvents(result.Chunks.Len() - 1)(result) // one service event per non-system chunk
+
+		for _, chunk := range result.Chunks {
+			// 1 service event per chunk
+			chunk.ServiceEventCount = unittest.PtrTo(uint16(1))
+		}
+		result.SystemChunk().ServiceEventCount = unittest.PtrTo(uint16(0))
+
+		// should return one service event per non-system chunk
+		for chunkIndex := 0; chunkIndex < result.Chunks.Len()-1; chunkIndex++ {
+			serviceEvents := result.ServiceEventsByChunk(uint64(chunkIndex))
+			assert.Equal(t, result.ServiceEvents[chunkIndex:chunkIndex+1], serviceEvents)
+		}
+		// should return empty list for system chunk
+		assert.Len(t, result.ServiceEventsByChunk(result.SystemChunk().Index), 0)
+	})
+
+	// NOTE: service events in non-system chunks is unsupported by the old data model
+	t.Run("service events in all chunks", func(t *testing.T) {
+		result := unittest.ExecutionResultFixture()
+		unittest.WithServiceEvents(result.Chunks.Len())(result) // one service event per chunk
+
+		for _, chunk := range result.Chunks {
+			// 1 service event per chunk
+			chunk.ServiceEventCount = unittest.PtrTo(uint16(1))
+		}
+
+		// should return one service event per chunk
+		for chunkIndex := 0; chunkIndex < result.Chunks.Len(); chunkIndex++ {
+			serviceEvents := result.ServiceEventsByChunk(uint64(chunkIndex))
+			assert.Equal(t, result.ServiceEvents[chunkIndex:chunkIndex+1], serviceEvents)
+		}
+	})
+}
diff --git a/module/chunks/chunkVerifier.go b/module/chunks/chunkVerifier.go
index e71bc8b63e8..f110aa8b2b4 100644
--- a/module/chunks/chunkVerifier.go
+++ b/module/chunks/chunkVerifier.go
@@ -291,14 +291,13 @@ func (fcv *ChunkVerifier) verifyTransactionsInContext(
 		return nil, chmodels.NewCFInvalidEventsCollection(chunk.EventCollection, eventsHash, chIndex, execResID, events)
 	}
 
-	if systemChunk {
-		equal, err := result.ServiceEvents.EqualTo(serviceEvents)
-		if err != nil {
-			return nil, fmt.Errorf("error while comparing service events: %w", err)
-		}
-		if !equal {
-			return nil, chmodels.CFInvalidServiceSystemEventsEmitted(result.ServiceEvents, serviceEvents, chIndex, execResID)
-		}
+	serviceEventsInChunk := result.ServiceEventsByChunk(chunk.Index)
+	equal, err := serviceEventsInChunk.EqualTo(serviceEvents)
+	if err != nil {
+		return nil, fmt.Errorf("error while comparing service events: %w", err)
+	}
+	if !equal {
+		return nil, chmodels.CFInvalidServiceSystemEventsEmitted(serviceEventsInChunk, serviceEvents, chIndex, execResID)
 	}
 
 	// Applying chunk updates to the partial trie.	This returns the expected
diff --git a/module/chunks/chunkVerifier_test.go b/module/chunks/chunkVerifier_test.go
index 4ea50cb3fed..6ac2edd8366 100644
--- a/module/chunks/chunkVerifier_test.go
+++ b/module/chunks/chunkVerifier_test.go
@@ -31,6 +31,7 @@ import (
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
+// eventsList is the set of events emitted by each transaction, by default
 var eventsList = flow.EventsList{
 	{
 		Type:             "event.someType",
@@ -58,7 +59,8 @@ var testChain = flow.Emulator
 var epochSetupEvent, _ = unittest.EpochSetupFixtureByChainID(testChain)
 var epochCommitEvent, _ = unittest.EpochCommitFixtureByChainID(testChain)
 
-var systemEventsList = []flow.Event{
+// serviceEventsList is the list of service events emitted by default.
+var serviceEventsList = []flow.Event{
 	epochSetupEvent,
 }
 
@@ -72,6 +74,10 @@ type ChunkVerifierTestSuite struct {
 	verifier *chunks.ChunkVerifier
 	ledger   *completeLedger.Ledger
 
+	// Below, snapshots and outputs map transaction scripts to execution artifacts
+	// Test cases can inject a script when constructing a chunk, then associate
+	// it with the desired execution artifacts by adding entries to these maps.
+	// If no entry exists, then the default snapshot/output is used.
 	snapshots map[string]*snapshot.ExecutionSnapshot
 	outputs   map[string]fvm.ProcedureOutput
 }
@@ -139,7 +145,7 @@ func TestChunkVerifier(t *testing.T) {
 
 // TestHappyPath tests verification of the baseline verifiable chunk
 func (s *ChunkVerifierTestSuite) TestHappyPath() {
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	spockSecret, err := s.verifier.Verify(vch)
@@ -151,7 +157,7 @@ func (s *ChunkVerifierTestSuite) TestHappyPath() {
 func (s *ChunkVerifierTestSuite) TestMissingRegisterTouchForUpdate() {
 	unittest.SkipUnless(s.T(), unittest.TEST_DEPRECATED, "Check new partial ledger for missing keys")
 
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// remove the second register touch
@@ -166,7 +172,7 @@ func (s *ChunkVerifierTestSuite) TestMissingRegisterTouchForUpdate() {
 func (s *ChunkVerifierTestSuite) TestMissingRegisterTouchForRead() {
 	unittest.SkipUnless(s.T(), unittest.TEST_DEPRECATED, "Check new partial ledger for missing keys")
 
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// remove the second register touch
@@ -181,7 +187,7 @@ func (s *ChunkVerifierTestSuite) TestMissingRegisterTouchForRead() {
 // the state commitment computed after updating the partial trie
 // doesn't match the one provided by the chunks
 func (s *ChunkVerifierTestSuite) TestWrongEndState() {
-	meta := s.GetTestSetup(s.T(), "wrongEndState", false)
+	meta := s.GetTestSetup(s.T(), "wrongEndState", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// modify calculated end state, which is different from the one provided by the vch
@@ -201,7 +207,7 @@ func (s *ChunkVerifierTestSuite) TestWrongEndState() {
 // of failed transaction. if a transaction fails, it should
 // still change the state commitment.
 func (s *ChunkVerifierTestSuite) TestFailedTx() {
-	meta := s.GetTestSetup(s.T(), "failedTx", false)
+	meta := s.GetTestSetup(s.T(), "failedTx", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// modify the FVM output to include a failing tx. the input already has a failing tx, but we need to
@@ -223,7 +229,7 @@ func (s *ChunkVerifierTestSuite) TestFailedTx() {
 // TestEventsMismatch tests verification behavior in case
 // of emitted events not matching chunks
 func (s *ChunkVerifierTestSuite) TestEventsMismatch() {
-	meta := s.GetTestSetup(s.T(), "eventsMismatch", false)
+	meta := s.GetTestSetup(s.T(), "eventsMismatch", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// add an additional event to the list of events produced by FVM
@@ -245,8 +251,8 @@ func (s *ChunkVerifierTestSuite) TestEventsMismatch() {
 
 // TestServiceEventsMismatch tests verification behavior in case
 // of emitted service events not matching chunks'
-func (s *ChunkVerifierTestSuite) TestServiceEventsMismatch() {
-	meta := s.GetTestSetup(s.T(), "doesn't matter", true)
+func (s *ChunkVerifierTestSuite) TestServiceEventsMismatch_SystemChunk() {
+	meta := s.GetTestSetup(s.T(), "doesn't matter", true, true)
 	vch := meta.RefreshChunkData(s.T())
 
 	// modify the list of service events produced by FVM
@@ -257,6 +263,7 @@ func (s *ChunkVerifierTestSuite) TestServiceEventsMismatch() {
 	s.snapshots[string(serviceTxBody.Script)] = &snapshot.ExecutionSnapshot{}
 	s.outputs[string(serviceTxBody.Script)] = fvm.ProcedureOutput{
 		ComputationUsed:        computationUsed,
+		ServiceEvents:          unittest.EventsFixture(1),
 		ConvertedServiceEvents: flow.ServiceEventList{*epochCommitServiceEvent},
 		Events:                 meta.ChunkEvents,
 	}
@@ -268,8 +275,8 @@ func (s *ChunkVerifierTestSuite) TestServiceEventsMismatch() {
 }
 
 // TestServiceEventsAreChecked ensures that service events are in fact checked
-func (s *ChunkVerifierTestSuite) TestServiceEventsAreChecked() {
-	meta := s.GetTestSetup(s.T(), "doesn't matter", true)
+func (s *ChunkVerifierTestSuite) TestServiceEventsAreChecked_SystemChunk() {
+	meta := s.GetTestSetup(s.T(), "doesn't matter", true, true)
 	vch := meta.RefreshChunkData(s.T())
 
 	// setup the verifier output to include the correct data for the service events
@@ -282,9 +289,56 @@ func (s *ChunkVerifierTestSuite) TestServiceEventsAreChecked() {
 	assert.NoError(s.T(), err)
 }
 
+// Tests the case where a service event is emitted outside the system chunk
+// and the event computed by the VN does not match the Result.
+// NOTE: this test case relies on the ordering of transactions in generateCollection.
+func (s *ChunkVerifierTestSuite) TestServiceEventsMismatch_NonSystemChunk() {
+	script := "service event mismatch in non-system chunk"
+	meta := s.GetTestSetup(s.T(), script, false, true)
+	vch := meta.RefreshChunkData(s.T())
+
+	// modify the list of service events produced by FVM
+	// EpochSetup event is expected, but we emit EpochCommit here resulting in a chunk fault
+	epochCommitServiceEvent, err := convert.ServiceEvent(testChain, epochCommitEvent)
+	require.NoError(s.T(), err)
+
+	s.snapshots[script] = &snapshot.ExecutionSnapshot{}
+	// overwrite the expected output for our custom transaction, passing
+	// in the non-matching EpochCommit event (should cause validation failure)
+	s.outputs[script] = fvm.ProcedureOutput{
+		ComputationUsed:        computationUsed,
+		ConvertedServiceEvents: flow.ServiceEventList{*epochCommitServiceEvent},
+		Events:                 meta.ChunkEvents[:3], // 2 default event + EpochSetup
+	}
+
+	_, err = s.verifier.Verify(vch)
+
+	assert.Error(s.T(), err)
+	assert.True(s.T(), chunksmodels.IsChunkFaultError(err))
+	assert.IsType(s.T(), &chunksmodels.CFInvalidServiceEventsEmitted{}, err)
+}
+
+// Tests that service events are checked, when they appear outside the system chunk.
+// NOTE: this test case relies on the ordering of transactions in generateCollection.
+func (s *ChunkVerifierTestSuite) TestServiceEventsAreChecked_NonSystemChunk() {
+	script := "service event in non-system chunk"
+	meta := s.GetTestSetup(s.T(), script, false, true)
+	vch := meta.RefreshChunkData(s.T())
+
+	// setup the verifier output to include the correct data for the service events
+	output := generateDefaultOutput()
+	output.ConvertedServiceEvents = meta.ServiceEvents
+	output.Events = meta.ChunkEvents[:3] // 2 default events + 1 service event
+	s.outputs[script] = output
+
+	spockSecret, err := s.verifier.Verify(vch)
+	assert.NoError(s.T(), err)
+	assert.NotNil(s.T(), spockSecret)
+}
+
 // TestSystemChunkWithCollectionFails ensures verification fails for system chunks with collections
 func (s *ChunkVerifierTestSuite) TestSystemChunkWithCollectionFails() {
-	meta := s.GetTestSetup(s.T(), "doesn't matter", true)
+	meta := s.GetTestSetup(s.T(), "doesn't matter", true, true)
 
 	// add a collection to the system chunk
 	col := unittest.CollectionFixture(1)
@@ -301,7 +355,7 @@ func (s *ChunkVerifierTestSuite) TestSystemChunkWithCollectionFails() {
 // TestEmptyCollection tests verification behaviour if a
 // collection doesn't have any transaction.
 func (s *ChunkVerifierTestSuite) TestEmptyCollection() {
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 
 	// reset test to use an empty collection
 	collection := unittest.CollectionFixture(0)
@@ -323,7 +377,7 @@ func (s *ChunkVerifierTestSuite) TestEmptyCollection() {
 }
 
 func (s *ChunkVerifierTestSuite) TestExecutionDataBlockMismatch() {
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 
 	// modify Block in the ExecutionDataRoot
 	meta.ExecDataBlockID = unittest.IdentifierFixture()
@@ -337,7 +391,7 @@ func (s *ChunkVerifierTestSuite) TestExecutionDataBlockMismatch() {
 }
 
 func (s *ChunkVerifierTestSuite) TestExecutionDataChunkIdsLengthDiffers() {
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// add an additional ChunkExecutionDataID into the ExecutionDataRoot passed into Verify
@@ -350,7 +404,7 @@ func (s *ChunkVerifierTestSuite) TestExecutionDataChunkIdsLengthDiffers() {
 }
 
 func (s *ChunkVerifierTestSuite) TestExecutionDataChunkIdMismatch() {
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// modify one of the ChunkExecutionDataIDs passed into Verify
@@ -363,7 +417,7 @@ func (s *ChunkVerifierTestSuite) TestExecutionDataChunkIdMismatch() {
 }
 
 func (s *ChunkVerifierTestSuite) TestExecutionDataIdMismatch() {
-	meta := s.GetTestSetup(s.T(), "", false)
+	meta := s.GetTestSetup(s.T(), "", false, false)
 	vch := meta.RefreshChunkData(s.T())
 
 	// modify ExecutionDataID passed into Verify
@@ -456,13 +510,13 @@ func generateExecutionData(t *testing.T, blockID flow.Identifier, ced *execution
 	return executionDataID, executionDataRoot
 }
 
-func generateEvents(t *testing.T, isSystemChunk bool, collection *flow.Collection) (flow.EventsList, []flow.ServiceEvent) {
+func generateEvents(t *testing.T, collection *flow.Collection, includeServiceEvent bool) (flow.EventsList, []flow.ServiceEvent) {
 	var chunkEvents flow.EventsList
 	serviceEvents := make([]flow.ServiceEvent, 0)
 
 	// service events are also included as regular events
-	if isSystemChunk {
-		for _, e := range systemEventsList {
+	if includeServiceEvent {
+		for _, e := range serviceEventsList {
 			e := e
 			event, err := convert.ServiceEvent(testChain, e)
 			require.NoError(t, err)
@@ -500,6 +554,12 @@ func generateTransactionResults(t *testing.T, collection *flow.Collection) []flo
 	return txResults
 }
 
+// generateCollection generates a collection fixture that is predictable based on inputs.
+// Test cases in this file rely on the predictable pattern of collections generated here.
+// If this is a system chunk, we return a collection containing only the service transaction.
+// Otherwise, we return a collection with 5 transactions. Only the first of these 5 uses the input script.
+// The transaction script is the lookup key for determining the result of transaction execution,
+// so test cases can inject a desired transaction output associated with the input script.
 func generateCollection(t *testing.T, isSystemChunk bool, script string) *flow.Collection {
 	if isSystemChunk {
 		// the system chunk's data pack does not include the collection, but the execution data does.
@@ -510,11 +570,12 @@ func generateCollection(t *testing.T, isSystemChunk bool, script string) *flow.C
 	}
 
 	collectionSize := 5
-	magicTxIndex := 3
+	// add the user-specified transaction first
+	userSpecifiedTxIndex := 0
 
 	coll := unittest.CollectionFixture(collectionSize)
 	if script != "" {
-		coll.Transactions[magicTxIndex] = &flow.TransactionBody{Script: []byte(script)}
+		coll.Transactions[userSpecifiedTxIndex] = &flow.TransactionBody{Script: []byte(script)}
 	}
 
 	return &coll
@@ -540,7 +601,7 @@ func generateDefaultOutput() fvm.ProcedureOutput {
 	}
 }
 
-func (s *ChunkVerifierTestSuite) GetTestSetup(t *testing.T, script string, system bool) *testMetadata {
+func (s *ChunkVerifierTestSuite) GetTestSetup(t *testing.T, script string, system bool, includeServiceEvents bool) *testMetadata {
 	collection := generateCollection(t, system, script)
 	block := blockFixture(collection)
 
@@ -554,14 +615,9 @@ func (s *ChunkVerifierTestSuite) GetTestSetup(t *testing.T, script string, syste
 	}
 
 	// events
-	chunkEvents, serviceEvents := generateEvents(t, system, collection)
+	chunkEvents, serviceEvents := generateEvents(t, collection, includeServiceEvents)
 	// make sure this includes events even for the service tx
 	require.NotEmpty(t, chunkEvents)
-	if system {
-		require.Len(t, serviceEvents, 1)
-	} else {
-		require.Empty(t, serviceEvents)
-	}
 
 	// registerTouch and State setup
 	startState, proof, update := generateStateUpdates(t, s.ledger)
@@ -638,7 +694,9 @@ func (m *testMetadata) RefreshChunkData(t *testing.T) *verification.VerifiableCh
 			CollectionIndex: 0,
 			StartState:      flow.StateCommitment(m.StartState),
 			BlockID:         m.Header.ID(),
-			EventCollection: eventsMerkleRootHash,
+			// in these test cases, all defined service events correspond to the current chunk
+			ServiceEventCount: unittest.PtrTo(uint16(len(m.ServiceEvents))),
+			EventCollection:   eventsMerkleRootHash,
 		},
 		Index: 0,
 	}
diff --git a/utils/slices/slices.go b/utils/slices/slices.go
index d2333f2d5aa..a8ac7982467 100644
--- a/utils/slices/slices.go
+++ b/utils/slices/slices.go
@@ -1,6 +1,10 @@
 package slices
 
-import "sort"
+import (
+	"sort"
+
+	"golang.org/x/exp/constraints"
+)
 
 // Concat concatenates multiple []byte into one []byte with efficient one-time allocation.
 func Concat(slices [][]byte) []byte {
@@ -28,15 +32,25 @@ func EnsureByteSliceSize(b []byte, length int) []byte {
 	return stateBytes
 }
 
-// MakeRange returns a slice of int from [min, max]
-func MakeRange(min, max int) []int {
-	a := make([]int, max-min+1)
+// MakeRange returns a slice of numbers [min, max).
+// The range includes min and excludes max.
+func MakeRange[T constraints.Integer](min, max T) []T {
+	a := make([]T, max-min)
 	for i := range a {
-		a[i] = min + i
+		a[i] = min + T(i)
 	}
 	return a
 }
 
+// Fill constructs a slice of type T with length n. The slice is then filled with input "val".
+func Fill[T any](val T, n int) []T {
+	arr := make([]T, n)
+	for i := 0; i < n; i++ {
+		arr[i] = val
+	}
+	return arr
+}
+
 // AreStringSlicesEqual returns true if the two string slices are equal.
 func AreStringSlicesEqual(a, b []string) bool {
 	if len(a) != len(b) {
diff --git a/utils/unittest/encoding.go b/utils/unittest/encoding.go
new file mode 100644
index 00000000000..b2ab5fd0fa3
--- /dev/null
+++ b/utils/unittest/encoding.go
@@ -0,0 +1,29 @@
+package unittest
+
+import (
+	"testing"
+
+	"github.com/fxamacker/cbor/v2"
+	"github.com/stretchr/testify/require"
+)
+
+// EncodeDecodeDifferentVersions emulates the situation where a peer running software version A receives
+// a message from a sender running software version B, where the format of the message may have been upgraded between
+// the different software versions. This method works irrespective whether version A or B is the older/newer version
+// (also allowing that both versions are the same; in this degenerate edge-case the old and new format would be the same).
+//
+// This function works by encoding src using CBOR, then decoding the result into dst.
+// Compatible fields as defined by CBOR will be copied into dst; in-compatible fields
+// may be omitted.
+func EncodeDecodeDifferentVersions(t *testing.T, src, dst any) {
+	bz, err := cbor.Marshal(src)
+	require.NoError(t, err)
+	err = cbor.Unmarshal(bz, dst)
+	require.NoError(t, err)
+}
+
+// PtrTo returns a pointer to the input. Useful for concisely constructing
+// a reference-typed argument to a function or similar.
+func PtrTo[T any](target T) *T {
+	return &target
+}
diff --git a/utils/unittest/fixtures.go b/utils/unittest/fixtures.go
index 166330b1466..15bc9593e62 100644
--- a/utils/unittest/fixtures.go
+++ b/utils/unittest/fixtures.go
@@ -19,6 +19,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	sdk "github.com/onflow/flow-go-sdk"
+
 	hotstuff "github.com/onflow/flow-go/consensus/hotstuff/model"
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/access/rest/util"
@@ -1325,9 +1326,10 @@ func ChunkFixture(
 ) *flow.Chunk {
 	chunk := &flow.Chunk{
 		ChunkBody: flow.ChunkBody{
-			CollectionIndex:      collectionIndex,
-			StartState:           StateCommitmentFixture(),
-			EventCollection:      IdentifierFixture(),
+			CollectionIndex: collectionIndex,
+			StartState:      StateCommitmentFixture(),
+			EventCollection: IdentifierFixture(),
+			//ServiceEventCount:    PtrTo[uint16](0),
 			TotalComputationUsed: 4200,
 			NumberOfTransactions: 42,
 			BlockID:              blockID,