Add fork choice store pruning test (leanEthereum#604)

* Add fork choice store pruning test

* test(fc): add fork choice store pruning test

Add a fork choice filler for store pruning on finalization.

- Add a devnet fork choice test covering pruning after finalization advances
  to slot 3
- Verify stale attestation data targeting slots 1, 2, and 3 is removed
- Verify attestation data targeting slots 4 and 5 is preserved
- Extend StoreChecks with exact target-slot assertions for raw, new, and
  known attestation pools
- Update fork choice attestation steps to retain raw gossip signatures so
  pruning can be tested end to end

Closes leanEthereum#567

Co-Authored-By: Itodo-S <71233924+Itodo-S@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Itodo-S

packages/testing/src/consensus_testing/test_fixtures/fork_choice.py

@@ -289,6 +289,7 @@ def make_fixture(self) -> Self:
                         store = store.on_gossip_attestation(
                             signed_attestation,
                             scheme=LEAN_ENV_TO_SCHEMES[self.lean_env],
+                            is_aggregator=step.is_aggregator,
                         )
 
                     case GossipAggregatedAttestationStep():

packages/testing/src/consensus_testing/test_types/step_types.py

@@ -149,6 +149,15 @@ class AttestationStep(BaseForkChoiceStep):
     The framework fills in the attestation data and signature during make_fixture().
     """
 
+    is_aggregator: bool = False
+    """
+    Whether the node holds the aggregator role for this attestation.
+
+    Only aggregator nodes store gossip signatures in the raw signature pool.
+    Defaults to False so existing fillers preserve the behavior where gossip
+    attestations are validated but not stored.
+    """
+
     _filled_attestation: SignedAttestation | None = PrivateAttr(default=None)
     """The filled SignedAttestation, processed through the spec."""
 

packages/testing/src/consensus_testing/test_types/store_checks.py

@@ -180,6 +180,30 @@ class StoreChecks(CamelModel):
     attestation_checks: list[AttestationCheck] | None = None
     """Optional list of attestation content checks for specific validators."""
 
+    attestation_signature_target_slots: list[Slot] | None = None
+    """
+    Expected target slots present in attestation_signatures.
+
+    Compares the exact set of target checkpoint slots keyed in the raw gossip
+    signature map, independent of how many validators signed each target.
+    """
+
+    latest_new_aggregated_target_slots: list[Slot] | None = None
+    """
+    Expected target slots present in latest_new_aggregated_payloads.
+
+    Compares the exact set of target checkpoint slots keyed in the pending
+    aggregated proof map.
+    """
+
+    latest_known_aggregated_target_slots: list[Slot] | None = None
+    """
+    Expected target slots present in latest_known_aggregated_payloads.
+
+    Compares the exact set of target checkpoint slots keyed in the accepted
+    aggregated proof map.
+    """
+
     block_attestation_count: int | None = None
     """
     Expected number of aggregated attestations in the block body.
@@ -351,6 +375,24 @@ def _resolve(label: str) -> Bytes32:
                     )
                 check.validate_attestation(extracted[check.validator], label, step_index)
 
+        if "attestation_signature_target_slots" in fields:
+            assert self.attestation_signature_target_slots is not None
+            actual = sorted({data.target.slot for data in store.attestation_signatures})
+            expected = sorted(self.attestation_signature_target_slots)
+            _check("attestation_signatures.target_slots", actual, expected)
+
+        if "latest_new_aggregated_target_slots" in fields:
+            assert self.latest_new_aggregated_target_slots is not None
+            actual = sorted({data.target.slot for data in store.latest_new_aggregated_payloads})
+            expected = sorted(self.latest_new_aggregated_target_slots)
+            _check("latest_new_aggregated_payloads.target_slots", actual, expected)
+
+        if "latest_known_aggregated_target_slots" in fields:
+            assert self.latest_known_aggregated_target_slots is not None
+            actual = sorted({data.target.slot for data in store.latest_known_aggregated_payloads})
+            expected = sorted(self.latest_known_aggregated_target_slots)
+            _check("latest_known_aggregated_payloads.target_slots", actual, expected)
+
         # Block body attestation count
         if "block_attestation_count" in fields:
             if filled_block is None: