harness.rs

//! Test harness for marshal variants.
//!
//! This module provides a trait-based abstraction that allows writing tests once
//! and running them against both the standard and coding marshal variants.

use crate::{
    marshal::{
        coding::{
            shards,
            types::{coding_config_for_participants, CodedBlock},
            Coding,
        },
        config::Config,
        core::{Actor, Mailbox},
        mocks::{application::Application, block::Block},
        resolver::p2p as resolver,
        standard::Standard,
        Identifier,
    },
    simplex::{
        scheme::bls12381_threshold::vrf as bls12381_threshold_vrf,
        types::{Activity, Context, Finalization, Finalize, Notarization, Notarize, Proposal},
    },
    types::{coding::Commitment, Epoch, Epocher, FixedEpocher, Height, Round, View, ViewDelta},
    Heightable, Reporter,
};
use commonware_broadcast::buffered;
use commonware_coding::{CodecConfig, ReedSolomon};
use commonware_cryptography::{
    bls12381::primitives::variant::MinPk,
    certificate::{mocks::Fixture, ConstantProvider, Provider, Scheme as _},
    ed25519::{PrivateKey, PublicKey},
    sha256::{Digest as Sha256Digest, Sha256},
    Committable, Digest as DigestTrait, Digestible, Hasher as _, Signer,
};
use commonware_p2p::simulated::{self, Link, Network, Oracle};
use commonware_parallel::Sequential;
use commonware_runtime::{buffer::paged::CacheRef, deterministic, Clock, Metrics, Quota, Runner};
use commonware_storage::{
    archive::{immutable, prunable},
    translator::EightCap,
};
use commonware_utils::{test_rng_seeded, vec::NonEmptyVec, NZUsize, NZU16, NZU64};
use futures::StreamExt;
use rand::{
    seq::{IteratorRandom, SliceRandom},
    Rng,
};
use std::{
    collections::BTreeMap,
    future::Future,
    num::{NonZeroU16, NonZeroU32, NonZeroU64, NonZeroUsize},
    time::{Duration, Instant},
};
use tracing::info;

// Common type aliases
pub type D = Sha256Digest;
pub type K = PublicKey;
pub type Ctx = Context<D, K>;
pub type B = Block<D, Ctx>;
pub type V = MinPk;
pub type S = bls12381_threshold_vrf::Scheme<K, V>;
pub type P = ConstantProvider<S, Epoch>;

// Coding variant type aliases (uses Commitment in context)
pub type CodingCtx = Context<Commitment, K>;
pub type CodingB = Block<D, CodingCtx>;

// Common test constants
pub const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
pub const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
pub const NAMESPACE: &[u8] = b"test";
pub const NUM_VALIDATORS: u32 = 4;
pub const QUORUM: u32 = 3;
pub const NUM_BLOCKS: u64 = 160;
pub const BLOCKS_PER_EPOCH: NonZeroU64 = NZU64!(20);
pub const LINK: Link = Link {
    latency: Duration::from_millis(100),
    jitter: Duration::from_millis(1),
    success_rate: 1.0,
};
pub const UNRELIABLE_LINK: Link = Link {
    latency: Duration::from_millis(200),
    jitter: Duration::from_millis(50),
    success_rate: 0.7,
};
pub const TEST_QUOTA: Quota = Quota::per_second(NonZeroU32::MAX);

/// A provider that always returns `None`, modeling an application that
/// has pruned all epoch state.
#[derive(Clone)]
pub struct EmptyProvider;

impl Provider for EmptyProvider {
    type Scope = Epoch;
    type Scheme = S;

    fn scoped(&self, _scope: Epoch) -> Option<std::sync::Arc<S>> {
        None
    }
}

/// Default leader key for tests.
pub fn default_leader() -> K {
    PrivateKey::from_seed(0).public_key()
}

/// Create a raw test block with a derived context.
pub fn make_raw_block(parent: D, height: Height, timestamp: u64) -> B {
    let parent_view = height
        .previous()
        .map(|h| View::new(h.get()))
        .unwrap_or(View::zero());
    let context = Ctx {
        round: Round::new(Epoch::zero(), View::new(height.get())),
        leader: default_leader(),
        parent: (parent_view, parent),
    };
    B::new::<Sha256>(context, parent, height, timestamp)
}

/// Setup network for tests with an initial participant peer set.
pub async fn setup_network_with_participants<I>(
    context: deterministic::Context,
    tracked_peer_sets: NonZeroUsize,
    participants: I,
) -> Oracle<K, deterministic::Context>
where
    I: IntoIterator<Item = K>,
{
    let (network, oracle) = Network::new_with_peers(
        context.with_label("network"),
        simulated::Config {
            max_size: 1024 * 1024,
            disconnect_on_block: true,
            tracked_peer_sets,
        },
        participants,
    )
    .await;
    network.start();
    oracle
}

/// Setup network links between peers.
pub async fn setup_network_links(
    oracle: &mut Oracle<K, deterministic::Context>,
    peers: &[K],
    link: Link,
) {
    for p1 in peers.iter() {
        for p2 in peers.iter() {
            if p2 == p1 {
                continue;
            }
            let _ = oracle.add_link(p1.clone(), p2.clone(), link.clone()).await;
        }
    }
}

/// Result of setting up a validator.
pub struct ValidatorSetup<H: TestHarness> {
    pub application: Application<H::ApplicationBlock>,
    pub mailbox: Mailbox<S, H::Variant>,
    pub extra: H::ValidatorExtra,
    pub height: Height,
    pub actor_handle: commonware_runtime::Handle<()>,
}

/// Per-validator handle for test operations.
pub struct ValidatorHandle<H: TestHarness> {
    pub mailbox: Mailbox<S, H::Variant>,
    pub extra: H::ValidatorExtra,
}

impl<H: TestHarness> Clone for ValidatorHandle<H> {
    fn clone(&self) -> Self {
        Self {
            mailbox: self.mailbox.clone(),
            extra: self.extra.clone(),
        }
    }
}

/// A test harness that abstracts over marshal variant differences.
pub trait TestHarness: 'static + Sized {
    /// The application block type.
    /// Note: We require `Digestible<Digest = D>` so generic test functions can use
    /// `subscribe_by_digest` which expects the block's digest type.
    type ApplicationBlock: crate::Block + Digestible<Digest = D> + Clone + Send + 'static;

    /// The marshal variant type.
    type Variant: crate::marshal::core::Variant<
        ApplicationBlock = Self::ApplicationBlock,
        Commitment = Self::Commitment,
    >;

    /// The block type used in test operations.
    type TestBlock: Heightable + Clone + Send;

    /// Additional per-validator state (e.g., shards mailbox for coding).
    type ValidatorExtra: Clone + Send;

    /// The commitment type for consensus certificates.
    type Commitment: DigestTrait;

    /// Setup a single validator with all necessary infrastructure.
    fn setup_validator(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
    ) -> impl Future<Output = ValidatorSetup<Self>> + Send;

    /// Setup a single validator with custom acknowledgement pipeline settings.
    fn setup_validator_with(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
        max_pending_acks: NonZeroUsize,
        application: Application<Self::ApplicationBlock>,
    ) -> impl Future<Output = ValidatorSetup<Self>> + Send;

    /// Create a test block from parent and height.
    fn genesis_parent_commitment(num_participants: u16) -> Self::Commitment;

    /// Create a test block from parent and height.
    fn make_test_block(
        parent: D,
        parent_commitment: Self::Commitment,
        height: Height,
        timestamp: u64,
        num_participants: u16,
    ) -> Self::TestBlock;

    /// Get the commitment from a test block.
    fn commitment(block: &Self::TestBlock) -> Self::Commitment;

    /// Get the digest from a test block.
    fn digest(block: &Self::TestBlock) -> D;

    /// Get the height from a test block.
    fn height(block: &Self::TestBlock) -> Height;

    /// Propose a block (broadcast to network).
    fn propose(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) -> impl Future<Output = ()> + Send;

    /// Mark a block as verified.
    fn verify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
        all_handles: &mut [ValidatorHandle<Self>],
    ) -> impl Future<Output = ()> + Send;

    /// Mark a block as certified.
    fn certify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) -> impl Future<Output = bool> + Send;

    /// Create a finalization certificate.
    fn make_finalization(
        proposal: Proposal<Self::Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Finalization<S, Self::Commitment>;

    /// Create a notarization certificate.
    fn make_notarization(
        proposal: Proposal<Self::Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Notarization<S, Self::Commitment>;

    /// Report a finalization to the mailbox.
    fn report_finalization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        finalization: Finalization<S, Self::Commitment>,
    ) -> impl Future<Output = ()> + Send;

    /// Report a notarization to the mailbox.
    fn report_notarization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        notarization: Notarization<S, Self::Commitment>,
    ) -> impl Future<Output = ()> + Send;

    /// Get the timeout duration for the finalize test.
    fn finalize_timeout() -> Duration;

    /// Setup validator for pruning test with prunable archives.
    #[allow(clippy::type_complexity)]
    fn setup_prunable_validator(
        context: deterministic::Context,
        oracle: &Oracle<K, deterministic::Context>,
        validator: K,
        schemes: &[S],
        partition_prefix: &str,
        page_cache: CacheRef,
    ) -> impl Future<
        Output = (
            Mailbox<S, Self::Variant>,
            Self::ValidatorExtra,
            Application<Self::ApplicationBlock>,
        ),
    > + Send;

    /// Verify a block for the pruning test (simpler than full verify).
    fn verify_for_prune(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) -> impl Future<Output = ()> + Send;
}

fn contract_runner(seed: u64) -> deterministic::Runner {
    deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(seed)
            .with_timeout(Some(Duration::from_secs(30))),
    )
}

fn restart_cycles_for_seed(seed: u64) -> usize {
    let mut rng = test_rng_seeded(seed);
    rng.gen_range(2..=4)
}

struct HailstormValidator<H: TestHarness> {
    application: Application<H::ApplicationBlock>,
    handle: ValidatorHandle<H>,
    actor_handle: commonware_runtime::Handle<()>,
}

type CanonicalEntry<H> = (Height, D, Finalization<S, <H as TestHarness>::Commitment>);
type CanonicalChain<H> = Vec<CanonicalEntry<H>>;

struct HailstormState<'a, H: TestHarness> {
    validators: &'a mut [Option<HailstormValidator<H>>],
    canonical: &'a mut CanonicalChain<H>,
    parent: &'a mut D,
    parent_commitment: &'a mut H::Commitment,
    participants: &'a [K],
    schemes: &'a [S],
    propagation_delay: Duration,
}

fn active_validator_indices<H: TestHarness>(
    validators: &[Option<HailstormValidator<H>>],
) -> Vec<usize> {
    validators
        .iter()
        .enumerate()
        .filter_map(|(idx, validator)| validator.as_ref().map(|_| idx))
        .collect()
}

async fn wait_for_validator_height<H: TestHarness>(
    context: &mut deterministic::Context,
    validator: &HailstormValidator<H>,
    height: Height,
    expected_digest: D,
    expected_finalization: &Finalization<S, H::Commitment>,
    label: &str,
) {
    loop {
        let block = validator.handle.mailbox.get_block(height).await;
        let finalization = validator.handle.mailbox.get_finalization(height).await;
        if let (Some(block), Some(finalization)) = (block, finalization) {
            assert_eq!(
                block.digest(),
                expected_digest,
                "{label}: wrong block digest at height {}",
                height.get()
            );
            assert_eq!(
                finalization.round(),
                expected_finalization.round(),
                "{label}: wrong finalization round at height {}",
                height.get()
            );
            assert_eq!(
                finalization.proposal.payload,
                expected_finalization.proposal.payload,
                "{label}: wrong finalization payload at height {}",
                height.get()
            );
            break;
        }
        context.sleep(Duration::from_millis(10)).await;
    }
}

async fn assert_validator_matches_canonical<H: TestHarness>(
    validator: &HailstormValidator<H>,
    canonical: &[CanonicalEntry<H>],
    label: &str,
) {
    let delivered = validator.application.blocks();
    for (height, block) in delivered {
        let (_, expected_digest, _) = canonical
            .iter()
            .find(|(expected_height, _, _)| *expected_height == height)
            .unwrap_or_else(|| {
                panic!(
                    "{label}: unexpected delivered block at height {}",
                    height.get()
                )
            });
        assert_eq!(
            block.digest(),
            *expected_digest,
            "{label}: application delivered wrong digest at height {}",
            height.get()
        );
    }

    if let Some((height, digest)) = validator.application.tip() {
        let (_, expected_digest, _) = canonical
            .iter()
            .find(|(expected_height, _, _)| *expected_height == height)
            .unwrap_or_else(|| {
                panic!(
                    "{label}: unexpected delivered tip at height {}",
                    height.get()
                )
            });
        assert_eq!(
            digest,
            *expected_digest,
            "{label}: application reported wrong tip digest at height {}",
            height.get()
        );
    }

    for (height, expected_digest, expected_finalization) in canonical {
        let stored_block = validator
            .handle
            .mailbox
            .get_block(*height)
            .await
            .unwrap_or_else(|| {
                panic!(
                    "{label}: missing finalized block at height {}",
                    height.get()
                )
            });
        assert_eq!(
            stored_block.digest(),
            *expected_digest,
            "{label}: stored wrong block digest at height {}",
            height.get()
        );

        let stored_finalization = validator
            .handle
            .mailbox
            .get_finalization(*height)
            .await
            .unwrap_or_else(|| panic!("{label}: missing finalization at height {}", height.get()));
        assert_eq!(
            stored_finalization.round(),
            expected_finalization.round(),
            "{label}: stored wrong finalization round at height {}",
            height.get()
        );
        assert_eq!(
            stored_finalization.proposal.payload,
            expected_finalization.proposal.payload,
            "{label}: stored wrong finalization payload at height {}",
            height.get()
        );
    }

    if let Some((height, digest, _)) = canonical.last() {
        assert_eq!(
            validator.handle.mailbox.get_info(Identifier::Latest).await,
            Some((*height, *digest)),
            "{label}: latest info should match the canonical tip",
        );
    }
}

async fn assert_active_validators_match_canonical<H: TestHarness>(
    validators: &[Option<HailstormValidator<H>>],
    canonical: &[CanonicalEntry<H>],
) {
    for idx in active_validator_indices(validators) {
        let validator = validators[idx]
            .as_ref()
            .expect("active validator should be present");
        assert_validator_matches_canonical(validator, canonical, &format!("validator_{idx}")).await;
    }
}

async fn advance_hailstorm_to<H: TestHarness>(
    target: u64,
    context: &mut deterministic::Context,
    state: &mut HailstormState<'_, H>,
) {
    for height_value in (state.canonical.len() as u64 + 1)..=target {
        let height = Height::new(height_value);
        let active = active_validator_indices(state.validators);
        let proposer_idx = active[context.gen_range(0..active.len())];
        let verifier_count = usize::min(QUORUM as usize, active.len());
        let verifier_indices = active
            .iter()
            .copied()
            .filter(|idx| *idx != proposer_idx)
            .choose_multiple(context, verifier_count.saturating_sub(1));
        let block = H::make_test_block(
            *state.parent,
            *state.parent_commitment,
            height,
            height_value,
            state.participants.len() as u16,
        );
        let round = Round::new(Epoch::zero(), View::new(height_value));
        let proposal = Proposal {
            round,
            parent: height
                .previous()
                .map(|previous| View::new(previous.get()))
                .unwrap_or(View::zero()),
            payload: H::commitment(&block),
        };
        let expected_digest = H::digest(&block);
        let finalization = H::make_finalization(proposal.clone(), state.schemes, QUORUM);

        {
            let proposer = state.validators[proposer_idx]
                .as_mut()
                .expect("proposer should be active");
            H::propose(&mut proposer.handle, round, &block).await;
            H::report_notarization(
                &mut proposer.handle.mailbox,
                H::make_notarization(proposal, state.schemes, QUORUM),
            )
            .await;
        }

        for verifier_idx in verifier_indices.iter().copied() {
            let verifier = state.validators[verifier_idx]
                .as_mut()
                .expect("verifier should be active");
            H::verify(&mut verifier.handle, round, &block, &mut []).await;
        }

        context.sleep(state.propagation_delay).await;

        for idx in active_validator_indices(state.validators) {
            let validator = state.validators[idx]
                .as_mut()
                .expect("validator should remain active");
            H::report_finalization(&mut validator.handle.mailbox, finalization.clone()).await;
        }

        state
            .canonical
            .push((height, expected_digest, finalization));
        *state.parent = expected_digest;
        *state.parent_commitment = H::commitment(&block);

        let (_, _, expected_finalization) = state
            .canonical
            .last()
            .expect("canonical chain should contain the new height");
        for idx in active_validator_indices(state.validators) {
            let validator = state.validators[idx]
                .as_ref()
                .expect("validator should be active");
            wait_for_validator_height(
                context,
                validator,
                height,
                expected_digest,
                expected_finalization,
                &format!("validator_{idx}"),
            )
            .await;
        }
    }

    assert_active_validators_match_canonical(state.validators, state.canonical).await;
}

/// Stress marshal with repeated validator crashes and recoveries while a
/// canonical finalized chain continues to advance.
pub fn hailstorm<H: TestHarness>(
    seed: u64,
    shutdowns: usize,
    interval: u64,
    max_down: usize,
    link: Link,
) -> String {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(seed)
            .with_timeout(Some(H::finalize_timeout())),
    );
    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.clone(), NZUsize!(3), participants.clone())
                .await;
        let propagation_delay = link.latency;
        setup_network_links(&mut oracle, &participants, link.clone()).await;

        let mut validators = Vec::new();
        for (idx, validator) in participants.iter().enumerate() {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{idx}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[idx].clone()),
            )
            .await;
            validators.push(Some(HailstormValidator::<H> {
                application: setup.application,
                handle: ValidatorHandle {
                    mailbox: setup.mailbox,
                    extra: setup.extra,
                },
                actor_handle: setup.actor_handle,
            }));
        }

        let mut canonical = CanonicalChain::<H>::new();
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let mut target_height = 0u64;
        let max_interval = interval.max(1);
        let max_down = max_down.max(1);

        for shutdown_idx in 0..shutdowns {
            let leadup = context.gen_range(1..=max_interval);
            target_height += leadup;
            let mut state = HailstormState {
                validators: &mut validators,
                canonical: &mut canonical,
                parent: &mut parent,
                parent_commitment: &mut parent_commitment,
                participants: &participants,
                schemes: &schemes,
                propagation_delay,
            };
            advance_hailstorm_to(target_height, &mut context, &mut state).await;

            let active = active_validator_indices(&validators);
            let down_limit = usize::min(max_down, active.len().saturating_sub(1));
            let down_count = down_limit.max(1);
            let down_count = context.gen_range(1..=down_count);
            let mut selected = active
                .iter()
                .copied()
                .choose_multiple(&mut context, down_count);
            selected.sort_unstable();
            let persisted_height = target_height;
            for idx in selected.iter().copied() {
                let crashed = validators[idx]
                    .take()
                    .expect("selected validator should be active");
                crashed.actor_handle.abort();
                let _ = crashed.actor_handle.await;
            }
            info!(
                seed,
                shutdown_idx,
                ?selected,
                down_count,
                persisted_height,
                leadup,
                "marshal hailstorm shutdown"
            );

            let downtime = context.gen_range(1..=max_interval);
            target_height += downtime;
            let mut state = HailstormState {
                validators: &mut validators,
                canonical: &mut canonical,
                parent: &mut parent,
                parent_commitment: &mut parent_commitment,
                participants: &participants,
                schemes: &schemes,
                propagation_delay,
            };
            advance_hailstorm_to(target_height, &mut context, &mut state).await;

            for idx in selected.iter().copied() {
                let restarted = H::setup_validator(
                    context.with_label(&format!("validator_{idx}_restart_{shutdown_idx}")),
                    &mut oracle,
                    participants[idx].clone(),
                    ConstantProvider::new(schemes[idx].clone()),
                )
                .await;
                assert_eq!(
                    restarted.height,
                    Height::new(persisted_height),
                    "validator {idx} should recover its persisted finalized height before replay"
                );

                let mut restarted = HailstormValidator::<H> {
                    application: restarted.application,
                    handle: ValidatorHandle {
                        mailbox: restarted.mailbox,
                        extra: restarted.extra,
                    },
                    actor_handle: restarted.actor_handle,
                };
                for (_, _, finalization) in canonical.iter().skip(persisted_height as usize) {
                    H::report_finalization(&mut restarted.handle.mailbox, finalization.clone())
                        .await;
                }
                validators[idx] = Some(restarted);
            }

            for idx in selected.iter().copied() {
                let validator = validators[idx]
                    .as_ref()
                    .expect("restarted validator should be active");
                for (height, digest, finalization) in canonical.iter() {
                    wait_for_validator_height(
                        &mut context,
                        validator,
                        *height,
                        *digest,
                        finalization,
                        &format!("validator_{idx}_restarted"),
                    )
                    .await;
                }
            }
            assert_active_validators_match_canonical(&validators, &canonical).await;
            info!(
                seed,
                shutdown_idx,
                ?selected,
                target_height,
                downtime,
                "marshal hailstorm recovered"
            );
        }

        context.auditor().state()
    })
}

/// Contract: `marshal.proposed(...)=true` means the block survives an
/// immediate crash and repeated recoveries.
pub fn proposed_success_implies_recoverable_after_restart<H: TestHarness>(
    seeds: impl IntoIterator<Item = u64>,
) {
    for seed in seeds {
        let Fixture {
            participants,
            schemes,
            ..
        } = bls12381_threshold_vrf::fixture::<V, _>(
            &mut test_rng_seeded(seed),
            NAMESPACE,
            NUM_VALIDATORS,
        );

        let me = participants[0].clone();
        let provider = ConstantProvider::new(schemes[0].clone());
        let round = Round::new(Epoch::zero(), View::new(1));
        let block = H::make_test_block(
            Sha256::hash(b""),
            H::genesis_parent_commitment(NUM_VALIDATORS as u16),
            Height::new(1),
            100,
            NUM_VALIDATORS as u16,
        );
        let digest = H::digest(&block);
        let recovery_cycles = restart_cycles_for_seed(seed);

        let (_, mut checkpoint) = contract_runner(seed).start_and_recover({
            let participants = participants.clone();
            let me = me.clone();
            let provider = provider.clone();
            let block = block.clone();
            move |context| async move {
                let mut oracle = setup_network_with_participants(
                    context.clone(),
                    NZUsize!(1),
                    participants.clone(),
                )
                .await;
                let setup = H::setup_validator(
                    context.with_label("validator_0"),
                    &mut oracle,
                    me.clone(),
                    provider.clone(),
                )
                .await;
                let mut handle = ValidatorHandle::<H> {
                    mailbox: setup.mailbox,
                    extra: setup.extra,
                };
                H::propose(&mut handle, round, &block).await;
            }
        });

        for cycle in 0..recovery_cycles {
            let ((), next_checkpoint) =
                deterministic::Runner::from(checkpoint).start_and_recover({
                    let participants = participants.clone();
                    let me = me.clone();
                    let provider = provider.clone();
                    move |context| async move {
                        let mut oracle = setup_network_with_participants(
                            context.clone(),
                            NZUsize!(1),
                            participants.clone(),
                        )
                        .await;
                        let restarted = H::setup_validator(
                            context.with_label(&format!("validator_0_restart_{cycle}")),
                            &mut oracle,
                            me.clone(),
                            provider.clone(),
                        )
                        .await;
                        assert!(
                        restarted.mailbox.get_block(&digest).await.is_some(),
                        "marshal.proposed() returning true must imply the block is recoverable \
                         after restart (seed={seed}, cycle={cycle})"
                    );
                    }
                });
            checkpoint = next_checkpoint;
        }
    }
}

/// Contract: `marshal.verified(...)=true` means the block survives an
/// immediate crash and repeated recoveries.
pub fn verified_success_implies_recoverable_after_restart<H: TestHarness>(
    seeds: impl IntoIterator<Item = u64>,
) {
    for seed in seeds {
        let Fixture {
            participants,
            schemes,
            ..
        } = bls12381_threshold_vrf::fixture::<V, _>(
            &mut test_rng_seeded(seed),
            NAMESPACE,
            NUM_VALIDATORS,
        );

        let me = participants[0].clone();
        let provider = ConstantProvider::new(schemes[0].clone());
        let round = Round::new(Epoch::zero(), View::new(1));
        let block = H::make_test_block(
            Sha256::hash(b""),
            H::genesis_parent_commitment(NUM_VALIDATORS as u16),
            Height::new(1),
            100,
            NUM_VALIDATORS as u16,
        );
        let digest = H::digest(&block);
        let recovery_cycles = restart_cycles_for_seed(seed);

        let (_, mut checkpoint) = contract_runner(seed).start_and_recover({
            let participants = participants.clone();
            let me = me.clone();
            let provider = provider.clone();
            let block = block.clone();
            move |context| async move {
                let mut oracle = setup_network_with_participants(
                    context.clone(),
                    NZUsize!(1),
                    participants.clone(),
                )
                .await;
                let setup = H::setup_validator(
                    context.with_label("validator_0"),
                    &mut oracle,
                    me.clone(),
                    provider.clone(),
                )
                .await;
                let mut handle = ValidatorHandle::<H> {
                    mailbox: setup.mailbox,
                    extra: setup.extra,
                };
                let mut peers: [ValidatorHandle<H>; 0] = [];
                H::verify(&mut handle, round, &block, &mut peers).await;
            }
        });

        for cycle in 0..recovery_cycles {
            let ((), next_checkpoint) =
                deterministic::Runner::from(checkpoint).start_and_recover({
                    let participants = participants.clone();
                    let me = me.clone();
                    let provider = provider.clone();
                    move |context| async move {
                        let mut oracle = setup_network_with_participants(
                            context.clone(),
                            NZUsize!(1),
                            participants.clone(),
                        )
                        .await;
                        let restarted = H::setup_validator(
                            context.with_label(&format!("validator_0_restart_{cycle}")),
                            &mut oracle,
                            me.clone(),
                            provider.clone(),
                        )
                        .await;
                        assert!(
                        restarted.mailbox.get_block(&digest).await.is_some(),
                        "marshal.verified() returning true must imply the block is recoverable \
                         after restart (seed={seed}, cycle={cycle})"
                    );
                    }
                });
            checkpoint = next_checkpoint;
        }
    }
}

/// Regression: when the same block is verified at an earlier view and later
/// certified at a much later view (epoch-boundary reproposal), both writes
/// must land so retention can prune the earlier view without losing the
/// block. A naive "skip the sibling write if the block's digest is already
/// present in the other archive" optimization is unsafe because the two
/// archives prune per-view on the same boundary: if the block lives only in
/// `verified_blocks[V_early]` and never gets written to
/// `notarized_blocks[V_late]`, advancing retention past V_early drops the
/// block even though V_late is still within the window.
pub fn certify_at_later_view_survives_earlier_view_pruning<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            participants[0].clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let application = setup.application;
        let mut handle = ValidatorHandle::<H> {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // A repeated block that we will verify at an early view and certify
        // at a later view. Its height is intentionally well beyond the chain
        // we'll drive below, so it never enters the finalized archive via
        // gap repair and lives solely in the prunable caches.
        let repeated = H::make_test_block(
            Sha256::hash(b""),
            H::genesis_parent_commitment(NUM_VALIDATORS as u16),
            Height::new(5_000),
            9_999,
            NUM_VALIDATORS as u16,
        );
        let repeated_digest = H::digest(&repeated);

        // Negative control: a verify-only block at the same early view. Because
        // it is never certified, it lives solely in `verified_blocks[V=1]` and
        // must disappear once retention pruning advances past V=1. Asserting it
        // is gone confirms the prune actually fires at the expected floor, so
        // the `repeated` survivor assertion below is genuinely load-bearing.
        let orphan = H::make_test_block(
            Sha256::hash(b"orphan"),
            H::genesis_parent_commitment(NUM_VALIDATORS as u16),
            Height::new(6_000),
            9_998,
            NUM_VALIDATORS as u16,
        );
        let orphan_digest = H::digest(&orphan);

        // Verify `repeated` at V=1, then certify at V=25 (reproposal-style gap).
        let v_early = Round::new(Epoch::zero(), View::new(1));
        let v_late = Round::new(Epoch::zero(), View::new(25));
        let mut peers: [ValidatorHandle<H>; 0] = [];
        H::verify(&mut handle, v_early, &repeated, &mut peers).await;
        assert!(
            H::certify(&mut handle, v_late, &repeated).await,
            "certify must ack"
        );

        // Verify `orphan` at V=1 only (no certify).
        H::verify(&mut handle, v_early, &orphan, &mut peers).await;

        // Drive the finalized chain forward to advance `last_processed_round`
        // past V=1's retention boundary but not past V=25's. With
        // view_retention_timeout=10 and prunable_items_per_section=10,
        // processing views 1..=21 leaves `oldest_allowed=10` in both prunable
        // archives. V=1 is dropped, V=25 is retained.
        const CHAIN_LEN: u64 = 21;
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(NUM_VALIDATORS as u16);
        for i in 1..=CHAIN_LEN {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                NUM_VALIDATORS as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::zero(), View::new(i));
            H::propose(&mut handle, round, &block).await;
            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;
            parent = digest;
            parent_commitment = commitment;
        }
        while (application.blocks().len() as u64) < CHAIN_LEN {
            context.sleep(Duration::from_millis(10)).await;
        }
        context.sleep(Duration::from_millis(100)).await;

        // Negative control: the verify-only orphan at V=1 must be gone, which
        // proves retention pruning actually evicted V=1 at the expected floor.
        assert!(
            handle.mailbox.get_block(&orphan_digest).await.is_none(),
            "verify-only block at V=1 must be evicted by retention pruning"
        );

        // The repeated block must still be retrievable: verified_blocks[V=1]
        // has been pruned, but notarized_blocks[V=25] still holds it.
        let recovered = handle.mailbox.get_block(&repeated_digest).await;
        assert!(
            recovered.is_some(),
            "block certified at V=25 must survive retention pruning of V=1"
        );
        assert_eq!(recovered.unwrap().digest(), repeated_digest);
    });
}

/// Regression: when a leader equivocates, a validator may verify one block
/// (A) and then certify a different block (B) at the same round. `verified()`
/// and `certified()` must write to distinct archives so both blocks are
/// retained and retrievable; otherwise the second write collides on the same
/// prunable-archive index (`skip_if_index_exists=true`) and is silently
/// dropped despite the mailbox returning success.
pub fn certify_persists_equivocated_block<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            participants[0].clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle::<H> {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        let round = Round::new(Epoch::zero(), View::new(1));
        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(NUM_VALIDATORS as u16);

        // Two distinct blocks at the same height/round (leader equivocation):
        // distinct timestamps yield distinct digests.
        let block_a = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            NUM_VALIDATORS as u16,
        );
        let digest_a = H::digest(&block_a);
        let block_b = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            2,
            NUM_VALIDATORS as u16,
        );
        let digest_b = H::digest(&block_b);
        assert_ne!(digest_a, digest_b, "test requires distinct digests");

        let mut peers: [ValidatorHandle<H>; 0] = [];
        H::verify(&mut handle, round, &block_a, &mut peers).await;
        assert!(
            H::certify(&mut handle, round, &block_b).await,
            "certified must ack"
        );

        let got_a = handle.mailbox.get_block(&digest_a).await;
        assert!(
            got_a.is_some(),
            "verified block A must be persisted in verified_blocks"
        );
        assert_eq!(got_a.unwrap().digest(), digest_a);
        let got_b = handle.mailbox.get_block(&digest_b).await;
        assert!(
            got_b.is_some(),
            "certified block B must be persisted despite a verify at the same round"
        );
        assert_eq!(got_b.unwrap().digest(), digest_b);
    });
}

/// Contract: once marshal has delivered a finalized block to the application,
/// that finalized block and its certificate must already be durable.
pub fn delivery_visibility_implies_recoverable_after_restart<H: TestHarness>(
    seeds: impl IntoIterator<Item = u64>,
) {
    for seed in seeds {
        let Fixture {
            participants,
            schemes,
            ..
        } = bls12381_threshold_vrf::fixture::<V, _>(
            &mut test_rng_seeded(seed),
            NAMESPACE,
            NUM_VALIDATORS,
        );

        let me = participants[0].clone();
        let provider = ConstantProvider::new(schemes[0].clone());
        let application = Application::<H::ApplicationBlock>::manual_ack();
        let round = Round::new(Epoch::zero(), View::new(1));
        let block = H::make_test_block(
            Sha256::hash(b""),
            H::genesis_parent_commitment(NUM_VALIDATORS as u16),
            Height::new(1),
            100,
            NUM_VALIDATORS as u16,
        );
        let finalization = H::make_finalization(
            Proposal::new(round, View::zero(), H::commitment(&block)),
            &schemes,
            QUORUM,
        );
        let recovery_cycles = restart_cycles_for_seed(seed);

        let (_, mut checkpoint) = contract_runner(seed).start_and_recover({
            let participants = participants.clone();
            let me = me.clone();
            let provider = provider.clone();
            let application = application.clone();
            let block = block.clone();
            let finalization = finalization.clone();
            move |context| async move {
                let mut oracle = setup_network_with_participants(
                    context.clone(),
                    NZUsize!(1),
                    participants.clone(),
                )
                .await;
                let setup = H::setup_validator_with(
                    context.with_label("validator_0"),
                    &mut oracle,
                    me.clone(),
                    provider.clone(),
                    NZUsize!(1),
                    application.clone(),
                )
                .await;
                let mut mailbox = setup.mailbox;
                let mut handle = ValidatorHandle::<H> {
                    mailbox: mailbox.clone(),
                    extra: setup.extra,
                };
                let mut peers: [ValidatorHandle<H>; 0] = [];
                H::verify(&mut handle, round, &block, &mut peers).await;
                H::report_finalization(&mut mailbox, finalization.clone()).await;

                let height = application.acknowledged().await;
                assert_eq!(
                    height,
                    Height::new(1),
                    "expected the first delivered finalized block to become visible at height 1 \
                     before restart (seed={seed})"
                );
            }
        });

        for cycle in 0..recovery_cycles {
            let expected_round = finalization.round();
            let ((), next_checkpoint) =
                deterministic::Runner::from(checkpoint).start_and_recover({
                    let participants = participants.clone();
                    let me = me.clone();
                    let provider = provider.clone();
                    move |context| async move {
                        let mut oracle = setup_network_with_participants(
                            context.clone(),
                            NZUsize!(1),
                            participants.clone(),
                        )
                        .await;
                        let restarted = H::setup_validator(
                            context.with_label(&format!("validator_0_restart_{cycle}")),
                            &mut oracle,
                            me.clone(),
                            provider.clone(),
                        )
                        .await;
                        let recovered = restarted.mailbox.get_block(Height::new(1)).await.expect(
                            "delivered finalized block must be recoverable after restart \
                             (seed={seed}, cycle={cycle})",
                        );
                        assert_eq!(
                            recovered.height(),
                            Height::new(1),
                            "restart should recover the delivered finalized block by height \
                         (seed={seed}, cycle={cycle})"
                        );
                        assert_eq!(
                            restarted
                                .mailbox
                                .get_finalization(Height::new(1))
                                .await
                                .expect(
                                    "delivered finalization must be recoverable after restart \
                                 (seed={seed}, cycle={cycle})",
                                )
                                .round(),
                            expected_round,
                            "restart should recover the delivered finalization by height \
                         (seed={seed}, cycle={cycle})"
                        );
                    }
                });
            checkpoint = next_checkpoint;
        }
    }
}

// =============================================================================
// Standard Harness Implementation
// =============================================================================

/// Standard variant test harness.
pub struct StandardHarness;

impl TestHarness for StandardHarness {
    type ApplicationBlock = B;
    type Variant = Standard<B>;
    type TestBlock = B;
    type ValidatorExtra = buffered::Mailbox<K, B>;
    type Commitment = D;

    async fn setup_validator(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
    ) -> ValidatorSetup<Self> {
        Self::setup_validator_with(
            context,
            oracle,
            validator,
            provider,
            NZUsize!(1),
            Application::default(),
        )
        .await
    }

    async fn setup_validator_with(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
        max_pending_acks: NonZeroUsize,
        application: Application<Self::ApplicationBlock>,
    ) -> ValidatorSetup<Self> {
        let config = Config {
            provider,
            epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
            mailbox_size: 100,
            view_retention_timeout: ViewDelta::new(10),
            max_repair: NZUsize!(10),
            max_pending_acks,
            block_codec_config: (),
            partition_prefix: format!("validator-{}", validator.clone()),
            prunable_items_per_section: NZU64!(10),
            replay_buffer: NZUsize!(1024),
            key_write_buffer: NZUsize!(1024),
            value_write_buffer: NZUsize!(1024),
            page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
            strategy: Sequential,
        };
        let control = oracle.control(validator.clone());
        let backfill = control.register(1, TEST_QUOTA).await.unwrap();
        let resolver_cfg = resolver::Config {
            public_key: validator.clone(),
            peer_provider: oracle.manager(),
            blocker: oracle.control(validator.clone()),
            mailbox_size: config.mailbox_size,
            initial: Duration::from_secs(1),
            timeout: Duration::from_secs(2),
            fetch_retry_timeout: Duration::from_millis(100),
            priority_requests: false,
            priority_responses: false,
        };
        let resolver = resolver::init(&context, resolver_cfg, backfill);

        let broadcast_config = buffered::Config {
            public_key: validator.clone(),
            mailbox_size: config.mailbox_size,
            deque_size: 10,
            priority: false,
            codec_config: (),
            peer_provider: oracle.manager(),
        };
        let (broadcast_engine, buffer) = buffered::Engine::new(context.clone(), broadcast_config);
        let network = control.register(2, TEST_QUOTA).await.unwrap();
        broadcast_engine.start(network);

        let start = Instant::now();
        let finalizations_by_height = immutable::Archive::init(
            context.with_label("finalizations_by_height"),
            immutable::Config {
                metadata_partition: format!(
                    "{}-finalizations-by-height-metadata",
                    config.partition_prefix
                ),
                freezer_table_partition: format!(
                    "{}-finalizations-by-height-freezer-table",
                    config.partition_prefix
                ),
                freezer_table_initial_size: 64,
                freezer_table_resize_frequency: 10,
                freezer_table_resize_chunk_size: 10,
                freezer_key_partition: format!(
                    "{}-finalizations-by-height-freezer-key",
                    config.partition_prefix
                ),
                freezer_key_page_cache: config.page_cache.clone(),
                freezer_value_partition: format!(
                    "{}-finalizations-by-height-freezer-value",
                    config.partition_prefix
                ),
                freezer_value_target_size: 1024,
                freezer_value_compression: None,
                ordinal_partition: format!(
                    "{}-finalizations-by-height-ordinal",
                    config.partition_prefix
                ),
                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");
        info!(elapsed = ?start.elapsed(), "restored finalizations by height archive");

        let start = Instant::now();
        let finalized_blocks = immutable::Archive::init(
            context.with_label("finalized_blocks"),
            immutable::Config {
                metadata_partition: format!(
                    "{}-finalized_blocks-metadata",
                    config.partition_prefix
                ),
                freezer_table_partition: format!(
                    "{}-finalized_blocks-freezer-table",
                    config.partition_prefix
                ),
                freezer_table_initial_size: 64,
                freezer_table_resize_frequency: 10,
                freezer_table_resize_chunk_size: 10,
                freezer_key_partition: format!(
                    "{}-finalized_blocks-freezer-key",
                    config.partition_prefix
                ),
                freezer_key_page_cache: config.page_cache.clone(),
                freezer_value_partition: format!(
                    "{}-finalized_blocks-freezer-value",
                    config.partition_prefix
                ),
                freezer_value_target_size: 1024,
                freezer_value_compression: None,
                ordinal_partition: format!("{}-finalized_blocks-ordinal", config.partition_prefix),
                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");
        info!(elapsed = ?start.elapsed(), "restored finalized blocks archive");

        let (actor, mailbox, height) = Actor::init(
            context.clone(),
            finalizations_by_height,
            finalized_blocks,
            config,
        )
        .await;
        let actor_handle = actor.start(application.clone(), buffer.clone(), resolver);

        ValidatorSetup {
            application,
            mailbox,
            extra: buffer,
            height,
            actor_handle,
        }
    }

    fn genesis_parent_commitment(_num_participants: u16) -> D {
        Sha256::hash(b"")
    }

    fn make_test_block(
        parent: D,
        _parent_commitment: D,
        height: Height,
        timestamp: u64,
        _num_participants: u16,
    ) -> B {
        make_raw_block(parent, height, timestamp)
    }

    fn commitment(block: &B) -> D {
        block.digest()
    }

    fn digest(block: &B) -> D {
        block.digest()
    }

    fn height(block: &B) -> Height {
        block.height()
    }

    async fn propose(handle: &mut ValidatorHandle<Self>, round: Round, block: &B) {
        assert!(handle.mailbox.proposed(round, block.clone()).await);
    }

    async fn verify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &B,
        _all_handles: &mut [ValidatorHandle<Self>],
    ) {
        assert!(handle.mailbox.verified(round, block.clone()).await);
    }

    async fn certify(handle: &mut ValidatorHandle<Self>, round: Round, block: &B) -> bool {
        handle.mailbox.certified(round, block.clone()).await
    }

    fn make_finalization(proposal: Proposal<D>, schemes: &[S], quorum: u32) -> Finalization<S, D> {
        let finalizes: Vec<_> = schemes
            .iter()
            .take(quorum as usize)
            .map(|scheme| Finalize::sign(scheme, proposal.clone()).unwrap())
            .collect();
        Finalization::from_finalizes(&schemes[0], &finalizes, &Sequential).unwrap()
    }

    fn make_notarization(proposal: Proposal<D>, schemes: &[S], quorum: u32) -> Notarization<S, D> {
        let notarizes: Vec<_> = schemes
            .iter()
            .take(quorum as usize)
            .map(|scheme| Notarize::sign(scheme, proposal.clone()).unwrap())
            .collect();
        Notarization::from_notarizes(&schemes[0], &notarizes, &Sequential).unwrap()
    }

    async fn report_finalization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        finalization: Finalization<S, D>,
    ) {
        mailbox.report(Activity::Finalization(finalization)).await;
    }

    async fn report_notarization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        notarization: Notarization<S, D>,
    ) {
        mailbox.report(Activity::Notarization(notarization)).await;
    }

    fn finalize_timeout() -> Duration {
        Duration::from_secs(600)
    }

    async fn setup_prunable_validator(
        context: deterministic::Context,
        oracle: &Oracle<K, deterministic::Context>,
        validator: K,
        schemes: &[S],
        partition_prefix: &str,
        page_cache: CacheRef,
    ) -> (
        Mailbox<S, Self::Variant>,
        Self::ValidatorExtra,
        Application<B>,
    ) {
        let control = oracle.control(validator.clone());
        let provider = ConstantProvider::new(schemes[0].clone());
        let config = Config {
            provider,
            epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
            mailbox_size: 100,
            view_retention_timeout: 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: page_cache.clone(),
            strategy: Sequential,
        };

        let backfill = control.register(0, TEST_QUOTA).await.unwrap();
        let resolver_cfg = resolver::Config {
            public_key: validator.clone(),
            peer_provider: oracle.manager(),
            blocker: control.clone(),
            mailbox_size: config.mailbox_size,
            initial: Duration::from_secs(1),
            timeout: Duration::from_secs(2),
            fetch_retry_timeout: Duration::from_millis(100),
            priority_requests: false,
            priority_responses: false,
        };
        let resolver = resolver::init(&context, resolver_cfg, backfill);

        let broadcast_config = buffered::Config {
            public_key: validator.clone(),
            mailbox_size: config.mailbox_size,
            deque_size: 10,
            priority: false,
            codec_config: (),
            peer_provider: oracle.manager(),
        };
        let (broadcast_engine, buffer) = buffered::Engine::new(context.clone(), broadcast_config);
        let network = control.register(1, TEST_QUOTA).await.unwrap();
        broadcast_engine.start(network);

        let finalizations_by_height = prunable::Archive::init(
            context.with_label("finalizations_by_height"),
            prunable::Config {
                translator: EightCap,
                key_partition: format!("{}-finalizations-by-height-key", partition_prefix),
                key_page_cache: page_cache.clone(),
                value_partition: format!("{}-finalizations-by-height-value", partition_prefix),
                compression: None,
                codec_config: S::certificate_codec_config_unbounded(),
                items_per_section: NZU64!(10),
                key_write_buffer: config.key_write_buffer,
                value_write_buffer: config.value_write_buffer,
                replay_buffer: config.replay_buffer,
            },
        )
        .await
        .expect("failed to initialize finalizations by height archive");

        let finalized_blocks = prunable::Archive::init(
            context.with_label("finalized_blocks"),
            prunable::Config {
                translator: EightCap,
                key_partition: format!("{}-finalized-blocks-key", partition_prefix),
                key_page_cache: page_cache.clone(),
                value_partition: format!("{}-finalized-blocks-value", partition_prefix),
                compression: None,
                codec_config: config.block_codec_config,
                items_per_section: NZU64!(10),
                key_write_buffer: config.key_write_buffer,
                value_write_buffer: config.value_write_buffer,
                replay_buffer: config.replay_buffer,
            },
        )
        .await
        .expect("failed to initialize finalized blocks archive");

        let (actor, mailbox, _) = Actor::init(
            context.clone(),
            finalizations_by_height,
            finalized_blocks,
            config,
        )
        .await;
        let application = Application::<B>::default();
        actor.start(application.clone(), buffer.clone(), resolver);

        (mailbox, buffer, application)
    }

    async fn verify_for_prune(handle: &mut ValidatorHandle<Self>, round: Round, block: &B) {
        assert!(handle.mailbox.verified(round, block.clone()).await);
    }
}

/// Inline wrapper test harness for standard marshal behavior.
pub struct InlineHarness;

impl TestHarness for InlineHarness {
    type ApplicationBlock = <StandardHarness as TestHarness>::ApplicationBlock;
    type Variant = <StandardHarness as TestHarness>::Variant;
    type TestBlock = <StandardHarness as TestHarness>::TestBlock;
    type ValidatorExtra = <StandardHarness as TestHarness>::ValidatorExtra;
    type Commitment = <StandardHarness as TestHarness>::Commitment;

    async fn setup_validator(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
    ) -> ValidatorSetup<Self> {
        let setup = StandardHarness::setup_validator(context, oracle, validator, provider).await;
        ValidatorSetup {
            application: setup.application,
            mailbox: setup.mailbox,
            extra: setup.extra,
            height: setup.height,
            actor_handle: setup.actor_handle,
        }
    }

    async fn setup_validator_with(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
        max_pending_acks: NonZeroUsize,
        application: Application<Self::ApplicationBlock>,
    ) -> ValidatorSetup<Self> {
        let setup = StandardHarness::setup_validator_with(
            context,
            oracle,
            validator,
            provider,
            max_pending_acks,
            application,
        )
        .await;
        ValidatorSetup {
            application: setup.application,
            mailbox: setup.mailbox,
            extra: setup.extra,
            height: setup.height,
            actor_handle: setup.actor_handle,
        }
    }

    fn genesis_parent_commitment(num_participants: u16) -> Self::Commitment {
        StandardHarness::genesis_parent_commitment(num_participants)
    }

    fn make_test_block(
        parent: D,
        parent_commitment: Self::Commitment,
        height: Height,
        timestamp: u64,
        num_participants: u16,
    ) -> Self::TestBlock {
        StandardHarness::make_test_block(
            parent,
            parent_commitment,
            height,
            timestamp,
            num_participants,
        )
    }

    fn commitment(block: &Self::TestBlock) -> Self::Commitment {
        StandardHarness::commitment(block)
    }

    fn digest(block: &Self::TestBlock) -> D {
        StandardHarness::digest(block)
    }

    fn height(block: &Self::TestBlock) -> Height {
        StandardHarness::height(block)
    }

    async fn propose(handle: &mut ValidatorHandle<Self>, round: Round, block: &Self::TestBlock) {
        StandardHarness::propose(
            &mut ValidatorHandle::<StandardHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
        )
        .await;
    }

    async fn verify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
        _all_handles: &mut [ValidatorHandle<Self>],
    ) {
        StandardHarness::verify(
            &mut ValidatorHandle::<StandardHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
            &mut [],
        )
        .await;
    }

    async fn certify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) -> bool {
        StandardHarness::certify(
            &mut ValidatorHandle::<StandardHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
        )
        .await
    }

    fn make_finalization(
        proposal: Proposal<Self::Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Finalization<S, Self::Commitment> {
        StandardHarness::make_finalization(proposal, schemes, quorum)
    }

    fn make_notarization(
        proposal: Proposal<Self::Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Notarization<S, Self::Commitment> {
        StandardHarness::make_notarization(proposal, schemes, quorum)
    }

    async fn report_finalization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        finalization: Finalization<S, Self::Commitment>,
    ) {
        StandardHarness::report_finalization(mailbox, finalization).await;
    }

    async fn report_notarization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        notarization: Notarization<S, Self::Commitment>,
    ) {
        StandardHarness::report_notarization(mailbox, notarization).await;
    }

    fn finalize_timeout() -> Duration {
        StandardHarness::finalize_timeout()
    }

    async fn setup_prunable_validator(
        context: deterministic::Context,
        oracle: &Oracle<K, deterministic::Context>,
        validator: K,
        schemes: &[S],
        partition_prefix: &str,
        page_cache: CacheRef,
    ) -> (
        Mailbox<S, Self::Variant>,
        Self::ValidatorExtra,
        Application<Self::ApplicationBlock>,
    ) {
        StandardHarness::setup_prunable_validator(
            context,
            oracle,
            validator,
            schemes,
            partition_prefix,
            page_cache,
        )
        .await
    }

    async fn verify_for_prune(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) {
        StandardHarness::verify_for_prune(
            &mut ValidatorHandle::<StandardHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
        )
        .await;
    }
}

/// Deferred wrapper test harness for standard marshal behavior.
pub struct DeferredHarness;

impl TestHarness for DeferredHarness {
    type ApplicationBlock = <InlineHarness as TestHarness>::ApplicationBlock;
    type Variant = <InlineHarness as TestHarness>::Variant;
    type TestBlock = <InlineHarness as TestHarness>::TestBlock;
    type ValidatorExtra = <InlineHarness as TestHarness>::ValidatorExtra;
    type Commitment = <InlineHarness as TestHarness>::Commitment;

    async fn setup_validator(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
    ) -> ValidatorSetup<Self> {
        let setup = InlineHarness::setup_validator(context, oracle, validator, provider).await;
        ValidatorSetup {
            application: setup.application,
            mailbox: setup.mailbox,
            extra: setup.extra,
            height: setup.height,
            actor_handle: setup.actor_handle,
        }
    }

    async fn setup_validator_with(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
        max_pending_acks: NonZeroUsize,
        application: Application<Self::ApplicationBlock>,
    ) -> ValidatorSetup<Self> {
        let setup = InlineHarness::setup_validator_with(
            context,
            oracle,
            validator,
            provider,
            max_pending_acks,
            application,
        )
        .await;
        ValidatorSetup {
            application: setup.application,
            mailbox: setup.mailbox,
            extra: setup.extra,
            height: setup.height,
            actor_handle: setup.actor_handle,
        }
    }

    fn genesis_parent_commitment(num_participants: u16) -> Self::Commitment {
        InlineHarness::genesis_parent_commitment(num_participants)
    }

    fn make_test_block(
        parent: D,
        parent_commitment: Self::Commitment,
        height: Height,
        timestamp: u64,
        num_participants: u16,
    ) -> Self::TestBlock {
        InlineHarness::make_test_block(
            parent,
            parent_commitment,
            height,
            timestamp,
            num_participants,
        )
    }

    fn commitment(block: &Self::TestBlock) -> Self::Commitment {
        InlineHarness::commitment(block)
    }

    fn digest(block: &Self::TestBlock) -> D {
        InlineHarness::digest(block)
    }

    fn height(block: &Self::TestBlock) -> Height {
        InlineHarness::height(block)
    }

    async fn propose(handle: &mut ValidatorHandle<Self>, round: Round, block: &Self::TestBlock) {
        InlineHarness::propose(
            &mut ValidatorHandle::<InlineHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
        )
        .await;
    }

    async fn verify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
        _all_handles: &mut [ValidatorHandle<Self>],
    ) {
        InlineHarness::verify(
            &mut ValidatorHandle::<InlineHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
            &mut [],
        )
        .await;
    }

    async fn certify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) -> bool {
        InlineHarness::certify(
            &mut ValidatorHandle::<InlineHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
        )
        .await
    }

    fn make_finalization(
        proposal: Proposal<Self::Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Finalization<S, Self::Commitment> {
        InlineHarness::make_finalization(proposal, schemes, quorum)
    }

    fn make_notarization(
        proposal: Proposal<Self::Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Notarization<S, Self::Commitment> {
        InlineHarness::make_notarization(proposal, schemes, quorum)
    }

    async fn report_finalization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        finalization: Finalization<S, Self::Commitment>,
    ) {
        InlineHarness::report_finalization(mailbox, finalization).await;
    }

    async fn report_notarization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        notarization: Notarization<S, Self::Commitment>,
    ) {
        InlineHarness::report_notarization(mailbox, notarization).await;
    }

    fn finalize_timeout() -> Duration {
        InlineHarness::finalize_timeout()
    }

    async fn setup_prunable_validator(
        context: deterministic::Context,
        oracle: &Oracle<K, deterministic::Context>,
        validator: K,
        schemes: &[S],
        partition_prefix: &str,
        page_cache: CacheRef,
    ) -> (
        Mailbox<S, Self::Variant>,
        Self::ValidatorExtra,
        Application<Self::ApplicationBlock>,
    ) {
        InlineHarness::setup_prunable_validator(
            context,
            oracle,
            validator,
            schemes,
            partition_prefix,
            page_cache,
        )
        .await
    }

    async fn verify_for_prune(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &Self::TestBlock,
    ) {
        InlineHarness::verify_for_prune(
            &mut ValidatorHandle::<InlineHarness> {
                mailbox: handle.mailbox.clone(),
                extra: handle.extra.clone(),
            },
            round,
            block,
        )
        .await;
    }
}

// =============================================================================
// Coding Harness Implementation
// =============================================================================

/// Coding variant test harness.
pub struct CodingHarness;

type CodingVariant = Coding<CodingB, ReedSolomon<Sha256>, Sha256, K>;
type ShardsMailbox = shards::Mailbox<CodingB, ReedSolomon<Sha256>, Sha256, K>;

/// Genesis blocks use a special coding config that doesn't actually encode.
pub const GENESIS_CODING_CONFIG: commonware_coding::Config = commonware_coding::Config {
    minimum_shards: NZU16!(1),
    extra_shards: NZU16!(1),
};

/// Create a genesis Commitment (all zeros for digests, genesis config).
pub fn genesis_commitment() -> Commitment {
    Commitment::from((
        D::EMPTY,
        D::EMPTY,
        Sha256Digest::EMPTY,
        GENESIS_CODING_CONFIG,
    ))
}

/// Create a test block with a Commitment-based context.
pub fn make_coding_block(context: CodingCtx, parent: D, height: Height, timestamp: u64) -> CodingB {
    CodingB::new::<Sha256>(context, parent, height, timestamp)
}

impl TestHarness for CodingHarness {
    type ApplicationBlock = CodingB;
    type Variant = CodingVariant;
    type TestBlock = CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>;
    type ValidatorExtra = ShardsMailbox;
    type Commitment = Commitment;

