test(consensus): cover in-block recursive merge of overlapping pool proofs

The block builder's collapse-by-data branch in build_block merges
multiple single-message aggregates for the same AttestationData via
SingleMessageAggregate.aggregate(children=..., raw_xmss=[]).  This is
the only proposal-time path that constructs a recursive single-message
aggregate, and no existing fork-choice vector reached it because the
local aggregator absorbs same-data proofs before block build time.

Gossiping two aggregated proofs with overlapping participant sets at
slot 2 interval 3 (past the aggregate phase) bypasses the local
aggregator, so both proofs migrate side by side into the known pool at
slot 2 interval 4.  The greedy picker takes both because the second
still adds one uncovered validator on top of the first, exercising
the merge branch on overlapping children rather than the unrealistic
fully disjoint shape.

Author: unnawut

tests/consensus/lstar/fc/test_block_production.py

@@ -415,6 +415,101 @@ def test_produce_block_includes_pending_attestations(
     )
 
 
+def test_block_builder_merges_proofs_to_attestation(
+    fork_choice_test: ForkChoiceTestFiller,
+) -> None:
+    """
+    Two overlapping proofs for the same data merge into one in-block attestation.
+
+    Scenario
+    --------
+    Chain: genesis(0) -> block_1(1) -> block_2(2)
+
+    At slot 2, interval 3 (past the aggregate phase), two separate aggregators
+    each gossip an aggregated attestation for the same AttestationData over
+    slot_2 -> block_2:
+
+    - aggregator A: validators {0, 1, 2}
+    - aggregator B: validators {1, 2, 3}  (overlaps A on {1, 2})
+
+    Gossiping after interval 2 stops the local aggregator from absorbing them
+    on this slot. Both proofs land in the new pool side by side.
+    Slot 2, interval 4 then migrates both into the known pool.
+
+    Slot 3 block production reads from the known pool.
+    The greedy picker takes the first proof, then takes the second because it
+    still adds one uncovered validator on top of the first.
+    The builder merges the two proofs into one via SingleMessageAggregate.aggregate
+    with two children.
+
+    Expected
+    --------
+    - 1 aggregated attestation in the block body.
+    - Participants = {0, 1, 2, 3} (union of the two source proofs).
+    """
+    # Slot 2 interval 3 sits after the aggregate phase but before acceptance.
+    # Gossiped aggregates accumulate in the new pool without local pre-merging.
+    after_aggregate_interval = 2 * int(INTERVALS_PER_SLOT) + 3
+    after_aggregate_time = math.ceil(
+        after_aggregate_interval * int(MILLISECONDS_PER_INTERVAL) / 1000
+    )
+    # Slot 3 boundary. Crosses slot 2 interval 4, which migrates new -> known.
+    next_slot_time = 3 * int(SECONDS_PER_SLOT)
+
+    fork_choice_test(
+        steps=[
+            BlockStep(
+                block=BlockSpec(slot=Slot(1), label="block_1"),
+                checks=StoreChecks(head_slot=Slot(1)),
+            ),
+            BlockStep(
+                block=BlockSpec(slot=Slot(2), label="block_2"),
+                checks=StoreChecks(head_slot=Slot(2)),
+            ),
+            TickStep(time=after_aggregate_time),
+            GossipAggregatedAttestationStep(
+                attestation=AggregatedAttestationSpec(
+                    validator_indices=[
+                        ValidatorIndex(0),
+                        ValidatorIndex(1),
+                        ValidatorIndex(2),
+                    ],
+                    slot=Slot(2),
+                    target_slot=Slot(2),
+                    target_root_label="block_2",
+                ),
+            ),
+            GossipAggregatedAttestationStep(
+                attestation=AggregatedAttestationSpec(
+                    validator_indices=[
+                        ValidatorIndex(1),
+                        ValidatorIndex(2),
+                        ValidatorIndex(3),
+                    ],
+                    slot=Slot(2),
+                    target_slot=Slot(2),
+                    target_root_label="block_2",
+                ),
+            ),
+            TickStep(time=next_slot_time),
+            BlockStep(
+                block=BlockSpec(slot=Slot(3), label="block_3"),
+                checks=StoreChecks(
+                    head_slot=Slot(3),
+                    block_attestation_count=1,
+                    block_attestations=[
+                        AggregatedAttestationCheck(
+                            participants={0, 1, 2, 3},
+                            attestation_slot=Slot(2),
+                            target_slot=Slot(2),
+                        ),
+                    ],
+                ),
+            ),
+        ],
+    )
+
+
 def test_block_builder_recovers_finality_after_non_zero_boundary_stall(
     fork_choice_test: ForkChoiceTestFiller,
 ) -> None: