//! 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. Its commitment records are bounded by a window that evicts records with the lowest
//! rounds first.
//!
//! # Message Sizes
//!
//! ```text
//! Backfill block, in a block or notarized backfill response:
//!
//! +---------------------------------------------------------+-------------------+
//! | application block (up to the application block bound)   | coding config (4) |
//! +---------------------------------------------------------+-------------------+
//! ```
//!
//! Backfill fetches whole blocks while the buffer sends shards, so the network that carries
//! backfill must admit a complete block in one response.
//!
//! # 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, Handoff, Heightable, Relay,
        Reporter,
        marshal::{
            ancestry::{Ancestry, BlockProvider},
            coding::{
                Coding, Marshaled, MarshaledConfig, shards,
                types::{
                    CodedBlock, Shard, StoredCodedBlock, coding_config_for_participants,
                    hash_context,
                },
            },
            config::{Config, Start},
            core::{self, Processed, durability::Durable as _},
            mocks::{
                application::Application,
                block::encode_block,
                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,
                },
                store::{Op, Recording},
                verifying::{GatedVerifyingApp, MockVerifyingApp},
            },
            resolver::handler,
        },
        simplex::{
            Plan,
            scheme::bls12381_threshold::vrf as bls12381_threshold_vrf,
            types::{Activity, Finalization, Proposal},
        },
        types::{Epoch, Epocher, FixedEpocher, Height, Round, View, ViewDelta, coding::Commitment},
    };
    use bytes::Bytes;
    use commonware_actor::{Feedback, mailbox};
    use commonware_codec::{DecodeExt as _, Encode, FixedSize};
    use commonware_coding::{Config as CodingConfig, ReedSolomon, Scheme as _};
    use commonware_cryptography::{
        Committable, Digested, Digestible, Hasher,
        certificate::{ConstantProvider, Membership, Scheme as _, 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, Outcome, Resolution, Resolver, TargetedResolver};
    use commonware_runtime::{
        Clock, Metrics, Runner, Supervisor as _, buffer::paged::CacheRef, deterministic,
        utils::reschedule,
    };
    use commonware_storage::archive::immutable;
    use commonware_utils::{
        NZU16, NZU64, NZUsize,
        channel::oneshot::{self, error::TryRecvError},
        sync::Mutex,
        vec::NonEmptyVec,
    };
    use futures::StreamExt;
    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 TestShard = Shard<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 serves inserted blocks and records subscriptions without resolving
    /// them.
    #[derive(Clone, Default)]
    struct RecordingCodingBuffer {
        blocks: Arc<Mutex<Vec<Arc<TestCodedBlock>>>>,
        evict_on_next_hit: Arc<Mutex<bool>>,
        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 insert_transient(&self, block: TestCodedBlock) {
            self.blocks.lock().push(Arc::new(block));
            *self.evict_on_next_hit.lock() = true;
        }

        fn contains(&self, commitment: TestCommitment) -> bool {
            self.blocks
                .lock()
                .iter()
                .any(|block| block.commitment() == commitment)
        }

        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>> {
            let mut blocks = self.blocks.lock();
            let index = blocks
                .iter()
                .position(|block| block.commitment() == commitment)?;
            if std::mem::take(&mut *self.evict_on_next_hit.lock()) {
                return Some(blocks.remove(index));
            }
            Some(Arc::clone(&blocks[index]))
        }

        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 send(&self, round: Round, block: Arc<TestCodedBlock>, recipients: Recipients<K>) {
            self.sends.lock().push((round, block, recipients));
        }
    }

    /// Records the ancestry consumed by verification, accepting only after reaching genesis.
    #[derive(Clone, Default)]
    struct WalkingVerifyingApp {
        blocks: Arc<Mutex<Vec<D>>>,
    }

    impl crate::Application<deterministic::Context> for WalkingVerifyingApp {
        type Block = CodingB;
        type Context = CodingCtx;
        type SigningScheme = S;
        type Input = ();

        async fn propose(
            &mut self,
            _context: (deterministic::Context, CodingCtx),
            _ancestry: impl Ancestry<CodingB>,
            _input: (),
        ) -> Option<Digested<CodingB>> {
            None
        }

        async fn verify(
            &mut self,
            _context: (deterministic::Context, CodingCtx),
            mut ancestry: impl Ancestry<CodingB>,
        ) -> bool {
            while let Some(block) = ancestry.next().await {
                self.blocks.lock().push(block.digest());
                if block.height() == Height::zero() {
                    return true;
                }
            }
            false
        }
    }

    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<Outcome>>>>,
        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"
            );
        }

        /// Delivers `value` for `fetch` and returns whether marshal accepted it.
        async fn deliver(&self, fetch: CodingFetchRecord, value: Bytes) -> bool {
            let (response, receiver) = oneshot::channel();
            assert!(
                self.sender
                    .as_ref()
                    .expect("resolver sender missing")
                    .enqueue(handler::Message::Deliver {
                        delivery: Delivery {
                            key: fetch.key,
                            subscribers: NonEmptyVec::new((
                                fetch.subscriber,
                                tracing::Span::none()
                            )),
                        },
                        value,
                        response,
                    })
                    .accepted()
            );
            receiver.await.expect("delivery response missing")
                == Outcome::Resolved(Resolution::Complete)
        }

        /// Returns whether marshal accepted the newest 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")
                == Outcome::Resolved(Resolution::Complete)
        }

        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
        }
    }

    type Finalizations =
        immutable::Archive<deterministic::Context, D, Finalization<S, TestCommitment>>;
    type FinalizedBlocks = immutable::Archive<
        deterministic::Context,
        D,
        StoredCodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>,
    >;

    /// Builds a marshal actor configuration for tests.
    fn test_config(
        partition_prefix: &str,
        provider: ConstantProvider<S, Epoch>,
    ) -> Config<
        ConstantProvider<S, Epoch>,
        FixedEpocher,
        Sequential,
        CodingB,
        Arc<TestCodedBlock>,
        TestCommitment,
    > {
        Config {
            provider,
            epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
            start: Start::Genesis(CodingHarness::genesis_block(NUM_VALIDATORS as u16).into()),
            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::new(harness::PAGE_SIZE, harness::PAGE_CACHE_SIZE),
            strategy: Sequential,
        }
    }

    /// Initializes immutable finalized stores for tests.
    async fn immutable_finalized_stores(
        context: &deterministic::Context,
        partition_prefix: &str,
        config: &Config<
            ConstantProvider<S, Epoch>,
            FixedEpocher,
            Sequential,
            CodingB,
            Arc<TestCodedBlock>,
            TestCommitment,
        >,
    ) -> (Finalizations, FinalizedBlocks) {
        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");
        (finalizations_by_height, finalized_blocks)
    }

    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 = test_config(partition_prefix, provider);
        let (finalizations_by_height, finalized_blocks) =
            immutable_finalized_stores(&context, partition_prefix, &config).await;
        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,
            1024 * 1024,
        )
        .await;
        let control = oracle.control(me.clone());
        let shard_config: shards::Config<_, _, _, _, _, _> = shards::Config {
            scheme_provider: provider,
            blocker: control.clone(),
            block_codec_config: (),
            strategy: Sequential,
            mailbox_size: NZUsize!(10),
            peer_buffer_size: NZUsize!(64),
            records: harness::RECORDS,
            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 = encode_block(candidate, coding_config);
        (candidate_ctx, coded_candidate)
    }

    /// Rebinds the shards of [`missing_candidate`] to a commitment that claims another digest.
    /// The shards verify against the commitment's coding root, but the decoded block fails
    /// validation.
    fn undecodable(me: K) -> (TestCommitment, Vec<TestShard>) {
        let (_, candidate) = missing_candidate(me);
        let real = candidate.commitment();
        let commitment = TestCommitment::from((
            Sha256::hash(&[b"undecodable"]),
            real.root(),
            real.context(),
            real.config(),
        ));
        let shards = (0..NUM_VALIDATORS as u16)
            .map(|index| {
                let mut shard = candidate.shard(index).expect("missing shard");
                shard.commitment = commitment;
                shard
            })
            .collect();
        (commitment, shards)
    }

    /// Builds `length` coded blocks above genesis, each proposed at the view
    /// matching its height and naming the block below it as its parent.
    fn coding_chain(leader: K, length: u64) -> Vec<(Round, TestCodedBlock)> {
        let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
        let genesis = genesis_block();
        let mut parent_view = View::zero();
        let mut parent_commitment = genesis_coding_commitment(&genesis);
        let mut parent_digest = genesis.digest();
        let mut chain = Vec::new();
        for height in 1..=length {
            let round = Round::new(Epoch::zero(), View::new(height));
            let ctx = CodingCtx {
                round,
                leader: leader.clone(),
                parent: (parent_view, parent_commitment),
            };
            let block = make_coding_block(ctx, parent_digest, Height::new(height), height * 100);
            let coded: TestCodedBlock = encode_block(block.clone(), coding_config);
            parent_view = round.view();
            parent_commitment = coded.commitment();
            parent_digest = coded.digest();
            chain.push((round, coded));
        }
        chain
    }

    #[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 = encode_block(candidate.clone(), dishonest_config);
            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_eq!(
                response_rx.await.unwrap(),
                Outcome::Invalid,
                "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 child = Digested::precomputed(child.clone(), child.digest());
            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 mut 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"
            );
            assert!(
                matches!(verify_rx.try_recv(), Err(TryRecvError::Empty)),
                "a signer's verification waits for its assigned shard"
            );
            drop(verify_rx);
        });
    }

    /// A committee member without a signing share casts no notarize vote, so verifying a
    /// proposal resolves without waiting for the member's assigned shard.
    #[test_traced("WARN")]
    fn test_coding_verify_member_accepts_without_assigned_shard() {
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture {
                participants,
                verifier,
                ..
            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let me = participants[0].clone();
            let member = S::member(
                NAMESPACE,
                verifier.participants().clone(),
                verifier.polynomial().clone(),
                &me,
            )
            .expect("fixture participant must be a member");
            assert!(matches!(member.me(), Some(Membership::Member(_))));
            let provider = ConstantProvider::new(member);
            let (marshal, _resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-verify-member",
                provider.clone(),
                RecordingCodingBuffer::default(),
            )
            .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);

            // No shard reaches the member, so prompt acceptance shows that verification did not
            // wait for the member's assigned shard.
            let verify_rx = marshaled.verify(candidate_ctx, commitment).await;
            select! {
                result = verify_rx => {
                    assert!(
                        result.expect("verify must report a result"),
                        "the member must accept the proposal"
                    );
                },
                _ = context.sleep(Duration::from_secs(1)) => {
                    panic!("member verification must not wait for an assigned shard");
                },
            }
        });
    }

    /// 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"
            );
        });
    }

    /// Registers a certification gate at view 5 of epoch 0 and at `notarized`, has the wrapper
    /// report a notarization at `notarized`, and returns whether the gate at view 5 of epoch 0
    /// remains.
    fn epoch_zero_gate_remains(notarized: Round) -> bool {
        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 (marshal, _resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("actor_stack"),
                "coding-epoch-gates",
                provider.clone(),
                RecordingCodingBuffer::default(),
            )
            .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) = missing_candidate(me);
            let commitment = candidate.commitment();
            let earlier_round = Round::new(Epoch::zero(), View::new(5));
            for round in [earlier_round, notarized] {
                let (_, gate) = oneshot::channel();
                marshaled.gates().insert(round, commitment, gate);
            }
            let proposal = Proposal::new(notarized, View::zero(), commitment);
            let notarization = CodingHarness::make_notarization(proposal, &schemes, QUORUM);
            marshaled.report(Activity::Notarization(notarization));

            // Certification of the notarized round follows its notarization, so the report
            // must keep its gate.
            assert!(
                marshaled.gates().take(notarized, commitment).is_some(),
                "the notarized round must keep its gate"
            );
            marshaled.gates().take(earlier_round, commitment).is_some()
        })
    }

    /// An activity of a later epoch shows that the engine of every earlier epoch was retired,
    /// so reporting it releases the gates of earlier epochs without waiting for a finalization.
    #[test_traced("WARN")]
    fn test_coding_later_epoch_activity_releases_earlier_epoch() {
        let notarized = Round::new(Epoch::new(1), View::new(1));
        assert!(!epoch_zero_gate_remains(notarized));
    }

    /// Within an epoch only a finalization prunes, so a notarization at a later view keeps the
    /// gates of earlier views.
    #[test_traced("WARN")]
    fn test_coding_same_epoch_activity_keeps_gates() {
        let notarized = Round::new(Epoch::zero(), View::new(7));
        assert!(epoch_zero_gate_remains(notarized));
    }

    /// Pending verification must own a transiently buffered block before certification waits
    /// on its gate. Otherwise a cache hit can suppress the round-bound fetch before the
    /// verifier installs its local subscription, and eviction leaves certification pending
    /// indefinitely.
    #[test_traced("WARN")]
    fn test_coding_certify_retains_transient_buffer_hit_through_eviction() {
        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-transient-buffer",
                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: marshal.clone(),
                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;

            // The next lookup returns ownership while removing the buffer entry, modeling
            // shard engine eviction.
            buffer.insert_transient(candidate);
            let _verify_rx = marshaled.verify(candidate_ctx, commitment).await;
            let certify_rx = marshaled.certify(round, commitment).await;

            // This request is ordered after the verification wait and the certification hint
            // in the marshal mailbox. Once it returns, the one-shot buffer hit and eviction have
            // both occurred.
            marshal.get_processed().await;
            assert!(
                !buffer.contains(commitment),
                "the buffered block must be evicted before verification completes"
            );

            assert!(
                certify_rx.await.expect("certify result missing"),
                "certify should succeed via the shared verification gate"
            );
        });
    }

    #[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);
        }
    }

    /// A longer storm crosses the epoch boundary while validators are down, so
    /// recovery backfills certificates from a later epoch than the one the
    /// restarted validator last stored.
    #[test_group("slow")]
    #[test_traced("WARN")]
    fn test_coding_hailstorm_restarts_across_epochs() {
        let _ = harness::hailstorm::<CodingHarness>(13, 5, 5, 2, LINK);
    }

    #[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>();
    }

    /// Dispatch uses the finalized coded block without rereading the finalized archive.
    #[test_traced("WARN")]
    fn test_coding_dispatch_delivers_staged_block_without_archive_read() {
        const PARTITION_PREFIX: &str = "coding-staged-dispatch";
        let runner = deterministic::Runner::timed(Duration::from_secs(30));
        runner.start(|mut context| async move {
            let Fixture { schemes, .. } =
                bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
            let config = test_config(PARTITION_PREFIX, ConstantProvider::new(schemes[0].clone()));
            let (finalizations_by_height, finalized_blocks) =
                immutable_finalized_stores(&context, PARTITION_PREFIX, &config).await;
            let finalized_blocks = Recording::new(finalized_blocks, |block| {
                std::ptr::from_ref(block.inner()).addr()
            });
            let ops = finalized_blocks.ops();
            let (actor, mut mailbox, _) = core::Actor::init(
                context.child("actor"),
                finalizations_by_height,
                finalized_blocks,
                config,
            )
            .await;
            let (resolver_rx, resolver) = RecordingResolver::holding(context.child("resolver"));
            let application = Application::<CodingB>::manual_ack();
            let _actor_handle = actor.start(
                application.clone(),
                RecordingCodingBuffer::default(),
                (resolver_rx, resolver),
            );
            assert_eq!(application.acknowledged().await, Height::zero());

            // Store a verified block, then finalize it through consensus.
            let genesis = genesis_block();
            let round = Round::new(Epoch::zero(), View::new(1));
            let ctx = CodingCtx {
                round,
                leader: default_leader(),
                parent: (View::zero(), genesis_coding_commitment(&genesis)),
            };
            let block = make_coding_block(ctx, genesis.digest(), Height::new(1), 100);
            let coded: TestCodedBlock = encode_block(
                block.clone(),
                coding_config_for_participants(NUM_VALIDATORS as u16),
            );
            assert!(mailbox.verified(round, coded.clone()).await);
            let finalization = CodingHarness::make_finalization(
                Proposal::new(round, View::zero(), coded.commitment()),
                &schemes,
                QUORUM,
            );
            CodingHarness::report_finalization(&mut mailbox, finalization).await;
            assert_eq!(application.acknowledged().await, Height::new(1));

            let delivered = application
                .blocks()
                .get(&Height::new(1))
                .cloned()
                .expect("finalized block dispatched");
            assert_eq!(delivered.digest(), block.digest());
            let ops = ops.lock();
            let written = ops
                .iter()
                .position(|op| matches!(op, Op::Put(height, _) if *height == Height::new(1)))
                .expect("finalized block written");
            assert_eq!(
                ops[written],
                Op::Put(Height::new(1), Arc::as_ptr(&delivered).addr()),
                "storage must share the block payload dispatched to the application"
            );
            assert!(
                !ops[written..].contains(&Op::Get(Some(Height::new(1)))),
                "dispatch must not read the finalized block back from the archive: {ops:?}"
            );
        });
    }

    #[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_floor_retains_processed_predecessor() {
        harness::floor_retains_processed_predecessor::<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(),
                1024 * 1024,
            )
            .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 = encode_block(block.clone(), coding_config);
            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"
            );
        });
    }

    /// Gap repair archives a finalized block's ancestor that only the shard engine holds.
    ///
    /// The parent (height 1) and descendant (height 2) are cached only in the shard engine, and
    /// only the descendant has a finalization.
    #[test_traced("WARN")]
    fn test_coding_repair_reads_buffer_only_ancestor() {
        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(),
                1024 * 1024,
            )
            .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 marshal = setup.mailbox;
            let shards = 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);

            // Cache both blocks in the shard engine only.
            shards.proposed(Round::new(Epoch::new(0), View::new(1)), parent_block);
            shards.proposed(Round::new(Epoch::new(0), View::new(2)), descendant_block);
            assert!(shards.get(parent_commitment).await.is_some());
            assert!(shards.get(descendant_commitment).await.is_some());

            // Report finalization for the descendant only. The parent has no finalization, so
            // repair must walk the descendant's parent link and read the parent from the shard
            // engine.
            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 marshal, descendant_finalization).await;

            // Archiving the descendant runs repair, which archives the parent.
            while marshal.get_block(Height::new(2)).await.is_none() {
                reschedule().await;
            }
            let parent = marshal.get_block(Height::new(1)).await;
            assert_eq!(
                parent.map(|block| block.commitment()),
                Some(parent_commitment),
                "repair must archive the parent from the shard engine"
            );
        });
    }

    #[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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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), created 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 = encode_block(block_a.clone(), coding_config);
            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, as 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 = encode_block(block_b.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(block.clone(), coding_config);
                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 = encode_block(boundary_block.clone(), coding_config);
            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, since 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(),
                1024 * 1024,
            )
            .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 = encode_block(block.clone(), coding_config);
            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)) => {},
            }
        })
    }

    /// The shard engine's window can evict a commitment's record on either side of
    /// re-proposal verification. The re-proposal must remain live through core marshal's
    /// verified cache, while a `discovered` announcement after eviction must recreate shard
    /// state needed before notarization.
    #[test_traced("WARN")]
    fn test_coding_reproposal_recreates_shard_state_after_eviction() {
        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(),
                1024 * 1024,
            )
            .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);

            // Discovers a window of commitments in the views starting at `first`.
            let fill = |first: u64| {
                for view in first..first + harness::RECORDS.get() as u64 {
                    let commitment = TestCommitment::from((
                        Sha256::hash(&[&view.to_be_bytes()]),
                        Sha256::hash(&[b"filler_root"]),
                        Sha256::hash(&[b"filler_context"]),
                        coding_config,
                    ));
                    shards.discovered(
                        commitment,
                        participants[1].clone(),
                        Round::new(Epoch::new(0), View::new(view)),
                    );
                }
            };

            // 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 =
                encode_block(boundary_block.clone(), coding_config);
            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());
            assert!(shards.get(boundary_commitment).await.is_some());

            // Records for a window of later views evict the cached block.
            fill(boundary_height.get() + 1);
            assert!(shards.get(boundary_commitment).await.is_none());

            // Re-propose the boundary block above those views 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(40));
            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 eviction"
            );

            // 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");
                },
            }

            // Records for a window of later views evict the recreated shard state, so a new
            // assigned-shard subscription waits. The block persisted in core marshal remains,
            // and certification and caller-owned core subscriptions stay live through that
            // cache.
            fill(reproposal_round.view().get() + 1);
            let mut evicted = shards.subscribe_assigned_shard_verified(boundary_commitment);
            assert!(shards.get(boundary_commitment).await.is_none());
            assert!(matches!(
                evicted.try_recv(),
                Err(oneshot::error::TryRecvError::Empty)
            ));

            let block = marshal
                .subscribe_by_commitment(boundary_commitment, core::CommitmentFallback::Wait)
                .await
                .expect("core block subscription closed after shard eviction");
            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 eviction"
            );
        })
    }

    /// 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(),
                1024 * 1024,
            )
            .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 = encode_block(block.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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> =
                encode_block(block_a.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(boundary_block.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 mut senders = Vec::new();
            for peer in &participants[1..3] {
                let (sender, _) = oracle
                    .control(peer.clone())
                    .register(2, TEST_QUOTA)
                    .await
                    .unwrap();
                senders.push(sender);
            }
            setup_network_links(&mut oracle, &participants[..3], LINK).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 decodable block.
            let (missing_payload, missing_shards) = undecodable(me.clone());
            let round = Round::new(Epoch::zero(), View::new(1));
            let reproposal_context = CodingCtx {
                round,
                leader: me.clone(),
                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;

            // Once the certification gate task's subscription reaches the shard engine, a
            // failed reconstruction of the payload closes it.
            marshal.get_processed().await;
            shards.notarized(missing_payload, round);
            for (sender, shard) in senders.iter_mut().zip(&missing_shards[1..]) {
                sender.send(Recipients::One(me.clone()), shard.encode(), true);
            }

            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);
        })
    }

    /// A core commitment subscription closes when the shard engine fails to reconstruct the
    /// commitment's block.
    #[test_traced("WARN")]
    fn test_core_subscription_closes_when_coding_reconstruction_fails() {
        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(),
                1024 * 1024,
            )
            .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 mut senders = Vec::new();
            for peer in &participants[1..3] {
                let (sender, _) = oracle
                    .control(peer.clone())
                    .register(2, TEST_QUOTA)
                    .await
                    .unwrap();
                senders.push(sender);
            }
            setup_network_links(&mut oracle, &participants[..3], LINK).await;
            let marshal = setup.mailbox;
            let shards = setup.extra;

            // The commitment's shards verify, but the block they decode fails validation.
            let (missing_commitment, missing_shards) = undecodable(me.clone());
            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 },
            );

            // Once the underlying buffer subscription reaches the shard engine, a failed
            // reconstruction of the commitment closes it.
            marshal.get_processed().await;
            shards.notarized(missing_commitment, round);
            for (sender, shard) in senders.iter_mut().zip(&missing_shards[1..]) {
                sender.send(Recipients::One(me.clone()), shard.encode(), true);
            }

            // The core actor closes the subscription.
            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 failed reconstruction");
                },
            }
        })
    }

    #[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(),
                1024 * 1024,
            )
            .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().await != Some(Processed::Block(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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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 = encode_block(block.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(honest_parent.clone(), coding_config);
            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 = encode_block(malicious_block.clone(), coding_config);
            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 = encode_block(malicious_block2.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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 = encode_block(child.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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 = encode_block(child.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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 = encode_block(child.clone(), coding_config);
            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, which must fail.
        let malformed_bytes = [0u8; <TestCommitment as FixedSize>::SIZE];
        let result = TestCommitment::read(&mut commonware_codec::Copying(&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 mut 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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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 = encode_block(block.clone(), coding_config);
            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(),
                1024 * 1024,
            )
            .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 = encode_block(certified.clone(), coding_config);
            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 = encode_block(notarized.clone(), coding_config);
            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 = encode_block(block.clone(), coding_config);
            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 config.
        // The finalized decode takes the root from the commitment, so the wire
        // config must still be checked against it and the block rejected.
        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(),
                1024 * 1024,
            )
            .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> =
                encode_block(block1.clone(), coding_config_a);
            let commitment_a = coded_block_a.commitment();

            let coded_block_b: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                encode_block(block1.clone(), coding_config_b);
            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 coding config 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")]
    fn test_coding_finalized_block_delivery_skips_recoding() {
        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 mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "finalized-block-delivery",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;

            let (candidate_ctx, candidate) = missing_candidate(participants[0].clone());
            let commitment = candidate.commitment();
            let subscription =
                mailbox.subscribe_by_commitment(commitment, core::CommitmentFallback::Wait);
            context.sleep(Duration::from_millis(100)).await;

            // The finalization names the commitment, so the fetched block is
            // rebuilt from it without recomputing the coding root.
            let finalization = CodingHarness::make_finalization(
                Proposal::new(candidate_ctx.round, View::zero(), commitment),
                &schemes,
                QUORUM,
            );
            resolver.respond_to_next_fetch(candidate.encode());
            CodingHarness::report_finalization(&mut mailbox, finalization).await;

            let delivered = subscription.await.expect("subscription dropped");
            assert_eq!(delivered.commitment(), commitment);
            assert!(
                delivered.shard(0).is_none(),
                "finalized delivery should not recompute shards"
            );
            assert!(
                resolver.wait_for_delivery_response().await,
                "finalized delivery should validate"
            );
            assert!(
                resolver.fetches().iter().any(|fetch| matches!(
                    (&fetch.key, &fetch.subscriber),
                    (
                        handler::Key::Block(requested),
                        handler::Annotation::Finalized(handler::Finalized::ByRound { .. }),
                    ) if *requested == commitment
                )),
                "finalization should fetch the block by commitment"
            );
            assert!(mailbox.get_block(Height::new(1)).await.is_some());
        });
    }

    #[test_traced("WARN")]
    fn test_coding_coalesced_block_delivery_trusts_finalized_subscriber() {
        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"),
                "coalesced-block-delivery",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;
            let resolver_tx = resolver
                .sender
                .clone()
                .expect("recording resolver should keep its sender");

            let (_, candidate) = missing_candidate(participants[0].clone());
            let commitment = candidate.commitment();
            let height = Height::new(1);
            let subscription =
                mailbox.subscribe_by_commitment(commitment, core::CommitmentFallback::Wait);
            context.sleep(Duration::from_millis(100)).await;

            // One key can carry an ancestry subscriber without certification
            // evidence and a finalized-chain subscriber at once. The finalized
            // subscriber alone binds the coding root, so the shared delivery
            // decodes without recomputing it and lands in the finalized archive.
            let mut subscribers = NonEmptyVec::new((
                handler::Annotation::Untrusted { height },
                tracing::Span::none(),
            ));
            subscribers.push((
                handler::Annotation::Finalized(handler::Finalized::ByHeight { height }),
                tracing::Span::none(),
            ));
            let (response, response_rx) = oneshot::channel();
            assert!(
                resolver_tx
                    .enqueue(handler::Message::Deliver {
                        delivery: Delivery {
                            key: handler::Key::Block(commitment),
                            subscribers,
                        },
                        value: candidate.encode(),
                        response,
                    })
                    .accepted()
            );
            assert_eq!(
                response_rx.await.unwrap(),
                Outcome::Resolved(Resolution::Complete),
                "coalesced delivery should validate"
            );

            let delivered = subscription.await.expect("subscription dropped");
            assert_eq!(delivered.commitment(), commitment);
            assert!(
                delivered.shard(0).is_none(),
                "coalesced delivery should not recompute shards"
            );
            assert!(mailbox.get_block(height).await.is_some());
        });
    }

    #[test_traced("WARN")]
    fn test_coding_finalized_block_delivery_rejects_digest_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 (mut mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "finalized-block-digest-mismatch",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;

            let (candidate_ctx, candidate) = missing_candidate(participants[0].clone());
            let commitment = candidate.commitment();
            let other = make_coding_block(
                candidate_ctx.clone(),
                genesis_block().digest(),
                Height::new(1),
                101,
            );
            let other: TestCodedBlock = encode_block(
                other.clone(),
                coding_config_for_participants(NUM_VALIDATORS as u16),
            );
            assert_ne!(other.digest(), candidate.digest());

            // The finalized decode skips the root, so the digest check is what
            // rejects a different block under the same coding config.
            let finalization = CodingHarness::make_finalization(
                Proposal::new(candidate_ctx.round, View::zero(), commitment),
                &schemes,
                QUORUM,
            );
            resolver.respond_to_next_fetch(other.encode());
            CodingHarness::report_finalization(&mut mailbox, finalization).await;
            context.sleep(Duration::from_millis(100)).await;

            assert!(
                !resolver.wait_for_delivery_response().await,
                "digest mismatch should be rejected"
            );
            assert!(mailbox.get_block(Height::new(1)).await.is_none());
        });
    }

    /// Finalized height deliveries bind the bundled bytes to the signed block digest.
    #[test_traced("WARN")]
    fn test_coding_finalized_height_delivery_rejects_digest_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 partition_prefix = "finalized-height-digest-mismatch";
            let config = test_config(partition_prefix, ConstantProvider::new(schemes[0].clone()));
            let (finalizations_by_height, finalized_blocks) =
                immutable_finalized_stores(&context, partition_prefix, &config).await;
            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 application = Application::<CodingB>::manual_ack();
            let _actor_handle = actor.start(
                application.clone(),
                RecordingCodingBuffer::default(),
                (resolver_rx, resolver.clone()),
            );
            assert_eq!(application.acknowledged().await, Height::zero());

            let (candidate_ctx, honest) = missing_candidate(participants[0].clone());
            let height = honest.height();
            let other = make_coding_block(candidate_ctx.clone(), honest.parent(), height, 101);
            let other: TestCodedBlock = encode_block(
                other.clone(),
                coding_config_for_participants(NUM_VALIDATORS as u16),
            );
            let finalization = CodingHarness::make_finalization(
                Proposal::new(candidate_ctx.round, View::zero(), honest.commitment()),
                &schemes,
                QUORUM,
            );
            assert!(finalization.verify(&mut context, &schemes[0], &Sequential));
            assert_eq!(honest.height(), other.height());
            assert_eq!(honest.context(), other.context());
            assert_ne!(honest.inner().encode(), other.inner().encode());
            let delivered = CodingB::decode(other.inner().encode()).expect("valid block encoding");
            assert_eq!(&delivered, other.inner());
            assert_eq!(
                finalization.proposal.payload.block(),
                honest.inner().digest()
            );
            assert_ne!(finalization.proposal.payload.block(), delivered.digest());
            let other_digest = other.digest();

            let mut subscriptions = [
                mailbox
                    .subscribe_by_commitment(honest.commitment(), core::CommitmentFallback::Wait),
                mailbox.subscribe_by_commitment(other.commitment(), core::CommitmentFallback::Wait),
                mailbox.subscribe_by_digest(honest.digest(), core::DigestFallback::Wait),
                mailbox.subscribe_by_digest(other.digest(), core::DigestFallback::Wait),
            ];
            let processed = mailbox.get_processed().await;
            let tip = application.tip();
            let (response, response_rx) = oneshot::channel();
            assert!(
                resolver
                    .sender
                    .as_ref()
                    .expect("recording resolver should keep its sender")
                    .enqueue(handler::Message::Deliver {
                        delivery: Delivery {
                            key: handler::Key::Finalized { height },
                            subscribers: NonEmptyVec::new((
                                handler::Annotation::Finalized(handler::Finalized::ByHeight {
                                    height,
                                }),
                                tracing::Span::none(),
                            )),
                        },
                        value: (finalization, delivered).encode(),
                        response,
                    })
                    .accepted()
            );
            assert_eq!(
                response_rx.await.expect("delivery response missing"),
                Outcome::Invalid,
                "a valid finalization must reject different bundled block bytes"
            );

            assert!(mailbox.get_block(height).await.is_none());
            assert!(mailbox.get_block(&honest.digest()).await.is_none());
            assert!(mailbox.get_block(&other_digest).await.is_none());
            assert!(mailbox.get_finalization(height).await.is_none());
            assert_eq!(mailbox.get_processed().await, processed);
            for subscription in &mut subscriptions {
                assert!(matches!(
                    subscription.try_recv(),
                    Err(oneshot::error::TryRecvError::Empty)
                ));
            }
            assert_eq!(application.blocks().len(), 1);
            assert!(!application.blocks().contains_key(&height));
            assert!(application.pending_ack_heights().is_empty());
            assert_eq!(application.tip(), tip);
        });
    }

    #[test_traced("WARN")]
    fn test_coding_ancestry_block_delivery_recomputes_root() {
        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 mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "ancestry-block-delivery",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;

            // Certification persists the candidate before the application verdict.
            // That write alone must not authenticate its ancestry, so each fetch
            // below still proves its coding root by re-encoding.
            let mut chain = coding_chain(participants[0].clone(), 3);
            let (round, candidate) = chain.pop().expect("candidate");
            let notarization = CodingHarness::make_notarization(
                Proposal::new(round, View::new(2), candidate.commitment()),
                &schemes,
                QUORUM,
            );
            mailbox.report(Activity::Notarization(notarization));
            assert!(mailbox.certified(round, candidate).await);
            let (_, top) = chain.pop().expect("top");
            let (_, bottom) = chain.pop().expect("bottom");
            for (block, height) in [(top, 2), (bottom, 1)] {
                resolver.respond_to_next_fetch(block.encode());
                let delivered = mailbox
                    .subscribe_by_commitment(
                        block.commitment(),
                        core::CommitmentFallback::FetchByCommitment {
                            height: Height::new(height),
                        },
                    )
                    .await
                    .expect("subscription dropped");
                assert_eq!(delivered.commitment(), block.commitment());
                assert!(
                    delivered.shard(0).is_some(),
                    "ancestry delivery should recompute shards"
                );
                assert!(
                    resolver.wait_for_delivery_response().await,
                    "ancestry delivery should validate"
                );
            }
            let fetches = resolver.fetches();
            assert_eq!(fetches.len(), 2);
            assert!(fetches.iter().all(|fetch| matches!(
                (&fetch.key, &fetch.subscriber),
                (
                    handler::Key::Block(_),
                    handler::Annotation::Untrusted { .. }
                )
            )));
        });
    }

    #[test_traced("WARN")]
    fn test_coding_optimistic_ancestry_conflicts_with_certified_cache() {
        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 (mut mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "optimistic-ancestry-conflict",
                provider.clone(),
                RecordingCodingBuffer::default(),
            )
            .await;

            // Cache a certified ancestor at height 2 without finalizing its branch
            let mut certified_chain = coding_chain(participants[0].clone(), 3);
            let (round, certified_child) = certified_chain.pop().unwrap();
            let (_, certified_ancestor) = certified_chain.pop().unwrap();
            assert!(mailbox.verified(round, certified_child.clone()).await);
            mailbox.report(Activity::Certification(CodingHarness::make_notarization(
                Proposal::new(round, View::new(2), certified_child.commitment()),
                &schemes,
                QUORUM,
            )));
            resolver.respond_to_next_fetch(certified_ancestor.encode());
            let certified = mailbox
                .subscribe_by_commitment(
                    certified_ancestor.commitment(),
                    core::CommitmentFallback::FetchByCommitment {
                        height: certified_ancestor.height(),
                    },
                )
                .await
                .unwrap();
            assert!(resolver.wait_for_delivery_response().await);
            assert!(certified.shard(0).is_none());
            assert_eq!(
                resolver.fetches()[0].subscriber,
                handler::Annotation::Certified {
                    height: Height::new(2)
                }
            );

            // Later views extend a competing branch at the same heights
            let mut chain = Vec::new();
            let mut parent_view = View::zero();
            let mut parent_commitment = genesis_coding_commitment(&genesis_block());
            for height in 1..=4 {
                let round = Round::new(Epoch::zero(), View::new(height + 10));
                let block: TestCodedBlock = {
                    let block = make_coding_block(
                        CodingCtx {
                            round,
                            leader: participants[0].clone(),
                            parent: (parent_view, parent_commitment),
                        },
                        parent_commitment.block(),
                        Height::new(height),
                        height * 100,
                    );
                    encode_block(
                        block.clone(),
                        coding_config_for_participants(NUM_VALIDATORS as u16),
                    )
                };
                parent_view = round.view();
                parent_commitment = block.commitment();
                chain.push(block);
            }
            let candidate = &chain[3];
            let parent = &chain[2];
            assert_ne!(chain[1].digest(), certified_ancestor.digest());

            // Only the candidate and immediate parent are available before verification
            assert!(
                mailbox
                    .verified(parent.context().round, parent.clone())
                    .await
            );
            assert!(
                mailbox
                    .verified(candidate.context().round, candidate.clone())
                    .await
            );
            let cached = mailbox
                .subscribe_by_digest(parent.digest(), core::DigestFallback::Wait)
                .await
                .unwrap();
            assert_eq!(cached.commitment(), parent.commitment());
            let shards =
                start_shard_mailbox(context.child("shards"), participants, provider.clone()).await;
            let application = WalkingVerifyingApp::default();
            let mut marshaled = Marshaled::new(
                context.child("marshaled"),
                MarshaledConfig {
                    application: application.clone(),
                    marshal: mailbox.clone(),
                    shards,
                    scheme_provider: provider,
                    epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                    strategy: Sequential,
                },
            );
            let _shard_verdict = marshaled
                .verify(candidate.context(), candidate.commitment())
                .await;

            // Neither the certified fork nor a cached untrusted child authenticates these fetches
            for (index, ancestor) in [&chain[1], &chain[0]].into_iter().enumerate() {
                while resolver.fetches().len() < index + 2 {
                    reschedule().await;
                }
                let fetch = resolver.fetches()[index + 1].clone();
                assert_eq!(fetch.key, handler::Key::Block(ancestor.commitment()));
                assert_eq!(
                    fetch.subscriber,
                    handler::Annotation::Untrusted {
                        height: ancestor.height()
                    }
                );
                let mut subscription = mailbox
                    .subscribe_by_commitment(ancestor.commitment(), core::CommitmentFallback::Wait);
                let _ = mailbox.get_processed().await;

                if index == 0 {
                    assert!(
                        !resolver
                            .deliver(fetch.clone(), certified_ancestor.encode())
                            .await
                    );
                    assert!(matches!(
                        subscription.try_recv(),
                        Err(oneshot::error::TryRecvError::Empty)
                    ));
                }
                assert!(resolver.deliver(fetch, ancestor.encode()).await);
                let delivered = subscription.await.unwrap();
                assert_eq!(delivered.commitment(), ancestor.commitment());
                assert!(
                    delivered.shard(0).is_some(),
                    "untrusted ancestry must be recoded"
                );
            }

            // Both forks survive in storage and are retrieved by their own digests
            for ancestor in [&certified_ancestor, &chain[1]] {
                let stored = mailbox
                    .get_block(&ancestor.digest())
                    .await
                    .expect("both ancestors at the same height must remain cached");
                assert_eq!(stored.commitment(), ancestor.commitment());
            }
            assert!(
                marshaled
                    .certify(candidate.context().round, candidate.commitment())
                    .await
                    .await
                    .unwrap()
            );
            let expected: Vec<_> = chain
                .iter()
                .rev()
                .map(Digestible::digest)
                .chain(std::iter::once(genesis_block().digest()))
                .collect();
            assert_eq!(*application.blocks.lock(), expected);

            // Finalization follows the competing branch despite the certified block at height 2
            mailbox.report(Activity::Finalization(CodingHarness::make_finalization(
                Proposal::new(
                    candidate.context().round,
                    parent.context().round.view(),
                    candidate.commitment(),
                ),
                &schemes,
                QUORUM,
            )));
            while mailbox.get_processed().await != Some(Processed::Block(candidate.height())) {
                reschedule().await;
            }
            for block in &chain {
                let finalized = mailbox.get_block(block.height()).await.unwrap();
                assert_eq!(finalized.commitment(), block.commitment());
            }
            assert_eq!(resolver.fetches().len(), 3);
        });
    }

    #[test_traced("WARN")]
    fn test_coding_certified_ancestry_delivery_skips_recoding() {
        for (cached_parent, by_digest) in [(false, false), (true, false), (true, true)] {
            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 mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                    context.child("validator"),
                    "certified-ancestry-delivery",
                    ConstantProvider::new(schemes[0].clone()),
                    RecordingCodingBuffer::default(),
                )
                .await;

                // Build genesis <- grandparent <- parent <- child, verify the child
                // locally, and have consensus report that it was certified.
                // Certification implies its ancestors are certified.
                let mut chain = coding_chain(participants[0].clone(), 3);
                let (child_round, child) = chain.pop().expect("child");
                let (parent_round, parent) = chain.pop().expect("parent");
                let (_, grandparent) = chain.pop().expect("grandparent");
                if cached_parent {
                    assert!(mailbox.verified(parent_round, parent.clone()).await);
                }
                assert!(mailbox.verified(child_round, child.clone()).await);
                let notarization = CodingHarness::make_notarization(
                    Proposal::new(child_round, View::new(2), child.commitment()),
                    &schemes,
                    QUORUM,
                );
                mailbox.report(Activity::Certification(notarization));

                // Both local retrieval and remote delivery preserve certification evidence
                if !cached_parent {
                    resolver.respond_to_next_fetch(parent.encode());
                }
                let subscription = if by_digest {
                    mailbox.subscribe_by_digest(parent.digest(), core::DigestFallback::Wait)
                } else {
                    mailbox.subscribe_by_commitment(
                        parent.commitment(),
                        core::CommitmentFallback::FetchByCommitment {
                            height: Height::new(2),
                        },
                    )
                };
                let delivered = subscription.await.expect("parent subscription dropped");
                assert!(
                    delivered.shard(0).is_none(),
                    "certified parent should not recompute shards"
                );
                if cached_parent {
                    assert!(resolver.fetches().is_empty());
                } else {
                    assert!(resolver.wait_for_delivery_response().await);
                }

                // Resolving the parent extends the evidence to the grandparent
                resolver.respond_to_next_fetch(grandparent.encode());
                let delivered = mailbox
                    .subscribe_by_commitment(
                        grandparent.commitment(),
                        core::CommitmentFallback::FetchByCommitment {
                            height: Height::new(1),
                        },
                    )
                    .await
                    .expect("grandparent subscription dropped");
                assert!(
                    delivered.shard(0).is_none(),
                    "certified grandparent should not recompute shards"
                );
                assert!(resolver.wait_for_delivery_response().await);

                let certified: Vec<Height> = resolver
                    .fetches()
                    .iter()
                    .filter_map(|fetch| match fetch.subscriber {
                        handler::Annotation::Certified { height } => Some(height),
                        _ => None,
                    })
                    .collect();
                let expected = if cached_parent {
                    vec![Height::new(1)]
                } else {
                    vec![Height::new(2), Height::new(1)]
                };
                assert_eq!(certified, expected);
            });
        }
    }

    #[test_traced("WARN")]
    fn test_coding_certification_retention_follows_finalized_tip() {
        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 mailbox, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "certification-retention",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;

            // Start a certified parent fetch before the child is finalized
            let mut chain = coding_chain(participants[0].clone(), 3);
            let (round, child) = chain.pop().expect("child");
            let (_, parent) = chain.pop().expect("parent");
            let (_, grandparent) = chain.pop().expect("grandparent");
            assert!(mailbox.verified(round, child.clone()).await);
            let proposal = Proposal::new(round, View::new(2), child.commitment());
            let notarization = CodingHarness::make_notarization(proposal.clone(), &schemes, QUORUM);
            mailbox.report(Activity::Certification(notarization.clone()));
            let fallback = core::CommitmentFallback::FetchByCommitment {
                height: parent.height(),
            };
            let mut subscriptions =
                vec![mailbox.subscribe_by_commitment(parent.commitment(), fallback)];
            let _ = mailbox.get_processed().await;
            let fetch = resolver.fetches().pop().expect("parent fetch missing");
            assert_eq!(
                fetch.subscriber,
                handler::Annotation::Certified {
                    height: parent.height()
                }
            );

            // Finalization advances the tip while missing ancestors hold back processing
            mailbox.report(Activity::Finalization(CodingHarness::make_finalization(
                proposal, &schemes, QUORUM,
            )));
            assert_eq!(
                mailbox.get_info(child.height()).await,
                Some((child.height(), child.digest()))
            );
            subscriptions.push(mailbox.subscribe_by_commitment(parent.commitment(), fallback));

            // A late certification must not restore evidence below the tip
            mailbox.report(Activity::Certification(notarization));
            subscriptions.push(mailbox.subscribe_by_commitment(parent.commitment(), fallback));
            let _ = mailbox.get_processed().await;

            // The original fetch can still arrive, but must not retain older ancestry evidence
            let (response, response_rx) = oneshot::channel();
            assert!(
                resolver
                    .sender
                    .as_ref()
                    .expect("resolver sender missing")
                    .enqueue(handler::Message::Deliver {
                        delivery: Delivery {
                            key: fetch.key,
                            subscribers: NonEmptyVec::new((
                                fetch.subscriber,
                                tracing::Span::none()
                            )),
                        },
                        value: parent.encode(),
                        response,
                    })
                    .accepted()
            );
            assert_eq!(
                response_rx.await.expect("delivery response missing"),
                Outcome::Resolved(Resolution::Complete)
            );
            subscriptions.push(mailbox.subscribe_by_commitment(
                grandparent.commitment(),
                core::CommitmentFallback::FetchByCommitment {
                    height: grandparent.height(),
                },
            ));
            assert!(
                mailbox
                    .get_processed()
                    .await
                    .map_or(Height::zero(), Processed::height)
                    < grandparent.height()
            );
            let annotations: Vec<_> = resolver
                .fetches()
                .into_iter()
                .filter_map(|fetch| {
                    matches!(
                        fetch.subscriber,
                        handler::Annotation::Certified { .. }
                            | handler::Annotation::Untrusted { .. }
                    )
                    .then_some(fetch.subscriber)
                })
                .collect();
            assert_eq!(
                annotations,
                vec![
                    handler::Annotation::Certified {
                        height: parent.height()
                    },
                    handler::Annotation::Untrusted {
                        height: parent.height()
                    },
                    handler::Annotation::Untrusted {
                        height: parent.height()
                    },
                    handler::Annotation::Untrusted {
                        height: grandparent.height()
                    },
                ]
            );
        });
    }

    #[test_traced("WARN")]
    fn test_coding_notarized_delivery_recomputes_root() {
        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"),
                "notarized-block-delivery",
                ConstantProvider::new(schemes[0].clone()),
                RecordingCodingBuffer::default(),
            )
            .await;

            let (candidate_ctx, candidate) = missing_candidate(participants[0].clone());
            let commitment = candidate.commitment();
            let round = candidate_ctx.round;
            let notarization = CodingHarness::make_notarization(
                Proposal::new(round, View::zero(), commitment),
                &schemes,
                QUORUM,
            );

            // A notarization does not bind the coding root to the block, so the
            // delivery proves it by re-encoding the block.
            resolver.respond_to_next_fetch((notarization, candidate).encode());
            let subscription = mailbox.subscribe_by_commitment(
                commitment,
                core::CommitmentFallback::FetchByRound { round },
            );

            let delivered = subscription.await.expect("subscription dropped");
            assert_eq!(delivered.commitment(), commitment);
            assert!(
                delivered.shard(0).is_some(),
                "notarized delivery should recompute shards"
            );
            assert!(
                resolver.wait_for_delivery_response().await,
                "notarized delivery should validate"
            );
        });
    }

    #[test_traced("WARN")]
    fn test_coding_untrusted_delivery_rejects_mismatched_root() {
        for notarized in [false, true] {
            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"),
                    "untrusted-delivery-mismatched-root",
                    ConstantProvider::new(schemes[0].clone()),
                    RecordingCodingBuffer::default(),
                )
                .await;

                let (candidate_ctx, candidate) = missing_candidate(participants[0].clone());
                let config = candidate.config();
                let other: TestCodedBlock = {
                    let block = make_coding_block(
                        candidate_ctx.clone(),
                        genesis_block().digest(),
                        Height::new(1),
                        101,
                    );
                    encode_block(block, config)
                };
                let expected = candidate.commitment();
                let commitment = TestCommitment::from((
                    expected.block(),
                    other.commitment().root(),
                    expected.context(),
                    config,
                ));
                assert_ne!(commitment.root(), expected.root());

                // A leader can obtain notarize votes with valid assigned shards even
                // though their root encodes different bytes from the named block.
                for (index, shard) in other.shards(&Sequential).iter().enumerate() {
                    ReedSolomon::<Sha256>::check(
                        &config,
                        &commitment.root(),
                        index as u16,
                        shard,
                        &Sequential,
                    )
                    .expect("assigned shard should verify against the advertised root");
                }

                let (value, fallback) = if notarized {
                    let round = candidate_ctx.round;
                    let notarization = CodingHarness::make_notarization(
                        Proposal::new(round, View::zero(), commitment),
                        &schemes,
                        QUORUM,
                    );
                    (
                        (notarization, candidate.clone()).encode(),
                        core::CommitmentFallback::FetchByRound { round },
                    )
                } else {
                    (
                        candidate.encode(),
                        core::CommitmentFallback::FetchByCommitment {
                            height: Height::new(1),
                        },
                    )
                };
                resolver.respond_to_next_fetch(value);
                let mut subscription = mailbox.subscribe_by_commitment(commitment, fallback);
                while resolver.fetches().is_empty() {
                    reschedule().await;
                }
                assert!(
                    !resolver.wait_for_delivery_response().await,
                    "mismatched coding root should be rejected (notarized={notarized})"
                );
                assert!(matches!(
                    subscription.try_recv(),
                    Err(oneshot::error::TryRecvError::Empty)
                ));
                assert!(mailbox.get_block(&candidate.digest()).await.is_none());
            });
        }
    }

    #[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 = encode_block(floor_block.clone(), coding_config);
            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;
        })
    }

    /// The default prepare declines after the marshal's checks fetch the parent.
    #[test_traced("WARN")]
    fn test_coding_default_prepare_declines() {
        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 (marshal, _resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "default-prepare",
                provider.clone(),
                RecordingCodingBuffer::default(),
            )
            .await;
            let shards = start_shard_mailbox(
                context.child("shards"),
                participants.clone(),
                provider.clone(),
            )
            .await;
            let cfg = MarshaledConfig {
                application: WalkingVerifyingApp::default(),
                marshal,
                shards,
                scheme_provider: provider,
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("declining"), cfg);
            let genesis = CodingHarness::genesis_block(NUM_VALIDATORS as u16);
            let ctx = CodingCtx {
                round: Round::new(Epoch::zero(), View::new(1)),
                leader: participants[0].clone(),
                parent: (View::zero(), genesis.commitment()),
            };
            let decision = marshaled
                .prepare(ctx)
                .await
                .await
                .expect("prepare decision missing");
            assert_eq!(decision, Handoff::Wait, "the default prepare declines");
            assert!(
                context
                    .encode()
                    .contains("declining_parent_fetch_duration_count 1"),
                "a declining prepare still fetches the parent"
            );
        });
    }

    /// 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(),
                1024 * 1024,
            )
            .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(),
                1024 * 1024,
            )
            .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 = encode_block(parent.clone(), coding_config);
            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 = encode_block(child.clone(), coding_config);
            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 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();
            let _ = marshal_actor_handle.await;
            drop(marshaled);
            drop(marshal);
            drop(shards);

            // Restart from the same partition. The block must be durably persisted, or 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(),
                1024 * 1024,
            )
            .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 = encode_block::<_, ReedSolomon<Sha256>, Sha256>(
                block_to_propose.clone(),
                coding_config,
            )
            .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();
            let _ = marshal_actor_handle.await;
            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 only the staged proposal 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(),
                1024 * 1024,
            )
            .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 = encode_block::<_, ReedSolomon<Sha256>, Sha256>(
                block_to_propose.clone(),
                coding_config,
            )
            .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"
            );
        });
    }

    /// A prepare relay sends a held candidate without storing it, so an abandoned candidate
    /// costs no storage write, and a replacement candidate for the same round is sent the same
    /// way. The propose relay that locks the replacement in stores it without sending it again
    /// and completes the durability handshake.
    #[test_traced("WARN")]
    fn test_coding_prepare_relay_sends_once_and_propose_stores() {
        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("validator"),
                "prepare-relay",
                provider.clone(),
                buffer.clone(),
            )
            .await;
            let shards = start_shard_mailbox(
                context.child("shards"),
                participants.clone(),
                provider.clone(),
            )
            .await;

            let me = participants[0].clone();
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
            let genesis = genesis_block();
            let round = Round::new(Epoch::zero(), View::new(1));
            let ctx = CodingCtx {
                round,
                leader: me,
                parent: (View::zero(), genesis_coding_commitment(&genesis)),
            };

            // Stage two candidates for the round as prepare would. The mock application
            // builds one fixed block, so each candidate is staged by its own wrapper.
            let mut candidates = Vec::new();
            for data in [100, 200] {
                let block = make_coding_block(ctx.clone(), genesis.digest(), Height::new(1), data);
                let digest = block.digest();
                let commitment =
                    encode_block::<_, ReedSolomon<Sha256>, Sha256>(block.clone(), coding_config)
                        .commitment();
                let mock_app: MockVerifyingApp<CodingB, S> =
                    MockVerifyingApp::new().with_propose_result(block);
                let cfg = MarshaledConfig {
                    application: mock_app,
                    marshal: marshal.clone(),
                    shards: shards.clone(),
                    scheme_provider: provider.clone(),
                    epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                    strategy: Sequential,
                };
                let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);
                let staged = marshaled
                    .propose(ctx.clone())
                    .await
                    .await
                    .expect("propose should produce a commitment");
                assert_eq!(staged, commitment);
                candidates.push((marshaled, commitment, digest));
            }
            let (mut abandoned, abandoned_commitment, abandoned_digest) = candidates.remove(0);
            let (mut replacement, replacement_commitment, replacement_digest) =
                candidates.remove(0);

            // The early relay sends each held candidate and stores neither.
            let feedback = abandoned.broadcast(abandoned_commitment, Plan::Prepare { round });
            assert!(matches!(feedback, Feedback::Ok));
            let feedback = replacement.broadcast(replacement_commitment, Plan::Prepare { round });
            assert!(matches!(feedback, Feedback::Ok));
            while buffer.sends.lock().len() < 2 {
                reschedule().await;
            }
            assert!(
                marshal.get_verified(round).await.is_none(),
                "an early relay must not store a candidate"
            );

            // The lock-in stores the replacement without sending it again, and certification
            // resolves through the staged durability handshake.
            let feedback = replacement.broadcast(replacement_commitment, Plan::Propose { round });
            assert!(matches!(feedback, Feedback::Ok));
            assert!(
                replacement
                    .certify(round, replacement_commitment)
                    .await
                    .await
                    .expect("certify result missing"),
                "lock-in handshake must resolve certification durably"
            );
            assert_eq!(
                marshal
                    .get_verified(round)
                    .await
                    .map(|block| block.commitment()),
                Some(replacement_commitment),
                "lock-in must store the replacement"
            );
            assert!(
                marshal.get_block(&abandoned_digest).await.is_none(),
                "an abandoned candidate is never stored"
            );
            let sends = buffer.sends.lock();
            assert_eq!(sends.len(), 2, "each candidate is sent once");
            for ((sent_round, block, recipients), digest) in
                sends.iter().zip([abandoned_digest, replacement_digest])
            {
                assert_eq!(*sent_round, round);
                assert_eq!(block.digest(), digest);
                assert!(matches!(recipients, Recipients::All));
            }
        });
    }

    /// A candidate whose shards a prepare relay sent and that certification claimed before the
    /// lock-in goes out once. The propose relay that follows finds nothing staged and sends
    /// nothing.
    #[test_traced("WARN")]
    fn test_coding_propose_relay_after_claim_sends_once() {
        // The runner timeout is only a hang guard.
        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("validator"),
                "propose-relay-claim",
                provider.clone(),
                buffer.clone(),
            )
            .await;
            let shards = start_shard_mailbox(
                context.child("shards"),
                participants.clone(),
                provider.clone(),
            )
            .await;
            let (ctx, candidate) = missing_candidate(participants[0].clone());
            let round = ctx.round;
            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new()
                    .with_propose_result(candidate.inner().clone()),
                marshal: marshal.clone(),
                shards,
                scheme_provider: provider,
                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 should produce a commitment");
            assert_eq!(commitment, candidate.commitment());

            // The early relay sends the candidate's shards. Each lookup below goes through the
            // same marshal mailbox as the relays, so it observes every send they requested.
            let feedback = marshaled.broadcast(commitment, Plan::Prepare { round });
            assert!(matches!(feedback, Feedback::Ok));
            assert!(
                marshal.get_verified(round).await.is_none(),
                "an early relay must not store the candidate"
            );
            assert_eq!(buffer.sends.lock().len(), 1, "early relay must send once");

            // Certification claims the candidate and persists it.
            assert!(
                marshaled
                    .certify(round, commitment)
                    .await
                    .await
                    .expect("certify result missing"),
                "certification must persist the claimed candidate"
            );

            // The lock-in finds nothing staged and must not send the candidate again.
            let feedback = marshaled.broadcast(commitment, Plan::Propose { round });
            assert!(matches!(feedback, Feedback::Ok));
            assert_eq!(
                marshal
                    .get_verified(round)
                    .await
                    .map(|block| block.commitment()),
                Some(commitment),
                "certification must store the candidate"
            );
            assert_eq!(
                buffer.sends.lock().len(),
                1,
                "the lock-in must not send a claimed candidate again"
            );
        });
    }

    /// A prepare request builds on a parent that only our own vote may attest to, so it waits
    /// for the parent locally and draws no peer fetch. Local reconstruction or the parent's
    /// certification delivers it and completes the prepare. An ordinary propose fetches the
    /// parent by round.
    #[test_traced("WARN")]
    fn test_coding_prepare_waits_for_parent_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 buffer = RecordingCodingBuffer::default();
            let (marshal, resolver, _actor_handle) = start_coding_actor_with_recording(
                context.child("validator"),
                "prepare-parent-wait",
                provider.clone(),
                buffer.clone(),
            )
            .await;
            let shards = start_shard_mailbox(
                context.child("shards"),
                participants.clone(),
                provider.clone(),
            )
            .await;
            let coding_config = coding_config_for_participants(NUM_VALIDATORS as u16);
            let (parent_ctx, parent) = missing_candidate(participants[1].clone());
            let parent_round = parent_ctx.round;
            let ctx = CodingCtx {
                round: Round::new(Epoch::zero(), parent_round.view().next()),
                leader: participants[0].clone(),
                parent: (parent_round.view(), parent.commitment()),
            };
            let child = make_coding_block(ctx.clone(), parent.digest(), Height::new(2), 7);
            let child_commitment =
                encode_block::<_, ReedSolomon<Sha256>, Sha256>(child.clone(), coding_config)
                    .commitment();
            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new()
                    .with_handoff(Handoff::Stage(()))
                    .with_propose_result(child),
                marshal: marshal.clone(),
                shards,
                scheme_provider: provider,
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);

            // The actor issues any peer fetch for a missing block before it registers the local
            // wait, so a registered wait shows the prepare's parent lookup has been handled. The
            // runner timeout is only a hang guard for this loop.
            let decision = marshaled.prepare(ctx.clone()).await;
            while buffer.commitment_subscription_count() == 0 {
                reschedule().await;
            }
            assert!(
                resolver.fetches().is_empty(),
                "prepare must not fetch its parent"
            );

            // An ordinary propose on the same parent fetches it by round. The runner timeout is
            // only a hang guard for this loop.
            let proposal = marshaled.propose(ctx).await;
            while resolver.fetches().is_empty() {
                reschedule().await;
            }

            // Local delivery of the parent completes both requests without another fetch.
            assert!(marshal.verified(parent_round, parent).await);
            assert_eq!(
                decision.await.expect("prepare decision missing"),
                Handoff::Stage(child_commitment)
            );
            assert_eq!(
                proposal.await.expect("propose should produce a commitment"),
                child_commitment
            );
            let fetches = resolver.fetches();
            assert_eq!(fetches.len(), 1, "only the propose may fetch the parent");
            assert!(matches!(
                fetches[0].key,
                handler::Key::Notarized { round } if round == parent_round
            ));
        });
    }

    /// Coding validators vote on a block before reconstructing it, so a notarized or locally
    /// voted parent of a handoff can carry any height. A parent outside the request's epoch,
    /// up to the maximum height, yields no ancestry, as it can never certify, while a parent
    /// inside the epoch still builds.
    #[test_traced("WARN")]
    fn test_coding_prepare_skips_parent_outside_epoch() {
        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(),
                1024 * 1024,
            )
            .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 genesis = genesis_block();
            let genesis_parent_commitment = genesis_coding_commitment(&genesis);

            for (view, height, built) in [
                (1, Height::new(u64::MAX), false),
                (3, Height::new(3 * BLOCKS_PER_EPOCH.get()), false),
                (5, Height::new(1), true),
            ] {
                let parent_round = Round::new(Epoch::zero(), View::new(view));
                let parent_ctx = CodingCtx {
                    round: parent_round,
                    leader: participants[1].clone(),
                    parent: (View::zero(), genesis_parent_commitment),
                };
                let parent_block = make_coding_block(parent_ctx, genesis.digest(), height, view);
                let parent_digest = parent_block.digest();
                let parent = encode_block::<_, ReedSolomon<Sha256>, Sha256>(
                    parent_block.clone(),
                    coding_config,
                );
                let parent_commitment = parent.commitment();
                assert!(setup.mailbox.verified(parent_round, parent).await);

                let ctx = CodingCtx {
                    round: Round::new(Epoch::zero(), View::new(view + 1)),
                    leader: me.clone(),
                    parent: (View::new(view), parent_commitment),
                };
                let child = make_coding_block(ctx.clone(), parent_digest, Height::new(2), 7);
                let child_commitment =
                    encode_block::<_, ReedSolomon<Sha256>, Sha256>(child.clone(), coding_config)
                        .commitment();
                let cfg = MarshaledConfig {
                    application: MockVerifyingApp::<CodingB, S>::new()
                        .with_handoff(Handoff::Stage(()))
                        .with_propose_result(child),
                    marshal: setup.mailbox.clone(),
                    shards: setup.extra.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 decision = marshaled
                    .prepare(ctx)
                    .await
                    .await
                    .expect("prepare must return a decision");
                let expected = if built {
                    Handoff::Stage(child_commitment)
                } else {
                    Handoff::Wait
                };
                assert_eq!(decision, expected, "parent height {height:?}");
            }
        });
    }

    /// The genesis parent of a later epoch is the previous epoch's boundary block, whose
    /// height maps to the previous epoch. A handoff on that parent at view zero must still
    /// build the epoch's first block rather than decline as a parent outside the epoch.
    #[test_traced("WARN")]
    fn test_coding_prepare_builds_on_epoch_genesis() {
        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(),
                1024 * 1024,
            )
            .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 genesis = genesis_block();
            let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);

            // Store the last block of epoch 0, which is the genesis of epoch 1.
            let boundary_height = epocher.last(Epoch::zero()).expect("epoch 0 exists");
            let boundary_round = Round::new(Epoch::zero(), View::new(BLOCKS_PER_EPOCH.get()));
            let boundary_ctx = CodingCtx {
                round: boundary_round,
                leader: participants[1].clone(),
                parent: (View::zero(), genesis_coding_commitment(&genesis)),
            };
            let boundary_block =
                make_coding_block(boundary_ctx, genesis.digest(), boundary_height, 1);
            let boundary_digest = boundary_block.digest();
            let boundary = encode_block::<_, ReedSolomon<Sha256>, Sha256>(
                boundary_block.clone(),
                coding_config,
            );
            let boundary_commitment = boundary.commitment();
            assert!(setup.mailbox.verified(boundary_round, boundary).await);

            // The first proposal of epoch 1 names the boundary block at view zero.
            let ctx = CodingCtx {
                round: Round::new(Epoch::new(1), View::new(1)),
                leader: me.clone(),
                parent: (View::zero(), boundary_commitment),
            };
            let child = make_coding_block(ctx.clone(), boundary_digest, boundary_height.next(), 2);
            let child_commitment =
                encode_block::<_, ReedSolomon<Sha256>, Sha256>(child.clone(), coding_config)
                    .commitment();
            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new()
                    .with_handoff(Handoff::Stage(()))
                    .with_propose_result(child),
                marshal: setup.mailbox.clone(),
                shards: setup.extra.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher,
                strategy: Sequential,
            };
            let mut marshaled = Marshaled::new(context.child("marshaled"), cfg);
            let decision = marshaled
                .prepare(ctx)
                .await
                .await
                .expect("prepare must return a decision");
            assert_eq!(decision, Handoff::Stage(child_commitment));
        });
    }

    /// A leader that crashes between its relay broadcast and the journaling
    /// of its notarize vote holds a verified block for the round on restart.
    /// No vote names that block, so it cannot be notarized, and the restarted
    /// leader builds a fresh one on the recovered context whatever the stored
    /// block's own context says. The request is a handoff, which builds through
    /// the same checks as an ordinary proposal and answers under the
    /// application's decision. The relay persist stores the fresh block beside
    /// the stale one.
    ///
    /// Both blocks must then survive another restart. Reading them back
    /// afterwards checks storage itself, since the in-memory cache that serves
    /// a block right after it is sent does not survive the restart.
    #[test_traced("WARN")]
    fn test_prepare_rebuilds_despite_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(),
                1024 * 1024,
            )
            .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 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);

            // The stale block was built on a parent that replay has since
            // replaced, and its relay broadcast persisted it at the round.
            let round = Round::new(Epoch::zero(), View::new(2));
            let stale_ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (
                    View::new(1),
                    TestCommitment::from((
                        Sha256::hash(&[b"replaced-parent-block"]),
                        Sha256::hash(&[b"replaced-parent-inner"]),
                        Sha256::hash(&[b"replaced-parent-ctx"]),
                        coding_config,
                    )),
                ),
            };
            let stale = make_coding_block(stale_ctx, genesis.digest(), Height::new(1), 100);
            let stale_digest = stale.digest();
            let coded_stale: CodedBlock<_, ReedSolomon<Sha256>, Sha256> =
                encode_block(stale.clone(), coding_config);
            let stale_commitment = coded_stale.commitment();
            let (ack, persisted) = oneshot::channel();
            let _ = setup
                .mailbox
                .proposed(round, coded_stale, Recipients::All, ack);
            let sync = persisted.await.expect("stale block sync handle missing");
            assert!(sync.durable(round, "stale block").await);
            setup.crash().await;

            let setup = CodingHarness::setup_validator(
                context.child("restarted").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            let marshal = setup.mailbox.clone();
            assert_eq!(
                marshal
                    .get_verified(round)
                    .await
                    .map(|block| block.commitment()),
                Some(stale_commitment),
                "the restarted marshal must restore the stale block"
            );

            let ctx = CodingCtx {
                round,
                leader: me.clone(),
                parent: (View::zero(), genesis_parent_commitment),
            };
            let fresh = make_coding_block(ctx.clone(), genesis.digest(), Height::new(1), 200);
            let fresh_digest = fresh.digest();
            let fresh_commitment =
                encode_block::<_, ReedSolomon<Sha256>, Sha256>(fresh.clone(), coding_config)
                    .commitment();
            let mock_app: MockVerifyingApp<CodingB, S> = MockVerifyingApp::new()
                .with_handoff(Handoff::Vote(()))
                .with_propose_result(fresh);
            let cfg = MarshaledConfig {
                application: mock_app,
                marshal: marshal.clone(),
                shards: setup.extra.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 decision = marshaled
                .prepare(ctx)
                .await
                .await
                .expect("prepare must return a decision");
            assert_eq!(
                decision,
                Handoff::Vote(fresh_commitment),
                "a restarted leader must build a fresh block under the application's decision"
            );

            let _ = marshaled.broadcast(fresh_commitment, Plan::Propose { round });
            assert!(
                marshaled
                    .certify(round, fresh_commitment)
                    .await
                    .await
                    .expect("certify result missing"),
                "the fresh block must certify through the relay handshake"
            );
            assert_eq!(
                marshal
                    .get_verified(round)
                    .await
                    .map(|block| block.commitment()),
                Some(stale_commitment),
                "the stale block stays stored first at the round"
            );
            setup.crash().await;

            let setup = CodingHarness::setup_validator(
                context.child("recovered").with_attribute("index", 0),
                &mut oracle,
                me.clone(),
                ConstantProvider::new(schemes[0].clone()),
            )
            .await;
            for (label, digest) in [("stale", stale_digest), ("fresh", fresh_digest)] {
                assert!(
                    setup.mailbox.get_block(&digest).await.is_some(),
                    "the {label} block must be recoverable from storage"
                );
            }
            let cfg = MarshaledConfig {
                application: MockVerifyingApp::<CodingB, S>::new(),
                marshal: setup.mailbox.clone(),
                shards: setup.extra.clone(),
                scheme_provider: ConstantProvider::new(schemes[0].clone()),
                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
                strategy: Sequential,
            };
            let mut certifier = Marshaled::new(context.child("certifier"), cfg);
            assert!(
                certifier
                    .certify(round, fresh_commitment)
                    .await
                    .await
                    .expect("certify result missing"),
                "the fresh block must certify from storage after a restart"
            );
        });
    }

    /// 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 fetches the stored
    /// parent by its commitment on restart and re-proposes it, and the relay
    /// persist deduplicates against the pre-crash write.
    #[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(),
                1024 * 1024,
            )
            .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> =
                encode_block(boundary_block.clone(), coding_config);
            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");
                },
            }
        });
    }
}