//! Wrapper for consensus applications that handles epochs, erasure coding, and block dissemination.
//!
//! # Overview
//!
//! [`Marshaled`] is an adapter that wraps any [`Application`] implementation to handle
//! epoch transitions and erasure coded broadcast automatically. It intercepts consensus
//! operations (propose, verify, certify) and ensures blocks are only produced within valid epoch boundaries.
//!
//! # Epoch Boundaries
//!
//! An epoch is a fixed number of blocks (the `epoch_length`). When the last block in an epoch
//! is reached, this wrapper prevents new blocks from being built & proposed until the next epoch begins.
//! Instead, it re-proposes the boundary block to avoid producing blocks that would be pruned
//! by the epoch transition.
//!
//! # Erasure Coding
//!
//! This wrapper integrates with a variant of marshal that supports erasure coded broadcast. When a leader
//! proposes a new block, it is automatically erasure encoded and its shards are broadcasted to active
//! participants. When verifying a proposed block (the precondition for notarization), the wrapper
//! ensures the commitment's context digest matches the consensus context and, when this node can
//! sign, waits for validation of the shard assigned to it by the proposer. If that shard is valid,
//! the assigned shard is relayed to all other participants to aid in block reconstruction.
//!
//! A participant may still reconstruct the full block from gossiped shards before its assigned shard
//! arrives. That is sufficient for later certification and repair flows, but it is not treated as
//! notarization readiness: a signer only helps form a notarization once it has validated the
//! shard it is supposed to echo. A node that cannot sign casts no notarize vote, so its
//! verification resolves without waiting for a shard. The shard engine still verifies and relays
//! the assigned shard of a committee member without a share.
//!
//! During certification (the phase between notarization and finalization), the wrapper subscribes to
//! block reconstruction and validates epoch boundaries, parent commitment, height contiguity, and
//! that the block's embedded context matches the consensus context before allowing the block to be
//! certified. If certification fails, the voter can still emit a nullify vote to advance the view.
//!
//! # Usage
//!
//! Wrap your [`Application`] implementation with [`Marshaled::new`] and provide it to your
//! consensus engine for the [`Automaton`], [`Relay`], and [`Reporter`]. As the reporter, the
//! wrapper forwards consensus activity to marshal and releases the state it keeps for each round
//! once consensus finalizes that round. The wrapper handles all epoch logic transparently.
//!
//! ```rust,ignore
//! let cfg = MarshaledConfig {
//!     application: my_application,
//!     marshal: marshal_mailbox,
//!     shards: shard_mailbox,
//!     scheme_provider,
//!     epocher,
//!     strategy,
//! };
//! let application = Marshaled::new(context, cfg);
//! ```
//!
//! # Implementation Notes
//!
//! - Genesis blocks are handled specially: epoch 0 returns the application's genesis block,
//!   while subsequent epochs use the last block of the previous epoch as genesis
//! - Blocks are automatically verified to be within the current epoch
//!
//! # Notarization and Data Availability
//!
//! In rare crash cases, it is possible for a notarization certificate to exist without a block being
//! available to the honest parties (e.g., if the whole network crashed before receiving `f+1` shards
//! and the proposer went permanently offline). In this case, `certify` may remain pending while it
//! waits for the unavailable block. Simplex may time out and nullify the view, but that timeout does
//! not resolve the certification request.
//!
//! For this reason, it should not be expected that every notarized payload will be certifiable due
//! to the lack of an available block. However, if even one honest and online party has the block,
//! they will attempt to forward it to others via marshal's resolver. This case is already present
//! in the event of a block that was proposed with invalid codec; Marshal will not be able to reconstruct
//! the block, and therefore won't serve it.
//!
//! ```text
//!                                      ┌───────────────────────────────────────────────────┐
//!                                      ▼                                                   │
//! ┌─────────────────────┐   ┌─────────────────────┐   ┌─────────────────────┐   ┌─────────────────────┐
//! │          B1         │◀──│          B2         │◀──│          B3         │XXX│          B4         │
//! └─────────────────────┘   └─────────────────────┘   └──────────┬──────────┘   └─────────────────────┘
//!                                                                │
//!                                                         Pending Certify
//! ```

use crate::{
    Application, Automaton, Block, CertifiableAutomaton, CertifiableBlock, Epochable, Handoff,
    Heightable, Relay, Reporter,
    marshal::{
        ancestry::Ancestry,
        application::{
            build::{self, Resolved},
            gates::{self, GateOutcome, Gates},
            relay,
            validation::{
                Stage, is_block_in_expected_epoch, is_inferred_reproposal_at_certify,
                is_valid_reproposal_at_verify,
            },
        },
        coding::{
            Coding, shards,
            types::{CodedBlock, coding_config_for_participants, hash_context},
            validation::{ProposalError, validate_block, validate_proposal},
        },
        core,
    },
    simplex::{
        Plan,
        scheme::Scheme,
        types::{Activity, Context},
    },
    types::{Epoch, Epocher, Round, View, coding::Commitment},
};
use commonware_actor::Feedback;
use commonware_codec::Codec;
use commonware_coding::{Config as CodingConfig, Scheme as CodingScheme};
use commonware_cryptography::{
    Committable, Digested, Digestible, Hasher,
    certificate::{Provider, Scheme as _, Verifier},
};
use commonware_macros::select;
use commonware_parallel::Strategy;
use commonware_runtime::{
    Clock, Metrics, Spawner,
    telemetry::{
        metrics::{
            MetricsExt as _,
            histogram::{Buckets, Timed},
        },
        traces::TracedExt as _,
    },
};
use commonware_utils::channel::{fallible::OneshotExt, oneshot};
use futures::future::Either;
use std::{future::Future, sync::Arc};
use tracing::{Instrument as _, debug, info_span, warn};