    async fn setup_validator(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
    ) -> ValidatorSetup<Self> {
        Self::setup_validator_with(
            context,
            oracle,
            validator,
            provider,
            NZUsize!(1),
            Application::default(),
        )
        .await
    }

    async fn setup_validator_with(
        context: deterministic::Context,
        oracle: &mut Oracle<K, deterministic::Context>,
        validator: K,
        provider: P,
        max_pending_acks: NonZeroUsize,
        application: Application<Self::ApplicationBlock>,
    ) -> ValidatorSetup<Self> {
        let config = Config {
            provider: provider.clone(),
            epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
            mailbox_size: 100,
            view_retention_timeout: ViewDelta::new(10),
            max_repair: NZUsize!(10),
            max_pending_acks,
            block_codec_config: (),
            partition_prefix: format!("validator-{}", validator.clone()),
            prunable_items_per_section: NZU64!(10),
            replay_buffer: NZUsize!(1024),
            key_write_buffer: NZUsize!(1024),
            value_write_buffer: NZUsize!(1024),
            page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
            strategy: Sequential,
        };

        let control = oracle.control(validator.clone());
        let backfill = control.register(1, TEST_QUOTA).await.unwrap();
        let resolver_cfg = resolver::Config {
            public_key: validator.clone(),
            peer_provider: oracle.manager(),
            blocker: oracle.control(validator.clone()),
            mailbox_size: config.mailbox_size,
            initial: Duration::from_secs(1),
            timeout: Duration::from_secs(2),
            fetch_retry_timeout: Duration::from_millis(100),
            priority_requests: false,
            priority_responses: false,
        };
        let resolver = resolver::init(&context, resolver_cfg, backfill);

        let start = Instant::now();
        let finalizations_by_height = immutable::Archive::init(
            context.with_label("finalizations_by_height"),
            immutable::Config {
                metadata_partition: format!(
                    "{}-finalizations-by-height-metadata",
                    config.partition_prefix
                ),
                freezer_table_partition: format!(
                    "{}-finalizations-by-height-freezer-table",
                    config.partition_prefix
                ),
                freezer_table_initial_size: 64,
                freezer_table_resize_frequency: 10,
                freezer_table_resize_chunk_size: 10,
                freezer_key_partition: format!(
                    "{}-finalizations-by-height-freezer-key",
                    config.partition_prefix
                ),
                freezer_key_page_cache: config.page_cache.clone(),
                freezer_value_partition: format!(
                    "{}-finalizations-by-height-freezer-value",
                    config.partition_prefix
                ),
                freezer_value_target_size: 1024,
                freezer_value_compression: None,
                ordinal_partition: format!(
                    "{}-finalizations-by-height-ordinal",
                    config.partition_prefix
                ),
                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");
        info!(elapsed = ?start.elapsed(), "restored finalizations by height archive");

        let start = Instant::now();
        let finalized_blocks = immutable::Archive::init(
            context.with_label("finalized_blocks"),
            immutable::Config {
                metadata_partition: format!(
                    "{}-finalized_blocks-metadata",
                    config.partition_prefix
                ),
                freezer_table_partition: format!(
                    "{}-finalized_blocks-freezer-table",
                    config.partition_prefix
                ),
                freezer_table_initial_size: 64,
                freezer_table_resize_frequency: 10,
                freezer_table_resize_chunk_size: 10,
                freezer_key_partition: format!(
                    "{}-finalized_blocks-freezer-key",
                    config.partition_prefix
                ),
                freezer_key_page_cache: config.page_cache.clone(),
                freezer_value_partition: format!(
                    "{}-finalized_blocks-freezer-value",
                    config.partition_prefix
                ),
                freezer_value_target_size: 1024,
                freezer_value_compression: None,
                ordinal_partition: format!("{}-finalized_blocks-ordinal", config.partition_prefix),
                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");
        info!(elapsed = ?start.elapsed(), "restored finalized blocks archive");

        let shard_config: shards::Config<_, _, _, _, _, Sha256, _, _> = shards::Config {
            scheme_provider: provider.clone(),
            blocker: oracle.control(validator.clone()),
            shard_codec_cfg: CodecConfig {
                maximum_shard_size: 1024 * 1024,
            },
            block_codec_cfg: (),
            strategy: Sequential,
            mailbox_size: 10,
            peer_buffer_size: NZUsize!(64),
            background_channel_capacity: 1024,
            peer_provider: oracle.manager(),
        };
        let (shard_engine, shard_mailbox) = shards::Engine::new(context.clone(), shard_config);
        let network = control.register(2, TEST_QUOTA).await.unwrap();
        shard_engine.start(network);

        let (actor, mailbox, height) = Actor::init(
            context.clone(),
            finalizations_by_height,
            finalized_blocks,
            config,
        )
        .await;
        let actor_handle = actor.start(application.clone(), shard_mailbox.clone(), resolver);

        ValidatorSetup {
            application,
            mailbox,
            extra: shard_mailbox,
            height,
            actor_handle,
        }
    }

    fn make_test_block(
        parent: D,
        parent_commitment: Commitment,
        height: Height,
        timestamp: u64,
        num_participants: u16,
    ) -> CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256> {
        let parent_view = height
            .previous()
            .map(|h| View::new(h.get()))
            .unwrap_or(View::zero());
        let context = CodingCtx {
            round: Round::new(Epoch::zero(), View::new(height.get())),
            leader: default_leader(),
            parent: (parent_view, parent_commitment),
        };
        let raw = CodingB::new::<Sha256>(context, parent, height, timestamp);
        let coding_config = coding_config_for_participants(num_participants);
        CodedBlock::new(raw, coding_config, &Sequential)
    }

    fn genesis_parent_commitment(_num_participants: u16) -> Commitment {
        genesis_commitment()
    }

    fn commitment(block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>) -> Commitment {
        block.commitment()
    }

    fn digest(block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>) -> D {
        block.digest()
    }

    fn height(block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>) -> Height {
        block.height()
    }

    async fn propose(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>,
    ) {
        assert!(handle.mailbox.proposed(round, block.clone()).await);
    }

    async fn verify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>,
        _all_handles: &mut [ValidatorHandle<Self>],
    ) {
        assert!(handle.mailbox.verified(round, block.clone()).await);
    }

    async fn certify(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>,
    ) -> bool {
        handle.mailbox.certified(round, block.clone()).await
    }

    fn make_finalization(
        proposal: Proposal<Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Finalization<S, Commitment> {
        let finalizes: Vec<_> = schemes
            .iter()
            .take(quorum as usize)
            .map(|scheme| Finalize::sign(scheme, proposal.clone()).unwrap())
            .collect();
        Finalization::from_finalizes(&schemes[0], &finalizes, &Sequential).unwrap()
    }

    fn make_notarization(
        proposal: Proposal<Commitment>,
        schemes: &[S],
        quorum: u32,
    ) -> Notarization<S, Commitment> {
        let notarizes: Vec<_> = schemes
            .iter()
            .take(quorum as usize)
            .map(|scheme| Notarize::sign(scheme, proposal.clone()).unwrap())
            .collect();
        Notarization::from_notarizes(&schemes[0], &notarizes, &Sequential).unwrap()
    }

    async fn report_finalization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        finalization: Finalization<S, Commitment>,
    ) {
        mailbox.report(Activity::Finalization(finalization)).await;
    }

    async fn report_notarization(
        mailbox: &mut Mailbox<S, Self::Variant>,
        notarization: Notarization<S, Commitment>,
    ) {
        mailbox.report(Activity::Notarization(notarization)).await;
    }

    fn finalize_timeout() -> Duration {
        Duration::from_secs(900)
    }

    async fn setup_prunable_validator(
        context: deterministic::Context,
        oracle: &Oracle<K, deterministic::Context>,
        validator: K,
        schemes: &[S],
        partition_prefix: &str,
        page_cache: CacheRef,
    ) -> (
        Mailbox<S, Self::Variant>,
        Self::ValidatorExtra,
        Application<CodingB>,
    ) {
        let control = oracle.control(validator.clone());
        let provider = ConstantProvider::new(schemes[0].clone());
        let config = Config {
            provider: provider.clone(),
            epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
            mailbox_size: 100,
            view_retention_timeout: 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: page_cache.clone(),
            strategy: Sequential,
        };

        let backfill = control.register(0, TEST_QUOTA).await.unwrap();
        let resolver_cfg = resolver::Config {
            public_key: validator.clone(),
            peer_provider: oracle.manager(),
            blocker: control.clone(),
            mailbox_size: config.mailbox_size,
            initial: Duration::from_secs(1),
            timeout: Duration::from_secs(2),
            fetch_retry_timeout: Duration::from_millis(100),
            priority_requests: false,
            priority_responses: false,
        };
        let resolver = resolver::init(&context, resolver_cfg, backfill);

        let shard_config: shards::Config<_, _, _, _, _, Sha256, _, _> = shards::Config {
            scheme_provider: provider.clone(),
            blocker: oracle.control(validator.clone()),
            shard_codec_cfg: CodecConfig {
                maximum_shard_size: 1024 * 1024,
            },
            block_codec_cfg: (),
            strategy: Sequential,
            mailbox_size: 10,
            peer_buffer_size: NZUsize!(64),
            background_channel_capacity: 1024,
            peer_provider: oracle.manager(),
        };
        let (shard_engine, shard_mailbox) = shards::Engine::new(context.clone(), shard_config);
        let network = control.register(1, TEST_QUOTA).await.unwrap();
        shard_engine.start(network);

        let finalizations_by_height = prunable::Archive::init(
            context.with_label("finalizations_by_height"),
            prunable::Config {
                translator: EightCap,
                key_partition: format!("{}-finalizations-by-height-key", partition_prefix),
                key_page_cache: page_cache.clone(),
                value_partition: format!("{}-finalizations-by-height-value", partition_prefix),
                compression: None,
                codec_config: S::certificate_codec_config_unbounded(),
                items_per_section: NZU64!(10),
                key_write_buffer: config.key_write_buffer,
                value_write_buffer: config.value_write_buffer,
                replay_buffer: config.replay_buffer,
            },
        )
        .await
        .expect("failed to initialize finalizations by height archive");

        let finalized_blocks = prunable::Archive::init(
            context.with_label("finalized_blocks"),
            prunable::Config {
                translator: EightCap,
                key_partition: format!("{}-finalized-blocks-key", partition_prefix),
                key_page_cache: page_cache.clone(),
                value_partition: format!("{}-finalized-blocks-value", partition_prefix),
                compression: None,
                codec_config: config.block_codec_config,
                items_per_section: NZU64!(10),
                key_write_buffer: config.key_write_buffer,
                value_write_buffer: config.value_write_buffer,
                replay_buffer: config.replay_buffer,
            },
        )
        .await
        .expect("failed to initialize finalized blocks archive");

        let (actor, mailbox, _) = Actor::init(
            context.clone(),
            finalizations_by_height,
            finalized_blocks,
            config,
        )
        .await;
        let application = Application::<CodingB>::default();
        actor.start(application.clone(), shard_mailbox.clone(), resolver);

        (mailbox, shard_mailbox, application)
    }

    async fn verify_for_prune(
        handle: &mut ValidatorHandle<Self>,
        round: Round,
        block: &CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>,
    ) {
        assert!(handle.mailbox.verified(round, block.clone()).await);
    }
}

// =============================================================================
// Generic Test Functions
// =============================================================================

/// Run the finalization test with the given parameters.
pub fn finalize<H: TestHarness>(seed: u64, link: Link, quorum_sees_finalization: bool) -> String {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(seed)
            .with_timeout(Some(H::finalize_timeout())),
    );
    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.clone(), NZUsize!(3), participants.clone())
                .await;

        let mut applications = BTreeMap::new();
        let mut handles = Vec::new();

        for (i, validator) in participants.iter().enumerate() {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            applications.insert(validator.clone(), setup.application);
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }

        setup_network_links(&mut oracle, &participants, link.clone()).await;

        let mut blocks = Vec::new();
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        for i in 1..=NUM_BLOCKS {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            parent = H::digest(&block);
            parent_commitment = H::commitment(&block);
            blocks.push(block);
        }

        let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);
        blocks.shuffle(&mut context);

        for block in blocks.iter() {
            let height = H::height(block);
            assert!(
                !height.is_zero(),
                "genesis block should not have been generated"
            );

            let bounds = epocher.containing(height).unwrap();
            let round = Round::new(bounds.epoch(), View::new(height.get()));

            let actor_index: usize = (height.get() % (NUM_VALIDATORS as u64)) as usize;
            let mut handle = handles[actor_index].clone();
            H::propose(&mut handle, round, block).await;
            H::verify(&mut handle, round, block, &mut handles).await;

            context.sleep(link.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(height.previous().unwrap().get()),
                payload: H::commitment(block),
            };
            let notarization = H::make_notarization(proposal.clone(), &schemes, QUORUM);
            H::report_notarization(&mut handle.mailbox, notarization).await;

            let fin = H::make_finalization(proposal, &schemes, QUORUM);
            if quorum_sees_finalization {
                let do_finalize = context.gen_bool(0.2);
                for (i, h) in handles
                    .iter_mut()
                    .choose_multiple(&mut context, NUM_VALIDATORS as usize)
                    .iter_mut()
                    .enumerate()
                {
                    if (do_finalize && i < QUORUM as usize)
                        || height.get() == NUM_BLOCKS
                        || height == bounds.last()
                    {
                        H::report_finalization(&mut h.mailbox, fin.clone()).await;
                    }
                }
            } else {
                for h in handles.iter_mut() {
                    if context.gen_bool(0.2)
                        || height.get() == NUM_BLOCKS
                        || height == bounds.last()
                    {
                        H::report_finalization(&mut h.mailbox, fin.clone()).await;
                    }
                }
            }
        }

        let mut finished = false;
        while !finished {
            context.sleep(Duration::from_secs(1)).await;
            if applications.len() != NUM_VALIDATORS as usize {
                continue;
            }
            finished = true;
            for app in applications.values() {
                if app.blocks().len() != NUM_BLOCKS as usize {
                    finished = false;
                    break;
                }
                let Some((height, _)) = app.tip() else {
                    finished = false;
                    break;
                };
                if height.get() < NUM_BLOCKS {
                    finished = false;
                    break;
                }
            }
        }

        context.auditor().state()
    })
}

/// Test that marshal can pipeline application acknowledgements up to the configured backlog.
pub fn ack_pipeline_backlog<H: TestHarness>() {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(0xA11CE)
            .with_timeout(Some(Duration::from_secs(120))),
    );
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let validator = participants[0].clone();
        let application = Application::<H::ApplicationBlock>::manual_ack();
        let setup = H::setup_validator_with(
            context.with_label("validator_0"),
            &mut oracle,
            validator,
            ConstantProvider::new(schemes[0].clone()),
            NZUsize!(3),
            application,
        )
        .await;
        let application = setup.application;
        let mut handles = vec![ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        }];
        let mut handle = handles[0].clone();

        let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(NUM_VALIDATORS as u16);
        for i in 1..=5 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                NUM_VALIDATORS as u16,
            );
            let commitment = H::commitment(&block);
            parent = H::digest(&block);
            parent_commitment = commitment;
            let round = Round::new(
                epocher.containing(H::height(&block)).unwrap().epoch(),
                View::new(i),
            );
            H::verify(&mut handle, round, &block, &mut handles).await;
            let proposal = Proposal {
                round,
                parent: View::new(i.saturating_sub(1)),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;
        }

        // Backlog should fill to configured capacity before any ack is released.
        while application.blocks().len() < 3 || application.pending_ack_heights().len() < 3 {
            context.sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(
            application.pending_ack_heights(),
            vec![Height::new(1), Height::new(2), Height::new(3)]
        );
        assert!(!application.blocks().contains_key(&Height::new(4)));
        assert!(!application.blocks().contains_key(&Height::new(5)));

        // Releasing acks should preserve FIFO order and allow further dispatch.
        for expected in 1..=5 {
            let expected = Height::new(expected);
            while application.pending_ack_heights().first().copied() != Some(expected) {
                context.sleep(Duration::from_millis(10)).await;
            }
            let acknowledged = application
                .acknowledge_next()
                .expect("pending ack should be present");
            assert_eq!(acknowledged, expected);
        }

        // All finalized blocks should eventually be delivered after draining the backlog.
        while application.blocks().len() < 5 || !application.pending_ack_heights().is_empty() {
            context.sleep(Duration::from_millis(10)).await;
        }
    });
}

/// Test that batched pending-ack progress survives restart.
pub fn ack_pipeline_backlog_persists_on_restart<H: TestHarness>() {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(0xA11CF)
            .with_timeout(Some(Duration::from_secs(120))),
    );
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let validator = participants[0].clone();
        let application = Application::<H::ApplicationBlock>::manual_ack();
        let setup = H::setup_validator_with(
            context.with_label("validator_0"),
            &mut oracle,
            validator.clone(),
            ConstantProvider::new(schemes[0].clone()),
            NZUsize!(3),
            application,
        )
        .await;
        let application = setup.application;
        let mut handles = vec![ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        }];
        let mut handle = handles[0].clone();

        let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(NUM_VALIDATORS as u16);
        for i in 1..=3 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                NUM_VALIDATORS as u16,
            );
            let commitment = H::commitment(&block);
            parent = H::digest(&block);
            parent_commitment = commitment;
            let round = Round::new(
                epocher.containing(H::height(&block)).unwrap().epoch(),
                View::new(i),
            );
            H::verify(&mut handle, round, &block, &mut handles).await;
            let proposal = Proposal {
                round,
                parent: View::new(i.saturating_sub(1)),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;
        }

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

        // Acknowledge all pending blocks without yielding so marshal can drain
        // them in one ack arm and sync metadata once.
        assert_eq!(application.acknowledge_next(), Some(Height::new(1)));
        assert_eq!(application.acknowledge_next(), Some(Height::new(2)));
        assert_eq!(application.acknowledge_next(), Some(Height::new(3)));

        // Yield to marshal.
        context.sleep(Duration::from_millis(10)).await;

        // Assert that the application has processed up to height 3.
        assert_eq!(
            application.tip().map(|(height, _)| height),
            Some(Height::new(3))
        );

        // Restart marshal and confirm the processed height restored from metadata.
        let restart = H::setup_validator_with(
            context.with_label("validator_0_restart"),
            &mut oracle,
            validator,
            ConstantProvider::new(schemes[0].clone()),
            NZUsize!(3),
            Application::manual_ack(),
        )
        .await;
        assert_eq!(restart.height, Height::new(3));
    });
}

/// Test sync height floor.
pub fn sync_height_floor<H: TestHarness>() {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(0xFF)
            .with_timeout(Some(Duration::from_secs(300))),
    );
    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.clone(), NZUsize!(3), participants.clone())
                .await;

        let mut applications = BTreeMap::new();
        let mut handles = Vec::new();

        // Skip first validator
        for (i, validator) in participants.iter().enumerate().skip(1) {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            applications.insert(validator.clone(), setup.application);
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }

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

        let mut blocks = Vec::new();
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        for i in 1..=NUM_BLOCKS {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            parent = H::digest(&block);
            parent_commitment = H::commitment(&block);
            blocks.push(block);
        }

        let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);

        for block in blocks.iter() {
            let height = H::height(block);
            assert!(
                !height.is_zero(),
                "genesis block should not have been generated"
            );

            let bounds = epocher.containing(height).unwrap();
            let round = Round::new(bounds.epoch(), View::new(height.get()));

            let actor_index: usize = (height.get() % (applications.len() as u64)) as usize;
            let mut handle = handles[actor_index].clone();
            H::propose(&mut handle, round, block).await;
            H::verify(&mut handle, round, block, &mut handles).await;

            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(height.previous().unwrap().get()),
                payload: H::commitment(block),
            };
            let notarization = H::make_notarization(proposal.clone(), &schemes, QUORUM);
            H::report_notarization(&mut handle.mailbox, notarization).await;

            let fin = H::make_finalization(proposal, &schemes, QUORUM);
            for h in handles.iter_mut() {
                H::report_finalization(&mut h.mailbox, fin.clone()).await;
            }
        }

        let mut finished = false;
        while !finished {
            context.sleep(Duration::from_secs(1)).await;
            finished = true;
            for app in applications.values().skip(1) {
                if app.blocks().len() != NUM_BLOCKS as usize {
                    finished = false;
                    break;
                }
                let Some((height, _)) = app.tip() else {
                    finished = false;
                    break;
                };
                if height.get() < NUM_BLOCKS {
                    finished = false;
                    break;
                }
            }
        }

        // Create the first validator now
        let validator = participants.first().unwrap();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            validator.clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let app = setup.application;
        let mut mailbox = setup.mailbox;

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

        const NEW_SYNC_FLOOR: u64 = 100;
        let second_handle = &mut handles[1];
        let latest_finalization = second_handle
            .mailbox
            .get_finalization(Height::new(NUM_BLOCKS))
            .await
            .unwrap();

        mailbox.set_floor(Height::new(NEW_SYNC_FLOOR)).await;
        H::report_finalization(&mut mailbox, latest_finalization).await;

        let mut finished = false;
        while !finished {
            context.sleep(Duration::from_secs(1)).await;
            finished = true;
            if app.blocks().len() != (NUM_BLOCKS - NEW_SYNC_FLOOR) as usize {
                finished = false;
                continue;
            }
            let Some((height, _)) = app.tip() else {
                finished = false;
                continue;
            };
            if height.get() < NUM_BLOCKS {
                finished = false;
                continue;
            }
        }

        for height in 1..=NUM_BLOCKS {
            let block = mailbox
                .get_block(Identifier::Height(Height::new(height)))
                .await;
            if height <= NEW_SYNC_FLOOR {
                assert!(block.is_none());
            } else {
                assert_eq!(block.unwrap().height().get(), height);
            }
        }
    })
}

/// Test pruning of finalized archives.
pub fn prune_finalized_archives<H: TestHarness>() {
    let runner = deterministic::Runner::new(
        deterministic::Config::new().with_timeout(Some(Duration::from_secs(120))),
    );
    runner.start(|mut context| async move {
        let Fixture {
            participants,
            schemes,
            ..
        } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
        let oracle =
            setup_network_with_participants(context.clone(), NZUsize!(1), participants.clone())
                .await;

        let validator = participants[0].clone();
        let partition_prefix = format!("prune-test-{}", validator.clone());
        let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE);

        let init_marshal = |ctx: deterministic::Context| {
            let validator = validator.clone();
            let schemes = schemes.clone();
            let partition_prefix = partition_prefix.clone();
            let page_cache = page_cache.clone();
            let oracle = &oracle;
            async move {
                H::setup_prunable_validator(
                    ctx,
                    oracle,
                    validator,
                    &schemes,
                    &partition_prefix,
                    page_cache,
                )
                .await
            }
        };

        let (mut mailbox, extra, application) = init_marshal(context.with_label("init")).await;
        let _ = extra; // Used by CodingHarness, silence warning for StandardHarness

        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(NUM_VALIDATORS as u16);
        let epocher = FixedEpocher::new(BLOCKS_PER_EPOCH);
        for i in 1..=20u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                NUM_VALIDATORS as u16,
            );
            let commitment = H::commitment(&block);
            parent = H::digest(&block);
            parent_commitment = commitment;
            let bounds = epocher.containing(Height::new(i)).unwrap();
            let round = Round::new(bounds.epoch(), View::new(i));

            let mut handle = ValidatorHandle {
                mailbox: mailbox.clone(),
                extra: extra.clone(),
            };
            H::verify_for_prune(&mut handle, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut mailbox, finalization).await;
        }

        while application.blocks().len() < 20 {
            context.sleep(Duration::from_millis(10)).await;
        }

        for i in 1..=20u64 {
            assert!(
                mailbox.get_block(Height::new(i)).await.is_some(),
                "block {i} should exist before pruning"
            );
            assert!(
                mailbox.get_finalization(Height::new(i)).await.is_some(),
                "finalization {i} should exist before pruning"
            );
        }

        mailbox.prune(Height::new(25)).await;
        context.sleep(Duration::from_millis(50)).await;
        for i in 1..=20u64 {
            assert!(
                mailbox.get_block(Height::new(i)).await.is_some(),
                "block {i} should still exist after pruning above floor"
            );
        }

        mailbox.prune(Height::new(10)).await;
        context.sleep(Duration::from_millis(100)).await;
        for i in 1..10u64 {
            assert!(
                mailbox.get_block(Height::new(i)).await.is_none(),
                "block {i} should be pruned"
            );
            assert!(
                mailbox.get_finalization(Height::new(i)).await.is_none(),
                "finalization {i} should be pruned"
            );
        }

        for i in 10..=20u64 {
            assert!(
                mailbox.get_block(Height::new(i)).await.is_some(),
                "block {i} should still exist after pruning"
            );
            assert!(
                mailbox.get_finalization(Height::new(i)).await.is_some(),
                "finalization {i} should still exist after pruning"
            );
        }

        mailbox.prune(Height::new(20)).await;
        context.sleep(Duration::from_millis(100)).await;
        for i in 10..20u64 {
            assert!(
                mailbox.get_block(Height::new(i)).await.is_none(),
                "block {i} should be pruned after second prune"
            );
            assert!(
                mailbox.get_finalization(Height::new(i)).await.is_none(),
                "finalization {i} should be pruned after second prune"
            );
        }

        assert!(
            mailbox.get_block(Height::new(20)).await.is_some(),
            "block 20 should still exist"
        );
        assert!(
            mailbox.get_finalization(Height::new(20)).await.is_some(),
            "finalization 20 should still exist"
        );

        drop(mailbox);
        drop(extra);
        let (mailbox, _extra, _application) = init_marshal(context.with_label("restart")).await;

        for i in 1..20u64 {
            assert!(
                mailbox.get_block(Height::new(i)).await.is_none(),
                "block {i} should still be pruned after restart"
            );
            assert!(
                mailbox.get_finalization(Height::new(i)).await.is_none(),
                "finalization {i} should still be pruned after restart"
            );
        }

        assert!(
            mailbox.get_block(Height::new(20)).await.is_some(),
            "block 20 should still exist after restart"
        );
        assert!(
            mailbox.get_finalization(Height::new(20)).await.is_some(),
            "finalization 20 should still exist after restart"
        );
    })
}

