feat(forks): add SigScheme capability and @requires marker (Stage 7 of leanEthereum#686) (leanEthereum#715)

* feat(forks): add SigScheme capability and @requires marker (Stage 7 of leanEthereum#686)

Introduces the first fork-level capability and a pytest marker to gate
tests on capability presence.

Capability
----------
- New `SigScheme` runtime-checkable Protocol in `forks/capabilities.py`
  asserts a `sig_scheme: ClassVar[GeneralizedXmssScheme]` attribute.
- `LstarSpec` binds `sig_scheme = TARGET_SIGNATURE_SCHEME` so
  `isinstance(LstarSpec(), SigScheme)` returns True.
- The three spec methods that previously took
  `scheme=TARGET_SIGNATURE_SCHEME` (`verify_signatures`,
  `on_gossip_attestation`, `on_block`) drop the parameter and read
  `self.sig_scheme` directly. The capability becomes the runtime source
  of truth.

Marker
------
- `requires(*capabilities)` pytest marker, registered in
  `pytest_plugins/filler.py`. Composes (AND) with the existing
  `valid_from` / `valid_until` / `valid_at` fork-range markers.
- `_check_markers_valid_for_fork` instantiates the active spec once and
  runs `isinstance(spec, capability)` per required capability.
- A `requires(...)` helper in `framework.markers` works around pytest's
  auto-detect-class shortcut (which trips on Protocol args to
  `@pytest.mark.requires(...)`).

Tests
-----
- 11 unit tests in `tests/lean_spec/forks/test_capabilities.py` cover
  the Protocol and the dispatch helper (composition with the
  fork-range markers, multiple `@requires` markers, error path for
  non-runtime_checkable Protocol).
- One smoke filler test in
  `tests/consensus/lstar/test_capability_gating.py` exercises the
  marker through pytest's live collection: one test marked with
  SigScheme runs, one marked with a synthetic absent capability is
  deselected.

Filler scheme override
----------------------
The three filler call sites that previously passed
`scheme=LEAN_ENV_TO_SCHEMES[self.lean_env]` to spec methods (in
`test_fixtures/fork_choice.py` and `test_types/block_spec.py`) drop the
kwarg. The PR description has the trade-off note and revert path if
that override is in fact needed somewhere we missed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(forks): tighten capability marker and apply review feedback

- Rename the marker helper to requires_capability and validate
  runtime-checkable Protocols at call time (fail at import, not
  at collection)
- Cache the fork spec instance in marker dispatch instead of
  constructing it once per test
- Re-export the capabilities namespace from lean_spec.forks so
  future capabilities don't need new import-site edits
- Register valid_from / valid_at / requires markers in pyproject
  so unit tests can build real pytest Marks under strict-markers
- Drop the hand-rolled Mark stand-in in tests; build real Marks
  via the MarkDecorator path; drop is True / is False on bool
  predicates
- Tighten docstrings per project style (no paragraph blocks, no
  backtick references, no internal-name references)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(forks): tighten dispatcher-guard test docstring

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Thomas Coratger <60488569+tcoratger@users.noreply.github.com>

Author: 3 people

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

@@ -25,7 +25,6 @@
 from lean_spec.types import Slot, Uint64, ValidatorIndex
 
 from ..keys import (
-    LEAN_ENV_TO_SCHEMES,
     XmssKeyManager,
 )
 from ..test_types import (
@@ -330,11 +329,7 @@ def make_fixture(self) -> Self:
 
                         # Process the block through Store.
                         # This validates, applies state transition, and updates the store's head.
-                        store = spec.on_block(
-                            store,
-                            signed_block,
-                            scheme=LEAN_ENV_TO_SCHEMES[self.lean_env],
-                        )
+                        store = spec.on_block(store, signed_block)
 
                     case AttestationStep():
                         # Process a gossip attestation.
@@ -351,7 +346,6 @@ def make_fixture(self) -> Self:
                         store = spec.on_gossip_attestation(
                             store,
                             signed_attestation,
-                            scheme=LEAN_ENV_TO_SCHEMES[self.lean_env],
                             is_aggregator=step.is_aggregator,
                         )
 

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

@@ -29,7 +29,7 @@
 from lean_spec.subspecs.xmss.containers import Signature
 from lean_spec.types import Bytes32, CamelModel, Slot, ValidatorIndex, ValidatorIndices
 
-from ..keys import LEAN_ENV_TO_SCHEMES, XmssKeyManager, create_dummy_signature
+from ..keys import XmssKeyManager, create_dummy_signature
 from .aggregated_attestation_spec import AggregatedAttestationSpec
 
 
@@ -448,7 +448,6 @@ def build_signed_block_with_store(
                     data=attestation.data,
                     signature=signature,
                 ),
-                scheme=LEAN_ENV_TO_SCHEMES[lean_env],
                 is_aggregator=True,
             )
 

packages/testing/src/framework/__init__.py

@@ -4,3 +4,7 @@
 This module provides base classes and utilities that are common across
 both consensus and execution layer testing.
 """
+
+from .markers import requires_capability
+
+__all__ = ["requires_capability"]

packages/testing/src/framework/markers.py

@@ -0,0 +1,34 @@
+"""Helper for the capability-requirement pytest marker."""
+
+import pytest
+
+
+def requires_capability(*capabilities: type) -> pytest.MarkDecorator:
+    """Build a capability-requirement marker over one or more Protocols.
+
+    Why a helper is needed at all:
+
+    - Pytest treats a single class argument to a marker as the
+      thing being decorated, not as marker data.
+    - That makes pytest try to instantiate the class.
+    - Protocols can't be instantiated, so applying the marker
+      directly to a Protocol raises TypeError at import.
+
+    What this helper does:
+
+    - Passes the capability through as marker data instead of as
+      the decoration target.
+    - Validates each argument up front, so non-Protocol classes
+      and Protocols missing the runtime-checkable decorator fail
+      at import rather than at test collection.
+
+    Raises:
+        TypeError: If any argument is not a runtime-checkable Protocol.
+    """
+    for cap in capabilities:
+        if not getattr(cap, "_is_runtime_protocol", False):
+            raise TypeError(
+                f"requires_capability expects @runtime_checkable Protocols; "
+                f"got {getattr(cap, '__name__', cap)!r}"
+            )
+    return pytest.mark.requires.with_args(*capabilities)

packages/testing/src/framework/pytest_plugins/filler.py

@@ -1,5 +1,6 @@
 """Layer-agnostic pytest plugin for generating Ethereum test fixtures."""
 
+import functools
 import importlib
 import json
 import shutil
@@ -12,6 +13,13 @@
 import pytest
 
 
+@functools.cache
+def _spec_instance_for(fork_class: type) -> Any:
+    """Build the active fork's spec instance once and reuse across collection."""
+    spec_class_method: Any = fork_class.spec_class  # ty: ignore[unresolved-attribute]
+    return spec_class_method()()
+
+
 class FixtureCollector:
     """Collects generated fixtures and writes them to disk."""
 
@@ -226,6 +234,11 @@ def pytest_configure(config: pytest.Config) -> None:
         "markers",
         "valid_at(fork): specifies at which fork a test case is valid",
     )
+    config.addinivalue_line(
+        "markers",
+        "requires(*capabilities): only collect when the active fork "
+        "advertises every listed runtime-checkable Protocol",
+    )
 
     # Get options
     output_dir = Path(config.getoption("--output"))
@@ -313,6 +326,14 @@ def _check_markers_valid_for_fork(
     """Check if test markers indicate validity for the given fork.
 
     Shared logic for both collection-time and parametrization-time fork filtering.
+
+    Composition rules:
+
+    - Fork-range markers form an intersection across kinds and a union
+      within a kind.
+    - The exact-fork marker short-circuits to a single-fork match.
+    - The capability marker AND-composes on top of either branch — the
+      active fork must satisfy every listed capability Protocol.
     """
     has_valid_from = False
     has_valid_until = False
@@ -321,6 +342,7 @@ def _check_markers_valid_for_fork(
     valid_from_forks = []
     valid_until_forks = []
     valid_at_forks = []
+    required_capabilities: list[type] = []
 
     for marker in markers:
         if marker.name == "valid_from":
@@ -341,12 +363,21 @@ def _check_markers_valid_for_fork(
                 target_fork = get_fork_by_name(fork_name)
                 if target_fork:
                     valid_at_forks.append(target_fork)
+        elif marker.name == "requires":
+            required_capabilities.extend(marker.args)
+
+    def _capability_check() -> bool:
+        """Active fork must structurally satisfy every required capability."""
+        if not required_capabilities:
+            return True
+        spec = _spec_instance_for(fork_class)
+        return all(isinstance(spec, cap) for cap in required_capabilities)
 
-    if not (has_valid_from or has_valid_until or has_valid_at):
+    if not (has_valid_from or has_valid_until or has_valid_at or required_capabilities):
         return True
 
     if has_valid_at:
-        return fork_class in valid_at_forks
+        return fork_class in valid_at_forks and _capability_check()
 
     from_valid = True
     if has_valid_from:
@@ -356,7 +387,7 @@ def _check_markers_valid_for_fork(
     if has_valid_until:
         until_valid = any(fork_class <= until_fork for until_fork in valid_until_forks)
 
-    return from_valid and until_valid
+    return from_valid and until_valid and _capability_check()
 
 
 def _is_test_item_valid_for_fork(

pyproject.toml

@@ -106,7 +106,10 @@ addopts = [
 ]
 markers = [
     "slow: marks tests as slow (deselect with '-m \"not slow\"')",
+    "valid_from: marks tests as valid from a specific fork version",
     "valid_until: marks tests as valid until a specific fork version",
+    "valid_at: marks tests as valid only at a specific fork version",
+    "requires: marks tests as requiring one or more fork capabilities",
     "interop: integration tests for multiple leanSpec nodes",
     "num_validators: number of validators for interop test cluster",
 ]

src/lean_spec/forks/__init__.py

@@ -1,5 +1,7 @@
 """Multi-fork dispatch layer for leanSpec consensus specification."""
 
+from . import capabilities
+from .capabilities import SigScheme
 from .lstar.containers import (
     AggregatedAttestation,
     Attestation,
@@ -48,6 +50,7 @@
     "ForkRegistry",
     "LstarSpec",
     "LstarStore",
+    "SigScheme",
     "SignedAggregatedAttestation",
     "SignedAttestation",
     "SignedBlock",
@@ -57,4 +60,5 @@
     "Store",
     "Validator",
     "Validators",
+    "capabilities",
 ]

src/lean_spec/forks/capabilities.py

@@ -0,0 +1,16 @@
+"""Optional structural capabilities a fork may advertise."""
+
+from typing import ClassVar, Protocol, runtime_checkable
+
+from lean_spec.subspecs.xmss.interface import GeneralizedXmssScheme
+
+
+@runtime_checkable
+class SigScheme(Protocol):
+    """Fork advertising a generalized XMSS signature scheme.
+
+    - The runtime check only verifies the attribute is present.
+    - The static type contract is enforced by the type checker.
+    """
+
+    sig_scheme: ClassVar[GeneralizedXmssScheme]

src/lean_spec/forks/lstar/spec.py

@@ -75,6 +75,9 @@ class LstarSpec(ForkProtocol):
 
     previous: ClassVar[type[ForkProtocol] | None] = None
 
+    # Capabilities advertised by this fork.
+    sig_scheme: ClassVar[GeneralizedXmssScheme] = TARGET_SIGNATURE_SCHEME
+
     state_class: type[State] = State
     block_class: type[Block] = Block
     block_body_class: type[BlockBody] = BlockBody
@@ -787,7 +790,6 @@ def verify_signatures(
         self,
         signed_block: SignedBlock,
         validators: Validators,
-        scheme: GeneralizedXmssScheme = TARGET_SIGNATURE_SCHEME,
     ) -> bool:
         """
         Verify all XMSS signatures in this signed block.
@@ -797,10 +799,11 @@ def verify_signatures(
         - Each body attestation is signed by participating validators
         - The proposer signed the block root with the proposal key
 
+        The signing scheme is read from this fork's capability.
+
         Args:
             signed_block: The signed block whose signatures are checked.
             validators: Validator registry providing public keys for verification.
-            scheme: XMSS signature scheme for verification.
 
         Returns:
             True if all signatures are valid.
@@ -862,7 +865,7 @@ def verify_signatures(
         block_root = hash_tree_root(block)
 
         try:
-            valid = scheme.verify(
+            valid = self.sig_scheme.verify(
                 proposer.get_proposal_pubkey(),
                 block.slot,
                 block_root,
@@ -1031,19 +1034,28 @@ def on_gossip_attestation(
         self,
         store: LstarStore,
         signed_attestation: SignedAttestation,
-        scheme: GeneralizedXmssScheme = TARGET_SIGNATURE_SCHEME,
         is_aggregator: bool = False,
     ) -> LstarStore:
         """Process a signed attestation received via gossip network.
 
         This method:
-        1. Verifies the XMSS signature
+
+        1. Verifies the XMSS signature using this fork's capability
         2. Stores the signature when the node is in aggregator mode
 
         Subnet filtering happens at the p2p subscription layer — only
         attestations from subscribed subnets reach this method. No
         additional subnet check is needed here.
 
+        Args:
+            store: The current forkchoice store.
+            signed_attestation: The signed attestation to process.
+            is_aggregator: True if the node is an aggregator.
+
+        Returns:
+            A new store with the attestation signature recorded when in
+            aggregator mode, otherwise the input store unchanged.
+
         Raises:
             ValueError: If validator not found in state.
             AssertionError: If signature verification fails.
@@ -1067,7 +1079,7 @@ def on_gossip_attestation(
             )
             public_key = key_state.validators[validator_id].get_attestation_pubkey()
 
-            assert scheme.verify(
+            assert self.sig_scheme.verify(
                 public_key, attestation_data.slot, hash_tree_root(attestation_data), signature
             ), "Signature verification failed"
 
@@ -1160,16 +1172,18 @@ def on_block(
         self,
         store: LstarStore,
         signed_block: SignedBlock,
-        scheme: GeneralizedXmssScheme = TARGET_SIGNATURE_SCHEME,
     ) -> LstarStore:
         """Process a new block and update the forkchoice state.
 
         This method integrates a block into the forkchoice store by:
+
         1. Validating the block's parent exists
         2. Computing the post-state via the state transition function
         3. Processing attestations included in the block body (on-chain)
         4. Updating the forkchoice head
 
+        Signatures are verified using this fork's capability.
+
         Raises:
             AssertionError: If parent block/state not found in store.
         """
@@ -1196,7 +1210,7 @@ def on_block(
             )
 
             # Validate cryptographic signatures
-            valid_signatures = self.verify_signatures(signed_block, parent_state.validators, scheme)
+            valid_signatures = self.verify_signatures(signed_block, parent_state.validators)
 
             # Execute state transition function to compute post-block state
             post_state = self.state_transition(parent_state, block, valid_signatures)

tests/consensus/lstar/test_capability_gating.py

@@ -0,0 +1,39 @@
+"""Smoke tests for the capability-requirement marker dispatch."""
+
+from typing import ClassVar, Protocol, runtime_checkable
+
+import pytest
+from consensus_testing import StateExpectation, StateTransitionTestFiller, generate_pre_state
+from framework import requires_capability
+
+from lean_spec.forks import SigScheme
+from lean_spec.types import Slot
+
+pytestmark = pytest.mark.valid_until("Lstar")
+
+
+@runtime_checkable
+class _AbsentCapability(Protocol):
+    """A capability no real fork advertises."""
+
+    never_an_attribute_on_any_real_fork: ClassVar[object]
+
+
+@requires_capability(SigScheme)
+def test_runs_when_fork_advertises_sigscheme(
+    state_transition_test: StateTransitionTestFiller,
+) -> None:
+    """Lstar advertises the signature-scheme capability — this test runs."""
+    state_transition_test(
+        pre=generate_pre_state(),
+        blocks=[],
+        post=StateExpectation(slot=Slot(0)),
+    )
+
+
+@requires_capability(_AbsentCapability)
+def test_deselected_when_capability_absent(
+    state_transition_test: StateTransitionTestFiller,
+) -> None:
+    """No fork advertises the absent capability — this test must be deselected."""
+    raise AssertionError("this test was executed — capability-requirement deselection is broken")