refactor: enforce frozen natively in StrictBaseModel (leanEthereum#845)

Restore the frozen constraint that leanEthereum#789 removed, this time enforced once
in the base model instead of per class. Every spec type is now immutable
by default, with no opt-outs: the State accumulator and the fork-choice
Store are frozen too, and every remaining in-place mutation site is
converted back to the model_copy(update=...) functional style.

Follow-up to leanEthereum#842/leanEthereum#843, as discussed in the leanEthereum#843 review thread.

Changes:
- base.py: add frozen to StrictBaseModel; restore the pre-leanEthereum#789 docstring.
- Delete the 17 per-class model_config | {"frozen": True} overrides
  (block, checkpoint, attestation, aggregation, validator, xmss, eth2)
  now that the base enforces them.
- state_transition.py: process_slots rebinds through model_copy (the
  deepcopy barrier is no longer needed), process_block_header applies
  its updates atomically in one final copy, process_attestations
  returns a new state.
- fork_choice.py, timeline.py, validator_duties.py, aggregation.py:
  every store update flows through model_copy; dicts and inner sets are
  shallow-copied before growing so the caller's store is left untouched.
- node/chain/service.py, node/sync/service.py: rebind the store instead
  of patching it in place.
- xmss/interface.py: advance_preparation returns a rebuilt secret key.
- enr.py: from_rlp rebuilds the record with the computed node id.
- packages/testing + tests: all fixture-setup mutations converted to
  model_copy rebinding; helpers that mutated arguments now return the
  new instance.
- Restore test_frozen_rejects_assignment and the immutability wording
  in the SSZ patterns rule; add mirrored immutability tests for State
  and Store.

Validation:
- just check passes (ruff lint + format, ty, codespell, mdformat, lock)
- Unit suites: lstar spec 257 passed, node 1682 passed,
  crypto/ssz/enr/containers/base 1660 passed

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

Author: tcoratger

.claude/rules/ssz-patterns.md

@@ -29,7 +29,7 @@ When creating SSZ types, follow these established patterns:
 - These use Pydantic models with a `data` field for contents
 - Example: `MyList(data=[1, 2, 3])` *has* a list of data with SSZ serialization
 
-**Key principle**: If the type conceptually *holds* or *contains* other data, use SSZModel for consistent validation and SSZ encoding.
+**Key principle**: If the type conceptually *holds* or *contains* other data, use SSZModel for consistent validation and immutability.
 
 ## Modular Architecture
 

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

@@ -246,7 +246,9 @@ def make_fixture(self) -> Self:
 
         # Updating validators changes the state root.
         # We must also update the anchor block to match.
-        self.anchor_state.validators = Validators(data=updated_validators)
+        self.anchor_state = self.anchor_state.model_copy(
+            update={"validators": Validators(data=updated_validators)}
+        )
         self.anchor_block = self.anchor_block.model_copy(
             update={"state_root": hash_tree_root(self.anchor_state)}
         )

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

@@ -540,8 +540,7 @@ def build_signed_block_with_store(
         }
         for attestation_data, proofs in merged_store.latest_known_aggregated_payloads.items():
             merged_known.setdefault(attestation_data, set()).update(proofs)
-        caller_store.latest_known_aggregated_payloads = merged_known
-        store = caller_store
+        store = caller_store.model_copy(update={"latest_known_aggregated_payloads": merged_known})
 
         # Append forced attestations that bypass the builder's MAX cap.
         # Each entry is signed and aggregated so the block carries valid proofs.

src/lean_spec/base.py

@@ -48,15 +48,17 @@ def to_json(self, **kwargs: Any) -> dict[str, Any]:
 
 class StrictBaseModel(CamelModel):
     """
-    Strict base model for all spec types.
+    Immutable, strict base model for all spec types.
 
-    Adds two constraints on top of CamelModel:
+    Adds three constraints on top of CamelModel:
 
+    - Frozen: attribute assignment after construction raises
     - Extra forbidden: unknown fields rejected at construction
     - Strict: no implicit type coercion
     """
 
     model_config = CamelModel.model_config | {
         "extra": "forbid",
+        "frozen": True,
         "strict": True,
     }

src/lean_spec/node/chain/service.py

@@ -110,7 +110,9 @@ async def _tick_to(self, target_interval: Interval) -> list[SignedAggregatedAtte
         # Acceptance for the jumped slots waits for the final slot's tick.
         # That is safe: acceptance is a monotone pool merge, and the head recomputes from scratch.
         if target_interval - store.time > Interval(INTERVALS_PER_SLOT):
-            store.time = target_interval - Interval(INTERVALS_PER_SLOT)
+            store = store.model_copy(
+                update={"time": target_interval - Interval(INTERVALS_PER_SLOT)}
+            )
 
         # Tick remaining intervals one at a time.
         while store.time < target_interval:

src/lean_spec/node/networking/enr/enr.py

@@ -373,7 +373,7 @@ def from_rlp(cls, rlp_data: bytes) -> Self:
         # Compute and store node_id for routing/identification.
         node_id = enr.compute_node_id()
         if node_id is not None:
-            enr.node_id = node_id
+            enr = enr.model_copy(update={"node_id": node_id})
 
         return enr
 

src/lean_spec/node/networking/enr/eth2.py

@@ -35,8 +35,6 @@ class Eth2Data(StrictBaseModel):
     SSZ: fork_digest (4) + next_fork_version (4) + next_fork_epoch (8)
     """
 
-    model_config = StrictBaseModel.model_config | {"frozen": True}
-
     fork_digest: ForkDigest
     """Current active fork identifier (4 bytes)."""
 

src/lean_spec/node/sync/service.py

@@ -656,7 +656,7 @@ def _deconstruct_block_into_store(
             aggregates.append(SignedAggregatedAttestation(data=attestation_data, proof=combined))
 
         if aggregates:
-            store.latest_new_aggregated_payloads = new_payloads
+            store = store.model_copy(update={"latest_new_aggregated_payloads": new_payloads})
 
         return store, aggregates
 

src/lean_spec/spec/crypto/xmss/containers.py

@@ -42,8 +42,6 @@ class PublicKey(HexSerializedContainer):
 class Signature(HexSerializedContainer):
     """A single XMSS signature for one slot and message under one public key."""
 
-    model_config = Container.model_config | {"frozen": True}
-
     path: HashTreeOpening
     """Authentication path from the one-time key up to the Merkle root."""
 
@@ -138,8 +136,6 @@ class SecretKey(HexSerializedContainer):
 class KeyPair(StrictBaseModel):
     """A single XMSS public/secret pair returned by key generation."""
 
-    model_config = StrictBaseModel.model_config | {"frozen": True}
-
     public_key: PublicKey
     """Public key."""
 
@@ -161,8 +157,6 @@ class ValidatorKeyPair(StrictBaseModel):
     Two independent pairs let each role sign from its own Winternitz chains.
     """
 
-    model_config = StrictBaseModel.model_config | {"frozen": True}
-
     attestation_keypair: KeyPair
     """Key pair used to sign attestation data."""
 

src/lean_spec/spec/crypto/xmss/interface.py

@@ -398,10 +398,13 @@ def advance_preparation(self, secret_key: SecretKey) -> SecretKey:
         )
 
         # Phase 3: rotate the right tree into the left slot, advance the index.
-        secret_key.left_bottom_tree = secret_key.right_bottom_tree
-        secret_key.right_bottom_tree = new_right_bottom_tree
-        secret_key.left_bottom_tree_index = Uint64(left_index + 1)
-        return secret_key
+        return secret_key.model_copy(
+            update={
+                "left_bottom_tree": secret_key.right_bottom_tree,
+                "right_bottom_tree": new_right_bottom_tree,
+                "left_bottom_tree_index": Uint64(left_index + 1),
+            }
+        )
 
 
 PROD_SIGNATURE_SCHEME = GeneralizedXmssScheme(config=PROD_CONFIG, poseidon=POSEIDON)