/// The outcome of the coding proposal checks. A build carries the coding configuration of the
/// epoch's participants to the sealing step.
type CodingResolved<B, C, H, A> =
    Resolved<Commitment<B, C, H>, Arc<CodedBlock<B, C, H>>, A, CodingConfig>;

/// Configuration for initializing [`Marshaled`].
#[allow(clippy::type_complexity)]
pub struct MarshaledConfig<A, B, C, H, Z, S, ES>
where
    B: CertifiableBlock<Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>>
        + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    /// The underlying application to wrap.
    pub application: A,
    /// Mailbox for communicating with the marshal engine.
    pub marshal: core::Mailbox<Z::Scheme, Coding<B, C, H, <Z::Scheme as Verifier>::PublicKey>>,
    /// Mailbox for communicating with the shards engine.
    pub shards: shards::Mailbox<B, C, H, <Z::Scheme as Verifier>::PublicKey>,
    /// Provider for signing schemes scoped by epoch.
    pub scheme_provider: Z,
    /// Strategy for parallel operations.
    pub strategy: S,
    /// Strategy for determining epoch boundaries.
    pub epocher: ES,
}

/// An [`Application`] adapter that handles epoch transitions and erasure coded broadcast.
///
/// This wrapper intercepts consensus operations to enforce epoch boundaries. It prevents
/// blocks from being produced outside their valid epoch and handles the special case of
/// re-proposing boundary blocks during epoch transitions.
#[allow(clippy::type_complexity)]
pub struct Marshaled<E, A, B, C, H, Z, S, ES>
where
    A: Application<E>,
    B: CertifiableBlock<Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>>
        + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    context: Arc<E>,
    application: A,
    marshal: core::Mailbox<Z::Scheme, Coding<B, C, H, <Z::Scheme as Verifier>::PublicKey>>,
    shards: shards::Mailbox<B, C, H, <Z::Scheme as Verifier>::PublicKey>,
    scheme_provider: Z,
    epocher: ES,
    strategy: S,
    gates: Gates<Commitment<B, C, H>, Arc<CodedBlock<B, C, H>>>,

    build_duration: Timed,
    verify_duration: Timed,
    proposal_parent_fetch_duration: Timed,
    ancestor_fetch_duration: Timed,
    erasure_encode_duration: Timed,
}

impl<E, A, B, C, H, Z, S, ES> Clone for Marshaled<E, A, B, C, H, Z, S, ES>
where
    A: Application<E>,
    B: CertifiableBlock<Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>>
        + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    fn clone(&self) -> Self {
        Self {
            context: self.context.clone(),
            application: self.application.clone(),
            marshal: self.marshal.clone(),
            shards: self.shards.clone(),
            scheme_provider: self.scheme_provider.clone(),
            epocher: self.epocher.clone(),
            strategy: self.strategy.clone(),
            gates: self.gates.clone(),
            build_duration: self.build_duration.clone(),
            verify_duration: self.verify_duration.clone(),
            proposal_parent_fetch_duration: self.proposal_parent_fetch_duration.clone(),
            ancestor_fetch_duration: self.ancestor_fetch_duration.clone(),
            erasure_encode_duration: self.erasure_encode_duration.clone(),
        }
    }
}

