feat(validator): source votes from the head chain's justified checkpoint

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

@@ -62,14 +62,22 @@ class AttestationCheck(CamelModel):
     source_slot: Slot | None = None
     """Expected source checkpoint slot."""
 
+    source_root_label: str | None = None
+    """Expected source checkpoint root, named by label and resolved to a root."""
+
     target_slot: Slot | None = None
     """Expected target checkpoint slot."""
 
-    location: Literal["new", "known"]
-    """Which pool the attestation should sit in, the pending or the accepted pool."""
+    location: Literal["new", "known", "signatures"]
+    """Which pool the attestation should sit in: the pending pool, the accepted pool, or the raw
+    per-validator signature pool."""
 
     def validate_attestation(
-        self, attestation: AttestationData, location: str, step_index: int
+        self,
+        attestation: AttestationData,
+        location: str,
+        step_index: int,
+        expected_source_root: Bytes32 | None = None,
     ) -> None:
         """Validate attestation properties."""
         for field_name in self.model_fields_set & _ATTESTATION_SLOT_ACCESSORS.keys():
@@ -81,6 +89,13 @@ def validate_attestation(
                     f"{field_name} = {actual_slot}, expected {expected_slot}"
                 )
 
+        if expected_source_root is not None and attestation.source.root != expected_source_root:
+            raise AssertionError(
+                f"Step {step_index}: validator {self.validator} {location} "
+                f"source root = 0x{attestation.source.root.hex()}, "
+                f"expected 0x{expected_source_root.hex()}"
+            )
+
 
 class StoreChecks(SelectiveCheck):
     """Store state checks for fork choice tests, validating only the fields a test sets."""
@@ -293,32 +308,53 @@ def _resolve(label: str) -> Bytes32:
         if "attestation_checks" in fields:
             assert self.attestation_checks is not None
             for attestation_check in self.attestation_checks:
-                if attestation_check.location == "new":
-                    payloads = store.latest_new_aggregated_payloads
-                    label = "in latest_new"
-                else:
-                    payloads = store.latest_known_aggregated_payloads
-                    label = "in latest_known"
-
-                # Map each validator to its highest-slot vote across the raw pool.
+                # Map each validator to its highest-slot vote in the named pool.
                 #
                 # The checker inspects pool content before pruning, so no finality cutoff applies.
                 # On equal slots the first vote seen wins, matching the fork-choice rule.
                 extracted_attestations: dict[ValidatorIndex, AttestationData] = {}
-                for attestation_data, proofs in payloads.items():
-                    for proof in proofs:
-                        for participant_index in proof.participants.to_validator_indices():
-                            previous_vote = extracted_attestations.get(participant_index)
+                if attestation_check.location == "signatures":
+                    # The raw signature pool groups one entry per validator under each vote.
+                    label = "in attestation_signatures"
+                    for attestation_data, entries in store.attestation_signatures.items():
+                        for signature_entry in entries:
+                            voter_index = signature_entry.validator_index
+                            previous_vote = extracted_attestations.get(voter_index)
                             if previous_vote is None or previous_vote.slot < attestation_data.slot:
-                                extracted_attestations[participant_index] = attestation_data
+                                extracted_attestations[voter_index] = attestation_data
+                else:
+                    # The aggregated pools group proofs covering many validators under each vote.
+                    if attestation_check.location == "new":
+                        payloads = store.latest_new_aggregated_payloads
+                        label = "in latest_new_aggregated_payloads"
+                    else:
+                        payloads = store.latest_known_aggregated_payloads
+                        label = "in latest_known_aggregated_payloads"
+                    for attestation_data, proofs in payloads.items():
+                        for proof in proofs:
+                            for participant_index in proof.participants.to_validator_indices():
+                                previous_vote = extracted_attestations.get(participant_index)
+                                if (
+                                    previous_vote is None
+                                    or previous_vote.slot < attestation_data.slot
+                                ):
+                                    extracted_attestations[participant_index] = attestation_data
 
                 if attestation_check.validator not in extracted_attestations:
                     raise AssertionError(
                         f"Step {step_index}: validator {attestation_check.validator} not found "
-                        f"{label}_aggregated_payloads"
+                        f"{label}"
                     )
+                expected_source_root = (
+                    _resolve(attestation_check.source_root_label)
+                    if attestation_check.source_root_label is not None
+                    else None
+                )
                 attestation_check.validate_attestation(
-                    extracted_attestations[attestation_check.validator], label, step_index
+                    extracted_attestations[attestation_check.validator],
+                    label,
+                    step_index,
+                    expected_source_root,
                 )
 
         # Target slots keyed in each attestation pool

src/lean_spec/spec/forks/lstar/validator_duties.py

@@ -15,7 +15,7 @@
     ValidatorIndex,
 )
 from lean_spec.spec.forks.lstar.errors import RejectionReason, SpecRejectionError
-from lean_spec.spec.ssz import Uint64
+from lean_spec.spec.ssz import Bytes32, Uint64
 
 
 class ValidatorDutiesMixin(LstarSpecBase):
@@ -75,19 +75,31 @@ def get_attestation_target(self, store: LstarStore) -> Checkpoint:
         return Checkpoint(root=target_block_root, slot=target_block.slot)
 
     def produce_attestation_data(self, store: LstarStore, slot: Slot) -> AttestationData:
-        """Attestation data for the given slot: head, target, and the latest justified source."""
+        """Attestation data for the given slot: head, target, and the head's justified source."""
         head_checkpoint = Checkpoint(
             root=store.head,
             slot=store.blocks[store.head].slot,
         )
 
         target_checkpoint = self.get_attestation_target(store)
 