/// Regression test: delayed block backfill delivered after floor advancement must not crash.
///
/// This models a resolver peer that responds to `Request::Block` only after the
/// victim has advanced its floor and pruned finalized storage. The stale delivery
/// must be rejected and must not be persisted.
pub fn reject_stale_block_delivery_after_floor_update<H: TestHarness>() {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(0xBADC0DE)
            .with_timeout(Some(Duration::from_secs(120))),
    );
    runner.start(|mut context| async move {
        let Fixture {
            participants,
            schemes,
            ..
        } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);

        let victim = participants[0].clone();
        let attacker = participants[1].clone();
        let peers = vec![victim.clone(), attacker.clone()];
        let mut oracle =
            setup_network_with_participants(context.clone(), NZUsize!(1), peers.clone()).await;

        let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE);
        let (mut victim_mailbox, victim_extra, _victim_application) = H::setup_prunable_validator(
            context.with_label("victim"),
            &oracle,
            victim.clone(),
            &schemes,
            &format!("stale-floor-victim-{}", victim),
            page_cache.clone(),
        )
        .await;
        let (attacker_mailbox, attacker_extra, _attacker_application) =
            H::setup_prunable_validator(
                context.with_label("attacker"),
                &oracle,
                attacker.clone(),
                &schemes,
                &format!("stale-floor-attacker-{}", attacker),
                page_cache,
            )
            .await;
        let _ = victim_extra; // Used by CodingHarness, silence warning for StandardHarness.

        setup_network_links(&mut oracle, &peers, LINK).await;
        oracle
            .remove_link(attacker.clone(), victim.clone())
            .await
            .unwrap();

        // Make the attacker able to serve the block by commitment.
        let stale_height = Height::new(5);
        let round = Round::new(Epoch::zero(), View::new(stale_height.get()));
        let stale_block = H::make_test_block(
            Sha256::hash(b"stale-parent"),
            H::genesis_parent_commitment(NUM_VALIDATORS as u16),
            stale_height,
            stale_height.get(),
            NUM_VALIDATORS as u16,
        );
        let commitment = H::commitment(&stale_block);
        let mut attacker_handle = ValidatorHandle {
            mailbox: attacker_mailbox,
            extra: attacker_extra,
        };
        H::propose(&mut attacker_handle, round, &stale_block).await;
        let mut no_handles: Vec<ValidatorHandle<H>> = Vec::new();
        H::verify(
            &mut attacker_handle,
            round,
            &stale_block,
            no_handles.as_mut_slice(),
        )
        .await;

        // Trigger victim fetch for this block via finalization report.
        let proposal = Proposal {
            round,
            parent: View::new(stale_height.get() - 1),
            payload: commitment,
        };
        let finalization = H::make_finalization(proposal, &schemes, QUORUM);
        H::report_finalization(&mut victim_mailbox, finalization).await;

        // Let block requests get issued while responses are still blocked.
        context.sleep(Duration::from_millis(500)).await;

        // Advance floor beyond the stale block and prune.
        let floor = Height::new(10);
        victim_mailbox.set_floor(floor).await;
        // Barrier: mailbox messages are FIFO, so this confirms `set_floor`
        // has been processed before we re-enable the delayed delivery path.
        let _ = victim_mailbox.get_finalization(floor).await;

        // Restore attacker -> victim traffic so delayed resolver responses can arrive.
        oracle
            .add_link(attacker.clone(), victim.clone(), LINK)
            .await
            .unwrap();
        context.sleep(Duration::from_secs(3)).await;

        // Stale-but-valid delivery should not be considered Byzantine behavior.
        let blocked = oracle.blocked().await.unwrap();
        assert!(
            !blocked
                .iter()
                .any(|(blocker, blocked)| blocker == &victim && blocked == &attacker),
            "stale delivery below floor must not block the serving peer"
        );

        assert!(
            victim_mailbox.get_block(stale_height).await.is_none(),
            "stale block below floor must not be persisted"
        );
        assert!(
            victim_mailbox
                .get_finalization(stale_height)
                .await
                .is_none(),
            "stale finalization below floor must not be persisted"
        );
    });
}

/// Test basic block subscription delivery.
pub fn subscribe_basic_block_delivery<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let mut handles = Vec::new();
        for (i, validator) in participants.iter().enumerate() {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }
        let mut handle = handles[0].clone();

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

        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let digest = H::digest(&block);
        let commitment = H::commitment(&block);

        let subscription_rx = handle
            .mailbox
            .subscribe_by_digest(Some(Round::new(Epoch::zero(), View::new(1))), digest)
            .await;

        H::propose(&mut handle, Round::new(Epoch::zero(), View::new(1)), &block).await;
        H::verify(
            &mut handle,
            Round::new(Epoch::zero(), View::new(1)),
            &block,
            &mut handles,
        )
        .await;

        let proposal = Proposal {
            round: Round::new(Epoch::zero(), View::new(1)),
            parent: View::zero(),
            payload: commitment,
        };
        let notarization = H::make_notarization(proposal.clone(), &schemes, QUORUM);
        H::report_notarization(&mut handle.mailbox, notarization).await;

        let finalization = H::make_finalization(proposal, &schemes, QUORUM);
        H::report_finalization(&mut handle.mailbox, finalization).await;

        let received_block = subscription_rx.await.unwrap();
        assert_eq!(received_block.digest(), digest);
        assert_eq!(received_block.height().get(), 1);
    })
}

/// Test multiple subscriptions.
pub fn subscribe_multiple_subscriptions<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let mut handles = Vec::new();
        for (i, validator) in participants.iter().enumerate() {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }
        let mut handle = handles[0].clone();

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

        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block1 = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let block2 = H::make_test_block(
            H::digest(&block1),
            H::commitment(&block1),
            Height::new(2),
            2,
            participants.len() as u16,
        );
        let digest1 = H::digest(&block1);
        let digest2 = H::digest(&block2);

        let sub1_rx = handle
            .mailbox
            .subscribe_by_digest(Some(Round::new(Epoch::zero(), View::new(1))), digest1)
            .await;
        let sub2_rx = handle
            .mailbox
            .subscribe_by_digest(Some(Round::new(Epoch::zero(), View::new(2))), digest2)
            .await;
        let sub3_rx = handle
            .mailbox
            .subscribe_by_digest(Some(Round::new(Epoch::zero(), View::new(1))), digest1)
            .await;

        for (view, block) in [(1u64, &block1), (2, &block2)] {
            let round = Round::new(Epoch::zero(), View::new(view));
            H::propose(&mut handle, round, block).await;
            H::verify(&mut handle, round, block, &mut handles).await;

            let proposal = Proposal {
                round,
                parent: View::new(view.checked_sub(1).unwrap()),
                payload: H::commitment(block),
            };
            let notarization = H::make_notarization(proposal.clone(), &schemes, QUORUM);
            H::report_notarization(&mut handle.mailbox, notarization).await;

            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;
        }

        let received1_sub1 = sub1_rx.await.unwrap();
        let received2 = sub2_rx.await.unwrap();
        let received1_sub3 = sub3_rx.await.unwrap();

        assert_eq!(received1_sub1.digest(), digest1);
        assert_eq!(received2.digest(), digest2);
        assert_eq!(received1_sub3.digest(), digest1);
        assert_eq!(received1_sub1.height().get(), 1);
        assert_eq!(received2.height().get(), 2);
        assert_eq!(received1_sub3.height().get(), 1);
    })
}

/// Test canceled subscriptions.
pub fn subscribe_canceled_subscriptions<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let mut handles = Vec::new();
        for (i, validator) in participants.iter().enumerate() {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }
        let mut handle = handles[0].clone();

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

        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block1 = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let block2 = H::make_test_block(
            H::digest(&block1),
            H::commitment(&block1),
            Height::new(2),
            2,
            participants.len() as u16,
        );
        let digest1 = H::digest(&block1);
        let digest2 = H::digest(&block2);

        let sub1_rx = handle
            .mailbox
            .subscribe_by_digest(Some(Round::new(Epoch::zero(), View::new(1))), digest1)
            .await;
        let sub2_rx = handle
            .mailbox
            .subscribe_by_digest(Some(Round::new(Epoch::zero(), View::new(2))), digest2)
            .await;

        drop(sub1_rx);

        for (view, block) in [(1u64, &block1), (2, &block2)] {
            let round = Round::new(Epoch::zero(), View::new(view));
            H::propose(&mut handle, round, block).await;
            H::verify(&mut handle, round, block, &mut handles).await;

            let proposal = Proposal {
                round,
                parent: View::new(view.checked_sub(1).unwrap()),
                payload: H::commitment(block),
            };
            let notarization = H::make_notarization(proposal.clone(), &schemes, QUORUM);
            H::report_notarization(&mut handle.mailbox, notarization).await;

            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;
        }

        let received2 = sub2_rx.await.unwrap();
        assert_eq!(received2.digest(), digest2);
        assert_eq!(received2.height().get(), 2);
    })
}

/// Test blocks from different sources.
pub fn subscribe_blocks_from_different_sources<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let mut handles = Vec::new();
        for (i, validator) in participants.iter().enumerate() {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }
        let mut handle = handles[0].clone();

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

        let parent = Sha256::hash(b"");
        let n = participants.len() as u16;
        let block1 = H::make_test_block(
            parent,
            H::genesis_parent_commitment(n),
            Height::new(1),
            1,
            n,
        );
        let block2 = H::make_test_block(
            H::digest(&block1),
            H::commitment(&block1),
            Height::new(2),
            2,
            n,
        );
        let block3 = H::make_test_block(
            H::digest(&block2),
            H::commitment(&block2),
            Height::new(3),
            3,
            n,
        );
        let block4 = H::make_test_block(
            H::digest(&block3),
            H::commitment(&block3),
            Height::new(4),
            4,
            n,
        );
        let block5 = H::make_test_block(
            H::digest(&block4),
            H::commitment(&block4),
            Height::new(5),
            5,
            n,
        );

        let sub1_rx = handle
            .mailbox
            .subscribe_by_digest(None, H::digest(&block1))
            .await;
        let sub2_rx = handle
            .mailbox
            .subscribe_by_digest(None, H::digest(&block2))
            .await;
        let sub3_rx = handle
            .mailbox
            .subscribe_by_digest(None, H::digest(&block3))
            .await;
        let sub4_rx = handle
            .mailbox
            .subscribe_by_digest(None, H::digest(&block4))
            .await;
        let sub5_rx = handle
            .mailbox
            .subscribe_by_digest(None, H::digest(&block5))
            .await;

        // Block1: Broadcasted by the actor
        H::propose(
            &mut handle,
            Round::new(Epoch::zero(), View::new(1)),
            &block1,
        )
        .await;
        context.sleep(Duration::from_millis(20)).await;

        let received1 = sub1_rx.await.unwrap();
        assert_eq!(received1.digest(), H::digest(&block1));
        assert_eq!(received1.height().get(), 1);

        // Block2: Verified by the actor
        H::propose(
            &mut handle,
            Round::new(Epoch::zero(), View::new(2)),
            &block2,
        )
        .await;
        H::verify(
            &mut handle,
            Round::new(Epoch::zero(), View::new(2)),
            &block2,
            &mut handles,
        )
        .await;

        let received2 = sub2_rx.await.unwrap();
        assert_eq!(received2.digest(), H::digest(&block2));
        assert_eq!(received2.height().get(), 2);

        // Block3: Notarized by the actor
        let proposal3 = Proposal {
            round: Round::new(Epoch::zero(), View::new(3)),
            parent: View::new(2),
            payload: H::commitment(&block3),
        };
        let notarization3 = H::make_notarization(proposal3.clone(), &schemes, QUORUM);
        H::report_notarization(&mut handle.mailbox, notarization3).await;
        H::propose(
            &mut handle,
            Round::new(Epoch::zero(), View::new(3)),
            &block3,
        )
        .await;
        H::verify(
            &mut handle,
            Round::new(Epoch::zero(), View::new(3)),
            &block3,
            &mut handles,
        )
        .await;

        let received3 = sub3_rx.await.unwrap();
        assert_eq!(received3.digest(), H::digest(&block3));
        assert_eq!(received3.height().get(), 3);

        // Block4: Finalized by the actor
        let finalization4 = H::make_finalization(
            Proposal {
                round: Round::new(Epoch::zero(), View::new(4)),
                parent: View::new(3),
                payload: H::commitment(&block4),
            },
            &schemes,
            QUORUM,
        );
        H::report_finalization(&mut handle.mailbox, finalization4).await;
        H::propose(
            &mut handle,
            Round::new(Epoch::zero(), View::new(4)),
            &block4,
        )
        .await;
        H::verify(
            &mut handle,
            Round::new(Epoch::zero(), View::new(4)),
            &block4,
            &mut handles,
        )
        .await;

        let received4 = sub4_rx.await.unwrap();
        assert_eq!(received4.digest(), H::digest(&block4));
        assert_eq!(received4.height().get(), 4);

        // Block5: Finalized by the actor with notarization
        let proposal5 = Proposal {
            round: Round::new(Epoch::zero(), View::new(5)),
            parent: View::new(4),
            payload: H::commitment(&block5),
        };
        let notarization5 = H::make_notarization(proposal5.clone(), &schemes, QUORUM);
        H::report_notarization(&mut handle.mailbox, notarization5).await;
        let finalization5 = H::make_finalization(proposal5, &schemes, QUORUM);
        H::report_finalization(&mut handle.mailbox, finalization5).await;
        H::propose(
            &mut handle,
            Round::new(Epoch::zero(), View::new(5)),
            &block5,
        )
        .await;
        H::verify(
            &mut handle,
            Round::new(Epoch::zero(), View::new(5)),
            &block5,
            &mut handles,
        )
        .await;

        let received5 = sub5_rx.await.unwrap();
        assert_eq!(received5.digest(), H::digest(&block5));
        assert_eq!(received5.height().get(), 5);
    })
}

/// Test basic get_info queries for present and missing data.
pub fn get_info_basic_queries_present_and_missing<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let me = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            me,
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // Initially, no latest
        assert!(handle.mailbox.get_info(Identifier::Latest).await.is_none());

        // Before finalization, specific height returns None
        assert!(handle.mailbox.get_info(Height::new(1)).await.is_none());

        // Create and verify a block, then finalize it
        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let digest = H::digest(&block);
        let commitment = H::commitment(&block);
        let round = Round::new(Epoch::zero(), View::new(1));

        H::propose(&mut handle, round, &block).await;
        context.sleep(LINK.latency).await;

        let proposal = Proposal {
            round,
            parent: View::zero(),
            payload: commitment,
        };
        let finalization = H::make_finalization(proposal, &schemes, QUORUM);
        H::report_finalization(&mut handle.mailbox, finalization).await;

        // Latest should now be the finalized block
        assert_eq!(
            handle.mailbox.get_info(Identifier::Latest).await,
            Some((Height::new(1), digest))
        );

        // Height 1 now present
        assert_eq!(
            handle.mailbox.get_info(Height::new(1)).await,
            Some((Height::new(1), digest))
        );

        // Commitment should map to its height
        assert_eq!(
            handle.mailbox.get_info(&digest).await,
            Some((Height::new(1), digest))
        );

        // Missing height
        assert!(handle.mailbox.get_info(Height::new(2)).await.is_none());

        // Missing commitment
        let missing = Sha256::hash(b"missing");
        assert!(handle.mailbox.get_info(&missing).await.is_none());
    })
}

