refactor(forks/lstar): split spec.py into per-concern mixins

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

@@ -0,0 +1,120 @@
+"""Internal typing contract shared by the lstar spec mixins."""
+
+from abc import abstractmethod
+from collections.abc import Set as AbstractSet
+
+from lean_spec.spec.forks.lstar.containers import (
+    AggregatedAttestation,
+    AggregatedAttestations,
+    AttestationData,
+    Block,
+    BlockBody,
+    BlockHeader,
+    Config,
+    SignedAggregatedAttestation,
+    SignedBlock,
+    SingleMessageAggregate,
+    Slot,
+    State,
+    Store,
+    ValidatorIndex,
+    Validators,
+)
+from lean_spec.spec.ssz import Bytes32
+
+from ..protocol import ForkProtocol
+from .interval import Interval
+
+LstarStore = Store[State, Block]
+"""Concrete Store specialization owned by the lstar fork."""
+
+
+class LstarSpecContract(ForkProtocol):
+    """Shared typing contract for the lstar fork mixins.
+
+    Declares the concrete container factory types.
+    Declares the cross-mixin method surface.
+    Lets each mixin call siblings without importing the other mixins.
+    """
+
+    state_class: type[State]
+    block_class: type[Block]
+    block_body_class: type[BlockBody]
+    block_header_class: type[BlockHeader]
+    aggregated_attestations_class: type[AggregatedAttestations]
+    store_class: type[LstarStore]
+    attestation_data_class: type[AttestationData]
+    aggregated_attestation_class: type[AggregatedAttestation]
+    config_class: type[Config]
+
+    @abstractmethod
+    def process_slots(self, state: State, target_slot: Slot) -> State:
+        """Advance the state through empty slots up to the target slot."""
+        ...
+
+    @abstractmethod
+    def process_block(self, state: State, block: Block) -> State:
+        """Apply full block processing including header and body."""
+        ...
+
+    @abstractmethod
+    def state_transition(
+        self,
+        state: State,
+        block: Block,
+    ) -> State:
+        """Apply the complete state transition function for a block."""
+        ...
+
+    @abstractmethod
+    def build_block(
+        self,
+        state: State,
+        slot: Slot,
+        proposer_index: ValidatorIndex,
+        parent_root: Bytes32,
+        known_block_roots: AbstractSet[Bytes32],
+        aggregated_payloads: dict[AttestationData, set[SingleMessageAggregate]] | None = None,
+    ) -> tuple[Block, State, list[AggregatedAttestation], list[SingleMessageAggregate]]:
+        """Build a valid block on top of the given pre-state."""
+        ...
+
+    @abstractmethod
+    def verify_signatures(
+        self,
+        signed_block: SignedBlock,
+        validators: Validators,
+    ) -> bool:
+        """Verify the merged aggregate proof carried by a signed block."""
+        ...
+
+    @abstractmethod
+    def prune_stale_attestation_data(self, store: LstarStore) -> LstarStore:
+        """Remove attestation data that can no longer influence fork choice."""
+        ...
+
+    @abstractmethod
+    def accept_new_attestations(self, store: LstarStore) -> LstarStore:
+        """Migrate pending payloads into the known pool and update the head."""
+        ...
+
+    @abstractmethod
+    def update_safe_target(self, store: LstarStore) -> LstarStore:
+        """Update the deepest block with supermajority attestation weight."""
+        ...
+
+    @abstractmethod
+    def aggregate(self, store: LstarStore) -> tuple[LstarStore, list[SignedAggregatedAttestation]]:
+        """Combine raw validator votes into compact aggregated attestations."""
+        ...
+
+    @abstractmethod
+    def on_tick(
+        self,
+        store: LstarStore,
+        target_interval: Interval,
+        has_proposal: bool,
+        is_aggregator: bool = False,
+    ) -> tuple[LstarStore, list[SignedAggregatedAttestation]]:
+        """Advance store time to the target interval, performing interval actions."""
+        ...

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