+        # Source the vote from the head chain's own justified checkpoint.
+        # The store can advance its justified from a minority fork the head never extended.
+        # Voting with the head's justified keeps the source on the chain the vote extends.
+        justified_source = store.states[store.head].latest_justified
+
+        # Replace the placeholder genesis root with the real one.
+        if justified_source.root == Bytes32.zero():
+            justified_source = Checkpoint(root=store.head, slot=justified_source.slot)
+
+        # Sanity check: the source must be older or equal to the target.
+        assert justified_source.slot <= target_checkpoint.slot
+
         return self.attestation_data_class(
             slot=slot,
             head=head_checkpoint,
             target=target_checkpoint,
-            source=store.latest_justified,
+            source=justified_source,
         )
 
     def produce_block_with_signatures(

tests/consensus/lstar/fork_choice/test_attestation_source_divergence.py

@@ -5,9 +5,12 @@
 from consensus_testing import (
     AggregatedAttestationCheck,
     AggregatedAttestationSpec,
+    AttestationCheck,
+    AttestationStep,
     BlockSpec,
     BlockStep,
     ForkChoiceTestFiller,
+    GossipAttestationSpec,
     StoreChecks,
 )
 from lean_spec.spec.forks import Slot, ValidatorIndex
@@ -118,3 +121,109 @@ def test_justified_divergence_self_heals_in_next_block(
             ),
         ],
     )
+
+
+def test_honest_vote_sources_from_head_chain_not_store(
+    fork_choice_test: ForkChoiceTestFiller,
+) -> None:
+    """
+    An honest vote sources from the head chain's justified, not the store's.
+
+    Given
+    -----
+    - 4 validators; a slot needs 3 votes (2/3) to be justified.
+    - the chain:
+        genesis
+        - common(1)
+          - block_2(2) -> block_3(3)
+          - fork_4(4)
+    - block_3 includes V0's vote for block_2.
+    - fork_4 includes V1, V2, V3's votes for common.
+    - fork_4 reaches 3 votes, so it justifies slot 1.
+    - block_3 has only 1 vote, so it justifies nothing.
+    - the views diverge: store = common (slot 1), block_3 state = genesis (slot 0).
+    - head stays on block_3.
+
+    When
+    ----
+    - V0 gossips an honest vote at slot 4, produced from attestation duties on the block_3 head.
+
+    Then
+    ----
+    - the vote sits in the raw signature pool.
+    - its head is block_3 (slot 3), the current head.
+    - its source is genesis (slot 0), the head chain's own justified.
+    - its source is not common (slot 1), the store's justified.
+    """
+    fork_choice_test(
+        steps=[
+            BlockStep(
+                block=BlockSpec(slot=Slot(1), label="common"),
+                checks=StoreChecks(head_slot=Slot(1)),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(2), parent_label="common", label="block_2"),
+                checks=StoreChecks(head_slot=Slot(2)),
+            ),
+            BlockStep(
+                block=BlockSpec(
+                    slot=Slot(3),
+                    parent_label="block_2",
+                    label="block_3",
+                    attestations=[
+                        AggregatedAttestationSpec(
+                            validator_indices=[ValidatorIndex(0)],
+                            slot=Slot(2),
+                            target_slot=Slot(2),
+                            target_root_label="block_2",
+                        ),
+                    ],
+                ),
+                checks=StoreChecks(head_slot=Slot(3)),
+            ),
+            BlockStep(
+                block=BlockSpec(
+                    slot=Slot(4),
+                    parent_label="common",
+                    label="fork_4",
+                    attestations=[
+                        AggregatedAttestationSpec(
+                            validator_indices=[
+                                ValidatorIndex(1),
+                                ValidatorIndex(2),
+                                ValidatorIndex(3),
+                            ],
+                            slot=Slot(1),
+                            target_slot=Slot(1),
+                            target_root_label="common",
+                        ),
+                    ],
+                ),
+                checks=StoreChecks(
+                    head_slot=Slot(3),
+                    head_root_label="block_3",
+                    latest_justified_slot=Slot(1),
+                    latest_justified_root_label="common",
+                ),
+            ),
+            AttestationStep(
+                attestation=GossipAttestationSpec(
+                    validator_index=ValidatorIndex(0),
+                    slot=Slot(4),
+                    target_slot=Slot(3),
+                ),
+                is_aggregator=True,
+                checks=StoreChecks(
+                    attestation_checks=[
+                        AttestationCheck(
+                            validator=ValidatorIndex(0),
+                            location="signatures",
+                            head_slot=Slot(3),
+                            source_slot=Slot(0),
+                            source_root_label="genesis",
+                        ),
+                    ],
+                ),
+            ),
+        ],
+    )

tests/node/validator/test_service.py

@@ -937,7 +937,7 @@ async def test_attestation_data_references_correct_checkpoints(
         The attestation must reference:
         - head: the current chain head
         - target: the attestation target based on forkchoice
-        - source: the latest justified checkpoint
+        - source: the head chain's justified checkpoint
         """
         attestations_produced: list[SignedAttestation] = []
 
@@ -955,7 +955,9 @@ async def capture_attestation(attestation: SignedAttestation) -> None:
 
         store = real_sync_service.store
         expected_head_root = store.head
-        expected_source = store.latest_justified
+        # The head is the genesis anchor, whose state carries the zero-root justified placeholder.
+        # The produced source normalizes that placeholder to the real genesis root.
+        expected_source = Checkpoint(root=store.head, slot=Slot(0))
 
         for signed_attestation in attestations_produced:
             attestation_data = signed_attestation.data