impl<E, A, B, C, H, Z, S, ES> Marshaled<E, A, B, C, H, Z, S, ES>
where
    E: Spawner + Metrics + Clock,
    A: Application<
            E,
            Block = B,
            SigningScheme = Z::Scheme,
            Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
            Input = (),
        >,
    B: CertifiableBlock<Context = <A as Application<E>>::Context> + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    /// Creates a new [`Marshaled`] wrapper.
    ///
    /// # Panics
    ///
    /// Panics if the marshal metadata store cannot be initialized.
    pub fn new(context: E, cfg: MarshaledConfig<A, B, C, H, Z, S, ES>) -> Self {
        let MarshaledConfig {
            application,
            marshal,
            shards,
            scheme_provider,
            strategy,
            epocher,
        } = cfg;

        let build_histogram = context.histogram(
            "build_duration",
            "Histogram of time taken for the application to build a new block, in seconds",
            Buckets::LOCAL,
        );
        let build_duration = Timed::new(build_histogram);

        let verify_histogram = context.histogram(
            "verify_duration",
            "Histogram of time taken for the application to verify a block, in seconds",
            Buckets::LOCAL,
        );
        let verify_duration = Timed::new(verify_histogram);

        let parent_fetch_histogram = context.histogram(
            "parent_fetch_duration",
            "Histogram of time taken to fetch a parent block in proposal, in seconds",
            Buckets::LOCAL,
        );
        let proposal_parent_fetch_duration = Timed::new(parent_fetch_histogram);

        let ancestor_fetch_histogram = context.histogram(
            "ancestor_fetch_duration",
            "Histogram of time taken to fetch a block via the ancestry stream, in seconds",
            Buckets::LOCAL,
        );
        let ancestor_fetch_duration = Timed::new(ancestor_fetch_histogram);

        let erasure_histogram = context.histogram(
            "erasure_encode_duration",
            "Histogram of time taken to erasure encode a built block on the strategy, in seconds",
            Buckets::LOCAL,
        );
        let erasure_encode_duration = Timed::new(erasure_histogram);

        Self {
            context: Arc::new(context),
            application,
            marshal,
            shards,
            scheme_provider,
            strategy,
            epocher,
            gates: Gates::new(),

            build_duration,
            verify_duration,
            proposal_parent_fetch_duration,
            ancestor_fetch_duration,
            erasure_encode_duration,
        }
    }

    /// Verifies a proposed block within epoch boundaries.
    ///
    /// This method validates that:
    /// 1. The block is within the current epoch (unless it's a boundary block re-proposal)
    /// 2. Re-proposals are only allowed for the last block in an epoch
    /// 3. The block's parent digest matches the consensus context's expected parent
    /// 4. The block's height is exactly one greater than the parent's height
    /// 5. The block's embedded context digest matches the commitment
    /// 6. The block's embedded context matches the consensus context
    /// 7. The underlying application's verification logic passes
    ///
    /// Verification is spawned in a background task and returns a receiver that will contain
    /// the verification result.
    ///
    /// If `prefetched_block` is provided, it is used directly. This is useful in `certify` when
    /// we've already fetched the block to extract its embedded context. Otherwise, a local-only
    /// wait for the block is sent to marshal before this method returns.
    fn deferred_verify(
        &self,
        consensus_context: Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
        commitment: Commitment<B, C, H>,
        prefetched_block: Option<Arc<CodedBlock<B, C, H>>>,
        stage: Stage,
    ) -> oneshot::Receiver<GateOutcome> {
        let marshal = self.marshal.clone();
        let mut application = self.application.clone();
        let epocher = self.epocher.clone();
        let verify_duration = self.verify_duration.clone();
        let ancestor_fetch_duration = self.ancestor_fetch_duration.clone();

        // Verification needs the full block but waits only for local delivery. Certification starts
        // recovery only when the block is not available locally. If the shard engine evicts a
        // cached block before verification registers its wait, verification is left with neither
        // the block nor an active fetch. Register the wait before the caller publishes the gate so
        // it receives the cached block or is waiting when recovery delivers it.
        let candidate = prefetched_block.map_or_else(
            || {
                Either::Right(
                    marshal.subscribe_by_commitment(commitment, core::CommitmentFallback::Wait),
                )
            },
            Either::Left,
        );

        let (mut tx, rx) = oneshot::channel();
        let context = self
            .context
            .child("deferred_verify")
            .with_attribute("round", consensus_context.round);
        let span = info_span!(
            "marshal.coding.verify.deferred",
            round = %consensus_context.round,
            commitment = %commitment
        );
        context.spawn(move |runtime_context| {
            async move {
                let round = consensus_context.round;
                let (parent_view, parent_commitment) = consensus_context.parent;

                // Start the parent fetch immediately so it can proceed in parallel
                // with candidate reconstruction. The parent round comes from the
                // caller's context (the certified consensus context in verify, the
                // quorum-defended embedded context in certify), never from the
                // unverified child block.
                let parent_request = marshal.subscribe_by_commitment(
                    parent_commitment,
                    core::CommitmentFallback::FetchByRound {
                        round: Round::new(consensus_context.epoch(), parent_view),
                    },
                );

                // Get the candidate block either from the caller or by waiting for
                // local reconstruction. Candidate data remains local-only: a
                // notarization is not sufficient reason to request it from peers.
                let block = match candidate {
                    Either::Left(block) => block,
                    Either::Right(block_request) => select! {
                        _ = tx.closed() => {
                            debug!(
                                reason = "consensus dropped receiver",
                                "skipping verification"
                            );
                            return;
                        },
                        result = block_request => match result {
                            Ok(block) => block,
                            Err(_) => {
                                debug!(reason = "block unavailable", "skipping verification");
                                return;
                            }
                        },
                    },
                };

                // Start the candidate store immediately: it depends on neither the
                // parent fetch (which may hit the network) nor the verdict below.
                // Storing before validation is intentional: these caches provide
                // candidate availability/recovery, not a validity decision. This
                // task gates the finalize vote by resolving true only after both
                // app verification succeeds and the store is durable.
                let store = stage.store(&marshal, round, Arc::clone(&block));
                let verify = async {
                    // Await the parent fetch we started above.
                    let parent = select! {
                        _ = tx.closed() => {
                            debug!(
                                reason = "consensus dropped receiver",
                                "skipping verification"
                            );
                            return None;
                        },
                        result = parent_request => match result {
                            Ok(parent) => parent,
                            Err(_) => {
                                debug!(reason = "failed to fetch parent", "skipping verification");
                                return None;
                            }
                        },
                    };

                    if let Err(err) = validate_block(
                        &epocher,
                        block.as_ref(),
                        parent.as_ref(),
                        &consensus_context,
                        commitment,
                        parent_commitment,
                    ) {
                        debug!(
                            ?err,
                            expected_commitment = %commitment,
                            block_commitment = %block.commitment(),
                            expected_parent_commitment = %parent_commitment,
                            parent_commitment = %parent.commitment(),
                            expected_parent = %parent.digest(),
                            block_parent = %block.parent(),
                            parent_height = %parent.height(),
                            block_height = %block.height(),
                            "block failed coded invariant validation"
                        );
                        return Some(false);
                    }

                    let ancestry_stream = marshal.ancestor_stream(
                        Arc::new(runtime_context.child("ancestor_stream")),
                        [block.inner_digested(), parent.inner_digested()],
                        ancestor_fetch_duration,
                    );
                    let validity_request = application
                        .verify(
                            (
                                runtime_context.child("app_verify"),
                                consensus_context.clone(),
                            ),
                            ancestry_stream,
                        )
                        .instrument(info_span!(
                            "marshal.coding.application.verify",
                            round = %consensus_context.round,
                            commitment = %commitment,
                            parent_view = parent_view.traced(),
                            parent = %parent_commitment
                        ));

                    // If consensus drops the receiver, we can stop work early.
                    let timer = verify_duration.timer(&runtime_context);
                    let result = select! {
                        _ = tx.closed() => {
                            debug!(
                                reason = "consensus dropped receiver",
                                "skipping verification"
                            );
                            None
                        },
                        is_valid = validity_request => Some(is_valid),
                    };
                    timer.observe(&runtime_context);
                    result
                };
                let (verdict, durable) = futures::join!(verify, store);

                // Publish only when the block is both valid and durable. App-invalid
                // candidates may already be in the cache from the concurrent store above,
                // so the gate verdict is the authority for consensus progress.
                if let Some(application_valid) = gates::resolve(verdict, durable) {
                    tx.send_lossy(GateOutcome::Ready(application_valid));
                }
            }
            .instrument(span)
        });

        rx
    }

    fn certify_from_embedded_context(
        &self,
        round: Round,
        payload: Commitment<B, C, H>,
    ) -> oneshot::Receiver<bool> {
        // Certify may be reached without an earlier `verify`, so the shard
        // engine may not know the leader yet. A notarized commitment is still
        // enough to start reconstruction from sender-indexed gossip shards
        // already buffered for the commitment.
        self.shards.notarized(payload, round);

        // No in-progress task means we never verified this proposal locally.
        // We can use the block's embedded context to move to the next view. If a Byzantine
        // proposer embedded a malicious context, the f+1 honest validators from the notarizing quorum
        // will verify against the proper context and reject the mismatch, preventing a 2f+1
        // finalization quorum.
        //
        // We must fetch here rather than only wait for local reconstruction. A Byzantine
        // leader can send enough shards to just f+1 honest validators, collect enough honest
        // notarize votes to form a notarization, and leave the remaining honest validators
        // unable to reconstruct the block. Those validators need the notarized round to
        // recover and certify; otherwise they can remain stuck if the Byzantine validators
        // stop participating in the next view.
        //
        // Subscribe to the block and verify using its embedded context once available.
        debug!(
            ?round,
            ?payload,
            "subscribing to block for certification using embedded context"
        );
        let block_rx = self
            .marshal
            .subscribe_by_commitment(payload, core::CommitmentFallback::FetchByRound { round });
        let marshaled = self.clone();
        let shards = self.shards.clone();
        let (mut tx, rx) = oneshot::channel();
        let context = self.context.child("certify").with_attribute("round", round);
        context.spawn(move |_| {
            async move {
                let block = select! {
                    _ = tx.closed() => {
                        debug!(
                            reason = "consensus dropped receiver",
                            "skipping certification"
                        );
                        return;
                    },
                    result = block_rx => match result {
                        Ok(block) => block,
                        Err(_) => {
                            debug!(
                                ?payload,
                                reason = "failed to fetch block for certification",
                                "skipping certification"
                            );
                            return;
                        }
                    },
                };

                // Re-proposal detection for certify path: we don't have the consensus
                // context, only the block's embedded context from original proposal.
                // Infer re-proposal from:
                // 1. Block is at epoch boundary (only boundary blocks can be re-proposed)
                // 2. Certification round's view > embedded context's view (re-proposals
                //    retain their original embedded context, so a later view indicates
                //    the block was re-proposed)
                // 3. Same epoch (re-proposals don't cross epoch boundaries)
                let embedded_context = block.context();
                let is_reproposal = is_inferred_reproposal_at_certify(
                    &marshaled.epocher,
                    block.height(),
                    embedded_context.round,
                    round,
                );
                if is_reproposal {
                    // Certifier holds a notarization for this block, so route
                    // the write to the notarized cache. `certified` is
                    // idempotent, so crash-recovery double-invocation is safe.
                    if !marshaled.marshal.certified(round, block).await {
                        return;
                    }
                    tx.send_lossy(true);
                    return;
                }

                // Inform the shard engine of an externally proposed commitment.
                shards.discovered(
                    payload,
                    embedded_context.leader.clone(),
                    embedded_context.round,
                );

                // Use the block's embedded context for verification, passing the
                // prefetched block to avoid fetching it again inside deferred_verify.
                let verify_rx = marshaled.deferred_verify(
                    embedded_context,
                    payload,
                    Some(block),
                    Stage::Certified,
                );
                gates::forward(tx, verify_rx, |result| match result {
                    GateOutcome::Ready(result) => Some(result),
                    GateOutcome::Recover => None,
                })
                .await;
            }
            .instrument(info_span!(
                "marshal.coding.certify.embedded",
                round = %round,
                commitment = %payload
            ))
        });
        rx
    }
    fn certify_from_existing_task(
        &self,
        round: Round,
        payload: Commitment<B, C, H>,
        task: oneshot::Receiver<GateOutcome>,
    ) -> oneshot::Receiver<bool> {
        // `verify()` intentionally waits only for local candidate data. Once
        // certification starts, a notarization exists and the same pending
        // verifier must be unblocked by round-bound recovery if local
        // reconstruction never completes.
        self.shards.notarized(payload, round);
        self.marshal.hint_notarized(round, payload);

        // A completed gate either carries an applicable local verdict or requests
        // recovery. After an unclean restart the in-memory task is gone, which also
        // recovers via the embedded-context fetch path.
        let marshaled = self.clone();
        let (tx, rx) = oneshot::channel();
        let context = self
            .context
            .child("certify_existing")
            .with_attribute("round", round);
        context.spawn(move |_| {
            gates::drive(tx, task, round, payload, move || {
                marshaled.certify_from_embedded_context(round, payload)
            })
            .instrument(info_span!(
                "marshal.coding.certify.existing",
                round = %round,
                commitment = %payload
            ))
        });
        rx
    }

    /// Runs the proposal checks for `consensus_context` and yields what the proposal path does
    /// next: re-propose the epoch boundary block, decline a parent the marshal cannot build on,
    /// or build on the fetched parent.
    ///
    /// `parent_fallback` decides how a parent missing locally is acquired. A propose request
    /// fetches it by round. A prepare request builds on a parent that only our own shard vote may
    /// attest to, so it draws no peer fetch. Local reconstruction or the parent's certification,
    /// which acquires the block from peers, delivers it.
    ///
    /// The returned future borrows nothing from `self`, so it can move into the spawned prepare
    /// or propose task.
    fn checks(
        &self,
        consensus_context: Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
        parent_fallback: core::CommitmentFallback,
    ) -> impl Future<Output = CodingResolved<B, C, H, impl Ancestry<B>>> + Send + 'static {
        let scheme_provider = self.scheme_provider.clone();
        let context = self.context.clone();
        let marshal = self.marshal.clone();
        let epocher = self.epocher.clone();
        let proposal_parent_fetch_duration = self.proposal_parent_fetch_duration.clone();
        let ancestor_fetch_duration = self.ancestor_fetch_duration.clone();
        let build_duration = self.build_duration.clone();
        async move {
            let Some(scheme) = scheme_provider.scheme(consensus_context.epoch()) else {
                debug!(round = %consensus_context.round, "no scheme for epoch, skipping proposal");
                return Resolved::Skip;
            };
            let n_participants =
                u16::try_from(scheme.participants().len()).expect("too many participants");
            let coding_config = coding_config_for_participants(n_participants);

            // The parent for any consensus context is in the same epoch: the
            // boundary block of the previous epoch is the genesis block of the
            // current epoch.
            //
            // Proposal context carries the parent view and commitment but not
            // the parent height. The parent may sit above the finalized tip and
            // may still be uncertified, so a fetch must stay round-bound until the
            // block is returned.
            let (parent_view, parent_commitment) = consensus_context.parent;
            let parent_request =
                marshal.subscribe_by_commitment(parent_commitment, parent_fallback);
            let parent_timer = proposal_parent_fetch_duration.timer(&*context);
            let Ok(parent) = parent_request.await else {
                debug!(
                    ?parent_commitment,
                    reason = "failed to fetch parent block",
                    "skipping proposal"
                );
                return Resolved::Skip;
            };
            parent_timer.observe(&*context);

            // Coding validators vote before reconstructing a block, so an uncertified handoff
            // parent can carry any height. Build only on a parent whose height maps to the
            // request's epoch, since any other parent can never certify. A parent at view zero
            // is the epoch's genesis: for a later epoch that is the previous epoch's boundary
            // block, whose height maps to the previous epoch.
            if parent_view != View::zero()
                && !is_block_in_expected_epoch(&epocher, parent.height(), consensus_context.epoch())
            {
                debug!(
                    ?parent_commitment,
                    reason = "parent is outside the epoch",
                    "skipping proposal"
                );
                return Resolved::Skip;
            }

            // Special case: If the parent block is the last block in the epoch,
            // re-propose it so as not to produce any blocks that will be cut out
            // by the epoch transition.
            let last_in_epoch = epocher
                .last(consensus_context.epoch())
                .expect("current epoch should exist");
            if parent.height() == last_in_epoch {
                return Resolved::Reuse(parent.commitment(), parent);
            }

            let ancestor_stream = marshal.ancestor_stream(
                context.clone(),
                [parent.inner_digested()],
                ancestor_fetch_duration,
            );
            Resolved::Build(
                ancestor_stream,
                build_duration.timer(&*context),
                coding_config,
            )
        }
    }
}

