commonware_consensus/marshal/coding/

mod.rs

//! Ordered delivery of erasure-coded blocks.
//!
//! # Overview
//!
//! The coding marshal couples the consensus pipeline with erasure-coded block broadcast.
//! Blocks are produced by an application, encoded into [`types::Shard`]s, fanned out to peers, and
//! later reconstructed when a notarization or finalization proves that the data is needed.
//! Compared to [`super::standard`], this variant makes more efficient usage of the network's bandwidth
//! by spreading the load of block dissemination across all participants.
//!
//! # Components
//!
//! - [`crate::marshal::core::Actor`]: The unified marshal actor that orders finalized blocks,
//!   handles acknowledgements from the application, and requests repairs when gaps are detected.
//!   Used with [`Coding`] as the variant type parameter.
//! - [`crate::marshal::core::Mailbox`]: Accepts requests from other local subsystems and forwards
//!   them to the actor. Used with [`Coding`] as the variant type parameter.
//! - [`shards::Engine`]: Broadcasts shards, verifies locally held fragments, and reconstructs
//!   entire [`types::CodedBlock`]s on demand.
//! - [`crate::marshal::resolver`]: Issues outbound fetches to remote peers when marshal is missing
//!   a block, notarization, or finalization referenced by consensus.
//! - [`types`]: Defines commitments, distribution shards, and helper builders used across the
//!   module.
//! - [`Marshaled`]: Wraps an [`crate::Application`] implementation so it automatically enforces
//!   epoch boundaries and performs erasure encoding before a proposal leaves the application.
//!
//! # Data Flow
//!
//! 1. The application produces a block through [`Marshaled`], which encodes the payload and
//!    obtains a [`crate::types::coding::Commitment`] describing the shard layout.
//! 2. The block is broadcast via [`shards::Engine`]; each participant receives exactly one shard
//!    and reshares it to everyone else once it verifies the fragment.
//! 3. The actor ingests notarizations/finalizations from `simplex`, pulls reconstructed blocks
//!    from the shard engine or backfills them through [`crate::marshal::resolver`], and durably
//!    persists the ordered data.
//! 4. The actor reports finalized blocks to the node's [`crate::Reporter`] at-least-once and
//!    drives repair loops whenever notarizations reference yet-to-be-delivered payloads.
//!
//! # Storage and Repair
//!
//! Notarized data and certificates live in prunable archives managed internally, while finalized
//! blocks are migrated into immutable archives. Any gaps are filled by asking peers for specific
//! commitments through the resolver pipeline. The shard engine keeps only ephemeral, in-memory
//! caches; once a block is finalized it is evicted from the reconstruction map, reducing memory
//! pressure.
//!
//! # When to Use
//!
//! Choose this module when the consensus deployment wants erasure-coded dissemination with the
//! same ordering guarantees provided by [`super::standard`]. The API is a breaking change from
//! the standard marshal: applications must adapt to the coding-specific variant type and buffer
//! implementation required by this module.

pub mod shards;
pub mod types;
pub(crate) mod validation;

mod variant;
pub use variant::Coding;

mod marshaled;
pub use marshaled::{Marshaled, MarshaledConfig};

#[cfg(test)]
mod tests {
    use crate::{
        Automaton, Block, CertifiableAutomaton, CertifiableBlock, Relay,
        marshal::{
            ancestry::BlockProvider,
            coding::{
                Coding, Marshaled, MarshaledConfig, shards,
                types::{CodedBlock, coding_config_for_participants, hash_context},
            },
            config::{Config, Start},
            core,
            mocks::{
                application::Application,
                harness::{
                    self, BLOCKS_PER_EPOCH, CodingB, CodingCtx, CodingHarness, D, EmptyProvider, K,
                    LINK, NAMESPACE, NUM_VALIDATORS, QUORUM, S, TEST_QUOTA, TestHarness,
                    UNRELIABLE_LINK, V, default_leader, genesis_commitment, make_coding_block,
                    setup_network_links, setup_network_with_participants,
                },
                verifying::{GatedVerifyingApp, MockVerifyingApp},
            },
            resolver::handler,
        },
        simplex::{
            Plan, scheme::bls12381_threshold::vrf as bls12381_threshold_vrf, types::Proposal,
        },
        types::{Epoch, Epocher, FixedEpocher, Height, Round, View, ViewDelta, coding::Commitment},
    };
    use bytes::Bytes;
    use commonware_actor::{Feedback, mailbox};
    use commonware_codec::{Encode, FixedSize};
    use commonware_coding::{CodecConfig, Config as CodingConfig, ReedSolomon};
    use commonware_cryptography::{
        Committable, Digestible, Hasher,
        certificate::{ConstantProvider, Verifier as _, mocks::Fixture},
        sha256::Sha256,
    };
    use commonware_macros::{select, test_group, test_traced};
    use commonware_p2p::{Recipients, Sender as _};
    use commonware_parallel::Sequential;
    use commonware_resolver::{Delivery, Fetch, Resolver, TargetedResolver};
    use commonware_runtime::{
        Clock, Metrics, Runner, Supervisor as _, buffer::paged::CacheRef, deterministic,
    };
    use commonware_storage::archive::immutable;
    use commonware_utils::{
        NZU16, NZU64, NZUsize, channel::oneshot, sync::Mutex, vec::NonEmptyVec,
    };
    use std::{sync::Arc, time::Duration};

    type TestCodingVariant = Coding<CodingB, ReedSolomon<Sha256>, Sha256, K>;
    type TestCodedBlock = CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>;
    type TestCommitment = Commitment<CodingB, ReedSolomon<Sha256>, Sha256>;
    type CodingSendRecord = (Round, Arc<TestCodedBlock>, Recipients<K>);

    // Smallest valid coding config used to build trusted genesis commitments.
    const GENESIS_CODING_CONFIG: CodingConfig = CodingConfig {
        minimum_shards: NZU16!(1),
        extra_shards: NZU16!(1),
    };

    #[test]
    fn mailbox_provides_application_blocks() {
        fn assert_provider<P: BlockProvider<Block = CodingB>>() {}
        assert_provider::<core::Mailbox<S, TestCodingVariant>>();
    }

    /// A coding buffer that records subscriptions and never resolves them.
    #[derive(Clone, Default)]
    struct RecordingCodingBuffer {
        digest_subscriptions: Arc<Mutex<Vec<oneshot::Sender<Arc<TestCodedBlock>>>>>,
        commitment_subscriptions: Arc<Mutex<Vec<oneshot::Sender<Arc<TestCodedBlock>>>>>,
        sends: Arc<Mutex<Vec<CodingSendRecord>>>,
    }

    impl RecordingCodingBuffer {
        fn subscription_count(&self) -> usize {
            self.digest_subscriptions.lock().len() + self.commitment_subscriptions.lock().len()
        }

        fn commitment_subscription_count(&self) -> usize {
            self.commitment_subscriptions.lock().len()
        }
    }

    impl core::Buffer<TestCodingVariant> for RecordingCodingBuffer {
        type PublicKey = K;

        async fn find_by_digest(&self, _digest: D) -> Option<Arc<TestCodedBlock>> {
            None
        }

        async fn find_by_commitment(
            &self,
            _commitment: TestCommitment,
        ) -> Option<Arc<TestCodedBlock>> {
            None
        }

        fn subscribe_by_digest(
            &self,
            _digest: D,
        ) -> Option<oneshot::Receiver<Arc<TestCodedBlock>>> {
            let (sender, receiver) = oneshot::channel();
            self.digest_subscriptions.lock().push(sender);
            Some(receiver)
        }

        fn subscribe_by_commitment(
            &self,
            _commitment: TestCommitment,
        ) -> Option<oneshot::Receiver<Arc<TestCodedBlock>>> {
            let (sender, receiver) = oneshot::channel();
            self.commitment_subscriptions.lock().push(sender);
            Some(receiver)
        }

        fn retire(&self, _update: core::Retirement<TestCommitment>) {}

        fn send(&self, round: Round, block: Arc<TestCodedBlock>, recipients: Recipients<K>) {
            self.sends.lock().push((round, block, recipients));
        }
    }

    type CodingFetchRecord = Fetch<handler::Key<TestCommitment>, handler::Annotation>;
    type CodingTargetedFetch = (handler::Key<TestCommitment>, NonEmptyVec<K>);

    /// A resolver that records each fetch invocation; other methods are no-ops.
    #[derive(Clone, Default)]
    struct RecordingResolver {
        fetches: Arc<Mutex<Vec<CodingFetchRecord>>>,
        targeted: Arc<Mutex<Vec<CodingTargetedFetch>>>,
        auto_delivery: Arc<Mutex<Option<Bytes>>>,
        delivery_responses: Arc<Mutex<Vec<oneshot::Receiver<bool>>>>,
        sender: Option<mailbox::Sender<handler::Message<TestCommitment>>>,
    }

    impl RecordingResolver {
        fn holding(metrics: impl Metrics) -> (handler::Receiver<TestCommitment>, Self) {
            let (sender, receiver) = mailbox::new(metrics, NZUsize!(100));
            (
                handler::Receiver::new(receiver),
                Self {
                    fetches: Arc::new(Mutex::new(Vec::new())),
                    targeted: Arc::new(Mutex::new(Vec::new())),
                    auto_delivery: Arc::new(Mutex::new(None)),
                    delivery_responses: Arc::new(Mutex::new(Vec::new())),
                    sender: Some(sender),
                },
            )
        }

        fn record_fetch(&self, fetch: CodingFetchRecord) {
            self.fetches.lock().push(fetch.clone());
            let Some(value) = self.auto_delivery.lock().take() else {
                return;
            };
            let Some(sender) = &self.sender else {
                return;
            };
            let (response, response_rx) = oneshot::channel();
            self.delivery_responses.lock().push(response_rx);
            let _ = sender.enqueue(handler::Message::Deliver {
                delivery: Delivery {
                    key: fetch.key,
                    subscribers: NonEmptyVec::new((fetch.subscriber, tracing::Span::none())),
                },
                value,
                response,
            });
        }

        fn respond_to_next_fetch(&self, value: Bytes) {
            let replaced = self.auto_delivery.lock().replace(value);
            assert!(
                replaced.is_none(),
                "recording resolver already has an automatic delivery"
            );
        }

        async fn wait_for_delivery_response(&self) -> bool {
            let response = self
                .delivery_responses
                .lock()
                .pop()
                .expect("delivery response missing");
            response.await.expect("delivery response sender dropped")
        }

        fn fetches(&self) -> Vec<CodingFetchRecord> {
            self.fetches.lock().clone()
        }

        fn targeted(&self) -> Vec<CodingTargetedFetch> {
            self.targeted.lock().clone()
        }
    }

    impl Resolver for RecordingResolver {
        type Key = handler::Key<TestCommitment>;
        type Subscriber = handler::Annotation;

        fn fetch<F>(&mut self, fetch: F) -> Feedback
        where
            F: Into<Fetch<Self::Key, Self::Subscriber>> + Send,
        {
            self.record_fetch(fetch.into());
            Feedback::Ok
        }

        fn fetch_all<F>(&mut self, fetches: Vec<F>) -> Feedback
        where
            F: Into<Fetch<Self::Key, Self::Subscriber>> + Send,
        {
            for fetch in fetches {
                self.record_fetch(fetch.into());
            }
            Feedback::Ok
        }

        fn retain(
            &mut self,
            _predicate: impl Fn(&Self::Key, &Self::Subscriber) -> bool + Send + 'static,
        ) -> Feedback {
            Feedback::Ok
        }
    }

    impl TargetedResolver for RecordingResolver {
        type PublicKey = K;

        fn fetch_targeted(
            &mut self,
            fetch: impl Into<Fetch<Self::Key, Self::Subscriber>> + Send,
            targets: NonEmptyVec<Self::PublicKey>,
        ) -> Feedback {
            self.targeted.lock().push((fetch.into().key, targets));
            Feedback::Ok
        }

        fn fetch_all_targeted<F>(
            &mut self,
            fetches: Vec<(F, NonEmptyVec<Self::PublicKey>)>,
        ) -> Feedback
        where
            F: Into<Fetch<Self::Key, Self::Subscriber>> + Send,
        {
            let mut targeted = self.targeted.lock();
            for (fetch, targets) in fetches {
                targeted.push((fetch.into().key, targets));
            }
            Feedback::Ok
        }
    }