/// Test get_info latest progression with multiple finalizations.
pub fn get_info_latest_progression_multiple_finalizations<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let me = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            me,
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let mut digests = Vec::new();

        for i in 1..=5u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::zero(), View::new(i));

            H::propose(&mut handle, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;

            // Latest should always point to most recently finalized
            assert_eq!(
                handle.mailbox.get_info(Identifier::Latest).await,
                Some((Height::new(i), digest))
            );

            parent = digest;
            parent_commitment = commitment;
            digests.push(digest);
        }

        // Verify each height is accessible
        for (i, digest) in digests.iter().enumerate() {
            let height = Height::new(i as u64 + 1);
            assert_eq!(
                handle.mailbox.get_info(height).await,
                Some((height, *digest))
            );
        }
    })
}

/// Test get_block by height and latest.
pub fn get_block_by_height_and_latest<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let me = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            me,
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // Initially, no blocks
        assert!(handle
            .mailbox
            .get_block(Identifier::Height(Height::new(1)))
            .await
            .is_none());
        assert!(handle.mailbox.get_block(Identifier::Latest).await.is_none());

        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let mut blocks = Vec::new();

        for i in 1..=3u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::zero(), View::new(i));

            H::propose(&mut handle, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;

            parent = digest;
            parent_commitment = commitment;
            blocks.push((digest, block));
        }

        // Verify each block by height
        for (i, (digest, _block)) in blocks.iter().enumerate() {
            let height = Height::new(i as u64 + 1);
            let fetched = handle
                .mailbox
                .get_block(Identifier::Height(height))
                .await
                .unwrap();
            assert_eq!(fetched.digest(), *digest);
            assert_eq!(fetched.height(), height);
        }

        // Latest should be last block
        let latest = handle.mailbox.get_block(Identifier::Latest).await.unwrap();
        assert_eq!(latest.digest(), blocks[2].0);
        assert_eq!(latest.height(), Height::new(3));

        // Missing height
        assert!(handle
            .mailbox
            .get_block(Identifier::Height(Height::new(10)))
            .await
            .is_none());
    })
}

/// Test get_block by commitment from various sources.
pub fn get_block_by_commitment_from_sources_and_missing<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let me = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            me,
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // Create and finalize a block
        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let digest = H::digest(&block);
        let commitment = H::commitment(&block);
        let round = Round::new(Epoch::zero(), View::new(1));

        H::propose(&mut handle, round, &block).await;
        context.sleep(LINK.latency).await;

        let proposal = Proposal {
            round,
            parent: View::zero(),
            payload: commitment,
        };
        let finalization = H::make_finalization(proposal, &schemes, QUORUM);
        H::report_finalization(&mut handle.mailbox, finalization).await;

        // Get by commitment
        let fetched = handle.mailbox.get_block(&digest).await.unwrap();
        assert_eq!(fetched.digest(), digest);
        assert_eq!(fetched.height(), Height::new(1));

        // Missing commitment
        let missing = Sha256::hash(b"missing");
        assert!(handle.mailbox.get_block(&missing).await.is_none());
    })
}

/// Test get_finalization by height.
pub fn get_finalization_by_height<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let me = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            me,
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // Initially, no finalization
        assert!(handle
            .mailbox
            .get_finalization(Height::new(1))
            .await
            .is_none());

        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);

        for i in 1..=3u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::zero(), View::new(i));

            H::propose(&mut handle, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal.clone(), &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;

            // Verify finalization is retrievable
            let fin = handle
                .mailbox
                .get_finalization(Height::new(i))
                .await
                .unwrap();
            assert_eq!(fin.proposal.payload, commitment);
            assert_eq!(fin.round().view(), View::new(i));

            parent = digest;
            parent_commitment = commitment;
        }

        // Missing height
        assert!(handle
            .mailbox
            .get_finalization(Height::new(10))
            .await
            .is_none());
    })
}

/// Test hint_finalized triggers fetch.
pub fn hint_finalized_triggers_fetch<H: TestHarness>() {
    let runner = deterministic::Runner::new(
        deterministic::Config::new()
            .with_seed(42)
            .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.clone(), NZUsize!(3), participants.clone())
                .await;

        // Set up two validators
        let setup0 = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            participants[0].clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let app0 = setup0.application;
        let mut handle0 = ValidatorHandle {
            mailbox: setup0.mailbox,
            extra: setup0.extra,
        };

        let setup1 = H::setup_validator(
            context.with_label("validator_1"),
            &mut oracle,
            participants[1].clone(),
            ConstantProvider::new(schemes[1].clone()),
        )
        .await;
        let handle1: ValidatorHandle<H> = ValidatorHandle {
            mailbox: setup1.mailbox,
            extra: setup1.extra,
        };

        // Add links between validators
        setup_network_links(&mut oracle, &participants[..2], LINK).await;

        // Validator 0: Create and finalize blocks 1-5
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        for i in 1..=5u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::new(0), View::new(i));

            H::propose(&mut handle0, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle0.mailbox, finalization).await;

            parent = digest;
            parent_commitment = commitment;
        }

        // Wait for validator 0 to process all blocks
        while app0.blocks().len() < 5 {
            context.sleep(Duration::from_millis(10)).await;
        }

        // Validator 1 should not have block 5 yet
        assert!(handle1
            .mailbox
            .get_finalization(Height::new(5))
            .await
            .is_none());

        // Validator 1: hint that block 5 is finalized, targeting validator 0
        handle1
            .mailbox
            .hint_finalized(Height::new(5), NonEmptyVec::new(participants[0].clone()))
            .await;

        // Wait for the fetch to complete
        while handle1
            .mailbox
            .get_finalization(Height::new(5))
            .await
            .is_none()
        {
            context.sleep(Duration::from_millis(10)).await;
        }

        // Verify validator 1 now has the finalization
        let finalization = handle1
            .mailbox
            .get_finalization(Height::new(5))
            .await
            .expect("finalization should be fetched");
        assert_eq!(finalization.proposal.round.view(), View::new(5));
    })
}

/// Test ancestry stream.
pub fn ancestry_stream<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        let me = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            me,
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // Finalize blocks at heights 1-5
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        for i in 1..=5u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::zero(), View::new(i));

            H::propose(&mut handle, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;

            parent = digest;
            parent_commitment = commitment;
        }

        // Stream from latest -> height 1
        let (_, commitment) = handle.mailbox.get_info(Identifier::Latest).await.unwrap();
        let ancestry = handle.mailbox.ancestry((None, commitment)).await.unwrap();
        let blocks = ancestry.collect::<Vec<_>>().await;

        // Ensure correct delivery order: 5,4,3,2,1
        assert_eq!(blocks.len(), 5);
        (0..5).for_each(|i| {
            assert_eq!(blocks[i].height().get(), 5 - i as u64);
        });
    })
}

/// Test finalize same height different views.
pub fn finalize_same_height_different_views<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        // Set up two validators
        let mut handles = Vec::new();
        for (i, validator) in participants.iter().enumerate().take(2) {
            let setup = H::setup_validator(
                context.with_label(&format!("validator_{i}")),
                &mut oracle,
                validator.clone(),
                ConstantProvider::new(schemes[i].clone()),
            )
            .await;
            handles.push(ValidatorHandle {
                mailbox: setup.mailbox,
                extra: setup.extra,
            });
        }

        // Create block at height 1
        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let digest = H::digest(&block);
        let commitment = H::commitment(&block);

        // Both validators receive the block
        for handle in handles.iter_mut() {
            H::propose(handle, Round::new(Epoch::new(0), View::new(1)), &block).await;
        }
        context.sleep(LINK.latency).await;

        // Validator 0: Finalize with view 1
        let proposal_v1 = Proposal {
            round: Round::new(Epoch::new(0), View::new(1)),
            parent: View::new(0),
            payload: commitment,
        };
        let notarization_v1 = H::make_notarization(proposal_v1.clone(), &schemes, QUORUM);
        let finalization_v1 = H::make_finalization(proposal_v1.clone(), &schemes, QUORUM);
        H::report_notarization(&mut handles[0].mailbox, notarization_v1.clone()).await;
        H::report_finalization(&mut handles[0].mailbox, finalization_v1.clone()).await;

        // Validator 1: Finalize with view 2 (simulates receiving finalization from different view)
        let proposal_v2 = Proposal {
            round: Round::new(Epoch::new(0), View::new(2)), // Different view
            parent: View::new(0),
            payload: commitment, // Same block
        };
        let notarization_v2 = H::make_notarization(proposal_v2.clone(), &schemes, QUORUM);
        let finalization_v2 = H::make_finalization(proposal_v2.clone(), &schemes, QUORUM);
        H::report_notarization(&mut handles[1].mailbox, notarization_v2.clone()).await;
        H::report_finalization(&mut handles[1].mailbox, finalization_v2.clone()).await;

        // Wait for finalization processing
        context.sleep(Duration::from_millis(100)).await;

        // Verify both validators stored the block correctly
        let block0 = handles[0].mailbox.get_block(Height::new(1)).await.unwrap();
        let block1 = handles[1].mailbox.get_block(Height::new(1)).await.unwrap();
        assert_eq!(block0.digest(), digest);
        assert_eq!(block1.digest(), digest);

        // Verify both validators have finalizations stored
        let fin0 = handles[0]
            .mailbox
            .get_finalization(Height::new(1))
            .await
            .unwrap();
        let fin1 = handles[1]
            .mailbox
            .get_finalization(Height::new(1))
            .await
            .unwrap();

        // Verify the finalizations have the expected different views
        assert_eq!(fin0.proposal.payload, commitment);
        assert_eq!(fin0.round().view(), View::new(1));
        assert_eq!(fin1.proposal.payload, commitment);
        assert_eq!(fin1.round().view(), View::new(2));

        // Both validators can retrieve block by height
        assert_eq!(
            handles[0].mailbox.get_info(Height::new(1)).await,
            Some((Height::new(1), digest))
        );
        assert_eq!(
            handles[1].mailbox.get_info(Height::new(1)).await,
            Some((Height::new(1), digest))
        );

        // Test that a validator receiving BOTH finalizations handles it correctly
        H::report_finalization(&mut handles[0].mailbox, finalization_v2.clone()).await;
        H::report_finalization(&mut handles[1].mailbox, finalization_v1.clone()).await;
        context.sleep(Duration::from_millis(100)).await;

        // Validator 0 should still have the original finalization (v1)
        let fin0_after = handles[0]
            .mailbox
            .get_finalization(Height::new(1))
            .await
            .unwrap();
        assert_eq!(fin0_after.round().view(), View::new(1));

        // Validator 1 should still have the original finalization (v2)
        let fin1_after = handles[1]
            .mailbox
            .get_finalization(Height::new(1))
            .await
            .unwrap();
        assert_eq!(fin1_after.round().view(), View::new(2));
    })
}

/// Test init processed height.
pub fn init_processed_height<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

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

        // First session: create validator and finalize some blocks
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            validator.clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let app = setup.application;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };
        let initial_height = setup.height;

        // Initially should be zero (no blocks processed)
        assert_eq!(initial_height, Height::zero());

        // Finalize blocks 1-5
        let mut parent = Sha256::hash(b"");
        let mut parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        for i in 1..=5u64 {
            let block = H::make_test_block(
                parent,
                parent_commitment,
                Height::new(i),
                i,
                participants.len() as u16,
            );
            let digest = H::digest(&block);
            let commitment = H::commitment(&block);
            let round = Round::new(Epoch::zero(), View::new(i));

            H::propose(&mut handle, round, &block).await;
            context.sleep(LINK.latency).await;

            let proposal = Proposal {
                round,
                parent: View::new(i - 1),
                payload: commitment,
            };
            let finalization = H::make_finalization(proposal, &schemes, QUORUM);
            H::report_finalization(&mut handle.mailbox, finalization).await;

            parent = digest;
            parent_commitment = commitment;
        }

        // Wait for application to process all blocks
        while app.blocks().len() < 5 {
            context.sleep(Duration::from_millis(10)).await;
        }

        // Drop the handle to simulate shutdown
        drop(handle);

        // Second session: create new validator instance, should recover processed height
        let setup2 = H::setup_validator(
            context.with_label("validator_0_restart"),
            &mut oracle,
            validator.clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let recovered_height = setup2.height;

        // Should have recovered to height 5
        assert_eq!(recovered_height, Height::new(5));
    })
}

/// Test broadcast caches block.
pub fn broadcast_caches_block<H: TestHarness>() {
    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.clone(), NZUsize!(1), participants.clone())
                .await;

        // Set up one validator
        let validator = participants[0].clone();
        let setup = H::setup_validator(
            context.with_label("validator_0"),
            &mut oracle,
            validator.clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle = ValidatorHandle {
            mailbox: setup.mailbox,
            extra: setup.extra,
        };

        // Create block at height 1
        let parent = Sha256::hash(b"");
        let parent_commitment = H::genesis_parent_commitment(participants.len() as u16);
        let block = H::make_test_block(
            parent,
            parent_commitment,
            Height::new(1),
            1,
            participants.len() as u16,
        );
        let digest = H::digest(&block);
        let commitment = H::commitment(&block);

        // Broadcast the block
        H::propose(&mut handle, Round::new(Epoch::new(0), View::new(1)), &block).await;

        // Ensure the block is cached and retrievable
        handle
            .mailbox
            .get_block(&digest)
            .await
            .expect("block should be cached after broadcast");

        // Restart marshal, removing any in-memory cache
        let setup2 = H::setup_validator(
            context.with_label("validator_0_restart"),
            &mut oracle,
            validator.clone(),
            ConstantProvider::new(schemes[0].clone()),
        )
        .await;
        let mut handle2: ValidatorHandle<H> = ValidatorHandle {
            mailbox: setup2.mailbox,
            extra: setup2.extra,
        };

        // Put a notarization into the cache to re-initialize the ephemeral cache for the
        // first epoch.
        let notarization = H::make_notarization(
            Proposal {
                round: Round::new(Epoch::new(0), View::new(1)),
                parent: View::new(0),
                payload: commitment,
            },
            &schemes,
            QUORUM,
        );
        H::report_notarization(&mut handle2.mailbox, notarization).await;

        // Ensure the block is cached and retrievable
        handle2
            .mailbox
            .get_block(&digest)
            .await
            .expect("block should be cached after broadcast");
    })
}