impl<E, A, B, C, H, Z, S, ES> Automaton for Marshaled<E, A, B, C, H, Z, S, ES>
where
    E: Spawner + Metrics + Clock,
    A: Application<
            E,
            Block = B,
            SigningScheme = Z::Scheme,
            Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
            Input = (),
        >,
    B: CertifiableBlock<Context = <A as Application<E>>::Context> + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    type Digest = Commitment<B, C, H>;
    type Context = Context<Self::Digest, <Z::Scheme as Verifier>::PublicKey>;

    /// Proposes a new block or re-proposes the epoch boundary block.
    ///
    /// This method builds a new block from the underlying application unless the parent block
    /// is the last block in the current epoch. When at an epoch boundary, it re-proposes the
    /// boundary block to avoid creating blocks that would be invalidated by the epoch transition.
    ///
    /// The proposal operation is spawned in a background task and returns a receiver that will
    /// contain the proposed block's commitment when ready. The block is staged before the
    /// commitment is delivered and handed to marshal when consensus requests the relay
    /// broadcast, which persists it after the shards are sent. The resulting sync handle is
    /// awaited only at certification so it overlaps consensus voting. The commitment does not
    /// imply durability on its own. [`CertifiableAutomaton::certify`] awaits the registered
    /// certification gate before the finalize vote.
    #[allow(clippy::async_yields_async)]
    #[tracing::instrument(name = "marshal.coding.propose", level = "info", skip_all, fields(round = %consensus_context.round))]
    async fn propose(
        &mut self,
        consensus_context: Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
    ) -> oneshot::Receiver<Self::Digest> {
        let strategy = self.strategy.clone();
        let erasure_encode_duration = self.erasure_encode_duration.clone();
        let clock = self.context.clone();
        let (parent_view, parent_commitment) = consensus_context.parent;
        let checks = self.checks(
            consensus_context.clone(),
            core::CommitmentFallback::FetchByRound {
                round: Round::new(consensus_context.epoch(), parent_view),
            },
        );
        let span = info_span!(
            "marshal.coding.propose.task",
            round = %consensus_context.round
        );
        let round = consensus_context.round;
        build::propose(
            self.context.as_ref(),
            &self.application,
            self.gates.clone(),
            consensus_context,
            checks,
            span,
            move || {
                info_span!(
                    "marshal.coding.application.propose",
                    round = %round,
                    parent_view = parent_view.traced(),
                    parent = %parent_commitment
                )
            },
            move |block: Digested<B>, coding_config| async move {
                let erasure_timer = erasure_encode_duration.timer(&*clock);
                let coded = CodedBlock::encode(block, coding_config, &strategy).await;
                erasure_timer.observe(&*clock);
                (coded.commitment(), Arc::new(coded))
            },
        )
    }

    /// Verifies a received shard for a given round.
    ///
    /// This method validates that:
    /// 1. The coding configuration matches the expected configuration for the current scheme.
    /// 2. The commitment's context digest matches the consensus context (unless this is a re-proposal).
    /// 3. The shard is contained within the consensus commitment.
    ///
    /// Verification is spawned in a background task and returns a receiver that will contain
    /// the verification result. Additionally, this method kicks off deferred verification to
    /// start block verification early (hidden behind shard validity and network latency).
    #[allow(clippy::async_yields_async)]
    #[tracing::instrument(name = "marshal.coding.verify", level = "info", skip_all, fields(round = %consensus_context.round, commitment = %payload))]
    async fn verify(
        &mut self,
        consensus_context: Context<Self::Digest, <Z::Scheme as Verifier>::PublicKey>,
        payload: Self::Digest,
    ) -> oneshot::Receiver<bool> {
        // If there's no scheme for the current epoch, we cannot vote on the proposal.
        // Send back a receiver with a dropped sender.
        let Some(scheme) = self.scheme_provider.scheme(consensus_context.epoch()) else {
            debug!(
                round = %consensus_context.round,
                "no scheme for epoch, skipping verify"
            );
            let (_, rx) = oneshot::channel();
            return rx;
        };

        let n_participants =
            u16::try_from(scheme.participants().len()).expect("too many participants");
        let coding_config = coding_config_for_participants(n_participants);
        let is_reproposal = payload == consensus_context.parent.1;

        // Validate proposal-level invariants:
        // - coding config must match active participant set
        // - context digest must match unless this is a re-proposal
        let proposal_context = (!is_reproposal).then_some(&consensus_context);
        if let Err(err) = validate_proposal(payload, coding_config, proposal_context) {
            match err {
                ProposalError::CodingConfig => {
                    warn!(
                        round = %consensus_context.round,
                        got = ?payload.config(),
                        expected = ?coding_config,
                        "rejected proposal with unexpected coding configuration"
                    );
                }
                ProposalError::ContextDigest => {
                    let expected = hash_context::<H, _>(&consensus_context);
                    let got = payload.context();
                    warn!(
                        round = %consensus_context.round,
                        expected = ?expected,
                        got = ?got,
                        "rejected proposal with mismatched context digest"
                    );
                }
            }

            let (tx, rx) = oneshot::channel();
            tx.send_lossy(false);
            return rx;
        }

        // Re-proposals skip context-digest validation because the consensus context will point
        // at the prior epoch-boundary block while the embedded block context is from the
        // original proposal view.
        //
        // Re-proposals also skip shard-validity and deferred verification because:
        // 1. Consensus settles the block's validity when certifying the view that first carried it
        // 2. The parent-child height check would fail (parent IS the block)
        // 3. Waiting for shards could stall if the leader doesn't rebroadcast
        if is_reproposal {
            // Fetch the block to verify it's at the epoch boundary. This should be fast
            // since the parent block is typically already cached. A re-proposal names its
            // own parent, so the parent round is a certified round for this commitment and
            // lets a participant that never received the original proposal acquire it
            // instead of waiting for shards it cannot yet classify.
            let (parent_view, _) = consensus_context.parent;
            let block_rx = self.marshal.subscribe_by_commitment(
                payload,
                core::CommitmentFallback::FetchByRound {
                    round: Round::new(consensus_context.epoch(), parent_view),
                },
            );
            let marshal = self.marshal.clone();
            let shards = self.shards.clone();
            let epocher = self.epocher.clone();
            let round = consensus_context.round;
            let leader = consensus_context.leader;
            let gates = self.gates.clone();

            // Register a certification gate task synchronously before spawning work so
            // `certify` can always find it (no race with task startup).
            let (task_tx, task_rx) = oneshot::channel();
            gates.insert(round, payload, task_rx);

            let (mut tx, rx) = oneshot::channel();
            let context = self
                .context
                .child("verify_reproposal")
                .with_attribute("round", round);
            context.spawn(move |_| {
                async move {
                    let block = select! {
                        _ = tx.closed() => {
                            debug!(
                                reason = "consensus dropped receiver",
                                "skipping re-proposal verification"
                            );
                            return;
                        },
                        block = block_rx => match block {
                            Ok(block) => block,
                            Err(_) => {
                                debug!(
                                    ?payload,
                                    reason = "failed to fetch block for re-proposal verification",
                                    "skipping re-proposal verification"
                                );
                                // Fetch failure is an availability issue, not an explicit
                                // invalidity proof. Do not synthesize `false` here.
                                return;
                            }
                        },
                    };

                    // A rejection here is safe to publish as a gate verdict because
                    // the boundary check is intrinsic to `(round, commitment)`: it
                    // reads only the block's height and the round's epoch. The
                    // commitment also binds the original proposal context, so no
                    // honest notarization can form for this key under a conflicting
                    // header.
                    if !is_valid_reproposal_at_verify(&epocher, block.height(), round.epoch()) {
                        debug!(
                            height = %block.height(),
                            "re-proposal is not at epoch boundary"
                        );
                        task_tx.send_lossy(GateOutcome::Ready(false));
                        tx.send_lossy(false);
                        return;
                    }

                    // Announce the re-proposal only after the boundary check. A
                    // re-proposal's consensus round is not bound to the commitment, and
                    // the shard engine reads the participant set from the round's epoch,
                    // so announcing an unvalidated round would classify this block's
                    // shards against the wrong epoch.
                    shards.discovered(payload, leader, round);

                    // Valid re-proposal: notify the marshal and complete the
                    // certification gate task for `certify`.
                    let durable = marshal.verified(round, block).await;
                    if !durable {
                        return;
                    }
                    task_tx.send_lossy(GateOutcome::Ready(true));
                    tx.send_lossy(true);
                }
                .instrument(info_span!(
                    "marshal.coding.verify.reproposal",
                    round = %round,
                    commitment = %payload
                ))
            });
            return rx;
        }

        // Inform the shard engine of an externally proposed commitment. The context
        // digest validated above binds this round and leader to the commitment.
        self.shards.discovered(
            payload,
            consensus_context.leader.clone(),
            consensus_context.round,
        );

        // Kick off deferred verification early to hide verification latency behind
        // shard validity checks and network latency for collecting votes.
        //
        // The task's cancellation signal is the gate receiver registered below,
        // not consensus's verify receiver. Nullification advances the view and
        // drops the verify receiver without cancelling certification for it, so
        // deferred verification must survive that drop for certify to consume.
        let round = consensus_context.round;
        let task = self.deferred_verify(consensus_context, payload, None, Stage::Verified);
        self.gates.insert(round, payload, task);

        match scheme.me().and_then(|m| m.signer()) {
            Some(_) => {
                // Subscribe to assigned shard verification. For signers, this only
                // completes once the shard for our assigned index has been verified.
                // Reconstructing the block from peer gossip is useful for certification
                // later, but is not enough to emit a notarize vote.
                let validity_rx = self.shards.subscribe_assigned_shard_verified(payload);
                let (tx, rx) = oneshot::channel();
                let context = self
                    .context
                    .child("shard_validity_wait")
                    .with_attribute("round", round);
                context.spawn(move |_| {
                    async move {
                        gates::forward(tx, validity_rx, |()| Some(true)).await;
                    }
                    .instrument(info_span!(
                        "marshal.coding.verify.shard_validity",
                        round = %round,
                        commitment = %payload
                    ))
                });
                rx
            }
            None => {
                // A node that cannot sign casts no notarize vote, so it accepts the proposal
                // without waiting for its assigned shard.
                //
                // Later, when certifying, we will wait to receive the block from the network.
                let (tx, rx) = oneshot::channel();
                tx.send_lossy(true);
                rx
            }
        }
    }
}