    async fn start_coding_actor_with_recording(
        context: deterministic::Context,
        partition_prefix: &str,
        provider: ConstantProvider<S, Epoch>,
        buffer: RecordingCodingBuffer,
    ) -> (
        core::Mailbox<S, TestCodingVariant>,
        RecordingResolver,
        commonware_runtime::Handle<()>,
    ) {
        let config = Config {
            provider,
            epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
            start: Start::Genesis(CodingHarness::genesis_block(NUM_VALIDATORS as u16)),
            mailbox_size: NZUsize!(100),
            view_retention: ViewDelta::new(10),
            max_repair: NZUsize!(10),
            max_pending_acks: NZUsize!(1),
            block_codec_config: (),
            partition_prefix: partition_prefix.to_string(),
            prunable_items_per_section: NZU64!(10),
            replay_buffer: NZUsize!(1024),
            key_write_buffer: NZUsize!(1024),
            value_write_buffer: NZUsize!(1024),
            page_cache: CacheRef::from_pooler(
                &context,
                harness::PAGE_SIZE,
                harness::PAGE_CACHE_SIZE,
            ),
            strategy: Sequential,
        };

        let finalizations_by_height = immutable::Archive::init(
            context.child("finalizations_by_height"),
            immutable::Config {
                metadata_partition: format!("{partition_prefix}-finalizations-by-height-metadata"),
                freezer_table_partition: format!(
                    "{partition_prefix}-finalizations-by-height-freezer-table"
                ),
                freezer_table_initial_size: 64,
                freezer_table_resize_frequency: 10,
                freezer_table_resize_chunk_size: 10,
                freezer_key_partition: format!(
                    "{partition_prefix}-finalizations-by-height-freezer-key"
                ),
                freezer_key_page_cache: config.page_cache.clone(),
                freezer_value_partition: format!(
                    "{partition_prefix}-finalizations-by-height-freezer-value"
                ),
                freezer_value_target_size: 1024,
                freezer_value_compression: None,
                ordinal_partition: format!("{partition_prefix}-finalizations-by-height-ordinal"),
                items_per_section: NZU64!(10),
                codec_config: S::certificate_codec_config_unbounded(),
                replay_buffer: config.replay_buffer,
                freezer_key_write_buffer: config.key_write_buffer,
                freezer_value_write_buffer: config.value_write_buffer,
                ordinal_write_buffer: config.key_write_buffer,
            },
        )
        .await
        .expect("failed to initialize finalizations by height archive");

        let finalized_blocks = immutable::Archive::init(
            context.child("finalized_blocks"),
            immutable::Config {
                metadata_partition: format!("{partition_prefix}-finalized_blocks-metadata"),
                freezer_table_partition: format!(
                    "{partition_prefix}-finalized_blocks-freezer-table"
                ),
                freezer_table_initial_size: 64,
                freezer_table_resize_frequency: 10,
                freezer_table_resize_chunk_size: 10,
                freezer_key_partition: format!("{partition_prefix}-finalized_blocks-freezer-key"),
                freezer_key_page_cache: config.page_cache.clone(),
                freezer_value_partition: format!(
                    "{partition_prefix}-finalized_blocks-freezer-value"
                ),
                freezer_value_target_size: 1024,
                freezer_value_compression: None,
                ordinal_partition: format!("{partition_prefix}-finalized_blocks-ordinal"),
                items_per_section: NZU64!(10),
                codec_config: config.block_codec_config,
                replay_buffer: config.replay_buffer,
                freezer_key_write_buffer: config.key_write_buffer,
                freezer_value_write_buffer: config.value_write_buffer,
                ordinal_write_buffer: config.key_write_buffer,
            },
        )
        .await
        .expect("failed to initialize finalized blocks archive");

        let (actor, mailbox, _) = core::Actor::init(
            context.child("actor"),
            finalizations_by_height,
            finalized_blocks,
            config,
        )
        .await;
        let (resolver_rx, resolver) = RecordingResolver::holding(context.child("resolver"));
        let actor_handle = actor.start(
            Application::<CodingB>::default(),
            buffer,
            (resolver_rx, resolver.clone()),
        );
        (mailbox, resolver, actor_handle)
    }

    async fn start_shard_mailbox(
        context: deterministic::Context,
        participants: Vec<K>,
        provider: ConstantProvider<S, Epoch>,
    ) -> shards::Mailbox<CodingB, ReedSolomon<Sha256>, Sha256, K> {
        let me = participants[0].clone();
        let oracle =
            setup_network_with_participants(context.child("network"), NZUsize!(1), participants)
                .await;
        let control = oracle.control(me.clone());
        let shard_config: shards::Config<_, _, _, _, _, Sha256, _, _> = shards::Config {
            scheme_provider: provider,
            blocker: control.clone(),
            shard_codec_cfg: CodecConfig {
                maximum_shard_size: 1024 * 1024,
            },
            block_codec_cfg: (),
            strategy: Sequential,
            mailbox_size: NZUsize!(10),
            peer_buffer_size: NZUsize!(64),
            background_channel_capacity: NZUsize!(1024),
            peer_provider: oracle.manager(),
        };
        let (shard_engine, shard_mailbox) =
            shards::Engine::new(context.child("shards"), shard_config);
        let network = control.register(0, TEST_QUOTA).await.unwrap();
        shard_engine.start(network);
        shard_mailbox
    }