@@ -0,0 +1,135 @@
+"""Lstar fork — attestation aggregation."""
+
+from lean_spec.spec.crypto.merkleization import hash_tree_root
+from lean_spec.spec.forks.lstar.aggregation_select import select_greedily
+from lean_spec.spec.forks.lstar.containers import (
+    SignedAggregatedAttestation,
+    SingleMessageAggregate,
+)
+
+from ._contract import LstarSpecContract, LstarStore
+
+
+class AggregationMixin(LstarSpecContract):
+    """Attestation aggregation for the lstar fork."""
+
+    def aggregate(self, store: LstarStore) -> tuple[LstarStore, list[SignedAggregatedAttestation]]:
+        """Turn raw validator votes into compact aggregated attestations.
+
+        Validators cast individual signatures over gossip. Before those
+        votes can influence fork choice or be included in a block, they
+        must be combined into compact cryptographic proofs.
+
+        The store holds three pools of attestation evidence:
+
+        - **Gossip signatures**: individual validator votes arriving in real-time.
+        - **New payloads**: aggregated proofs from the current round, not yet
+          committed to the chain.
+        - **Known payloads**: previously accepted proofs, reusable as building
+          blocks for deeper aggregation.
+
+        For each unique piece of attestation data the algorithm proceeds in three phases:
+
+        1. **Select** — greedily pick existing proofs that maximize
+           validator coverage (new before known).
+        2. **Fill** — collect raw gossip signatures for any validators
+           not yet covered.
+        3. **Aggregate** — delegate to the XMSS subspec to produce a
+           single cryptographic proof.
+
+        After aggregation the store is updated:
+
+        - Consumed gossip signatures are removed.
+        - Newly produced proofs are recorded for future reuse.
+        """
+        validators = store.states[store.head].validators
+        gossip_signatures = store.attestation_signatures
+        new = store.latest_new_aggregated_payloads
+        known = store.latest_known_aggregated_payloads
+
+        new_aggregates: list[SignedAggregatedAttestation] = []
+
+        # Only attestation data with a new payload or a raw gossip signature
+        # can trigger aggregation. Known payloads alone cannot — they exist
+        # only to help extend coverage when combined with fresh evidence.
+        for data in new.keys() | gossip_signatures.keys():
+            # Phase 1: Select
+            #
+            # Start with the cheapest option: reuse proofs that already
+            # cover many validators.
+            #
+            # Child proofs are aggregated signatures from prior rounds.
+            # Selecting them first keeps the final proof tree shallow
+            # and avoids redundant cryptographic work.
+            #
+            # New payloads go first because they represent uncommitted
+            # work — known payloads fill remaining gaps.
+
+            child_proofs, covered = select_greedily(new.get(data), known.get(data))
+
+            # Phase 2: Fill
+            #
+            # For every validator not yet covered by a child proof,
+            # include its individual gossip signature.
+            #
+            # Sorting by validator index guarantees deterministic proof
+            # construction regardless of network arrival order.
+            raw_entries = [
+                (
+                    e.validator_index,
+                    validators[e.validator_index].get_attestation_public_key(),
+                    e.signature,
+                )
+                for e in sorted(gossip_signatures.get(data, set()), key=lambda e: e.validator_index)
+                if e.validator_index not in covered
+            ]
+
+            # The aggregation layer enforces a minimum: either at least one
+            # raw signature, or at least two child proofs to merge.
+            #
+            # A lone child proof is already a valid proof — nothing to do.
+            if not raw_entries and len(child_proofs) < 2:
+                continue
+
+            # Phase 3: Aggregate
+            #
+            # Build the recursive proof tree.
+            #
+            # Each child proof needs its participants' public keys so
+            # the XMSS prover can verify inner proofs while constructing
+            # the outer one.
+            children = [
+                (
+                    child,
+                    [
+                        validators[validator_index].get_attestation_public_key()
+                        for validator_index in child.participants.to_validator_indices()
+                    ],
+                )
+                for child in child_proofs
+            ]
+
+            # Hand everything to the XMSS subspec.
+            # Each fresh entry already carries its validator index alongside its key and signature.
+            # Out comes a single proof covering all selected validators.
+            proof = SingleMessageAggregate.aggregate(
+                children=children,
+                raw_xmss=raw_entries,
+                message=hash_tree_root(data),
+                slot=data.slot,
+            )
+            new_aggregates.append(SignedAggregatedAttestation(data=data, proof=proof))
+
+        # ── Store bookkeeping ────────────────────────────────────────
+        #
+        # Record freshly produced proofs so future rounds can reuse them.
+        # Remove gossip signatures that were consumed by this aggregation.
+        store.latest_new_aggregated_payloads = {}
+        for signed_attestation in new_aggregates:
+            store.latest_new_aggregated_payloads.setdefault(signed_attestation.data, set()).add(
+                signed_attestation.proof
+            )
+
+        for data in store.latest_new_aggregated_payloads:
+            store.attestation_signatures.pop(data, None)
+        return store, new_aggregates