impl<E, A, B, C, H, Z, S, ES> CertifiableAutomaton for Marshaled<E, A, B, C, H, Z, S, ES>
where
    E: Spawner + Metrics + Clock,
    A: Application<
            E,
            Block = B,
            SigningScheme = Z::Scheme,
            Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
            Input = (),
        >,
    B: CertifiableBlock<Context = <A as Application<E>>::Context> + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    /// Prepares a term-start proposal on an uncertified parent.
    ///
    /// Runs the same checks as [`Self::propose`], in the same order, except that a parent missing
    /// locally is awaited rather than fetched, and hands the application the parent's ancestry.
    /// When the checks find the epoch boundary block, the marshal re-proposes it without the
    /// application and answers [`Handoff::Stage`]. A round without a scheme answers
    /// [`Handoff::Wait`].
    #[allow(clippy::async_yields_async)]
    #[tracing::instrument(name = "marshal.coding.prepare", level = "info", skip_all, fields(round = %consensus_context.round))]
    async fn prepare(
        &mut self,
        consensus_context: Context<Self::Digest, <Z::Scheme as Verifier>::PublicKey>,
    ) -> oneshot::Receiver<Handoff<Self::Digest>> {
        let strategy = self.strategy.clone();
        let erasure_encode_duration = self.erasure_encode_duration.clone();
        let clock = self.context.clone();
        let (parent_view, parent_commitment) = consensus_context.parent;
        let checks = self.checks(consensus_context.clone(), core::CommitmentFallback::Wait);
        let span = info_span!(
            "marshal.coding.prepare.task",
            round = %consensus_context.round
        );
        let round = consensus_context.round;
        build::prepare(
            self.context.as_ref(),
            &self.application,
            self.gates.clone(),
            consensus_context,
            checks,
            span,
            move || {
                info_span!(
                    "marshal.coding.application.prepare",
                    round = %round,
                    parent_view = parent_view.traced(),
                    parent = %parent_commitment
                )
            },
            move |block: Digested<B>, coding_config| async move {
                let erasure_timer = erasure_encode_duration.timer(&*clock);
                let coded = CodedBlock::encode(block, coding_config, &strategy).await;
                erasure_timer.observe(&*clock);
                (coded.commitment(), Arc::new(coded))
            },
        )
    }

    #[allow(clippy::async_yields_async)]
    #[tracing::instrument(name = "marshal.coding.certify", level = "info", skip_all, fields(round = %round, commitment = %payload))]
    async fn certify(&mut self, round: Round, payload: Self::Digest) -> oneshot::Receiver<bool> {
        // First, check for an in-progress certification gate task.
        let task = self.gates.claim(round, payload, |block, ack| {
            self.marshal.verified_deferred(round, block, ack)
        });
        if let Some(task) = task {
            return self.certify_from_existing_task(round, payload, task);
        }

        self.certify_from_embedded_context(round, payload)
    }
}