    fn genesis_block() -> CodingB {
        let genesis_ctx = CodingCtx {
            round: Round::zero(),
            leader: default_leader(),
            parent: (View::zero(), genesis_commitment()),
        };
        make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0)
    }

    fn genesis_coding_commitment(block: &CodingB) -> TestCommitment {
        TestCommitment::from((
            block.digest(),
            block.digest(),
            hash_context::<Sha256, _>(&block.context()),
            GENESIS_CODING_CONFIG,
        ))
    }

    fn missing_candidate(me: K) -> (CodingCtx, TestCodedBlock) {
        let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
        let genesis = genesis_block();
        let genesis_parent_commitment = genesis_coding_commitment(&genesis);
        let round = Round::new(Epoch::zero(), View::new(1));
        let candidate_ctx = CodingCtx {
            round,
            leader: me,
            parent: (View::zero(), genesis_parent_commitment),
        };
        let candidate =
            make_coding_block(candidate_ctx.clone(), genesis.digest(), Height::new(1), 100);
        let coded_candidate: TestCodedBlock =
            CodedBlock::new(candidate, coding_config, &Sequential);
        (candidate_ctx, coded_candidate)
    }

    #[test_traced("WARN")]
    fn test_coding_batched_acks_retire_each_exact_commitment() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;
            let mut setup = CodingHarness::setup_validator_with(
                context.child("validator"),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
                NZUsize!(2),
                Application::manual_ack(),
            )
            .await;
            assert_eq!(setup.application.acknowledged().await, Height::zero());

            let mut parent = Sha256::hash(&[b""]);
            let mut parent_commitment =
                CodingHarness::genesis_parent_commitment(NUM_VALIDATORS as u16);
            let mut commitments = Vec::new();
            for height in 1..=2 {
                let round = Round::new(Epoch::zero(), View::new(height));
                let block = CodingHarness::make_test_block(
                    parent,
                    parent_commitment,
                    Height::new(height),
                    height,
                    NUM_VALIDATORS as u16,
                );
                let commitment = block.commitment();
                parent = block.digest();
                parent_commitment = commitment;
                commitments.push(commitment);

                setup.extra.proposed(
                    Round::new(Epoch::zero(), View::new(height + 10)),
                    block.clone(),
                );
                assert!(setup.extra.get(commitment).await.is_some());
                assert!(setup.mailbox.verified(round, block).await);
                CodingHarness::report_finalization(
                    &mut setup.mailbox,
                    CodingHarness::make_finalization(
                        Proposal {
                            round,
                            parent: View::new(height - 1),
                            payload: commitment,
                        },
                        &schemes,
                        QUORUM,
                    ),
                )
                .await;
            }

            while setup.application.pending_ack_heights() != vec![Height::new(1), Height::new(2)] {
                context.sleep(Duration::from_millis(10)).await;
            }
            assert_eq!(setup.application.acknowledge_next(), Some(Height::new(1)));
            assert_eq!(setup.application.acknowledge_next(), Some(Height::new(2)));

            while setup.extra.get(commitments[1]).await.is_some() {
                context.sleep(Duration::from_millis(10)).await;
            }
            assert!(setup.extra.get(commitments[0]).await.is_none());
        });
    }

    #[test_traced("WARN")]
    fn test_coding_floor_retires_only_superseded_ack_commitments() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;
            let mut setup = CodingHarness::setup_validator_with(
                context.child("validator"),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
                NZUsize!(3),
                Application::manual_ack(),
            )
            .await;
            assert_eq!(setup.application.acknowledged().await, Height::zero());

            let mut parent = Sha256::hash(&[b""]);
            let mut parent_commitment =
                CodingHarness::genesis_parent_commitment(NUM_VALIDATORS as u16);
            let mut commitments = Vec::new();
            let mut floor = None;
            for height in 1..=3 {
                let round = Round::new(Epoch::zero(), View::new(height));
                let block = CodingHarness::make_test_block(
                    parent,
                    parent_commitment,
                    Height::new(height),
                    height * 100,
                    NUM_VALIDATORS as u16,
                );
                let commitment = block.commitment();
                parent = block.digest();
                parent_commitment = commitment;
                commitments.push(commitment);

                // Every cache observation is newer than the floor, so only an exact
                // commitment retirement can remove it.
                setup.extra.proposed(
                    Round::new(Epoch::zero(), View::new(height + 10)),
                    block.clone(),
                );
                assert!(setup.mailbox.verified(round, block).await);
                let finalization = CodingHarness::make_finalization(
                    Proposal {
                        round,
                        parent: View::new(height - 1),
                        payload: commitment,
                    },
                    &schemes,
                    QUORUM,
                );
                if height == 2 {
                    floor = Some(finalization.clone());
                }
                CodingHarness::report_finalization(&mut setup.mailbox, finalization).await;
            }

            while setup.application.pending_ack_heights()
                != vec![Height::new(1), Height::new(2), Height::new(3)]
            {
                context.sleep(Duration::from_millis(10)).await;
            }

            setup
                .mailbox
                .set_floor(floor.expect("height 2 floor missing"));

            // The height-2 floor makes height 1 durable application progress. Heights 2 and 3
            // are re-dispatched, so their coding-buffer commitments remain live.
            while setup.mailbox.get_processed_height().await != Some(Height::new(1)) {
                context.sleep(Duration::from_millis(10)).await;
            }
            assert!(setup.extra.get(commitments[0]).await.is_none());
            assert!(setup.extra.get(commitments[1]).await.is_some());
            assert!(setup.extra.get(commitments[2]).await.is_some());
        });
    }

    #[test_traced("WARN")]
    fn test_coding_notarized_delivery_rejects_dishonest_payload_config() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let provider = ConstantProvider::new(schemes[0].clone());
            let honest_config = coding_config_for_participants(NUM_VALIDATORS as u16);
            let dishonest_config = coding_config_for_participants((NUM_VALIDATORS + 3) as u16);
            assert_ne!(honest_config, dishonest_config);

            let buffer = RecordingCodingBuffer::default();
            let (marshal, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-dishonest-payload-config",
                provider,
                buffer,
            )
            .await;
            let resolver_tx = resolver
                .sender
                .clone()
                .expect("recording resolver should keep its sender");

            let genesis = genesis_block();
            let round = Round::new(Epoch::zero(), View::new(1));
            let height = Height::new(1);
            let candidate_ctx = CodingCtx {
                round,
                leader: participants[0].clone(),
                parent: (View::zero(), genesis_coding_commitment(&genesis)),
            };
            let candidate = make_coding_block(candidate_ctx, genesis.digest(), height, 100);
            let dishonest_block: TestCodedBlock =
                CodedBlock::new(candidate.clone(), dishonest_config, &Sequential);
            let proposal = Proposal {
                round,
                parent: View::zero(),
                payload: dishonest_block.commitment(),
            };
            let notarization = CodingHarness::make_notarization(proposal, &schemes, QUORUM);

            let (response, response_rx) = oneshot::channel();
            assert!(
                resolver_tx
                    .enqueue(handler::Message::Deliver {
                        delivery: Delivery {
                            key: handler::Key::Notarized { round },
                            subscribers: NonEmptyVec::new((
                                handler::Annotation::Notarization { round },
                                tracing::Span::none(),
                            )),
                        },
                        value: (notarization, dishonest_block).encode(),
                        response,
                    })
                    .accepted()
            );
            assert!(
                !response_rx.await.unwrap(),
                "notarized delivery should reject a dishonest coding config"
            );

            context.sleep(Duration::from_millis(100)).await;
            assert!(
                marshal.get_block(height).await.is_none(),
                "dishonest deliveries must not store a finalized block"
            );
            assert!(
                marshal.get_finalization(height).await.is_none(),
                "dishonest deliveries must not archive a finalization"
            );
        });
    }

    #[test_traced("WARN")]
    fn test_coding_block_provider_parent_fetches_by_commitment() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let provider = ConstantProvider::new(schemes[0].clone());
            let buffer = RecordingCodingBuffer::default();
            let (marshal, _resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-provider-parent-commitment",
                provider,
                buffer.clone(),
            )
            .await;

            let (parent_ctx, parent) = missing_candidate(participants[0].clone());
            let child_ctx = CodingCtx {
                round: Round::new(Epoch::zero(), View::new(2)),
                leader: participants[0].clone(),
                parent: (parent_ctx.round.view(), parent.commitment()),
            };
            let child = make_coding_block(child_ctx, parent.digest(), Height::new(2), 200);
            let subscription = marshal.subscribe_parent(&child);

            context.sleep(Duration::from_millis(100)).await;
            assert_eq!(
                buffer.commitment_subscription_count(),
                1,
                "parent walkback should use the coding parent commitment"
            );
            drop(subscription);
        });
    }

    #[test_traced("WARN")]
    fn test_coding_verify_missing_candidate_waits_without_fetching() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let provider = ConstantProvider::new(schemes[0].clone());
            let me = participants[0].clone();
            let buffer = RecordingCodingBuffer::default();
            let (marshal, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-verify-missing-candidate",
                provider.clone(),
                buffer.clone(),
            )
            .await;
            let shards =
                start_shard_mailbox(context.child("shard_stack"), participants, provider.clone())
                    .await;
            let (candidate_ctx, candidate) = missing_candidate(me);
            let commitment = candidate.commitment();

            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new(),
                marshal,
                shards,
                scheme_provider: provider,
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let verify_rx = marshaled.verify(candidate_ctx, commitment).await;
            context.sleep(Duration::from_millis(100)).await;

            assert!(
                buffer.subscription_count() > 0,
                "missing candidate should register a local buffer wait"
            );
            assert!(
                resolver.fetches().is_empty(),
                "missing candidate verify must not fetch from peers"
            );
            assert!(
                resolver.targeted().is_empty(),
                "missing candidate verify must not issue targeted fetches"
            );
            drop(verify_rx);
        });
    }

    /// Crash-recovery shape: after an unclean shutdown, Simplex may recover a
    /// notarized commitment while marshal has no local certification gate task and no
    /// durable block. If enough shards or a peer response can provide the block,
    /// certification should fetch it by notarized round and persist it instead
    /// of treating the missing local copy as a hard failure.
    #[test_traced("WARN")]
    fn test_coding_certify_missing_candidate_fetches_by_round() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let provider = ConstantProvider::new(schemes[0].clone());
            let me = participants[0].clone();
            let buffer = RecordingCodingBuffer::default();
            let (marshal, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-certify-missing-candidate",
                provider.clone(),
                buffer.clone(),
            )
            .await;
            let shards =
                start_shard_mailbox(context.child("shard_stack"), participants, provider.clone())
                    .await;

            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new(),
                marshal,
                shards,
                scheme_provider: provider,
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let (candidate_ctx, candidate) = missing_candidate(me);
            let commitment = candidate.commitment();
            let round = candidate_ctx.round;
            let proposal = Proposal::new(round, View::zero(), commitment);
            let notarization = CodingHarness::make_notarization(proposal, &schemes, QUORUM);
            resolver.respond_to_next_fetch((notarization, candidate).encode());
            let certify_rx = marshaled.certify(round, commitment).await;

            let result = certify_rx.await.expect("certify result missing");
            assert!(result, "fetched notarized candidate should certify");
            assert!(
                resolver.wait_for_delivery_response().await,
                "notarized delivery should validate"
            );
            assert!(
                resolver.fetches().iter().any(|fetch| matches!(
                    (&fetch.key, &fetch.subscriber),
                    (
                        handler::Key::Notarized { round: request_round },
                        handler::Annotation::Notarization { round: subscriber_round },
                    ) if *request_round == round && *subscriber_round == round
                )),
                "certify should fetch notarized block by round"
            );

            assert!(
                buffer.subscription_count() > 0,
                "missing candidate should register a local buffer wait"
            );
            assert!(
                resolver.targeted().is_empty(),
                "missing candidate certify must not issue targeted fetches"
            );
        });
    }

    #[test_traced("WARN")]
    fn test_coding_certify_pending_verify_fetches_by_round() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let provider = ConstantProvider::new(schemes[0].clone());
            let me = participants[0].clone();
            let buffer = RecordingCodingBuffer::default();
            let (marshal, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-certify-pending-verify",
                provider.clone(),
                buffer,
            )
            .await;
            let shards =
                start_shard_mailbox(context.child("shard_stack"), participants, provider.clone())
                    .await;

            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new(),
                marshal,
                shards,
                scheme_provider: provider,
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let (candidate_ctx, candidate) = missing_candidate(me);
            let commitment = candidate.commitment();
            let round = candidate_ctx.round;
            let _verify_rx = marshaled.verify(candidate_ctx, commitment).await;

            let proposal = Proposal::new(round, View::zero(), commitment);
            let notarization = CodingHarness::make_notarization(proposal, &schemes, QUORUM);
            resolver.respond_to_next_fetch((notarization, candidate).encode());
            let certify_rx = marshaled.certify(round, commitment).await;

            let result = certify_rx.await.expect("certify result missing");
            assert!(
                result,
                "pending verify should complete after certification recovery"
            );
            assert!(
                resolver.wait_for_delivery_response().await,
                "notarized delivery should validate"
            );
            assert!(
                resolver.fetches().iter().any(|fetch| matches!(
                    (&fetch.key, &fetch.subscriber),
                    (
                        handler::Key::Notarized { round: request_round },
                        handler::Annotation::Notarization { round: subscriber_round },
                    ) if *request_round == round && *subscriber_round == round
                )),
                "certify should recover a pending verify by notarized round"
            );
            assert!(
                resolver.targeted().is_empty(),
                "certify recovery must not issue targeted fetches"
            );
        });
    }

    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_finalize_good_links() {
        for seed in 0..5 {
            let r1 = harness::finalize::<CodingHarness>(seed, LINK, false);
            let r2 = harness::finalize::<CodingHarness>(seed, LINK, false);
            assert_eq!(r1, r2);
        }
    }

    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_finalize_bad_links() {
        for seed in 0..5 {
            let r1 = harness::finalize::<CodingHarness>(seed, UNRELIABLE_LINK, false);
            let r2 = harness::finalize::<CodingHarness>(seed, UNRELIABLE_LINK, false);
            assert_eq!(r1, r2);
        }
    }

    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_finalize_good_links_quorum_sees_finalization() {
        for seed in 0..5 {
            let r1 = harness::finalize::<CodingHarness>(seed, LINK, true);
            let r2 = harness::finalize::<CodingHarness>(seed, LINK, true);
            assert_eq!(r1, r2);
        }
    }

    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_finalize_bad_links_quorum_sees_finalization() {
        for seed in 0..5 {
            let r1 = harness::finalize::<CodingHarness>(seed, UNRELIABLE_LINK, true);
            let r2 = harness::finalize::<CodingHarness>(seed, UNRELIABLE_LINK, true);
            assert_eq!(r1, r2);
        }
    }

    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_hailstorm_restarts() {
        for seed in 0..2 {
            let r1 = harness::hailstorm::<CodingHarness>(seed, 4, 4, 1, LINK);
            let r2 = harness::hailstorm::<CodingHarness>(seed, 4, 4, 1, LINK);
            assert_eq!(r1, r2);
        }
    }

    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_hailstorm_multi_restarts() {
        for seed in 0..2 {
            let r1 = harness::hailstorm::<CodingHarness>(seed, 4, 4, 2, LINK);
            let r2 = harness::hailstorm::<CodingHarness>(seed, 4, 4, 2, LINK);
            assert_eq!(r1, r2);
        }
    }

    #[test_traced("WARN")]
    fn test_coding_ack_pipeline_backlog() {
        harness::ack_pipeline_backlog::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_ack_pipeline_backlog_persists_on_restart() {
        harness::ack_pipeline_backlog_persists_on_restart::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_genesis_emitted_once() {
        harness::genesis_emitted_once::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_proposed_success_implies_recoverable_after_restart() {
        harness::proposed_success_implies_recoverable_after_restart::<CodingHarness>(0..16);
    }

    #[test_traced("WARN")]
    fn test_coding_verified_success_implies_recoverable_after_restart() {
        harness::verified_success_implies_recoverable_after_restart::<CodingHarness>(0..16);
    }

    #[test_traced("WARN")]
    fn test_coding_certified_success_implies_recoverable_after_restart() {
        harness::certified_success_implies_recoverable_after_restart::<CodingHarness>(0..16);
    }

    #[test_traced("WARN")]
    fn test_coding_delivery_visibility_implies_recoverable_after_restart() {
        harness::delivery_visibility_implies_recoverable_after_restart::<CodingHarness>(0..16);
    }

    #[test_traced("WARN")]
    fn test_coding_sync_height_floor() {
        harness::sync_height_floor::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_prune_finalized_archives() {
        harness::prune_finalized_archives::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_rejects_block_delivery_below_floor() {
        harness::reject_stale_block_delivery_after_floor_update::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_commitment_fetch_height_hint_mismatch_wakes_subscriber() {
        harness::commitment_fetch_height_hint_mismatch_wakes_subscriber::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_subscribe_basic_block_delivery() {
        harness::subscribe_basic_block_delivery::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_subscribe_multiple_subscriptions() {
        harness::subscribe_multiple_subscriptions::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_subscribe_canceled_subscriptions() {
        harness::subscribe_canceled_subscriptions::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_subscribe_blocks_from_different_sources() {
        harness::subscribe_blocks_from_different_sources::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_get_info_basic_queries_present_and_missing() {
        harness::get_info_basic_queries_present_and_missing::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_get_info_latest_progression_multiple_finalizations() {
        harness::get_info_latest_progression_multiple_finalizations::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_get_block_by_height_and_latest() {
        harness::get_block_by_height_and_latest::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_get_block_by_commitment_from_sources_and_missing() {
        harness::get_block_by_commitment_from_sources_and_missing::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_get_finalization_by_height() {
        harness::get_finalization_by_height::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_hint_finalized_triggers_fetch() {
        harness::hint_finalized_triggers_fetch::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_ancestry_stream() {
        harness::ancestry_stream::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_finalize_same_height_different_views() {
        harness::finalize_same_height_different_views::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_certify_persists_equivocated_block() {
        harness::certify_persists_equivocated_block::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_verified_after_restart_reverify_same_round_implies_recoverable() {
        harness::verified_after_restart_reverify_same_round_implies_recoverable::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_certify_after_restart_reverify_same_round_implies_recoverable() {
        harness::certify_after_restart_reverify_same_round_implies_recoverable::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_certify_at_later_view_survives_earlier_view_pruning() {
        harness::certify_at_later_view_survives_earlier_view_pruning::<CodingHarness>();
    }

    #[test_traced("WARN")]
    fn test_coding_certify_first_block_fetches_genesis_parent() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            let round = Round::new(Epoch::zero(), View::new(1));
            let block_ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::zero(), genesis_parent_commitment),
            };
            let block = make_coding_block(block_ctx.clone(), genesis.digest(), Height::new(1), 100);
            let coded_block = CodedBlock::new(block, coding_config, &Sequential);
            let commitment = coded_block.commitment();
            shards.proposed(round, coded_block);

            context.sleep(Duration::from_millis(10)).await;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal,
                shards,
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let shard_validity = marshaled
                .verify(block_ctx, commitment)
                .await
                .await
                .expect("verify result missing");
            assert!(shard_validity, "shard validity should pass");

            let certify_result = marshaled
                .certify(round, commitment)
                .await
                .await
                .expect("certify result missing");
            assert!(
                certify_result,
                "height-1 block should certify with genesis as parent"
            );
        });
    }

    /// Finalizing a descendant must not height-prune the shard-engine buffer before
    /// `try_repair_gaps` has consumed buffer-only ancestors.
    ///
    /// Places parent (height 1) and descendant (height 2) in the shard engine's
    /// reconstructed-block cache via `proposed()`, then reports a finalization
    /// for the descendant only.
    #[test_traced("WARN")]
    fn test_coding_store_finalization_does_not_prune_buffer_before_repair() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let mut handle = harness::ValidatorHandle::<CodingHarness> {
                mailbox: setup.mailbox,
                extra: setup.extra,
            };

            // Build a 2-block chain: parent at height 1, descendant at height 2.
            let parent_block = CodingHarness::make_test_block(
                Sha256::hash(&[b""]),
                CodingHarness::genesis_parent_commitment(NUM_VALIDATORS as u16),
                Height::new(1),
                1,
                NUM_VALIDATORS as u16,
            );
            let parent_digest = CodingHarness::digest(&parent_block);
            let parent_commitment = CodingHarness::commitment(&parent_block);

            let descendant_block = CodingHarness::make_test_block(
                parent_digest,
                parent_commitment,
                Height::new(2),
                2,
                NUM_VALIDATORS as u16,
            );
            let descendant_commitment = CodingHarness::commitment(&descendant_block);

            // Seed the shard engine's reconstructed-block cache with both blocks.
            CodingHarness::propose(
                &mut handle,
                Round::new(Epoch::new(0), View::new(1)),
                &parent_block,
            )
            .await;
            CodingHarness::propose(
                &mut handle,
                Round::new(Epoch::new(0), View::new(2)),
                &descendant_block,
            )
            .await;

            // Report finalization for the descendant only. The parent has no
            // finalization certificate: it must be archived by walking the
            // parent link from the descendant and sourcing the block from the
            // shard-engine buffer.
            let descendant_proposal = Proposal {
                round: Round::new(Epoch::new(0), View::new(2)),
                parent: View::new(1),
                payload: descendant_commitment,
            };
            let descendant_finalization =
                CodingHarness::make_finalization(descendant_proposal, &schemes, QUORUM);
            CodingHarness::report_finalization(&mut handle.mailbox, descendant_finalization).await;

            // Wait until the descendant is archived: that proves finalization processing
            // has completed, at which point the parent must already have been repaired
            // from the shard buffer.
            while handle.mailbox.get_block(Height::new(2)).await.is_none() {
                context.sleep(Duration::from_millis(10)).await;
            }

            let parent = handle.mailbox.get_block(Height::new(1)).await;
            assert!(
                parent.is_some(),
                "parent must be archived from shard buffer before height-prune evicts it"
            );
        });
    }

    #[test_traced("WARN")]
    fn test_coding_init_processed_height() {
        harness::init_processed_height::<CodingHarness>();
    }

    #[test_traced("INFO")]
    fn test_coding_broadcast_caches_block() {
        harness::broadcast_caches_block::<CodingHarness>();
    }

    /// Test that certifying a lower-view block after a higher-view block succeeds.
    ///
    /// This is a critical test for crash recovery scenarios where a validator may need
    /// to certify blocks in non-sequential view order.
    #[test_traced("INFO")]
    fn test_certify_lower_view_after_higher_view() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();

            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            // Create parent block at height 1
            let parent_ctx = CodingCtx {
                round: Round::new(Epoch::new(0), View::new(1)),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(1), 100);
            let parent_digest = parent.digest();
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(Round::new(Epoch::new(0), View::new(1)), coded_parent);

            // Block A at view 5 (height 2) - create with context matching what verify will receive
            let round_a = Round::new(Epoch::new(0), View::new(5));
            let context_a = CodingCtx {
                round: round_a,
                leader: me.clone(),
                parent: (View::new(1), parent_commitment),
            };
            let block_a = make_coding_block(context_a.clone(), parent_digest, Height::new(2), 200);
            let coded_block_a = CodedBlock::new(block_a.clone(), coding_config, &Sequential);
            let commitment_a = coded_block_a.commitment();
            shards.proposed(round_a, coded_block_a);

            // Block B at view 10 (height 2, different block same height - could happen with
            // different proposers or re-proposals)
            let round_b = Round::new(Epoch::new(0), View::new(10));
            let context_b = CodingCtx {
                round: round_b,
                leader: me.clone(),
                parent: (View::new(1), parent_commitment),
            };
            let block_b = make_coding_block(context_b.clone(), parent_digest, Height::new(2), 300);
            let coded_block_b = CodedBlock::new(block_b.clone(), coding_config, &Sequential);
            let commitment_b = coded_block_b.commitment();
            shards.proposed(round_b, coded_block_b);

            context.sleep(Duration::from_millis(10)).await;

            // Step 1: Verify block A at view 5
            let _ = marshaled.verify(context_a, commitment_a).await.await;

            // Step 2: Verify block B at view 10
            let _ = marshaled.verify(context_b, commitment_b).await.await;

            // Step 3: Certify block B at view 10 FIRST
            let certify_b = marshaled.certify(round_b, commitment_b).await;
            assert!(
                certify_b.await.unwrap(),
                "Block B certification should succeed"
            );

            // Step 4: Certify block A at view 5 - should succeed
            let certify_a = marshaled.certify(round_a, commitment_a).await;

            // Use select with timeout to detect never-resolving receiver
            select! {
                result = certify_a => {
                    assert!(result.unwrap(), "Block A certification should succeed");
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("Block A certification timed out");
                },
            }
        })
    }

    /// Regression test for re-proposal validation in optimistic_verify.
    ///
    /// Verifies that:
    /// 1. Valid re-proposals at epoch boundaries are accepted
    /// 2. Invalid re-proposals (not at epoch boundary) are rejected
    ///
    /// A re-proposal occurs when the parent digest equals the block being verified,
    /// meaning the same block is being proposed again in a new view.
    #[test_traced("INFO")]
    fn test_marshaled_reproposal_validation() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            setup_network_links(&mut oracle, &participants[..2], LINK).await;
            let reproposer_control = oracle.control(participants[1].clone());
            let (mut reproposer_sender, _reproposer_receiver) =
                reproposer_control.register(2, TEST_QUOTA).await.unwrap();
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            // Build a chain up to just below the epoch boundary (height 19 is the last
            // block in epoch 0 with BLOCKS_PER_EPOCH=20, since epoch 0 covers heights
            // 0-19), so the boundary block below chains onto height 18.
            let mut parent = genesis.digest();
            let mut last_view = View::zero();
            let mut last_commitment = genesis_commitment();

            // Capture a genuinely-chained non-boundary block (height 10) so Test 2
            // can re-propose one that validates cleanly under its own embedded
            // context. Without this, a certify rejection there could come from an
            // ancestry error rather than the re-proposal boundary gate under test.
            let mut non_boundary = None;
            for i in 1..BLOCKS_PER_EPOCH.get() - 1 {
                let round = Round::new(Epoch::new(0), View::new(i));
                let ctx = CodingCtx {
                    round,
                    leader: me.clone(),
                    parent: (last_view, last_commitment),
                };
                let block = make_coding_block(ctx.clone(), parent, Height::new(i), i * 100);
                let coded_block = CodedBlock::new(block.clone(), coding_config, &Sequential);
                last_commitment = coded_block.commitment();
                if i == 10 {
                    non_boundary = Some((View::new(i), last_commitment));
                }
                shards.proposed(round, coded_block);
                parent = block.digest();
                last_view = View::new(i);
            }
            let (non_boundary_view, non_boundary_commitment) =
                non_boundary.expect("chain includes a non-boundary block");

            // Create the epoch boundary block (height 19, last block in epoch 0)
            let boundary_height = Height::new(BLOCKS_PER_EPOCH.get() - 1);
            let boundary_round = Round::new(Epoch::new(0), View::new(boundary_height.get()));
            let boundary_context = CodingCtx {
                round: boundary_round,
                leader: me.clone(),
                parent: (last_view, last_commitment),
            };
            let boundary_block = make_coding_block(
                boundary_context.clone(),
                parent,
                boundary_height,
                boundary_height.get() * 100,
            );
            let coded_boundary =
                CodedBlock::new(boundary_block.clone(), coding_config, &Sequential);
            let boundary_commitment = coded_boundary.commitment();
            shards.discovered(
                boundary_commitment,
                boundary_context.leader.clone(),
                boundary_round,
            );
            shards.proposed(boundary_round, coded_boundary.clone());

            context.sleep(Duration::from_millis(10)).await;

            // Test 1: Valid re-proposal at epoch boundary should be accepted
            // Re-proposal context: parent digest equals the block being verified
            // Re-proposals happen within the same epoch when the parent is the last block
            //
            // In the coding marshal, verify() returns shard validity while deferred_verify
            // runs in the background. We call verify() to register the certification gate task,
            // then certify() returns the deferred_verify result.
            let reproposal_round = Round::new(Epoch::new(0), View::new(20));
            let reproposal_context = CodingCtx {
                round: reproposal_round,
                leader: participants[1].clone(),
                parent: (View::new(boundary_height.get()), boundary_commitment), // Parent IS the boundary block
            };

            // Call verify to kick off deferred verification.
            // We must await the verify result to ensure the certification gate task is
            // registered before calling certify.
            let shard_validity = marshaled
                .verify(reproposal_context.clone(), boundary_commitment)
                .await
                .await;
            assert!(
                shard_validity.unwrap(),
                "Re-proposal verify should return true for shard validity"
            );

            let assigned = shards.subscribe_assigned_shard_verified(boundary_commitment);
            let assigned_shard = coded_boundary
                .shard(0)
                .expect("missing assigned shard")
                .encode();
            reproposer_sender.send(Recipients::One(me.clone()), assigned_shard, true);
            select! {
                result = assigned => result.expect("assigned shard sender dropped"),
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("assigned shard from reproposer was not accepted");
                },
            }

            // Use certify to get the actual deferred_verify result
            let certify_result = marshaled
                .certify(reproposal_round, boundary_commitment)
                .await
                .await;
            assert!(
                certify_result.unwrap(),
                "Valid re-proposal at epoch boundary should be accepted"
            );

            // The same boundary commitment may be re-proposed again after the
            // first re-proposal certifies. Its parent view advances while its
            // parent payload remains the boundary commitment. Re-proposal
            // validation is intrinsic to the committed block and must return
            // the same result under both headers.
            let repeated_reproposal_round =
                Round::new(Epoch::new(0), View::new(boundary_height.get() + 2));
            let repeated_reproposal_context = CodingCtx {
                round: repeated_reproposal_round,
                leader: me.clone(),
                parent: (reproposal_round.view(), boundary_commitment),
            };
            let repeated_verify = marshaled
                .verify(repeated_reproposal_context, boundary_commitment)
                .await
                .await;
            assert!(
                repeated_verify.unwrap(),
                "Repeated re-proposal should remain valid as the parent view advances"
            );
            let repeated_certify = marshaled
                .certify(repeated_reproposal_round, boundary_commitment)
                .await
                .await;
            assert!(
                repeated_certify.unwrap(),
                "Repeated re-proposal certification should remain valid"
            );

            // Test 2: Invalid re-proposal (not at epoch boundary) should be
            // rejected. Re-propose the genuinely-chained height-10 block at view
            // 15. Because that block validates cleanly under its embedded context,
            // a certify rejection can only come from the re-proposal boundary gate,
            // not from an ancestry error.
            let invalid_reproposal_round = Round::new(Epoch::new(0), View::new(15));
            let invalid_reproposal_context = CodingCtx {
                round: invalid_reproposal_round,
                leader: me.clone(),
                parent: (non_boundary_view, non_boundary_commitment),
            };

            // Call verify to kick off deferred verification.
            // We must await the verify result to ensure the certification gate task is
            // registered before calling certify.
            let shard_validity = marshaled
                .verify(invalid_reproposal_context, non_boundary_commitment)
                .await
                .await;
            assert!(
                !shard_validity.unwrap(),
                "Invalid re-proposal verify should return false"
            );

            // Use certify to get the actual deferred_verify result
            let certify_result = marshaled
                .certify(invalid_reproposal_round, non_boundary_commitment)
                .await
                .await;
            assert!(
                !certify_result.unwrap(),
                "Invalid re-proposal (not at epoch boundary) should be rejected"
            );

            // Test 3: Re-proposal with mismatched epoch should be rejected
            // This is a regression test - re-proposals must be in the same epoch as the block.
            let cross_epoch_reproposal_round = Round::new(Epoch::new(1), View::new(20));
            let cross_epoch_reproposal_context = CodingCtx {
                round: cross_epoch_reproposal_round,
                leader: me.clone(),
                parent: (View::new(boundary_height.get()), boundary_commitment),
            };

            // Call verify to kick off deferred verification.
            // We must await the verify result to ensure the certification gate task is
            // registered before calling certify.
            let shard_validity = marshaled
                .verify(cross_epoch_reproposal_context.clone(), boundary_commitment)
                .await
                .await;
            assert!(
                !shard_validity.unwrap(),
                "Cross-epoch re-proposal verify should return false"
            );

            // Use certify to get the actual deferred_verify result
            let certify_result = marshaled
                .certify(cross_epoch_reproposal_round, boundary_commitment)
                .await
                .await;
            assert!(
                !certify_result.unwrap(),
                "Re-proposal with mismatched epoch should be rejected"
            );

            // Note: Tests for certify-only paths (crash recovery scenarios) are not included here
            // because they require multiple validators to reconstruct blocks from shards. In a
            // single-validator test setup, block reconstruction fails due to insufficient shards.
            // These paths are tested in integration tests with multiple validators.
        })
    }

    /// An invalid re-proposal must not open shard reconstruction for its payload.
    ///
    /// A re-proposal's consensus round is not bound to its payload because the context
    /// digest check is skipped: the block carries the context of its original proposal.
    /// The shard engine resolves the participant set from the announced round's epoch, so
    /// announcing before the boundary check lets a leader bind a commitment to a round it
    /// was never proposed in and misclassify shards from honest peers.
    #[test_traced("INFO")]
    fn test_invalid_reproposal_does_not_open_reconstruction() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            setup_network_links(&mut oracle, &participants[..2], LINK).await;
            let reproposer_control = oracle.control(participants[1].clone());
            let (mut reproposer_sender, _reproposer_receiver) =
                reproposer_control.register(2, TEST_QUOTA).await.unwrap();
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Store a mid-epoch block in core marshal without opening shard reconstruction.
            let original_round = Round::new(Epoch::new(0), View::new(5));
            let original_context = CodingCtx {
                round: original_round,
                leader: participants[1].clone(),
                parent: (View::new(4), genesis_commitment()),
            };
            let block = make_coding_block(
                original_context,
                Sha256::hash(&[b"parent"]),
                Height::new(5),
                500,
            );
            let coded_block: TestCodedBlock = CodedBlock::new(block, coding_config, &Sequential);
            let commitment = coded_block.commitment();
            assert!(marshal.verified(original_round, coded_block.clone()).await);
            assert!(shards.get(commitment).await.is_none());

            // Re-proposing the old block as epoch 1's genesis parent is invalid.
            // The payload is the parent, so `verify` takes the re-proposal path.
            let reproposal_round = Round::new(Epoch::new(1), View::new(1));
            let reproposal_context = CodingCtx {
                round: reproposal_round,
                leader: participants[1].clone(),
                parent: (View::zero(), commitment),
            };
            let assigned = shards.subscribe_assigned_shard_verified(commitment);
            let verdict = marshaled.verify(reproposal_context, commitment).await.await;
            assert!(!verdict.expect("re-proposal verdict missing"));

            // The re-proposer delivers this node's assigned shard. Without reconstruction
            // state it stays buffered, so assigned shard verification cannot complete.
            let assigned_shard = coded_block
                .shard(0)
                .expect("missing assigned shard")
                .encode();
            reproposer_sender.send(Recipients::One(me.clone()), assigned_shard, true);

            select! {
                _ = assigned => {
                    panic!("invalid re-proposal opened reconstruction for its payload");
                },
                _ = context.sleep(Duration::from_secs(1)) => {},
            }
        })
    }

    /// Exact commitment retirement (durable application progress) can occur on
    /// either side of re-proposal verification. The re-proposal must remain live
    /// through core marshal's verified cache, while a post-retirement `discovered`
    /// announcement must recreate shard state needed before notarization.
    #[test_traced("WARN")]
    fn test_coding_reproposal_recreates_shard_state_after_retirement() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            setup_network_links(&mut oracle, &participants[..2], LINK).await;
            let reproposer_control = oracle.control(participants[1].clone());
            let (mut reproposer_sender, _reproposer_receiver) =
                reproposer_control.register(2, TEST_QUOTA).await.unwrap();
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Build the epoch boundary block, store it in core marshal (the durable
            // backstop), and cache it in the shard engine.
            let boundary_height = Height::new(BLOCKS_PER_EPOCH.get() - 1);
            let boundary_round = Round::new(Epoch::new(0), View::new(boundary_height.get()));
            let boundary_context = CodingCtx {
                round: boundary_round,
                leader: me.clone(),
                parent: (View::new(boundary_height.get() - 1), genesis_commitment()),
            };
            let boundary_block = make_coding_block(
                boundary_context.clone(),
                Sha256::hash(&[b"parent"]),
                boundary_height,
                1900,
            );
            let coded_boundary: TestCodedBlock =
                CodedBlock::new(boundary_block, coding_config, &Sequential);
            let boundary_commitment = coded_boundary.commitment();
            assert!(
                marshal
                    .verified(boundary_round, coded_boundary.clone())
                    .await
            );
            shards.discovered(boundary_commitment, me.clone(), boundary_round);
            shards.proposed(boundary_round, coded_boundary.clone());
            context.sleep(Duration::from_millis(10)).await;
            assert!(shards.get(boundary_commitment).await.is_some());

            // Exact retirement with a round floor below the commitment's observed
            // round: only the exact commitment list can retire it.
            shards.retire(core::Retirement {
                round_floor: Round::new(Epoch::zero(), View::new(1)),
                exact_retirements: vec![boundary_commitment],
            });
            context.sleep(Duration::from_millis(10)).await;
            assert!(shards.get(boundary_commitment).await.is_none());

            // Re-propose the boundary block in the same epoch. Verification must
            // fetch the block from the core backstop and re-announce discovery.
            let reproposal_round = Round::new(Epoch::new(0), View::new(20));
            let reproposal_context = CodingCtx {
                round: reproposal_round,
                leader: participants[1].clone(),
                parent: (View::new(boundary_height.get()), boundary_commitment),
            };
            let verdict = marshaled
                .verify(reproposal_context, boundary_commitment)
                .await
                .await;
            assert!(
                verdict.expect("re-proposal verdict missing"),
                "re-proposal should verify from the core backstop after exact retirement"
            );

            // The recreated reconstruction state must accept this node's assigned
            // shard from the re-proposer (not the original leader).
            let assigned = shards.subscribe_assigned_shard_verified(boundary_commitment);
            let assigned_shard = coded_boundary
                .shard(0)
                .expect("missing assigned shard")
                .encode();
            reproposer_sender.send(Recipients::One(me.clone()), assigned_shard, true);
            select! {
                result = assigned => result.expect("assigned shard sender dropped"),
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("assigned shard was not accepted after state recreation");
                },
            }

            // Retiring again after verification removes the newly recreated shard
            // state, but not the block persisted in core marshal. Certification and
            // caller-owned core subscriptions must remain live through that cache.
            shards.retire(core::Retirement {
                round_floor: Round::new(Epoch::zero(), View::new(1)),
                exact_retirements: vec![boundary_commitment],
            });
            while shards.get(boundary_commitment).await.is_some() {
                context.sleep(Duration::from_millis(10)).await;
            }

            let block = marshal
                .subscribe_by_commitment(boundary_commitment, core::CommitmentFallback::Wait)
                .await
                .expect("core block subscription closed after shard retirement");
            assert_eq!(block.commitment(), boundary_commitment);

            let certify = marshaled
                .certify(reproposal_round, boundary_commitment)
                .await
                .await;
            assert!(
                certify.expect("certify result missing"),
                "re-proposal should certify after exact retirement"
            );
        })
    }

    /// A participant that never held the payload cannot wait for shards it
    /// cannot yet classify. Re-proposal verification must acquire the block by
    /// fetching at the parent's certified round and return the boundary verdict.
    #[test_traced("WARN")]
    fn test_coding_reproposal_verify_fetches_block_by_parent_round() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants[..2].iter().cloned(),
            )
            .await;

            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let v0_setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let v1_setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 1),
                &mut oracle,
                participants[1].clone(),
                ConstantProvider::new(schemes[1].clone()),
            )
            .await;
            setup_network_links(&mut oracle, &participants[..2], LINK).await;

            let mut v0_mailbox = v0_setup.mailbox;
            let v1_marshal = v1_setup.mailbox;
            let v1_shards = v1_setup.extra;

            // The boundary block of epoch 0, originally proposed at view 5.
            let original_round = Round::new(Epoch::new(0), View::new(5));
            let original_context = CodingCtx {
                round: original_round,
                leader: participants[0].clone(),
                parent: (View::new(4), genesis_commitment()),
            };
            let block = make_coding_block(
                original_context,
                Sha256::hash(&[b"parent"]),
                Height::new(BLOCKS_PER_EPOCH.get() - 1),
                1900,
            );
            let coded_block: TestCodedBlock = CodedBlock::new(block, coding_config, &Sequential);
            let commitment = coded_block.commitment();

            // Serving a fetch by round requires the block and the round's
            // notarization certificate.
            assert!(
                v0_mailbox
                    .verified(original_round, coded_block.clone())
                    .await
            );
            CodingHarness::report_notarization(
                &mut v0_mailbox,
                CodingHarness::make_notarization(
                    Proposal {
                        round: original_round,
                        parent: View::new(4),
                        payload: commitment,
                    },
                    &schemes,
                    QUORUM,
                ),
            )
            .await;

            // The re-proposal names its own parent, so verification targets the
            // parent round. Validator 1 never held the payload.
            assert!(v1_shards.get(commitment).await.is_none());
            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: v1_marshal.clone(),
                shards: v1_shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[1].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);
            let reproposal_context = CodingCtx {
                round: Round::new(Epoch::new(0), View::new(7)),
                leader: participants[1].clone(),
                parent: (View::new(5), commitment),
            };
            let verdict = marshaled.verify(reproposal_context, commitment).await.await;
            assert!(
                verdict.expect("re-proposal verdict missing"),
                "re-proposal should verify after fetching the block by parent round"
            );
        })
    }

    #[test_traced("WARN")]
    fn test_marshaled_rejects_mismatched_context_digest() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Create parent block at height 1 so the commitment is well-formed.
            let parent_ctx = CodingCtx {
                round: Round::new(Epoch::zero(), View::new(1)),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(1), 100);
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(Round::new(Epoch::zero(), View::new(1)), coded_parent);

            // Build a block with context A (commitment hash uses this context).
            let round_a = Round::new(Epoch::zero(), View::new(2));
            let context_a = CodingCtx {
                round: round_a,
                leader: me.clone(),
                parent: (View::new(1), parent_commitment),
            };
            let block_a = make_coding_block(context_a, parent.digest(), Height::new(2), 200);
            let coded_block_a: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                CodedBlock::new(block_a, coding_config, &Sequential);
            let commitment_a = coded_block_a.commitment();

            // Verify using a different consensus context B (hash mismatch).
            let round_b = Round::new(Epoch::zero(), View::new(3));
            let context_b = CodingCtx {
                round: round_b,
                leader: participants[1].clone(),
                parent: (View::new(1), parent_commitment),
            };

            let verify_rx = marshaled.verify(context_b, commitment_a).await;
            select! {
                result = verify_rx => {
                    assert!(
                        !result.unwrap(),
                        "mismatched context digest should be rejected"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("verify should reject mismatched context digest promptly");
                },
            }
        })
    }

    #[test_traced("WARN")]
    fn test_reproposal_certify_recovers_after_verify_receiver_drop() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            // Build a valid boundary re-proposal, but keep it unavailable until
            // after the optimistic verify receiver has been dropped.
            let boundary_height = Height::new(BLOCKS_PER_EPOCH.get() - 1);
            let boundary_round = Round::new(Epoch::zero(), View::new(boundary_height.get()));
            let boundary_context = CodingCtx {
                round: boundary_round,
                leader: me.clone(),
                parent: (View::zero(), genesis_commitment()),
            };
            let boundary_block = make_coding_block(
                boundary_context,
                genesis.digest(),
                boundary_height,
                boundary_height.get() * 100,
            );
            let coded_boundary = CodedBlock::new(boundary_block, coding_config, &Sequential);
            let boundary_commitment = coded_boundary.commitment();
            let reproposal_round = Round::new(Epoch::zero(), View::new(boundary_height.get() + 1));
            let reproposal_context = CodingCtx {
                round: reproposal_round,
                leader: me,
                parent: (View::new(boundary_height.get()), boundary_commitment),
            };

            // Start verify, then drop the receiver before the block is available.
            let verify_rx = marshaled
                .verify(reproposal_context, boundary_commitment)
                .await;
            drop(verify_rx);
            context.sleep(Duration::from_millis(10)).await;

            shards.proposed(boundary_round, coded_boundary);
            context.sleep(Duration::from_millis(10)).await;

            // Certify should not return the stale closed certification gate task; it
            // should recover through the embedded-context certification path.
            let certify_rx = marshaled
                .certify(reproposal_round, boundary_commitment)
                .await;
            select! {
                result = certify_rx => {
                    assert!(
                        result.expect("certify result missing"),
                        "certify should recover after verify receiver drop"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("certify should recover after verify receiver drop");
                },
            }
        })
    }

    #[test_traced("WARN")]
    fn test_reproposal_missing_block_does_not_synthesize_false() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Re-proposal payload with valid coding config, but no block available.
            let missing_payload = TestCommitment::from((
                Sha256::hash(&[b"missing_block"]),
                Sha256::hash(&[b"missing_root"]),
                Sha256::hash(&[b"missing_context"]),
                coding_config,
            ));
            let round = Round::new(Epoch::zero(), View::new(1));
            let reproposal_context = CodingCtx {
                round,
                leader: me,
                parent: (View::zero(), missing_payload),
            };

            // Verify must not synthesize `false` when the block cannot be fetched.
            let verify_rx = marshaled.verify(reproposal_context, missing_payload).await;

            // Ensure the certification gate task has registered its subscription, then
            // force cancellation by pruning the missing commitment.
            context.sleep(Duration::from_millis(100)).await;
            shards.retire(core::Retirement {
                round_floor: round,
                exact_retirements: vec![missing_payload],
            });

            select! {
                result = verify_rx => {
                    assert!(
                        result.is_err(),
                        "verify should resolve without explicit false when re-proposal block is unavailable"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("verify should resolve promptly when re-proposal block is unavailable");
                },
            }

            // Certify should not surface the closed certification gate task as the final result.
            // With no block available, it remains pending on the recovery path until the
            // certifier's caller times out or data arrives.
            let mut certify_rx = marshaled.certify(round, missing_payload).await;
            context.sleep(Duration::from_millis(100)).await;
            assert!(
                matches!(
                    certify_rx.try_recv(),
                    Err(commonware_utils::channel::oneshot::error::TryRecvError::Empty)
                ),
                "certify should remain pending without explicit false or stale cancellation"
            );
            drop(certify_rx);
        })
    }

    #[test_traced("WARN")]
    fn test_core_subscription_closes_when_coding_buffer_prunes_missing_commitment() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
            let missing_commitment = TestCommitment::from((
                Sha256::hash(&[b"missing_block"]),
                Sha256::hash(&[b"missing_root"]),
                Sha256::hash(&[b"missing_context"]),
                coding_config,
            ));
            let round = Round::new(Epoch::zero(), View::new(1));

            // Subscribe through the core actor. This internally subscribes to the
            // coding shard buffer and registers local waiters.
            let block_rx = marshal.subscribe_by_commitment(
                missing_commitment,
                core::CommitmentFallback::FetchByRound { round },
            );

            // Allow core actor to register the underlying buffer subscription.
            context.sleep(Duration::from_millis(100)).await;

            // Prune the missing commitment in the shard engine, which should cancel
            // the underlying buffer subscription.
            shards.retire(core::Retirement {
                round_floor: round,
                exact_retirements: vec![missing_commitment],
            });

            // The core actor must surface cancellation by closing the subscription,
            // not by panicking or leaving the waiter parked indefinitely.
            select! {
                result = block_rx => {
                    assert!(
                        result.is_err(),
                        "core subscription should close when coding buffer drops subscription"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("core subscription should resolve promptly after coding prune");
                },
            }
        })
    }

    #[test_traced("WARN")]
    fn test_coding_floor_preserves_registered_commitment_subscription() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let missing_round = Round::new(Epoch::zero(), View::new(1));
            let missing = CodingHarness::make_test_block(
                Sha256::hash(&[b""]),
                CodingHarness::genesis_parent_commitment(NUM_VALIDATORS as u16),
                Height::new(1),
                100,
                NUM_VALIDATORS as u16,
            );
            let missing_commitment = missing.commitment();
            shards.discovered(missing_commitment, participants[1].clone(), missing_round);

            let subscription =
                marshal.subscribe_by_commitment(missing_commitment, core::CommitmentFallback::Wait);
            context.sleep(Duration::from_millis(100)).await;

            let floor_round = Round::new(Epoch::zero(), View::new(2));
            let floor = CodingHarness::make_test_block(
                missing.digest(),
                missing_commitment,
                Height::new(2),
                200,
                NUM_VALIDATORS as u16,
            );
            let floor_commitment = floor.commitment();
            shards.proposed(floor_round, floor);
            assert!(shards.get(floor_commitment).await.is_some());
            marshal.set_floor(CodingHarness::make_finalization(
                Proposal {
                    round: floor_round,
                    parent: View::new(1),
                    payload: floor_commitment,
                },
                &schemes,
                QUORUM,
            ));

            while marshal.get_processed_height().await != Some(Height::new(2)) {
                context.sleep(Duration::from_millis(10)).await;
            }

            shards.proposed(Round::new(Epoch::zero(), View::new(3)), missing);
            select! {
                result = subscription => {
                    let block = result.expect("floor update closed commitment subscription");
                    assert_eq!(block.commitment(), missing_commitment);
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("commitment subscription did not resolve after local ingress");
                },
            }
        })
    }

    #[test_traced("WARN")]
    fn test_marshaled_rejects_unsupported_epoch() {
        #[derive(Clone)]
        struct LimitedEpocher {
            inner: FixedEpocher,
            max_epoch: u64,
        }

        impl Epocher for LimitedEpocher {
            fn containing(&self, height: Height) -> Option<crate::types::EpochInfo> {
                let bounds = self.inner.containing(height)?;
                if bounds.epoch().get() > self.max_epoch {
                    None
                } else {
                    Some(bounds)
                }
            }

            fn first(&self, epoch: Epoch) -> Option<Height> {
                if epoch.get() > self.max_epoch {
                    None
                } else {
                    self.inner.first(epoch)
                }
            }

            fn last(&self, epoch: Epoch) -> Option<Height> {
                if epoch.get() > self.max_epoch {
                    None
                } else {
                    self.inner.last(epoch)
                }
            }
        }

        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let limited_epocher = LimitedEpocher {
                inner: FixedEpocher::new(BLOCKS_PER_EPOCH),
                max_epoch: 0,
            };
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: limited_epocher,
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Create a parent block at height 19 (last block in epoch 0, which is supported)
            let parent_ctx = CodingCtx {
                round: Round::new(Epoch::zero(), View::new(19)),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(19), 1000);
            let parent_digest = parent.digest();
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(Round::new(Epoch::zero(), View::new(19)), coded_parent);

            // Create a block at height 20 (first block in epoch 1, which is NOT supported)
            let block_ctx = CodingCtx {
                round: Round::new(Epoch::new(1), View::new(20)),
                leader: default_leader(),
                parent: (View::new(19), parent_commitment),
            };
            let block = make_coding_block(block_ctx, parent_digest, Height::new(20), 2000);
            let coded_block = CodedBlock::new(block.clone(), coding_config, &Sequential);
            let block_commitment = coded_block.commitment();
            shards.proposed(Round::new(Epoch::new(1), View::new(20)), coded_block);

            context.sleep(Duration::from_millis(10)).await;

            // In the coding marshal, verify() returns shard validity while deferred_verify
            // runs in the background. We need to use certify() to get the deferred_verify result.
            let unsupported_round = Round::new(Epoch::new(1), View::new(20));
            let unsupported_context = CodingCtx {
                round: unsupported_round,
                leader: me.clone(),
                parent: (View::new(19), parent_commitment),
            };

            // Call verify to kick off deferred verification
            let _shard_validity = marshaled
                .verify(unsupported_context, block_commitment)
                .await;

            // Use certify to get the actual deferred_verify result
            let certify_result = marshaled
                .certify(unsupported_round, block_commitment)
                .await
                .await;

            assert!(
                !certify_result.unwrap(),
                "Block in unsupported epoch should be rejected"
            );
        })
    }

    #[test_traced("WARN")]
    fn test_marshaled_rejects_invalid_ancestry() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            // Create genesis block
            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            // Wrap with Marshaled verifier
            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Test case 1: Non-contiguous height
            //
            // We need both blocks in the same epoch.
            // With BLOCKS_PER_EPOCH=20: epoch 0 is heights 0-19, epoch 1 is heights 20-39
            //
            // Store honest parent at height 21 (epoch 1)
            let honest_parent_ctx = CodingCtx {
                round: Round::new(Epoch::new(1), View::new(21)),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let honest_parent = make_coding_block(
                honest_parent_ctx,
                genesis.digest(),
                Height::new(BLOCKS_PER_EPOCH.get() + 1),
                1000,
            );
            let parent_digest = honest_parent.digest();
            let coded_parent = CodedBlock::new(honest_parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(Round::new(Epoch::new(1), View::new(21)), coded_parent);

            // Byzantine proposer broadcasts malicious block at height 35
            // The block has the correct context (matching what consensus will provide)
            // but contains invalid content (non-contiguous height: 21 -> 35 instead of 21 -> 22)
            let byzantine_round = Round::new(Epoch::new(1), View::new(35));
            let byzantine_context = CodingCtx {
                round: byzantine_round,
                leader: me.clone(),
                parent: (View::new(21), parent_commitment), // Consensus says parent is at height 21
            };
            let malicious_block = make_coding_block(
                byzantine_context.clone(),
                parent_digest,
                Height::new(BLOCKS_PER_EPOCH.get() + 15), // Byzantine: non-contiguous height
                2000,
            );
            let coded_malicious =
                CodedBlock::new(malicious_block.clone(), coding_config, &Sequential);
            let malicious_commitment = coded_malicious.commitment();
            shards.proposed(byzantine_round, coded_malicious);

            // Small delay to ensure broadcast is processed
            context.sleep(Duration::from_millis(10)).await;

            // Marshaled.verify() kicks off deferred verification in the background.
            // The Marshaled verifier will:
            // 1. Fetch honest_parent (height 21) from marshal based on context.parent
            // 2. Fetch malicious_block (height 35) from marshal based on digest
            // 3. Validate height is contiguous (fail)
            // 4. Return false
            let _shard_validity = marshaled
                .verify(byzantine_context, malicious_commitment)
                .await;

            // Use certify to get the actual deferred_verify result
            let certify_result = marshaled
                .certify(byzantine_round, malicious_commitment)
                .await
                .await;

            assert!(
                !certify_result.unwrap(),
                "Byzantine block with non-contiguous heights should be rejected"
            );

            // Test case 2: Mismatched parent digest
            //
            // Create another malicious block with correct context and height
            // but referencing the wrong parent digest (genesis instead of honest_parent)
            let byzantine_round2 = Round::new(Epoch::new(1), View::new(22));
            let byzantine_context2 = CodingCtx {
                round: byzantine_round2,
                leader: me.clone(),
                parent: (View::new(21), parent_commitment), // Consensus says parent is at height 21
            };
            let malicious_block2 = make_coding_block(
                byzantine_context2.clone(),
                genesis.digest(), // Byzantine: wrong parent digest
                Height::new(BLOCKS_PER_EPOCH.get() + 2),
                3000,
            );
            let coded_malicious2 =
                CodedBlock::new(malicious_block2.clone(), coding_config, &Sequential);
            let malicious_commitment2 = coded_malicious2.commitment();
            shards.proposed(byzantine_round2, coded_malicious2);

            // Small delay to ensure broadcast is processed
            context.sleep(Duration::from_millis(10)).await;

            // Marshaled.verify() kicks off deferred verification in the background.
            // The Marshaled verifier will:
            // 1. Fetch honest_parent (height 21) from marshal based on context.parent
            // 2. Fetch malicious_block (height 22) from marshal based on digest
            // 3. Validate height is contiguous
            // 4. Validate parent commitment matches (fail)
            // 5. Return false
            let _shard_validity = marshaled
                .verify(byzantine_context2, malicious_commitment2)
                .await;

            // Use certify to get the actual deferred_verify result
            let certify_result = marshaled
                .certify(byzantine_round2, malicious_commitment2)
                .await
                .await;

            assert!(
                !certify_result.unwrap(),
                "Byzantine block with mismatched parent commitment should be rejected"
            );
        })
    }

    #[test_traced("WARN")]
    fn test_certify_without_prior_verify_crash_recovery() {
        // After a crash, consensus may call certify() without a prior verify().
        // The certify path should:
        //   1. Find no in-progress certification gate task
        //   2. Subscribe to the block from the shard engine
        //   3. Use the block's embedded context for deferred_verify
        //   4. Return Ok(true) for a valid block
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Create parent at height 1.
            let parent_round = Round::new(Epoch::zero(), View::new(1));
            let parent_ctx = CodingCtx {
                round: parent_round,
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(1), 100);
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(parent_round, coded_parent);

            // Create child at height 2.
            let child_round = Round::new(Epoch::zero(), View::new(2));
            let child_ctx = CodingCtx {
                round: child_round,
                leader: me.clone(),
                parent: (View::new(1), parent_commitment),
            };
            let child = make_coding_block(child_ctx, parent.digest(), Height::new(2), 200);
            let coded_child = CodedBlock::new(child, coding_config, &Sequential);
            let child_commitment = coded_child.commitment();
            shards.proposed(child_round, coded_child);

            context.sleep(Duration::from_millis(10)).await;

            // Call certify directly without any prior verify (simulating crash recovery).
            let certify_rx = marshaled.certify(child_round, child_commitment).await;
            select! {
                result = certify_rx => {
                    assert!(
                        result.unwrap(),
                        "certify without prior verify should succeed for valid block"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("certify should complete within timeout");
                },
            }
        })
    }

    /// Without a registered gate, certify runs the application against the
    /// block's embedded context. A live rejection there must reach consensus.
    #[test_traced("WARN")]
    fn test_certify_without_prior_verify_honors_application_rejection() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let mock_app: MockVerifyingApp<CodingB, S> =
                MockVerifyingApp::with_verify_result(false);
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Create parent at height 1.
            let parent_round = Round::new(Epoch::zero(), View::new(1));
            let parent_ctx = CodingCtx {
                round: parent_round,
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(1), 100);
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(parent_round, coded_parent);

            // Create child at height 2.
            let child_round = Round::new(Epoch::zero(), View::new(2));
            let child_ctx = CodingCtx {
                round: child_round,
                leader: me.clone(),
                parent: (View::new(1), parent_commitment),
            };
            let child = make_coding_block(child_ctx, parent.digest(), Height::new(2), 200);
            let coded_child = CodedBlock::new(child, coding_config, &Sequential);
            let child_commitment = coded_child.commitment();
            shards.proposed(child_round, coded_child);

            context.sleep(Duration::from_millis(10)).await;

            // No prior verify, so no gate exists and certify falls through to
            // the embedded-context path.
            let certify_rx = marshaled.certify(child_round, child_commitment).await;
            select! {
                result = certify_rx => {
                    assert!(
                        !result.expect("certify result missing"),
                        "certify must propagate the application rejection"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("certify should resolve promptly");
                },
            }
        })
    }

    /// A Byzantine leader can deliver assigned shards to only f+1 honest
    /// validators and form a notarization with its own vote, leaving no peer
    /// able to serve a full-block fetch. A validator that never saw the
    /// proposal then holds only buffered sender-indexed gossip shards, and
    /// draining them requires the reconstruction interest that `certify`
    /// registers with the shard engine (`shards.notarized`).
    #[test_traced("WARN")]
    fn test_certify_reconstructs_from_buffered_gossip_shards() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            // Register the peers' shard-channel senders and link the network.
            // No peer runs a marshal actor, so round-bound block fetches go
            // unanswered.
            let mut peer_senders = Vec::new();
            for peer in participants.iter().skip(1) {
                let (sender, _receiver) = oracle
                    .control(peer.clone())
                    .register(2, TEST_QUOTA)
                    .await
                    .unwrap();
                peer_senders.push(sender);
            }
            setup_network_links(&mut oracle, &participants, LINK).await;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // The parent is available locally. The notarized candidate is not:
            // this validator never saw its proposal.
            let parent_round = Round::new(Epoch::zero(), View::new(1));
            let parent_ctx = CodingCtx {
                round: parent_round,
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(1), 100);
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(parent_round, coded_parent);

            let child_round = Round::new(Epoch::zero(), View::new(2));
            let child_ctx = CodingCtx {
                round: child_round,
                leader: participants[1].clone(),
                parent: (View::new(1), parent_commitment),
            };
            let child = make_coding_block(child_ctx, parent.digest(), Height::new(2), 200);
            let coded_child: TestCodedBlock = CodedBlock::new(child, coding_config, &Sequential);
            let child_commitment = coded_child.commitment();

            // Deliver each peer's sender-indexed gossip shard. Without leader
            // discovery or reconstruction interest, the shard engine only
            // buffers them.
            for (i, sender) in peer_senders.iter_mut().enumerate() {
                let index = u16::try_from(i + 1).expect("peer index fits in u16");
                let shard = coded_child.shard(index).expect("missing shard").encode();
                sender.send(Recipients::One(me.clone()), shard, true);
            }
            context.sleep(Duration::from_millis(250)).await;

            // Certify must register reconstruction interest with the shard
            // engine, drain the buffered shards, and verify the reconstructed
            // block through its embedded context.
            let certify_rx = marshaled.certify(child_round, child_commitment).await;
            select! {
                result = certify_rx => {
                    assert!(
                        result.expect("certify result missing"),
                        "certify must reconstruct the candidate from buffered gossip shards"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("certify did not reconstruct from buffered gossip shards");
                },
            }
        })
    }

    /// Regression test: a Byzantine leader must not be able to crash honest nodes
    /// by proposing a `Commitment` with invalid `CodingConfig` bytes (e.g.
    /// zero-valued `NonZeroU16` fields). The fix validates the embedded config
    /// during deserialization so malformed commitments are rejected at the codec
    /// level before reaching `verify()`.
    #[test_traced("WARN")]
    fn test_malformed_commitment_config_rejected_at_deserialization() {
        use commonware_codec::{Encode, ReadExt};

        // Construct a Commitment with all-zero bytes (invalid CodingConfig:
        // minimum_shards=0, extra_shards=0). Serialize it and attempt to
        // deserialize -- this must fail.
        let malformed_bytes = [0u8; <TestCommitment as FixedSize>::SIZE];
        let result = TestCommitment::read(&mut &malformed_bytes[..]);
        assert!(
            result.is_err(),
            "deserialization of Commitment with zeroed CodingConfig must fail"
        );

        // A validly-constructed Commitment must still round-trip.
        let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
        let valid = TestCommitment::from((
            Sha256::hash(&[b"block"]),
            Sha256::hash(&[b"root"]),
            Sha256::hash(&[b"context"]),
            coding_config,
        ));
        let encoded = valid.encode();
        let decoded =
            TestCommitment::read(&mut &encoded[..]).expect("valid Commitment must deserialize");
        assert_eq!(valid, decoded);
    }

    #[test_traced("WARN")]
    fn test_certify_propagates_application_verify_failure() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            // 1) Set up a single validator marshal stack.
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            // 2) Force application verification to fail in deferred verification.
            let mock_app: MockVerifyingApp<CodingB, S> =
                MockVerifyingApp::with_verify_result(false);

            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let parent_round = Round::new(Epoch::zero(), View::new(1));
            let parent_context = CodingCtx {
                round: parent_round,
                leader: me.clone(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_context, genesis.digest(), Height::new(1), 100);
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(parent_round, coded_parent);

            // 3) Publish a valid child so optimistic verify can succeed.
            let round = Round::new(Epoch::zero(), View::new(2));
            let verify_context = CodingCtx {
                round,
                leader: me,
                parent: (View::new(1), parent_commitment),
            };
            let block =
                make_coding_block(verify_context.clone(), parent.digest(), Height::new(2), 200);
            let coded_block = CodedBlock::new(block, coding_config, &Sequential);
            let commitment = coded_block.commitment();
            shards.proposed(round, coded_block);

            context.sleep(Duration::from_millis(10)).await;

            let optimistic = marshaled.verify(verify_context, commitment).await;
            assert!(
                optimistic.await.expect("verify result missing"),
                "optimistic verify should pass pre-checks and schedule deferred verification"
            );

            // 4) Certify must observe the deferred application failure and return false.
            let certify = marshaled.certify(round, commitment).await;
            assert!(
                !certify.await.expect("certify result missing"),
                "certify should propagate deferred application verify failure"
            );
        })
    }

    /// Coding analog of the standard equivocation tests. The commitment binds
    /// the proposal context, so a proposal that names a different parent for
    /// the same commitment fails `validate_proposal` before any certification
    /// gate is registered. Certification of the notarized commitment then
    /// recovers through the block's embedded context.
    #[test_traced("WARN")]
    fn test_certify_not_poisoned_by_equivocating_parent_verify() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // The view-1 block: the parent this validator last certified.
            let certified_round = Round::new(Epoch::zero(), View::new(1));
            let certified_ctx = CodingCtx {
                round: certified_round,
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let certified = make_coding_block(certified_ctx, genesis.digest(), Height::new(1), 100);
            let certified_digest = certified.digest();
            let coded_certified = CodedBlock::new(certified, coding_config, &Sequential);
            let certified_commitment = coded_certified.commitment();
            shards.proposed(certified_round, coded_certified);

            // The view-2 block: notarized by the network but only nullified
            // here, so this validator never certified it.
            let notarized_round = Round::new(Epoch::zero(), View::new(2));
            let notarized_ctx = CodingCtx {
                round: notarized_round,
                leader: me.clone(),
                parent: (View::new(1), certified_commitment),
            };
            let notarized = make_coding_block(notarized_ctx, certified_digest, Height::new(2), 200);
            let notarized_digest = notarized.digest();
            let coded_notarized = CodedBlock::new(notarized, coding_config, &Sequential);
            let notarized_commitment = coded_notarized.commitment();
            shards.proposed(notarized_round, coded_notarized);

            // The view-3 block builds on the view-2 block.
            let round = Round::new(Epoch::zero(), View::new(3));
            let block_ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::new(2), notarized_commitment),
            };
            let block = make_coding_block(block_ctx, notarized_digest, Height::new(3), 300);
            let coded_block = CodedBlock::new(block, coding_config, &Sequential);
            let commitment = coded_block.commitment();
            shards.proposed(round, coded_block);

            context.sleep(Duration::from_millis(10)).await;

            // The equivocating proposal names the certified view-1 block as
            // parent, which this validator's parent selection accepts
            // (view 1 certified, view 2 nullified). The commitment's context
            // digest exposes the mismatch, so verify rejects it without
            // registering a certification gate.
            let equivocating_ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::new(1), certified_commitment),
            };
            let verify_rx = marshaled.verify(equivocating_ctx, commitment).await;
            select! {
                result = verify_rx => {
                    assert!(
                        !result.expect("verify result missing"),
                        "the equivocating proposal must not be notarized"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("verify should reject the equivocating proposal promptly");
                },
            }

            // The honest notarization for the same `(round, commitment)`
            // arrives. Certification recovers through the embedded context.
            let certify_rx = marshaled.certify(round, commitment).await;
            select! {
                result = certify_rx => {
                    assert!(
                        result.expect("certify result missing"),
                        "certify of the notarized commitment must succeed via the embedded context"
                    );
                },
                _ = context.sleep(Duration::from_secs(5)) => {
                    panic!("certify should resolve promptly");
                },
            }
        })
    }

    #[test_traced("WARN")]
    fn test_backfill_block_mismatched_commitment() {
        // Regression: when backfilling by Key::Block(commitment), a peer may return
        // a coded block with matching inner digest but a different coding commitment.
        // If a finalization for this digest is already cached, marshal must reject
        // the block unless V::commitment(block) matches the finalization payload.
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants[..2].iter().cloned(),
            )
            .await;

            let coding_config_a = coding_config_for_participants(NUM_VALIDATORS as u16);
            // Same total shards (4) but different min/extra split produces a different
            // coding root and config bytes, yielding a different commitment.
            let coding_config_b = commonware_coding::Config {
                minimum_shards: coding_config_a.minimum_shards.checked_add(1).unwrap(),
                extra_shards: NZU16!(coding_config_a.extra_shards.get() - 1),
            };

            let v0_setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                participants[0].clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let v1_setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 1),
                &mut oracle,
                participants[1].clone(),
                ConstantProvider::new(schemes[1].clone()),
            )
            .await;

            setup_network_links(&mut oracle, &participants[..2], LINK).await;

            let mut v0_mailbox = v0_setup.mailbox;
            let v1_mailbox = v1_setup.mailbox;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            let round1 = Round::new(Epoch::zero(), View::new(1));
            let block1_ctx = CodingCtx {
                round: round1,
                leader: participants[0].clone(),
                parent: (View::zero(), genesis_commitment()),
            };
            let block1 = make_coding_block(block1_ctx, genesis.digest(), Height::new(1), 100);

            let coded_block_a: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                CodedBlock::new(block1.clone(), coding_config_a, &Sequential);
            let commitment_a = coded_block_a.commitment();

            let coded_block_b: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                CodedBlock::new(block1.clone(), coding_config_b, &Sequential);
            let commitment_b = coded_block_b.commitment();

            assert_eq!(coded_block_a.digest(), coded_block_b.digest());
            assert_ne!(commitment_a, commitment_b);

            // Validator 1 proposes coded_block_b (same inner block, different coding).
            // This stores it in v1's shard engine and actor cache.
            assert!(v1_mailbox.verified(round1, coded_block_b.clone()).await);
            context.sleep(Duration::from_millis(100)).await;

            // Create finalization referencing commitment_a (the "correct" commitment).
            let proposal: Proposal<TestCommitment> = Proposal {
                round: round1,
                parent: View::zero(),
                payload: commitment_a,
            };
            let finalization = CodingHarness::make_finalization(proposal.clone(), &schemes, QUORUM);

            // Report finalization to v0. v0 doesn't have the block:
            //   - it fetches Key::Block(commitment)
            //   - v1 responds with coded_block_b (same digest, wrong commitment)
            //   - deliver path must reject because the response commitment does not
            //     match the request key
            CodingHarness::report_finalization(&mut v0_mailbox, finalization).await;

            // Wait for the fetch cycle to complete.
            context.sleep(Duration::from_secs(5)).await;

            // The mismatched block must not be stored.
            let stored = v0_mailbox.get_block(Height::new(1)).await;
            assert!(
                stored.is_none(),
                "v0 should reject backfilled block with mismatched commitment"
            );

            // Without the block, finalization should not be persisted by height yet.
            let stored_finalization = v0_mailbox.get_finalization(Height::new(1)).await;
            assert!(
                stored_finalization.is_none(),
                "finalization should not be archived until matching block is available"
            );
        })
    }

    #[test_traced("WARN")]
    #[should_panic(expected = "floor block parent commitment mismatch")]
    fn test_coding_floor_anchor_panics_on_parent_commitment_mismatch() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let (mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "floor-parent-commitment-mismatch",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;

            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
            let parent_round = Round::new(Epoch::zero(), View::new(1));
            let parent_context = CodingCtx {
                round: parent_round,
                leader: participants[0].clone(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent =
                make_coding_block(parent_context, Sha256::hash(&[b""]), Height::new(1), 100);

            let floor_round = Round::new(Epoch::zero(), View::new(2));
            let bad_context = CodingCtx {
                round: floor_round,
                leader: participants[0].clone(),
                parent: (View::new(1), genesis_commitment()),
            };
            let floor_block = make_coding_block(bad_context, parent.digest(), Height::new(2), 200);
            let coded_floor = CodedBlock::new(floor_block, coding_config, &Sequential);
            assert_ne!(coded_floor.parent(), coded_floor.context().parent.1.block());

            let finalization = CodingHarness::make_finalization(
                Proposal::new(
                    floor_round,
                    View::new(1),
                    CodingHarness::commitment(&coded_floor),
                ),
                &schemes,
                QUORUM,
            );
            resolver.respond_to_next_fetch(coded_floor.encode());
            mailbox.set_floor(finalization);
            context.sleep(Duration::from_secs(5)).await;
        })
    }

    /// When the scheme provider has no entry for the current epoch,
    /// `Marshaled::propose` and `Marshaled::verify` must return a dropped
    /// receiver (the consensus engine treats `RecvError` as "abstain").
    #[test_traced("WARN")]
    fn test_marshaled_missing_scheme_skips_propose_and_verify() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();

            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: setup.mailbox,
                shards: setup.extra,
                scheme_provider: EmptyProvider,
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let ctx = CodingCtx {
                round: Round::new(Epoch::zero(), View::new(1)),
                leader: me.clone(),
                parent: (View::zero(), genesis_commitment()),
            };

            // propose with a missing scheme returns a dropped sender
            let rx = marshaled.propose(ctx.clone()).await;
            assert!(rx.await.is_err());

            // verify with a missing scheme returns a dropped sender
            let rx = marshaled.verify(ctx, genesis_commitment()).await;
            assert!(rx.await.is_err());
        });
    }

    /// Regression: a validator must not vote finalize on a block that is not
    /// durably persisted. If `certify` resolves true, the block is on disk for
    /// this validator. We assert this by aborting the marshal actor the
    /// instant `certify` returns true, then restarting from the same partition.
    #[test_traced("WARN")]
    fn test_marshaled_certify_persists_block_before_resolving() {
        for seed in 0u64..16 {
            certify_persists_block_before_resolving_at(seed);
        }
    }

    fn certify_persists_block_before_resolving_at(seed: u64) {
        let runner = deterministic::Runner::new(
            deterministic::Config::new()
                .with_seed(seed)
                .with_timeout(Some(Duration::from_secs(60))),
        );
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;
            let marshal_actor_handle = setup.actor_handle;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);

            // Push parent (height 1) and child (height 2) into the shards
            // engine. These are reconstructable but NOT durably persisted.
            let parent_round = Round::new(Epoch::zero(), View::new(1));
            let parent_ctx = CodingCtx {
                round: parent_round,
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let parent = make_coding_block(parent_ctx, genesis.digest(), Height::new(1), 100);
            let coded_parent = CodedBlock::new(parent.clone(), coding_config, &Sequential);
            let parent_commitment = coded_parent.commitment();
            shards.proposed(parent_round, coded_parent);

            let child_round = Round::new(Epoch::zero(), View::new(2));
            let child_ctx = CodingCtx {
                round: child_round,
                leader: me.clone(),
                parent: (View::new(1), parent_commitment),
            };
            let child = make_coding_block(child_ctx.clone(), parent.digest(), Height::new(2), 200);
            let coded_child = CodedBlock::new(child.clone(), coding_config, &Sequential);
            let child_commitment = coded_child.commitment();
            let child_digest = coded_child.digest();
            shards.proposed(child_round, coded_child);

            context.sleep(Duration::from_millis(10)).await;

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Optimistic verify - returns shard validity (true).
            let shard_validity = marshaled
                .verify(child_ctx, child_commitment)
                .await
                .await
                .expect("verify result missing");
            assert!(shard_validity, "shard validity should pass");

            // Certify - this is the safety gate before finalize voting.
            let certify_result = marshaled
                .certify(child_round, child_commitment)
                .await
                .await
                .expect("certify result missing");
            assert!(certify_result, "certify should succeed");

            // Abort marshal immediately after certify returns to prove the
            // block is already persisted at that point.
            marshal_actor_handle.abort();
            drop(marshaled);
            drop(marshal);
            drop(shards);

            // Restart from the same partition. The block must be durably
            // persisted - otherwise the validator would have voted finalize
            // for a block it cannot serve from local storage.
            let setup2 = CodingHarness::setup_validator(
                context
                    .child("validator_restart")
                    .with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal2 = setup2.mailbox;

            let post_restart = marshal2.get_block(&child_digest).await;
            assert!(
                post_restart.is_some(),
                "certify resolved true, so block must be durably persisted"
            );
        });
    }

    /// Regression: a leader must be able to recover its own block across an unclean restart.
    /// `propose` registers a certification gate, so the leader establishes durability by
    /// certifying its own proposal. After certify, the block must survive restart even if
    /// `Relay::broadcast` never runs. This is the >= f+1 guarantee for the leader's own block.
    #[test_traced("WARN")]
    fn test_marshaled_proposed_block_persists_across_restart() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;
            let marshal_actor_handle = setup.actor_handle;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            // Build the block we want propose() to return. Its embedded context
            // uses the proper genesis commitment so the parent lookup matches
            // the cached genesis.
            let propose_round = Round::new(Epoch::zero(), View::new(1));
            let propose_context = CodingCtx {
                round: propose_round,
                leader: me.clone(),
                parent: (View::zero(), genesis_parent_commitment),
            };
            let block_to_propose = make_coding_block(
                propose_context.clone(),
                genesis.digest(),
                Height::new(1),
                100,
            );
            let block_digest = block_to_propose.digest();
            let expected_commitment = CodedBlock::<_, ReedSolomon<Sha256>, Sha256>::new(
                block_to_propose.clone(),
                coding_config,
                &Sequential,
            )
            .commitment();

            let mock_app: MockVerifyingApp<CodingB, S> =
                MockVerifyingApp::new().with_propose_result(block_to_propose);
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // Drive the leader-side propose path. `propose` stages the block
            // and returns the commitment. Durability is established by the
            // certify flush below.
            let commitment = marshaled
                .propose(propose_context)
                .await
                .await
                .expect("propose should produce a commitment");
            assert_eq!(commitment, expected_commitment);

            // The leader certifies its own proposal, which awaits the deferred propose
            // sync handle and establishes durability before the finalize vote.
            assert!(
                marshaled
                    .certify(propose_round, commitment)
                    .await
                    .await
                    .expect("certify result missing"),
                "certify must succeed for the leader's own proposal"
            );

            // Abort marshal after certify; the leader's own block must be durable.
            marshal_actor_handle.abort();
            drop(marshaled);
            drop(marshal);
            drop(shards);

            let setup2 = CodingHarness::setup_validator(
                context
                    .child("validator_restart")
                    .with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal2 = setup2.mailbox;

            // The proposer must recover its own block after restart. Without
            // the broadcast-path persistence fix, the block lived only in the
            // shards engine's in-memory cache and is now gone.
            let post_restart = marshal2.get_block(&block_digest).await;
            assert!(
                post_restart.is_some(),
                "proposer should recover its own block after restart"
            );
        });
    }

    /// A propose relay with a staged proposal must send it through the shard
    /// engine and complete the durability handshake. The freshly built block
    /// is nowhere persisted at broadcast time, so the forward fallback has
    /// nothing to serve: only the staged-hit path can seed the shard engine.
    #[test_traced("WARN")]
    fn test_marshaled_propose_relay_sends_staged_block() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            let propose_round = Round::new(Epoch::zero(), View::new(1));
            let propose_context = CodingCtx {
                round: propose_round,
                leader: me.clone(),
                parent: (View::zero(), genesis_parent_commitment),
            };
            let block_to_propose = make_coding_block(
                propose_context.clone(),
                genesis.digest(),
                Height::new(1),
                100,
            );
            let block_digest = block_to_propose.digest();
            let expected_commitment = CodedBlock::<_, ReedSolomon<Sha256>, Sha256>::new(
                block_to_propose.clone(),
                coding_config,
                &Sequential,
            )
            .commitment();

            let mock_app: MockVerifyingApp<CodingB, S> =
                MockVerifyingApp::new().with_propose_result(block_to_propose);
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let commitment = marshaled
                .propose(propose_context)
                .await
                .await
                .expect("propose should produce a commitment");
            assert_eq!(commitment, expected_commitment);

            // The relay must take the staged proposal and broadcast its
            // shards, seeding the shard engine's local cache.
            let subscription = shards.subscribe(commitment);
            let _ = marshaled.broadcast(
                commitment,
                Plan::Propose {
                    round: propose_round,
                },
            );
            let cached = subscription
                .await
                .expect("shard engine must cache the relayed proposal");
            assert_eq!(cached.digest(), block_digest);

            // The relayed proposal is persisted through the staged ack, so
            // certification resolves durably without a flush.
            assert!(
                marshaled
                    .certify(propose_round, commitment)
                    .await
                    .await
                    .expect("certify result missing"),
                "certify must succeed for the relayed proposal"
            );
        });
    }

    /// Regression: if marshal already holds a verified block for a round
    /// (say, persisted by a pre-crash propose whose notarize vote never
    /// reached the journal), a restarted leader's `propose` must return
    /// that block's commitment instead of rebuilding. The pre-crash
    /// commitment may already have been broadcast, so proposing a rebuilt
    /// block for the same round would equivocate. The recovered proposal
    /// must also be staged for the relay, so the broadcast re-sends its
    /// shards and certification resolves through the deduplicated
    /// re-persist.
    #[test_traced("WARN")]
    fn test_propose_reuses_verified_block_on_restart() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            let round = Round::new(Epoch::zero(), View::new(1));
            let ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::zero(), genesis_parent_commitment),
            };

            // Seed block A in marshal's verified cache for `round`.
            let block_a = make_coding_block(ctx.clone(), genesis.digest(), Height::new(1), 100);
            let coded_a: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                CodedBlock::new(block_a.clone(), coding_config, &Sequential);
            let commitment_a = coded_a.commitment();
            assert!(marshal.verified(round, coded_a).await);

            // The app cannot build (`propose` returns None) and its
            // verification never completes, so the assertions below hold
            // only if the stored block is reused as-is and certification
            // resolves through the durability gate registered by the
            // recovery staging.
            let (mock_app, verify_started, _release_verify): (GatedVerifyingApp<CodingB, S>, _, _) =
                GatedVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let commitment = marshaled
                .propose(ctx)
                .await
                .await
                .expect("propose must return a commitment");
            assert_eq!(
                commitment, commitment_a,
                "propose must reuse the block marshal already persisted for this round"
            );

            // The relay broadcast must find the recovered proposal staged and
            // re-persist it (a dedup no-op whose handle covers the pre-crash
            // write), resolving the certification gate registered by the
            // recovery path.
            let _ = marshaled.broadcast(commitment, Plan::Propose { round });
            let certify_rx = marshaled.certify(round, commitment).await;
            select! {
                result = certify_rx => {
                    assert!(
                        result.expect("certify result missing"),
                        "recovered proposal must certify through the relay handshake"
                    );
                },
                _ = verify_started => {
                    panic!("certifying a recovered proposal must not run app verification");
                },
            }
        });
    }

    /// Regression: a boundary re-proposal stores the parent block itself at
    /// the re-proposal round, under the parent's original embedded context.
    /// A leader that crashes after that relay broadcast must recognize the
    /// cached parent as the re-proposal on restart and propose it again,
    /// rather than skipping the round because its embedded context names an
    /// older round.
    #[test_traced("WARN")]
    fn test_propose_reuses_reproposed_boundary_block_on_restart() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            // Seed the boundary block at the re-proposal round, where the
            // pre-crash relay broadcast of the re-proposal persisted it.
            let boundary_height = Height::new(BLOCKS_PER_EPOCH.get() - 1);
            let boundary_round = Round::new(Epoch::zero(), View::new(boundary_height.get()));
            let boundary_ctx = CodingCtx {
                round: boundary_round,
                leader: default_leader(),
                parent: (View::zero(), genesis_parent_commitment),
            };
            let boundary_block =
                make_coding_block(boundary_ctx, genesis.digest(), boundary_height, 1900);
            let coded_boundary: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                CodedBlock::new(boundary_block, coding_config, &Sequential);
            let boundary_commitment = coded_boundary.commitment();
            let round = Round::new(Epoch::zero(), View::new(boundary_height.get() + 1));
            assert!(marshal.verified(round, coded_boundary).await);

            let ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::new(boundary_height.get()), boundary_commitment),
            };

            // The app cannot build and its verification never completes, so
            // the assertions below hold only if the cached parent is
            // re-proposed as-is.
            let (mock_app, verify_started, _release_verify): (GatedVerifyingApp<CodingB, S>, _, _) =
                GatedVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let commitment = marshaled
                .propose(ctx)
                .await
                .await
                .expect("propose must return a commitment");
            assert_eq!(
                commitment, boundary_commitment,
                "propose must re-propose the boundary block marshal already persisted for this round"
            );

            let _ = marshaled.broadcast(commitment, Plan::Propose { round });
            let certify_rx = marshaled.certify(round, commitment).await;
            select! {
                result = certify_rx => {
                    assert!(
                        result.expect("certify result missing"),
                        "re-proposed boundary block must certify through the relay handshake"
                    );
                },
                _ = verify_started => {
                    panic!("certifying a re-proposed boundary block must not run app verification");
                },
            }
        });
    }

    /// Regression: if a pre-crash leader persisted a verified block for a
    /// round but the simplex `Notarize` never reached the journal, replay
    /// can recover a `consensus_context` whose parent differs from the one
    /// the cached block was built against. The restarted leader must then
    /// drop the receiver so the voter nullifies the view via
    /// `MissingProposal`, rather than broadcasting the stale cached block
    /// under a header that peers will reject.
    #[test_traced("WARN")]
    fn test_propose_skips_when_verified_block_context_changed() {
        let runner = deterministic::Runner::timed(Duration::from_secs(60));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                schemes,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let mut oracle = setup_network_with_participants(
                context.child("network"),
                NZUsize!(1),
                participants.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);

            let setup = CodingHarness::setup_validator(
                context.child("validator").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            let genesis_ctx = CodingCtx {
                round: Round::zero(),
                leader: default_leader(),
                parent: (View::zero(), genesis_commitment()),
            };
            let genesis = make_coding_block(genesis_ctx, Sha256::hash(&[b""]), Height::zero(), 0);
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            // Stash a stale block built against genesis as its parent at round V=2.
            let round = Round::new(Epoch::zero(), View::new(2));
            let stale_ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::zero(), genesis_parent_commitment),
            };
            let stale_block = make_coding_block(stale_ctx, genesis.digest(), Height::new(1), 100);
            let stale_coded: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                CodedBlock::new(stale_block, coding_config, &Sequential);
            assert!(marshal.verified(round, stale_coded).await);

            // Simulate a replay where parent selection now points to a
            // different parent commitment than the cached block was built for.
            let new_parent_commitment = TestCommitment::from((
                Sha256::hash(&[b"different-parent-block"]),
                Sha256::hash(&[b"different-parent-inner"]),
                Sha256::hash(&[b"different-parent-ctx"]),
                coding_config,
            ));
            let new_ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::new(1), new_parent_commitment),
            };

            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new();
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: shards.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            let commitment_rx = marshaled.propose(new_ctx).await;
            assert!(
                commitment_rx.await.is_err(),
                "propose must drop the receiver when the cached block's context no longer matches"
            );
        });
    }
}