fix(lstar): separate protocol rejections from internal assertions

src/lean_spec/node/validator/service.py

@@ -26,6 +26,7 @@
     SignedAttestation,
     SignedBlock,
     Slot,
+    SpecRejectionError,
     ValidatorIndex,
 )
 from lean_spec.spec.forks.lstar.containers import MultiMessageAggregate, SingleMessageAggregate
@@ -209,8 +210,8 @@ async def _maybe_produce_block(self, slot: Slot) -> None:
             if self.on_block is not None:
                 await self.on_block(signed_block)
 
-        except AssertionError as exception:
-            # Slot-boundary races can fail proposer validation.
+        except SpecRejectionError as exception:
+            # Slot-boundary races can fail proposer validation with a rejection.
             # Skip block production; the attestation at interval 1 still happens.
             logger.debug(
                 "Block production skipped for validator %d at slot %d: %s",

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

@@ -111,19 +111,23 @@ class RejectionReason(StrEnum):
     """The input bytes cannot be decoded into the expected structure."""
 
 
-class SpecRejectionError(AssertionError):
+class SpecError(Exception):
+    """Base class for every error the spec raises."""
+
+
+class SpecRejectionError(SpecError):
     """
     A rejection carrying its language-neutral reason.
 
-    Subclassing the assertion error keeps existing rejection handlers working.
+    Clients match on the reason code, not the Python base class.
     """
 
     def __init__(self, reason: RejectionReason, message: str) -> None:
         """
         Bind the rejection to its reason.
 
         Args:
-            reason: Language-neutral reason clients assert against.
+            reason: Language-neutral reason clients match against.
             message: Human-readable explanation for logs and debugging.
         """
         super().__init__(message)

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

@@ -100,6 +100,8 @@ def create_store(
             SpecRejectionError: ANCHOR_STATE_ROOT_MISMATCH if the anchor block's
                 state root does not match the hash of the state.
         """
+        # Internal type guards: the anchor is always the concrete fork state and block.
+        # These narrow the generic protocol types; they are never a client rejection path.
         assert isinstance(state, State)
         assert isinstance(anchor_block, Block)
 

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

@@ -337,7 +337,8 @@ def process_attestations(
                 if target.slot > latest_justified.slot:
                     latest_justified = target
 
-                # The justifiable filter above guarantees an in-range index.
+                # Invariant: the already-justified skip drops targets at or below finalized.
+                # So the index is in range; the assert guards an internal fact, not a rejection.
                 justified_index = target.slot.justified_index_after(finalized_slot)
                 assert justified_index is not None
 
@@ -365,13 +366,12 @@ def process_attestations(
                     delta = int(finalized_slot - old_finalized_slot)
                     if delta > 0:
                         justified_slots = JustifiedSlots(data=justified_slots.data[delta:])
-                        assert all(root in root_to_slot for root in justifications), (
-                            "Justification root missing from root_to_slot"
-                        )
+                        # A root absent from the slot map is off-chain and cannot justify.
+                        # Drop such a tally, do not track or reject it.
                         justifications = {
                             root: votes
                             for root, votes in justifications.items()
-                            if root_to_slot[root] > finalized_slot
+                            if root in root_to_slot and root_to_slot[root] > finalized_slot
                         }
 
         # Re-pack the vote map into the flat SSZ layout, roots first.

tests/consensus/lstar/state_transition/test_finalization.py

@@ -12,12 +12,16 @@
 from lean_spec.spec.crypto.merkleization import hash_tree_root
 from lean_spec.spec.forks import Slot, ValidatorIndex
 from lean_spec.spec.forks.lstar.containers import (
+    BlockHeader,
+    Checkpoint,
+    HistoricalBlockHashes,
     JustificationRoots,
     JustificationValidators,
     JustifiedSlots,
+    State,
 )
 from lean_spec.spec.forks.lstar.spec import LstarSpec
-from lean_spec.spec.ssz import Boolean
+from lean_spec.spec.ssz import Boolean, Bytes32, Uint64
 
 pytestmark = pytest.mark.valid_until("Lstar")
 
@@ -1374,3 +1378,100 @@ def test_rebased_finalization_prunes_stale_votes_and_preserves_future_votes(
             ),
         ),
     )
+
+
+def test_finalization_rebase_drops_off_chain_pending_tally(
+    state_transition_test: StateTransitionTestFiller,
+) -> None:
+    """
+    Finalization drops a pending tally whose root is off the canonical chain.
+
+    Given
+    -----
+    - 4 validators; a slot needs 3 votes (2/3) to be justified.
+    - a hand-built pre-state at slot 3 over the chain:
+        genesis(0) -> block_1(1) -> block_2(2) -> block_3(3)
+    - slot 1 is justified, rooted at block_1.
+    - finalized is slot 0, rooted at the genesis anchor.
+    - a phantom root carries a pending tally.
+    - the phantom root appears nowhere in the chain history.
+    - the phantom tally holds V0 alone of 4.
+    - block_4 carries V0, V1, V2's vote from block_1 to block_2.
+
+    When
+    ----
+    - the chain processes block_4 on top of the pre-state.
+
+    Then
+    ----
+    - the state slot is 4.
+    - block_4's supermajority justifies slot 2.
+    - justified slot is 2, rooted at block_2.
+    - the adjacent source slot 1 finalizes.
+    - finalized slot is 1, rooted at block_1.
+    - the justified-slots bitfield is [True, False] relative to slot 1.
+    - the off-chain phantom root is dropped from the pending roots.
+    - the phantom tally is dropped from the pending voters.
+    """
+    genesis = build_genesis_state()
+
+    genesis_anchor_root = Bytes32(b"\x11" * 32)
+    block_1_root = Bytes32(b"\x22" * 32)
+    block_2_root = Bytes32(b"\x33" * 32)
+    phantom_justification_root = Bytes32(b"\xff" * 32)
+
+    pre = State(
+        config=genesis.config,
+        slot=Slot(3),
+        latest_block_header=BlockHeader(
+            slot=Slot(3),
+            proposer_index=ValidatorIndex.proposer_for_slot(Slot(3), Uint64(4)),
+            parent_root=block_2_root,
+            state_root=Bytes32.zero(),
+            body_root=genesis.latest_block_header.body_root,
+        ),
+        latest_justified=Checkpoint(root=block_1_root, slot=Slot(1)),
+        latest_finalized=Checkpoint(root=genesis_anchor_root, slot=Slot(0)),
+        historical_block_hashes=HistoricalBlockHashes(
+            data=[genesis_anchor_root, block_1_root, block_2_root]
+        ),
+        justified_slots=JustifiedSlots(data=[Boolean(True), Boolean(False)]),
+        validators=genesis.validators,
+        justifications_roots=JustificationRoots(data=[phantom_justification_root]),
+        justifications_validators=JustificationValidators(
+            data=[Boolean(True), Boolean(False), Boolean(False), Boolean(False)]
+        ),
+    )
+
+    state_transition_test(
+        pre=pre,
+        blocks=[
+            BlockSpec(
+                slot=Slot(4),
+                forced_attestations=[
+                    AggregatedAttestationSpec(
+                        validator_indices=[
+                            ValidatorIndex(0),
+                            ValidatorIndex(1),
+                            ValidatorIndex(2),
+                        ],
+                        slot=Slot(4),
+                        source_slot=Slot(1),
+                        source_root=block_1_root,
+                        target_slot=Slot(2),
+                        target_root=block_2_root,
+                    ),
+                ],
+            ),
+        ],
+        post=StateExpectation(
+            slot=Slot(4),
+            latest_justified_slot=Slot(2),
+            latest_justified_root=block_2_root,
+            latest_finalized_slot=Slot(1),
+            latest_finalized_root=block_1_root,
+            justified_slots=JustifiedSlots(data=[Boolean(True), Boolean(False)]),
+            justifications_roots=JustificationRoots(data=[]),
+            justifications_validators=JustificationValidators(data=[]),
+        ),
+    )

tests/node/validator/test_service.py

@@ -22,7 +22,7 @@
 from lean_spec.node.validator.registry import ValidatorEntry
 from lean_spec.spec.crypto.merkleization import hash_tree_root
 from lean_spec.spec.crypto.xmss import TARGET_SIGNATURE_SCHEME
-from lean_spec.spec.forks import Slot, ValidatorIndex
+from lean_spec.spec.forks import RejectionReason, Slot, SpecRejectionError, ValidatorIndex
 from lean_spec.spec.forks.lstar import Store
 from lean_spec.spec.forks.lstar.config import MILLISECONDS_PER_INTERVAL
 from lean_spec.spec.forks.lstar.containers import (
@@ -395,13 +395,13 @@ async def test_non_proposer_does_not_produce(
 
         assert blocks == []
 
-    async def test_assertion_error_is_logged_and_skipped(
+    async def test_rejection_is_logged_and_skipped(
         self,
         sync_service: SyncService,
         real_registry: ValidatorRegistry,
         caplog: pytest.LogCaptureFixture,
     ) -> None:
-        """Store AssertionError during block production is caught; no block emitted."""
+        """A proposer-validation rejection during block production is caught; no block emitted."""
         blocks: list[SignedBlock] = []
 
         service = ValidatorService(
@@ -414,7 +414,9 @@ async def test_assertion_error_is_logged_and_skipped(
         with patch.object(
             service.spec,
             "produce_block_with_signatures",
-            side_effect=AssertionError("mismatch"),
+            side_effect=SpecRejectionError(
+                RejectionReason.WRONG_PROPOSER, "not the scheduled proposer"
+            ),
         ):
             # Slot 0: proposer is validator 0 (0 % 8 = 0), which is in the registry.
             await service._maybe_produce_block(Slot(0))