impl<E, A, B, C, H, Z, S, ES> Relay for Marshaled<E, A, B, C, H, Z, S, ES>
where
    E: Send + Sync + 'static,
    A: Application<
            E,
            Block = B,
            Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
        >,
    B: CertifiableBlock<Context = <A as Application<E>>::Context> + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    type Digest = Commitment<B, C, H>;
    type PublicKey = <Z::Scheme as Verifier>::PublicKey;
    type Plan = Plan<Self::PublicKey>;

    fn broadcast(&mut self, commitment: Self::Digest, plan: Self::Plan) -> Feedback {
        // Coding variant does not support targeted forwarding. Peers reconstruct
        // blocks from erasure-coded shards.
        if matches!(plan, Plan::Forward { .. }) {
            return Feedback::Ok;
        }
        relay::broadcast(&self.gates, &self.marshal, commitment, plan)
    }
}

/// The wrapper is the consensus engine's reporter, as it is its automaton and relay: it forwards
/// every activity to marshal and first releases the state it keeps per round (certification
/// gates, staged proposals, and sent marks). A finalization releases its round and the rounds
/// below it. Any activity of a later epoch releases every earlier epoch, whose engine has been
/// retired, which covers finalized progress that marshal imported while no engine reported it.
/// The voter reports from the same task that requests the lock-in broadcast and certification,
/// so pruning is ordered with both.
impl<E, A, B, C, H, Z, S, ES> Reporter for Marshaled<E, A, B, C, H, Z, S, ES>
where
    E: Send + Sync + 'static,
    A: Application<
            E,
            Block = B,
            Context = Context<Commitment<B, C, H>, <Z::Scheme as Verifier>::PublicKey>,
        >,
    B: CertifiableBlock<Context = <A as Application<E>>::Context> + Codec,
    C: CodingScheme,
    H: Hasher,
    Z: Provider<Scope = Epoch, Scheme: Scheme<Commitment<B, C, H>>>,
    S: Strategy,
    ES: Epocher,
{
    type Activity = Activity<Z::Scheme, Commitment<B, C, H>>;

    fn report(&mut self, activity: Self::Activity) -> Feedback {
        let retired = match &activity {
            Activity::Finalization(finalization) => finalization.round(),
            activity => Round::new(activity.epoch(), View::zero()),
        };
        self.gates.retain_after(retired);
        self.marshal.report(activity)
    }
}

#[cfg(test)]
mod tests {
    use super::Marshaled;
    use crate::{
        marshal::{
            application::gates::Gates,
            coding::types::CodedBlock,
            mocks::{
                harness::{CodingB, P, S},
                verifying::MockVerifyingApp,
            },
        },
        types::{FixedEpocher, coding::Commitment},
    };
    use commonware_coding::ReedSolomon;
    use commonware_cryptography::sha256::Sha256;
    use commonware_parallel::Sequential;
    use commonware_runtime::deterministic;
    use std::sync::Arc;

    type TestCommitment = Commitment<CodingB, ReedSolomon<Sha256>, Sha256>;
    type TestCodedBlock = CodedBlock<CodingB, ReedSolomon<Sha256>, Sha256>;

    impl
        Marshaled<
            deterministic::Context,
            MockVerifyingApp<CodingB, S>,
            CodingB,
            ReedSolomon<Sha256>,
            Sha256,
            P,
            Sequential,
            FixedEpocher,
        >
    {
        /// Exposes the gates this wrapper prunes to the tests of the coding variant.
        pub(in crate::marshal::coding) const fn gates(
            &self,
        ) -> &Gates<TestCommitment, Arc<TestCodedBlock>> {
            &self.gates
        }
    }
}