miden-processor 1.0.0-rc.0

use alloc::{boxed::Box, collections::VecDeque, string::ToString, sync::Arc, vec::Vec};

use miden_air::trace::{
    RowIndex,
    chiplets::hasher::{HasherState, MERKLE_DEPTH_RANGE_SCALE, STATE_WIDTH},
};
use miden_core::{
    mast::{BasicBlockNode, ExecutableMastForest, MastNode, MastNodeExt, OpBatch},
    serde::{
        ByteReader, ByteWriter, Deserializable, DeserializationError, Serializable,
        read_bounded_len,
    },
};

use crate::{
    AceError, ContextId, ExecutionError, Felt, MIN_STACK_DEPTH, MemoryError, ONE, Word, ZERO,
    advice::AdviceError,
    continuation_stack::ContinuationStack,
    crypto::merkle::MerklePath,
    errors::OperationError,
    mast::{MastForestId, MastNodeId, SparseMastForest},
    processor::{
        AdviceProviderInterface, HasherInterface, MemoryInterface, Processor, SystemInterface,
    },
    trace::chiplets::{AEAD_STREAM_CYCLE_LEN, CircuitEvaluation, MAX_EVAL_CIRCUIT_INVOCATIONS},
    tracer::merkle_index_helper_values,
    utils::Idx,
};

mod serde;

#[cfg(test)]
mod serialization_tests;

// TRACE FRAGMENT CONTEXT
// ================================================================================================

/// Information required to build a core trace fragment (i.e. the system, decoder and stack
/// columns).
///
/// This struct is meant to be built by the processor, and consumed mutably by a core trace fragment
/// builder. That is, as core trace generation progresses, this struct can be mutated to represent
/// the generation context at any clock cycle within the fragment.
///
/// This struct is conceptually divided into 4 components:
/// 1. core trace state: the state of the processor at any clock cycle in the fragment, initialized
///    to the state at the first clock cycle in the fragment,
/// 2. execution replay: information needed to replay the execution of the processor for the
///    remainder of the fragment,
/// 3. continuation: a stack of continuations for the processor representing the nodes in the MAST
///    forest to execute when the current node is done executing,
/// 4. initial MAST forest: the MAST forest being executed at the start of the fragment (which can
///    change during execution when encountering an [`miden_core::mast::ExternalNode`] or
///    [`miden_core::mast::DynNode`]).
///
/// Note: the binary wire format is deliberately lossy because the embedded
/// [`ContinuationStack`](crate::continuation_stack::ContinuationStack) drops its debug metadata
/// (`package_debug_info` and `source_node_ids`), so the exact-equality round-trip test generated
/// by `serde_test` does not apply to this type.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct CoreTraceFragmentContext {
    pub state: CoreTraceState,
    pub replay: ExecutionReplay,
    /// Continuation stack with forest references encoded as [`MastForestId`]s into the
    /// `mast_forest_store` of the owning trace replay.
    pub continuation: ContinuationStack<MastForestId>,
    /// MAST forest active at the start of this fragment.
    pub initial_mast_forest_id: MastForestId,
}

// CORE TRACE STATE
// ================================================================================================

/// Subset of the processor state used to build the core trace (system, decoder and stack sets of
/// columns).
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct CoreTraceState {
    pub system: SystemState,
    pub decoder: DecoderState,
    pub stack: StackState,
}

// SYSTEM STATE
// ================================================================================================

/// The `SystemState` represents all the information needed to build one row of the System trace.
///
/// This struct captures the complete state of the system at a specific clock cycle, allowing for
/// reconstruction of the system trace during concurrent execution.
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct SystemState {
    /// Current clock cycle (row index in the trace)
    pub clk: RowIndex,

    /// Execution context ID - starts at 0 (root context), changes on CALL/SYSCALL operations
    pub ctx: ContextId,

    /// Hash of the function that initiated the current execution context
    /// - For root context: [ZERO; 4]
    /// - For CALL/DYNCALL contexts: hash of the called function
    /// - For SYSCALL contexts: hash remains from the calling function
    pub fn_hash: Word,

    /// Deferred root used for recording `log_deferred` calls.
    /// - Initially [ZERO; 4] (`TRUE_DIGEST`)
    /// - Updated with each deferred statement logged by `log_deferred`
    pub deferred_root: Word,
}

impl SystemState {
    /// Convenience constructor that creates a new `SystemState` from a `Processor`.
    pub(crate) fn from_processor<P: Processor>(processor: &P) -> Self {
        Self {
            clk: processor.system().clock(),
            ctx: processor.system().ctx(),
            fn_hash: processor.system().caller_hash(),
            deferred_root: processor.system().deferred_root(),
        }
    }
}

// DECODER STATE
// ================================================================================================

/// The subset of the decoder state required to build the trace.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct DecoderState {
    /// The value of the decoder's `addr` column.
    pub current_addr: Felt,
    /// The address of the current MAST node's parent.
    pub parent_addr: Felt,
}

impl DecoderState {
    /// This function is called when start executing a node (e.g. `JOIN`, `SPLIT`, etc). It emulates
    /// pushing a new node onto the block stack, and updates the decoder state to point to the
    /// current node in the block stack. Hence, the `current_addr` is set to the (replayed) address
    /// of the current node, and the `parent_addr` is set to the (replayed) address of the parent
    /// node (i.e. the node previously on top of the block stack).
    pub fn replay_node_start(
        &mut self,
        block_address_replay: &mut BlockAddressReplay,
        block_stack_replay: &mut BlockStackReplay,
    ) -> Result<(), ExecutionError> {
        self.current_addr = block_address_replay.replay_block_address()?;
        self.parent_addr = block_stack_replay.replay_node_start_parent_addr()?;
        Ok(())
    }

    /// This function is called when we hit an `END` operation, signaling the end of execution for a
    /// node. It updates the decoder state to point to the previous node in the block stack (which
    /// could be renamed to "node stack"), and returns the address of the node that just ended.
    pub fn replay_node_end(
        &mut self,
        block_stack_replay: &mut BlockStackReplay,
    ) -> Result<Felt, ExecutionError> {
        let node_end_data = block_stack_replay.replay_node_end()?;

        self.current_addr = node_end_data.prev_addr;
        self.parent_addr = node_end_data.prev_parent_addr;

        Ok(node_end_data.ended_node_addr)
    }
}

// STACK STATE
// ================================================================================================

/// This struct captures the state of the top 16 elements of the stack at a specific clock cycle;
/// that is, those elements that are written directly into the trace.
///
/// The stack trace consists of 19 columns total: 16 stack columns + 3 helper columns. The helper
/// columns (stack_depth, overflow_addr, and overflow_helper) are computed from the stack_depth and
/// last_overflow_addr fields.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct StackState {
    /// Top 16 stack slots (s0 to s15). These represent the top elements of the stack that are
    /// directly accessible.
    pub stack_top: [Felt; MIN_STACK_DEPTH], // 16 columns

    /// Current number of elements on the stack. It is guaranteed to be >= 16.
    stack_depth: usize,

    /// The last recorded overflow address for the stack - which is the clock cycle at which the
    /// last item was pushed to the overflow
    last_overflow_addr: Felt,
}

impl StackState {
    /// Creates a new StackState with the provided parameters.
    ///
    /// `stack_top` should be the top 16 elements of the stack stored in reverse order, i.e.,
    /// `stack_top[15]` is the topmost element (s0), and `stack_top[0]` is the bottommost element
    /// (s15).
    pub fn new(
        stack_top: [Felt; MIN_STACK_DEPTH],
        stack_depth: usize,
        last_overflow_addr: Felt,
    ) -> Self {
        Self {
            stack_top,
            stack_depth,
            last_overflow_addr,
        }
    }

    /// Returns the value at the specified index in the stack top.
    ///
    /// # Panics
    /// - if the index is greater than or equal to [MIN_STACK_DEPTH].
    pub fn get(&self, index: usize) -> Felt {
        self.stack_top[MIN_STACK_DEPTH - index - 1]
    }

    /// Returns the stack depth (b0 helper column)
    pub fn stack_depth(&self) -> usize {
        self.stack_depth
    }

    /// Returns the overflow address (b1 helper column) using the stack overflow replay
    pub fn overflow_addr(&self) -> Felt {
        self.last_overflow_addr
    }

    /// Returns the number of elements in the current context's overflow stack.
    pub fn num_overflow_elements_in_current_ctx(&self) -> usize {
        debug_assert!(self.stack_depth >= MIN_STACK_DEPTH);
        self.stack_depth - MIN_STACK_DEPTH
    }

    /// Pushes the given element onto the overflow stack at the provided clock cycle.
    pub fn push_overflow(&mut self, _element: Felt, clk: RowIndex) {
        self.stack_depth += 1;
        self.last_overflow_addr = clk.into();
    }

    /// Pops the top element from the overflow stack at the provided clock cycle, if any.
    ///
    /// If the overflow table is empty (i.e. stack depth is 16), the stack depth is unchanged, and
    /// None is returned.
    pub fn pop_overflow(
        &mut self,
        stack_overflow_replay: &mut StackOverflowReplay,
    ) -> Result<Option<Felt>, OperationError> {
        debug_assert!(self.stack_depth >= MIN_STACK_DEPTH);

        if self.stack_depth > MIN_STACK_DEPTH {
            let (stack_value, new_overflow_addr) = stack_overflow_replay.replay_pop_overflow()?;
            self.stack_depth -= 1;
            self.last_overflow_addr = new_overflow_addr;
            Ok(Some(stack_value))
        } else {
            self.last_overflow_addr = ZERO;
            Ok(None)
        }
    }

    /// Derives the denominator of the overflow helper (h0 helper column) from the current stack
    /// depth.
    ///
    /// It is expected that this values gets later inverted via batch inversion.
    pub fn overflow_helper(&self) -> Felt {
        let denominator = self.stack_depth() - MIN_STACK_DEPTH;
        Felt::new_unchecked(denominator as u64)
    }

    /// Starts a new execution context for this stack, resetting the stack depth to its minimum
    /// value, and last overflow address to 0.
    ///
    /// This has the effect of hiding the contents of the overflow table such that it appears as if
    /// the overflow table in the new context is empty.
    pub fn start_context(&mut self) {
        self.stack_depth = MIN_STACK_DEPTH;
        self.last_overflow_addr = ZERO;
    }

    /// Restores the prior context for this stack.
    ///
    /// This has the effect bringing back items previously hidden from the overflow table.
    pub fn restore_context(
        &mut self,
        stack_overflow_replay: &mut StackOverflowReplay,
    ) -> Result<(), OperationError> {
        let (stack_depth, last_overflow_addr) =
            stack_overflow_replay.replay_restore_context_overflow_addr()?;
        // Restore stack depth to the value from before the context switch (parallel to Process
        // Stack behavior)
        self.stack_depth = stack_depth;
        self.last_overflow_addr = last_overflow_addr;
        Ok(())
    }
}

/// Replay data necessary to build a trace fragment.
///
/// During execution, the processor records information to be replayed by the corresponding trace
/// generator. This is done due to the fact that the trace generators don't have access to some
/// components needed to produce those values, such as the memory chiplet, advice provider, etc. It
/// also packages up all the necessary data for trace generators to generate trace fragments, which
/// can be done on separate machines in parallel, for example.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct ExecutionReplay {
    pub block_stack: BlockStackReplay,
    pub execution_context: ExecutionContextReplay,
    pub stack_overflow: StackOverflowReplay,
    pub memory_reads: MemoryReadsReplay,
    pub advice: AdviceReplay,
    pub hasher: HasherResponseReplay,
    pub block_address: BlockAddressReplay,
    pub mast_forest_resolution: MastForestResolutionReplay,
}

// EXECUTION CONTEXT REPLAY
// ================================================================================================

#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct ExecutionContextReplay {
    /// Extra data needed to recover the state on an END operation specifically for
    /// CALL/SYSCALL/DYNCALL nodes (which start/end a new execution context).
    execution_contexts: VecDeque<ExecutionContextSystemInfo>,
}

impl ExecutionContextReplay {
    /// Records an execution context system info for a CALL/SYSCALL/DYNCALL operation.
    pub fn record_execution_context(&mut self, ctx_info: ExecutionContextSystemInfo) {
        self.execution_contexts.push_back(ctx_info);
    }

    /// Replays the next recorded execution context system info.
    pub fn replay_execution_context(
        &mut self,
    ) -> Result<ExecutionContextSystemInfo, OperationError> {
        self.execution_contexts
            .pop_front()
            .ok_or(OperationError::Internal("no execution context recorded"))
    }
}

// BLOCK STACK REPLAY
// ================================================================================================

/// Replay data for the block stack.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct BlockStackReplay {
    /// The parent address, recorded when a new node is started (JOIN, SPLIT, etc).
    node_start_parent_addr: VecDeque<Felt>,
    /// The data needed to recover the state on an END operation.
    node_end: VecDeque<NodeEndData>,
}

impl BlockStackReplay {
    /// Records the node's parent address
    pub fn record_node_start_parent_addr(&mut self, parent_addr: Felt) {
        self.node_start_parent_addr.push_back(parent_addr);
    }

    /// Records the necessary data needed to properly recover the state on an END operation.
    ///
    /// See [NodeEndData] for more details.
    pub fn record_node_end(
        &mut self,
        ended_node_addr: Felt,
        prev_addr: Felt,
        prev_parent_addr: Felt,
    ) {
        self.node_end.push_back(NodeEndData {
            ended_node_addr,
            prev_addr,
            prev_parent_addr,
        });
    }

    /// Replays the node's parent address
    pub fn replay_node_start_parent_addr(&mut self) -> Result<Felt, ExecutionError> {
        self.node_start_parent_addr
            .pop_front()
            .ok_or(ExecutionError::Internal("no node start parent address recorded"))
    }

    /// Replays the data needed to recover the state on an END operation.
    pub fn replay_node_end(&mut self) -> Result<NodeEndData, ExecutionError> {
        self.node_end
            .pop_front()
            .ok_or(ExecutionError::Internal("no node address and flags recorded"))
    }
}

/// The semantic block-stack entry kind consumed by an END operation.
#[derive(Debug, Clone, Copy)]
pub enum EndBlockStackEntryKind {
    /// An ordinary control-flow continuation.
    Continuation,
    /// A LOOP continuation, whose block-stack message carries the loop marker.
    LoopContinuation,
    /// A caller frame whose saved state is restored by this END.
    CallerFrame,
}

/// The flags written in the second word of the hasher state for END operations.
///
/// `entry_kind` is typed so trace construction cannot accidentally emit an invalid combination
/// such as a caller-frame END that is also marked as a LOOP END.
#[derive(Debug)]
pub struct EndBlockFlags {
    is_loop_body: bool,
    entry_kind: EndBlockStackEntryKind,
}

impl EndBlockFlags {
    /// Creates END flags for the specified block-stack entry kind.
    pub fn new(is_loop_body: bool, entry_kind: EndBlockStackEntryKind) -> Self {
        Self { is_loop_body, entry_kind }
    }

    /// Encodes the flags in their hasher-state column order.
    pub fn to_hasher_state_second_word(&self) -> Word {
        let is_loop_body = if self.is_loop_body { ONE } else { ZERO };
        let (is_loop, restores_caller_frame) = match self.entry_kind {
            EndBlockStackEntryKind::Continuation => (ZERO, ZERO),
            EndBlockStackEntryKind::LoopContinuation => (ONE, ZERO),
            EndBlockStackEntryKind::CallerFrame => (ZERO, ONE),
        };

        [is_loop_body, is_loop, restores_caller_frame, ZERO].into()
    }
}

/// The data needed to fully recover the state on an END operation.
///
/// We record `ended_node_addr` in order to be able to properly populate the trace row for the
/// node operation. Additionally, we record `prev_addr` and `prev_parent_addr` to allow emulating
/// peeking into the block stack, which is needed when processing REPEAT or RESPAN nodes.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct NodeEndData {
    /// the address of the node that is ending
    pub ended_node_addr: Felt,
    /// the address of the node sitting on top of the block stack after the END operation (or 0 if
    /// the block stack is empty)
    pub prev_addr: Felt,
    /// the parent address of the node sitting on top of the block stack after the END operation
    /// (or 0 if the block stack is empty)
    pub prev_parent_addr: Felt,
}

/// Data required to recover the state of an execution context when restoring it during an END
/// operation.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct ExecutionContextSystemInfo {
    pub parent_ctx: ContextId,
    pub parent_fn_hash: Word,
}

// MAST FOREST RESOLUTION REPLAY
// ================================================================================================

/// Records and replays the resolutions of [`crate::host::Host::get_mast_forest`].
///
/// These calls are made when encountering an [`miden_core::mast::ExternalNode`], or when
/// encountering a [`miden_core::mast::DynNode`] where the procedure hash on the stack refers to
/// a procedure not present in the current forest.
///
/// The forest reference is stored as a [`MastForestId`] into the `mast_forest_store` of the
/// trace replay that owns this data. This avoids holding a strong `Arc<MastForest>` reference per
/// resolution, allowing the trace generation context to deduplicate
/// forests across fragments.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct MastForestResolutionReplay {
    mast_forest_resolutions: VecDeque<(MastNodeId, MastForestId)>,
}

impl MastForestResolutionReplay {
    /// Records a resolution of a MastNodeId with the id of its associated MAST forest in the
    /// trace generation context's `mast_forest_store`.
    pub fn record_resolution(&mut self, node_id: MastNodeId, forest_id: MastForestId) {
        self.mast_forest_resolutions.push_back((node_id, forest_id));
    }

    /// Replays the next recorded MastForest resolution, returning both the node ID and the forest
    /// id.
    pub fn replay_resolution(&mut self) -> Result<(MastNodeId, MastForestId), ExecutionError> {
        self.mast_forest_resolutions
            .pop_front()
            .ok_or(ExecutionError::Internal("no MastForest resolutions recorded"))
    }

    pub(crate) fn iter_forest_ids(&self) -> impl Iterator<Item = MastForestId> + '_ {
        self.mast_forest_resolutions.iter().map(|(_node_id, forest_id)| *forest_id)
    }
}

// MEMORY REPLAY
// ================================================================================================

/// Records and replays all the reads made to memory, in which all elements and words read from
/// memory during a given fragment are recorded by the fast processor, and replayed by the main
/// trace fragment generators.
///
/// This is used to simulate memory reads in parallel trace generation without needing to actually
/// access the memory chiplet.
///
/// Elements/words read are stored with their addresses and are assumed to be read from the same
/// addresses that they were recorded at. This works naturally since the fast processor has exactly
/// the same access patterns as the main trace generators (which re-executes part of the program).
/// The read methods include debug assertions to verify address consistency.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct MemoryReadsReplay {
    elements_read: VecDeque<(Felt, Felt, ContextId, RowIndex)>,
    words_read: VecDeque<(Word, Felt, ContextId, RowIndex)>,
}

impl MemoryReadsReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------

    /// Records a read element from memory
    pub fn record_read_element(
        &mut self,
        element: Felt,
        addr: Felt,
        ctx: ContextId,
        clk: RowIndex,
    ) {
        self.elements_read.push_back((element, addr, ctx, clk));
    }

    /// Records a read word from memory
    pub fn record_read_word(&mut self, word: Word, addr: Felt, ctx: ContextId, clk: RowIndex) {
        self.words_read.push_back((word, addr, ctx, clk));
    }

    // ACCESSORS
    // --------------------------------------------------------------------------------

    pub fn replay_read_element(&mut self, addr: Felt) -> Result<Felt, MemoryError> {
        let (element, stored_addr, _ctx, _clk) = self
            .elements_read
            .pop_front()
            .ok_or(MemoryError::MemoryReadFailed("memory elements replay is empty".to_string()))?;
        debug_assert_eq!(stored_addr, addr, "Address mismatch: expected {addr}, got {stored_addr}");
        Ok(element)
    }

    pub fn replay_read_word(&mut self, addr: Felt) -> Result<Word, MemoryError> {
        let (word, stored_addr, _ctx, _clk) = self
            .words_read
            .pop_front()
            .ok_or(MemoryError::MemoryReadFailed("memory words replay is empty".to_string()))?;
        debug_assert_eq!(stored_addr, addr, "Address mismatch: expected {addr}, got {stored_addr}");
        Ok(word)
    }

    /// Returns an iterator over all recorded memory element reads, yielding tuples of
    /// (element, address, context ID, clock cycle).
    pub fn iter_read_elements(&self) -> impl Iterator<Item = (Felt, Felt, ContextId, RowIndex)> {
        self.elements_read.iter().copied()
    }

    /// Returns an iterator over all recorded memory word reads, yielding tuples of
    /// (word, address, context ID, clock cycle).
    pub fn iter_read_words(&self) -> impl Iterator<Item = (Word, Felt, ContextId, RowIndex)> {
        self.words_read.iter().copied()
    }

    /// Returns the total number of trace rows contributed by recorded reads, or `None` on overflow.
    pub fn num_accesses(&self) -> Option<usize> {
        self.elements_read.len().checked_add(self.words_read.len())
    }
}

/// Records and replays all the writes made to memory, in which all elements written to memory
/// throughout a program's execution are recorded by the fast processor.
///
/// This is separated from [MemoryReadsReplay] since writes are not needed for core trace generation
/// (as reads are), but only to be able to fully build the memory chiplet trace.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct MemoryWritesReplay {
    elements_written: VecDeque<(Felt, Felt, ContextId, RowIndex)>,
    words_written: VecDeque<(Word, Felt, ContextId, RowIndex)>,
}

impl MemoryWritesReplay {
    /// Records a write element to memory
    pub fn record_write_element(
        &mut self,
        element: Felt,
        addr: Felt,
        ctx: ContextId,
        clk: RowIndex,
    ) {
        self.elements_written.push_back((element, addr, ctx, clk));
    }

    /// Records a write word to memory
    pub fn record_write_word(&mut self, word: Word, addr: Felt, ctx: ContextId, clk: RowIndex) {
        self.words_written.push_back((word, addr, ctx, clk));
    }

    /// Returns an iterator over all recorded memory element writes, yielding tuples of
    /// (element, address, context ID, clock cycle).
    pub fn iter_elements_written(
        &self,
    ) -> impl Iterator<Item = &(Felt, Felt, ContextId, RowIndex)> {
        self.elements_written.iter()
    }

    /// Returns an iterator over all recorded memory word writes, yielding tuples of
    /// (word, address, context ID, clock cycle).
    pub fn iter_words_written(&self) -> impl Iterator<Item = &(Word, Felt, ContextId, RowIndex)> {
        self.words_written.iter()
    }

    /// Returns the total number of trace rows contributed by recorded writes, or `None` on
    /// overflow.
    pub fn num_accesses(&self) -> Option<usize> {
        self.elements_written.len().checked_add(self.words_written.len())
    }
}

impl MemoryInterface for MemoryReadsReplay {
    fn read_element(&mut self, _ctx: ContextId, addr: Felt) -> Result<Felt, MemoryError> {
        self.replay_read_element(addr)
    }

    fn read_word(
        &mut self,
        _ctx: ContextId,
        addr: Felt,
        _clk: RowIndex,
    ) -> Result<Word, MemoryError> {
        self.replay_read_word(addr)
    }

    fn write_element(
        &mut self,
        _ctx: ContextId,
        _addr: Felt,
        _element: Felt,
    ) -> Result<(), MemoryError> {
        Ok(())
    }

    fn write_word(
        &mut self,
        _ctx: ContextId,
        _addr: Felt,
        _clk: RowIndex,
        _word: Word,
    ) -> Result<(), MemoryError> {
        Ok(())
    }
}

// ADVICE REPLAY
// ================================================================================================

/// Implements a shim for the advice provider, in which all advice provider operations during a
/// given fragment are pre-recorded by the fast processor.
///
/// This is used to simulate advice provider interactions in parallel trace generation without
/// needing to actually access the advice provider. All advice provider operations are recorded
/// during fast execution and then replayed during parallel trace generation.
///
/// The shim records all operations with their parameters and results, and provides replay methods
/// that return the pre-recorded results. This works naturally since the fast processor has exactly
/// the same access patterns as the main trace generators (which re-executes part of the program).
/// The read methods include debug assertions to verify parameter consistency.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct AdviceReplay {
    // Stack operations
    stack_pops: VecDeque<Felt>,
    stack_word_pops: VecDeque<Word>,
    stack_dword_pops: VecDeque<[Word; 2]>,
}

impl AdviceReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------

    /// Records the value returned by a pop_stack operation
    pub fn record_pop_stack(&mut self, value: Felt) {
        self.stack_pops.push_back(value);
    }

    /// Records the word returned by a pop_stack_word operation
    pub fn record_pop_stack_word(&mut self, word: Word) {
        self.stack_word_pops.push_back(word);
    }

    /// Records the double word returned by a pop_stack_dword operation
    pub fn record_pop_stack_dword(&mut self, dword: [Word; 2]) {
        self.stack_dword_pops.push_back(dword);
    }

    // ACCESSORS (used during parallel trace generation)
    // --------------------------------------------------------------------------------

    /// Replays a pop_stack operation, returning the previously recorded value
    pub fn replay_pop_stack(&mut self) -> Result<Felt, ExecutionError> {
        self.stack_pops
            .pop_front()
            .ok_or(ExecutionError::Internal("no stack pop operations recorded"))
    }

    /// Replays a pop_stack_word operation, returning the previously recorded word
    pub fn replay_pop_stack_word(&mut self) -> Result<Word, ExecutionError> {
        self.stack_word_pops
            .pop_front()
            .ok_or(ExecutionError::Internal("no stack word pop operations recorded"))
    }

    /// Replays a pop_stack_dword operation, returning the previously recorded double word
    pub fn replay_pop_stack_dword(&mut self) -> Result<[Word; 2], ExecutionError> {
        self.stack_dword_pops
            .pop_front()
            .ok_or(ExecutionError::Internal("no stack dword pop operations recorded"))
    }
}

impl AdviceProviderInterface for AdviceReplay {
    fn pop_stack(&mut self) -> Result<Felt, AdviceError> {
        self.replay_pop_stack().map_err(|_| AdviceError::StackReadFailed)
    }

    fn pop_stack_word(&mut self) -> Result<Word, AdviceError> {
        self.replay_pop_stack_word().map_err(|_| AdviceError::StackReadFailed)
    }

    fn pop_stack_dword(&mut self) -> Result<[Word; 2], AdviceError> {
        self.replay_pop_stack_dword().map_err(|_| AdviceError::StackReadFailed)
    }

    /// Returns an empty Merkle path, as Merkle paths are ignored in parallel trace generation.
    fn get_merkle_path(
        &self,
        _root: Word,
        _depth: Felt,
        _index: Felt,
    ) -> Result<Option<MerklePath>, AdviceError> {
        Ok(None)
    }

    /// Returns an empty Merkle path and root, as they are ignored in parallel trace generation.
    fn update_merkle_node(
        &mut self,
        _root: Word,
        _depth: Felt,
        _index: Felt,
        _value: Word,
    ) -> Result<Option<MerklePath>, AdviceError> {
        Ok(None)
    }
}

// BITWISE REPLAY
// ================================================================================================

/// Enum representing the different bitwise operations that can be recorded.
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub enum BitwiseOp {
    U32And,
    U32Xor,
}

/// One 8-row AEAD stream entry to replay into the bitwise chiplet.
#[derive(Debug, PartialEq, Eq)]
pub struct AeadStreamReplayEntry {
    pub ctx: Felt,
    pub clk: Felt,
    pub src_ptr: Felt,
    pub dst_ptr: Felt,
    pub lane_base: Felt,
    pub plaintext: [Felt; 4],
    pub keystream: [Felt; 8],
    pub ciphertext: [Felt; 8],
}

/// Replay entry for the bitwise chiplet.
#[derive(Debug, PartialEq, Eq)]
pub enum BitwiseReplayEntry {
    U32(BitwiseOp, Felt, Felt),
    AeadStream(Box<AeadStreamReplayEntry>),
}

/// Replay data for bitwise operations.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct BitwiseReplay {
    entries: VecDeque<BitwiseReplayEntry>,
}

impl BitwiseReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------

    /// Records the operands of a u32and operation.
    pub fn record_u32and(&mut self, a: Felt, b: Felt) {
        self.entries.push_back(BitwiseReplayEntry::U32(BitwiseOp::U32And, a, b));
    }

    /// Records the operands of a u32xor operation.
    pub fn record_u32xor(&mut self, a: Felt, b: Felt) {
        self.entries.push_back(BitwiseReplayEntry::U32(BitwiseOp::U32Xor, a, b));
    }

    /// Records one 8-row AEAD stream entry.
    pub fn record_aead_stream(&mut self, entry: AeadStreamReplayEntry) {
        self.entries.push_back(BitwiseReplayEntry::AeadStream(Box::new(entry)));
    }

    /// Returns the number of trace rows contributed by the recorded operations, or `None` on
    /// overflow.
    pub fn trace_len(&self) -> Option<usize> {
        self.entries.iter().try_fold(0usize, |len, entry| {
            let entry_len = match entry {
                BitwiseReplayEntry::U32(..) => 1,
                BitwiseReplayEntry::AeadStream(..) => AEAD_STREAM_CYCLE_LEN,
            };
            len.checked_add(entry_len)
        })
    }
}

impl IntoIterator for BitwiseReplay {
    type Item = BitwiseReplayEntry;
    type IntoIter = <VecDeque<BitwiseReplayEntry> as IntoIterator>::IntoIter;

    /// Returns an iterator over all recorded u32 operations with their operands.
    fn into_iter(self) -> Self::IntoIter {
        self.entries.into_iter()
    }
}

// KERNEL REPLAY
// ================================================================================================

/// Replay data for kernel operations.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct KernelReplay {
    kernel_proc_accesses: VecDeque<Word>,
}

impl KernelReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------

    /// Records the procedure hash of a syscall.
    pub fn record_kernel_proc_access(&mut self, proc_hash: Word) {
        self.kernel_proc_accesses.push_back(proc_hash);
    }
}

impl IntoIterator for KernelReplay {
    type Item = Word;
    type IntoIter = <VecDeque<Word> as IntoIterator>::IntoIter;

    /// Returns an iterator over all recorded kernel procedure accesses.
    fn into_iter(self) -> Self::IntoIter {
        self.kernel_proc_accesses.into_iter()
    }
}

// ACE REPLAY
// ================================================================================================

/// Replay data for ACE operations.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct AceReplay {
    circuit_evaluations: VecDeque<(RowIndex, CircuitEvaluation)>,
}

impl AceReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------

    /// Records a circuit evaluation, rejecting invocations beyond the witness collection limit.
    pub fn record_circuit_evaluation(
        &mut self,
        circuit_eval: CircuitEvaluation,
    ) -> Result<(), AceError> {
        if self.circuit_evaluations.len() >= MAX_EVAL_CIRCUIT_INVOCATIONS as usize {
            return Err(AceError(format!(
                "number of recorded eval_circuit invocations cannot exceed {MAX_EVAL_CIRCUIT_INVOCATIONS}"
            )));
        }
        let clk = RowIndex::from(circuit_eval.clk());
        self.circuit_evaluations.push_back((clk, circuit_eval));
        Ok(())
    }

    /// Returns the total number of trace rows contributed by recorded evaluations, or `None` on
    /// overflow.
    pub fn trace_len(&self) -> Option<usize> {
        self.circuit_evaluations
            .iter()
            .try_fold(0usize, |len, (_, evaluation)| len.checked_add(evaluation.num_rows()))
    }
}

impl IntoIterator for AceReplay {
    type Item = (RowIndex, CircuitEvaluation);
    type IntoIter = <VecDeque<(RowIndex, CircuitEvaluation)> as IntoIterator>::IntoIter;

    /// Returns an iterator over all recorded circuit evaluations.
    fn into_iter(self) -> Self::IntoIter {
        self.circuit_evaluations.into_iter()
    }
}

// RANGE CHECKER REPLAY
// ================================================================================================

/// Values requested from the 16-bit range-check table by a single operation.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub enum RangeCheckReplayValues {
    Two([u16; 2]),
    Four([u16; 4]),
    Five([u16; 5]),
}

impl AsRef<[u16]> for RangeCheckReplayValues {
    fn as_ref(&self) -> &[u16] {
        match self {
            Self::Two(values) => values,
            Self::Four(values) => values,
            Self::Five(values) => values,
        }
    }
}

/// Replay data for range checking operations.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct RangeCheckerReplay {
    range_checks: VecDeque<RangeCheckReplayValues>,
}

impl RangeCheckerReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------

    /// Records the set of range checks which result from a u32 operation.
    pub fn record_range_check_u32(&mut self, u16_limbs: [u16; 4]) {
        self.range_checks.push_back(RangeCheckReplayValues::Four(u16_limbs));
    }

    /// Records the two range checks which enforce that a Merkle depth is in `[1, 64]`.
    pub fn record_merkle_depth(&mut self, depth: Felt) {
        let depth =
            u16::try_from(depth.as_canonical_u64()).expect("Merkle depth must fit in 16 bits");
        let scaled_depth = depth
            .checked_sub(1)
            .and_then(|depth_minus_one| depth_minus_one.checked_mul(MERKLE_DEPTH_RANGE_SCALE))
            .expect("Merkle depth must be in the range 1..=64");

        self.range_checks.push_back(RangeCheckReplayValues::Two([depth, scaled_depth]));
    }

    /// Records the five range checks which bind a Merkle index to its canonical field
    /// representative: the four limbs of `y = (p - 1 - index - bit) / 2`, plus twice the top
    /// limb to enforce `y < 2^63`.
    pub fn record_merkle_index(&mut self, index: Felt) {
        let (_, limbs) = merkle_index_helper_values(index);
        let doubled_top = limbs[3]
            .checked_mul(2)
            .expect("canonical Merkle-index witness top limb must be below 2^15");
        self.range_checks.push_back(RangeCheckReplayValues::Five([
            limbs[0],
            limbs[1],
            limbs[2],
            limbs[3],
            doubled_top,
        ]));
    }

    /// Records the two 16-bit limbs of `divisor - remainder - 1` for U32DIV.
    pub fn record_u32div_remainder_diff(&mut self, u16_limbs: [u16; 2]) {
        self.range_checks.push_back(RangeCheckReplayValues::Two(u16_limbs));
    }
}

impl IntoIterator for RangeCheckerReplay {
    type Item = RangeCheckReplayValues;
    type IntoIter = <VecDeque<RangeCheckReplayValues> as IntoIterator>::IntoIter;

    /// Returns an iterator over all range checks recorded during execution.
    fn into_iter(self) -> Self::IntoIter {
        self.range_checks.into_iter()
    }
}

// BLOCK ADDRESS REPLAY
// ================================================================================================

#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct BlockAddressReplay {
    /// Recorded hasher addresses from operations like hash_control_block, hash_basic_block, etc.
    block_addresses: VecDeque<Felt>,
}

impl BlockAddressReplay {
    /// Records the address associated with a `Hasher::hash_control_block` or
    /// `Hasher::hash_basic_block` operation.
    pub fn record_block_address(&mut self, addr: Felt) {
        self.block_addresses.push_back(addr);
    }

    /// Replays a `Hasher::hash_control_block` or `Hasher::hash_basic_block` operation, returning
    /// the pre-recorded address
    pub fn replay_block_address(&mut self) -> Result<Felt, ExecutionError> {
        self.block_addresses
            .pop_front()
            .ok_or(ExecutionError::Internal("no block address operations recorded"))
    }
}

// HASHER RESPONSE REPLAY
// ================================================================================================

/// Records and replays the response of requests made to the hasher chiplet during the execution of
/// a program.
///
/// The hasher responses are recorded during fast processor execution and then replayed during core
/// trace generation.
#[derive(Debug, Default, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct HasherResponseReplay {
    /// Recorded hasher operations from Compress requests.
    ///
    /// Each entry contains (address, output_state)
    compression_operations: VecDeque<(Felt, [Felt; 12])>,

    /// Recorded hasher operations from Merkle path verification operations.
    ///
    /// Each entry contains (address, computed_root)
    build_merkle_root_operations: VecDeque<(Felt, Word)>,

    /// Recorded hasher operations from Merkle root update operations.
    ///
    /// Each entry contains (address, old_root, new_root)
    mrupdate_operations: VecDeque<(Felt, Word, Word)>,
}

impl HasherResponseReplay {
    // MUTATIONS (populated by the fast processor)
    // --------------------------------------------------------------------------------------------

    /// Records a `Hasher::compress` operation with its address and compressed state.
    pub fn record_compression(&mut self, addr: Felt, hashed_state: [Felt; 12]) {
        self.compression_operations.push_back((addr, hashed_state));
    }

    /// Records a Merkle path verification with its address and computed root
    pub fn record_build_merkle_root(&mut self, addr: Felt, computed_root: Word) {
        self.build_merkle_root_operations.push_back((addr, computed_root));
    }

    /// Records a Merkle root update with its address, old root, and new root
    pub fn record_update_merkle_root(&mut self, addr: Felt, old_root: Word, new_root: Word) {
        self.mrupdate_operations.push_back((addr, old_root, new_root));
    }

    // ACCESSORS (used by parallel trace generators)
    // --------------------------------------------------------------------------------------------

    /// Replays a `Hasher::compress` operation, returning its address and compressed state.
    pub fn replay_compression(&mut self) -> Result<(Felt, [Felt; 12]), OperationError> {
        self.compression_operations
            .pop_front()
            .ok_or(OperationError::Internal("no compression operations recorded"))
    }

    /// Replays a Merkle path verification, returning the pre-recorded address and computed root
    pub fn replay_build_merkle_root(&mut self) -> Result<(Felt, Word), OperationError> {
        self.build_merkle_root_operations
            .pop_front()
            .ok_or(OperationError::Internal("no build merkle root operations recorded"))
    }

    /// Replays a Merkle root update, returning the pre-recorded address, old root, and new root
    pub fn replay_update_merkle_root(&mut self) -> Result<(Felt, Word, Word), OperationError> {
        self.mrupdate_operations
            .pop_front()
            .ok_or(OperationError::Internal("no mrupdate operations recorded"))
    }
}

impl HasherInterface for HasherResponseReplay {
    fn compress(&mut self, _state: HasherState) -> Result<(Felt, HasherState), OperationError> {
        self.replay_compression()
    }

    fn compress_aead_xof(
        &mut self,
        _ctx: ContextId,
        _clk: RowIndex,
        state: HasherState,
    ) -> Result<[Felt; 16], OperationError> {
        Ok(miden_core::chiplets::eidos_compression::compress_raw_xof_lanes(&state)
            .map(Felt::from_u32))
    }

    fn verify_merkle_root(
        &mut self,
        claimed_root: Word,
        _value: Word,
        _path: Option<&MerklePath>,
        _index: Felt,
        on_err: impl FnOnce() -> OperationError,
    ) -> Result<Felt, OperationError> {
        let (addr, computed_root) = self.replay_build_merkle_root()?;
        if claimed_root == computed_root {
            Ok(addr)
        } else {
            // If the hasher doesn't compute the same root (using the same path),
            // then it means that `node` is not the value currently in the tree at `index`
            Err(on_err())
        }
    }

    fn update_merkle_root(
        &mut self,
        claimed_old_root: Word,
        _old_value: Word,
        _new_value: Word,
        _path: Option<&MerklePath>,
        _index: Felt,
        on_err: impl FnOnce() -> OperationError,
    ) -> Result<(Felt, Word), OperationError> {
        let (address, old_root, new_root) = self.replay_update_merkle_root()?;

        if claimed_old_root == old_root {
            Ok((address, new_root))
        } else {
            Err(on_err())
        }
    }
}

/// Enum representing the different hasher operations that can be recorded, along with their
/// operands.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub enum HasherOp {
    Compress([Felt; STATE_WIDTH]),
    AeadXof(ContextId, RowIndex, [Felt; STATE_WIDTH]),
    HashControlBlock((Word, Word, Felt, Word)),
    /// `(forest_id, node_id, expected_hash)` — `forest_id` is an id into the
    /// `mast_forest_store` of the trace replay that owns this operation.
    HashBasicBlock((MastForestId, MastNodeId, Word)),
    BuildMerkleRoot((Word, MerklePath, Felt)),
    UpdateMerkleRoot((Word, Word, MerklePath, Felt)),
}

impl HasherOp {
    /// Resolves the five forest-free variants; `None` for [`HasherOp::HashBasicBlock`], which
    /// needs forest access.
    fn resolve_forest_free<'a>(self) -> Option<ResolvedHasherOp<'a>> {
        match self {
            HasherOp::Compress(s) => Some(ResolvedHasherOp::Compress(s)),
            HasherOp::AeadXof(ctx, clk, s) => Some(ResolvedHasherOp::AeadXof(ctx, clk, s)),
            HasherOp::HashControlBlock(x) => Some(ResolvedHasherOp::HashControlBlock(x)),
            HasherOp::HashBasicBlock(_) => None,
            HasherOp::BuildMerkleRoot(x) => Some(ResolvedHasherOp::BuildMerkleRoot(x)),
            HasherOp::UpdateMerkleRoot(x) => Some(ResolvedHasherOp::UpdateMerkleRoot(x)),
        }
    }
}

/// Basic-block data resolved for replay by the hasher-chiplet builder.
#[derive(Debug)]
pub enum ResolvedBasicBlockGroups<'a> {
    /// Buffered replay borrows batches from the finalized MAST forest.
    Borrowed(&'a [OpBatch]),
    /// Streamed replay owns the group hashes because it crosses a thread boundary.
    Owned(Vec<OpBatch>),
}

/// A hasher-chiplet request with all operands resolved, so it can be replayed without further
/// MAST forest lookups.
///
/// This is what the hasher-chiplet builder consumes: the buffered replay resolves its recorded
/// [`HasherOp`]s at drain time and borrows basic-block batches from the finalized forest, while the
/// streaming path resolves at record time and forwards owned data to the concurrent builder.
#[derive(Debug)]
pub enum ResolvedHasherOp<'a> {
    Compress([Felt; STATE_WIDTH]),
    AeadXof(ContextId, RowIndex, [Felt; STATE_WIDTH]),
    HashControlBlock((Word, Word, Felt, Word)),
    HashBasicBlock((ResolvedBasicBlockGroups<'a>, Word)),
    BuildMerkleRoot((Word, MerklePath, Felt)),
    UpdateMerkleRoot((Word, Word, MerklePath, Felt)),
}

/// Records and replays all the requests made to the hasher chiplet during the execution of a
/// program, for the purposes of generating the hasher chiplet's trace.
///
/// In buffered mode (the default) requests are recorded during fast-processor execution and
/// replayed during hasher chiplet trace generation. In streamed mode
/// ([`HasherRequestReplay::streamed`]) each request is resolved at record time and forwarded to
/// a hasher-chiplet builder running concurrently with execution; see
/// `FastProcessor::execute_and_build_trace_sync`.
#[derive(Debug, Default)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct HasherRequestReplay {
    sink: HasherOpSink,
}

#[derive(Debug)]
enum HasherOpSink {
    Buffered(VecDeque<HasherOp>),
    #[cfg(feature = "std")]
    Streamed(std::sync::mpsc::Sender<ResolvedHasherOp<'static>>),
}

impl PartialEq for HasherRequestReplay {
    fn eq(&self, other: &Self) -> bool {
        match (&self.sink, &other.sink) {
            (HasherOpSink::Buffered(lhs), HasherOpSink::Buffered(rhs)) => lhs == rhs,
            #[cfg(feature = "std")]
            (HasherOpSink::Streamed(_), HasherOpSink::Streamed(_)) => true,
            #[cfg(feature = "std")]
            _ => false,
        }
    }
}

impl Eq for HasherRequestReplay {}

impl Default for HasherOpSink {
    fn default() -> Self {
        Self::Buffered(VecDeque::new())
    }
}

impl HasherRequestReplay {
    /// Creates a replay that forwards each request, resolved, to the receiving end as it is
    /// recorded.
    ///
    /// Send failures are ignored: they mean the consumer stopped early, and its error surfaces
    /// when the caller joins it.
    #[cfg(feature = "std")]
    pub(crate) fn streamed(sender: std::sync::mpsc::Sender<ResolvedHasherOp<'static>>) -> Self {
        Self { sink: HasherOpSink::Streamed(sender) }
    }

    // `resolved` is consumed only by the std-only streamed arm; without std the buffered arm
    // stores the raw op and resolves later at drain.
    #[cfg_attr(not(feature = "std"), allow(unused_variables))]
    fn record_resolved(
        &mut self,
        op: HasherOp,
        resolved: impl FnOnce() -> ResolvedHasherOp<'static>,
    ) {
        match &mut self.sink {
            HasherOpSink::Buffered(ops) => ops.push_back(op),
            #[cfg(feature = "std")]
            HasherOpSink::Streamed(sender) => {
                let _ = sender.send(resolved());
            },
        }
    }

    /// Records a forest-free op: buffered sinks store it as is, streamed sinks resolve and
    /// forward it. [`HasherOp::HashBasicBlock`] must go through
    /// [`Self::record_hash_basic_block`] instead.
    fn record(&mut self, op: HasherOp) {
        match &mut self.sink {
            HasherOpSink::Buffered(ops) => ops.push_back(op),
            #[cfg(feature = "std")]
            HasherOpSink::Streamed(sender) => {
                let resolved =
                    op.resolve_forest_free().expect("`record` is only called with forest-free ops");
                let _ = sender.send(resolved);
            },
        }
    }

    /// Records a raw, unresolved op — test-only escape hatch for injecting
    /// malformed ids to exercise the drain-time resolution error paths.
    #[cfg(test)]
    pub(crate) fn record_raw(&mut self, op: HasherOp) {
        match &mut self.sink {
            HasherOpSink::Buffered(ops) => ops.push_back(op),
            #[cfg(feature = "std")]
            HasherOpSink::Streamed(_) => panic!("record_raw is for buffered replays"),
        }
    }

    /// Records a `Hasher::compress()` request.
    pub fn record_compress_input(&mut self, state: [Felt; STATE_WIDTH]) {
        self.record(HasherOp::Compress(state));
    }

    /// Records an AEAD-XOF compression request.
    pub fn record_aead_xof_input(
        &mut self,
        ctx: ContextId,
        clk: RowIndex,
        state: [Felt; STATE_WIDTH],
    ) {
        self.record(HasherOp::AeadXof(ctx, clk, state));
    }

    /// Records a `Hasher::hash_control_block()` request.
    pub fn record_hash_control_block(
        &mut self,
        h1: Word,
        h2: Word,
        domain: Felt,
        expected_hash: Word,
    ) {
        self.record(HasherOp::HashControlBlock((h1, h2, domain, expected_hash)));
    }

    /// Records a `Hasher::hash_basic_block()` request.
    ///
    /// `basic_block_node` must be the node identified by `(forest_id, node_id)`; streamed
    /// replays copy its per-batch group hashes so the concurrent builder needs no forest
    /// access.
    pub fn record_hash_basic_block(
        &mut self,
        forest_id: MastForestId,
        node_id: MastNodeId,
        basic_block_node: &BasicBlockNode,
    ) {
        let expected_hash = basic_block_node.digest();
        self.record_resolved(HasherOp::HashBasicBlock((forest_id, node_id, expected_hash)), || {
            ResolvedHasherOp::HashBasicBlock((
                ResolvedBasicBlockGroups::Owned(basic_block_node.op_batches().to_vec()),
                expected_hash,
            ))
        });
    }

    /// Records a `Hasher::build_merkle_root()` request.
    pub fn record_build_merkle_root(&mut self, leaf: Word, path: MerklePath, index: Felt) {
        self.record(HasherOp::BuildMerkleRoot((leaf, path, index)));
    }

    /// Records a `Hasher::update_merkle_root()` request.
    pub fn record_update_merkle_root(
        &mut self,
        old_value: Word,
        new_value: Word,
        path: MerklePath,
        index: Felt,
    ) {
        self.record(HasherOp::UpdateMerkleRoot((old_value, new_value, path, index)));
    }

    /// Returns the forest IDs referenced by buffered basic-block hash requests.
    ///
    /// Trace replay deserialization uses these IDs to validate that every request points into the
    /// replay's finalized MAST forest store. Streamed requests have already been delivered to the
    /// concurrent trace builder, so they yield no IDs here.
    pub(crate) fn iter_hash_basic_block_forest_ids(
        &self,
    ) -> impl Iterator<Item = MastForestId> + '_ {
        self.buffered_ops()
            .into_iter()
            .flat_map(|ops| ops.iter())
            .filter_map(|op| match op {
                HasherOp::HashBasicBlock((forest_id, _node_id, _expected_hash)) => Some(*forest_id),
                _ => None,
            })
    }

    fn buffered_ops(&self) -> Option<&VecDeque<HasherOp>> {
        match &self.sink {
            HasherOpSink::Buffered(ops) => Some(ops),
            #[cfg(feature = "std")]
            HasherOpSink::Streamed(_) => None,
        }
    }

    /// Drains the buffered requests as resolved ops, looking basic blocks up in the finalized
    /// forest store.
    ///
    /// A streamed replay must not reach this path — its ops were already delivered to the
    /// concurrent builder (and the overlap path swaps in an empty buffered replay via
    /// `take_hasher_replay`) — so it yields an error rather than silently producing an empty
    /// hasher chiplet.
    pub fn into_resolved_ops<'a>(
        self,
        mast_forest_store: &'a [Arc<SparseMastForest>],
    ) -> impl Iterator<Item = Result<ResolvedHasherOp<'a>, ExecutionError>> + 'a {
        let (streamed_err, ops) = match self.sink {
            HasherOpSink::Buffered(ops) => (None, ops),
            #[cfg(feature = "std")]
            HasherOpSink::Streamed(_) => (
                Some(ExecutionError::Internal(
                    "streamed hasher replay reached the buffered trace-build path",
                )),
                VecDeque::new(),
            ),
        };
        streamed_err.into_iter().map(Err).chain(ops.into_iter().map(move |op| match op {
            HasherOp::HashBasicBlock((forest_id, node_id, expected_hash)) => {
                let forest =
                    mast_forest_store.get(forest_id.to_usize()).ok_or(ExecutionError::Internal(
                        "MAST forest id in hasher replay out of range of mast_forest_store",
                    ))?;
                let node = forest
                    .get_node_by_id(node_id)
                    .ok_or(ExecutionError::Internal("invalid node ID in hasher replay"))?;
                let MastNode::Block(basic_block_node) = node else {
                    return Err(ExecutionError::Internal(
                        "expected basic block node in hasher replay",
                    ));
                };
                Ok(ResolvedHasherOp::HashBasicBlock((
                    ResolvedBasicBlockGroups::Borrowed(basic_block_node.op_batches()),
                    expected_hash,
                )))
            },
            op => Ok(op.resolve_forest_free().expect("all other variants are forest-free")),
        }))
    }
}

// STACK OVERFLOW REPLAY
// ================================================================================================

/// Implements a shim for stack overflow operations, in which all overflow values and addresses
/// during a given fragment are pre-recorded by the fast processor and replayed by the main trace
/// fragment generators.
///
/// This is used to simulate stack overflow functionality in parallel trace generation without
/// needing to maintain the actual overflow table. All overflow operations are recorded during
/// fast execution and then replayed during parallel trace generation.
///
/// The shim records overflow values (from pop operations) and overflow addresses (representing
/// the clock cycle of the last overflow update) and provides replay methods that return the
/// pre-recorded values. This works naturally since the fast processor has exactly the same
/// access patterns as the main trace generators.
#[derive(Debug, PartialEq, Eq)]
#[cfg_attr(
    all(feature = "arbitrary", test),
    miden_test_serialization_macros::serialization_test
)]
pub struct StackOverflowReplay {
    /// Recorded overflow values and overflow addresses from pop_overflow operations. Each entry
    /// represents a value that was popped from the overflow stack, and the overflow address of the
    /// entry at the top of the overflow stack *after* the pop operation.
    ///
    /// For example, given the following table:
    ///
    /// | Overflow Value | Overflow Address |
    /// |----------------|------------------|
    /// |      8         |         14       |
    /// |      2         |         16       |
    /// |      7         |         18       |
    ///
    /// a `pop_overflow()` operation would return (popped_value, prev_addr) = (7, 16).
    overflow_values: VecDeque<(Felt, Felt)>,

    /// Recorded (stack depth, overflow address) returned when restoring a context
    restore_context_info: VecDeque<(usize, Felt)>,
}

impl Default for StackOverflowReplay {
    fn default() -> Self {
        Self::new()
    }
}

impl StackOverflowReplay {
    /// Creates a new StackOverflowReplay with empty operation vectors
    pub fn new() -> Self {
        Self {
            overflow_values: VecDeque::new(),
            restore_context_info: VecDeque::new(),
        }
    }

    // MUTATORS
    // --------------------------------------------------------------------------------

    /// Records the value returned by a pop_overflow operation, along with the overflow address
    /// stored in the overflow table *after* the pop. That is, `new_overflow_addr` represents the
    /// clock cycle at which the value *before* `value` was added to the overflow table. See the
    /// docstring for the `overflow_values` field for more information.
    ///
    /// This *must* only be called if there is an actual value in the overflow table to pop; that
    /// is, don't call if the stack depth is 16.
    pub fn record_pop_overflow(&mut self, value: Felt, new_overflow_addr: Felt) {
        self.overflow_values.push_back((value, new_overflow_addr));
    }

    /// Records the overflow address when restoring a context
    pub fn record_restore_context_overflow_addr(&mut self, stack_depth: usize, addr: Felt) {
        self.restore_context_info.push_back((stack_depth, addr));
    }

    // ACCESSORS
    // --------------------------------------------------------------------------------

    /// Peeks at the next recorded pop_overflow operation, returning the previously recorded value
    /// and `new_overflow_addr` without removing it from the queue.
    pub fn peek_replay_pop_overflow(&self) -> Option<&(Felt, Felt)> {
        self.overflow_values.front()
    }

    /// Replays a pop_overflow operation, returning the previously recorded value and
    /// `new_overflow_addr`.
    ///
    /// This *must* only be called if there is an actual value in the overflow table to pop; that
    /// is, don't call if the stack depth is 16.
    ///
    /// See [Self::record_pop_overflow] for more details.
    pub fn replay_pop_overflow(&mut self) -> Result<(Felt, Felt), OperationError> {
        self.overflow_values
            .pop_front()
            .ok_or(OperationError::Internal("no overflow pop operations recorded"))
    }

    /// Replays the overflow address when restoring a context
    pub fn replay_restore_context_overflow_addr(
        &mut self,
    ) -> Result<(usize, Felt), OperationError> {
        self.restore_context_info
            .pop_front()
            .ok_or(OperationError::Internal("no overflow address operations recorded"))
    }
}

#[cfg(all(test, feature = "std"))]
mod tests {
    use core::mem::size_of;

    use super::*;

    #[test]
    fn end_block_flags_encode_each_entry_kind() {
        let cases = [
            (
                EndBlockFlags::new(false, EndBlockStackEntryKind::Continuation),
                [ZERO, ZERO, ZERO, ZERO].into(),
            ),
            (
                EndBlockFlags::new(true, EndBlockStackEntryKind::LoopContinuation),
                [ONE, ONE, ZERO, ZERO].into(),
            ),
            (
                EndBlockFlags::new(false, EndBlockStackEntryKind::CallerFrame),
                [ZERO, ZERO, ONE, ZERO].into(),
            ),
        ];

        for (flags, expected) in cases {
            assert_eq!(flags.to_hasher_state_second_word(), expected);
        }
    }

    #[test]
    fn bitwise_replay_trace_len_accounts_for_aead_stream_rows() {
        let mut replay = BitwiseReplay::default();
        replay.record_u32and(ZERO, ZERO);
        for lane_base in [ZERO, Felt::new_unchecked(8)] {
            replay.record_aead_stream(AeadStreamReplayEntry {
                ctx: ZERO,
                clk: ZERO,
                src_ptr: ZERO,
                dst_ptr: ZERO,
                lane_base,
                plaintext: [ZERO; 4],
                keystream: [ZERO; 8],
                ciphertext: [ZERO; 8],
            });
        }

        assert_eq!(replay.trace_len(), Some(1 + 2 * AEAD_STREAM_CYCLE_LEN));
        assert!(
            size_of::<BitwiseReplayEntry>() < size_of::<AeadStreamReplayEntry>(),
            "ordinary replay entries must not pay for the inline AEAD stream payload",
        );
    }

    /// A streamed replay reaching the buffered trace-build path is a logic error
    /// and must surface immediately, not as a silently empty hasher chiplet.
    #[test]
    fn streamed_replay_in_buffered_path_errors() {
        let (sender, _receiver) = std::sync::mpsc::channel();
        let replay = HasherRequestReplay::streamed(sender);
        let mut ops = replay.into_resolved_ops(&[]);
        assert!(matches!(ops.next(), Some(Err(ExecutionError::Internal(_)))));
        assert!(ops.next().is_none());
    }
}

// ARBITRARY (proptest)
// ================================================================================================

#[cfg(any(test, feature = "arbitrary"))]
mod arbitrary {
    extern crate alloc;

    use miden_air::trace::RowIndex;
    use proptest::prelude::*;

    use super::{
        AdviceReplay, BitwiseOp, BitwiseReplay, BlockAddressReplay, BlockStackReplay,
        CoreTraceState, DecoderState, ExecutionContextReplay, ExecutionContextSystemInfo,
        KernelReplay, MIN_STACK_DEPTH, MemoryReadsReplay, MemoryWritesReplay, NodeEndData,
        RangeCheckReplayValues, RangeCheckerReplay, StackState, SystemState,
    };
    use crate::{ContextId, Felt, Word};

    impl Arbitrary for SystemState {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // All fields are any-valid on the wire.
            (any::<RowIndex>(), any::<ContextId>(), any::<Word>(), any::<Word>())
                .prop_map(|(clk, ctx, fn_hash, deferred_root)| Self {
                    clk,
                    ctx,
                    fn_hash,
                    deferred_root,
                })
                .boxed()
        }
    }

    impl Arbitrary for DecoderState {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (any::<Felt>(), any::<Felt>())
                .prop_map(|(current_addr, parent_addr)| Self { current_addr, parent_addr })
                .boxed()
        }
    }

    impl Arbitrary for StackState {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // The reader rejects stack_depth < MIN_STACK_DEPTH; bound it by construction and
            // keep a small overflow margin so the depth stays representative and shrinking stays
            // cheap.
            (
                any::<[Felt; MIN_STACK_DEPTH]>(),
                (MIN_STACK_DEPTH..=(MIN_STACK_DEPTH + 8)),
                any::<Felt>(),
            )
                .prop_map(|(stack_top, stack_depth, last_overflow_addr)| {
                    Self::new(stack_top, stack_depth, last_overflow_addr)
                })
                .boxed()
        }
    }

    impl Arbitrary for NodeEndData {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (any::<Felt>(), any::<Felt>(), any::<Felt>())
                .prop_map(|(ended_node_addr, prev_addr, prev_parent_addr)| Self {
                    ended_node_addr,
                    prev_addr,
                    prev_parent_addr,
                })
                .boxed()
        }
    }

    impl Arbitrary for ExecutionContextSystemInfo {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (any::<ContextId>(), any::<Word>())
                .prop_map(|(parent_ctx, parent_fn_hash)| Self { parent_ctx, parent_fn_hash })
                .boxed()
        }
    }

    impl Arbitrary for ExecutionContextReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Populated through the public recording API (fields are private).
            proptest::collection::vec(any::<ExecutionContextSystemInfo>(), 0..=4)
                .prop_map(|contexts| {
                    let mut replay = Self::default();
                    for ctx_info in contexts {
                        replay.record_execution_context(ctx_info);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for BlockStackReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Populated through the public recording API (fields are private).
            (
                proptest::collection::vec(any::<Felt>(), 0..=4),
                proptest::collection::vec((any::<Felt>(), any::<Felt>(), any::<Felt>()), 0..=4),
            )
                .prop_map(|(parent_addrs, node_ends)| {
                    let mut replay = Self::default();
                    for parent_addr in parent_addrs {
                        replay.record_node_start_parent_addr(parent_addr);
                    }
                    for (ended_node_addr, prev_addr, prev_parent_addr) in node_ends {
                        replay.record_node_end(ended_node_addr, prev_addr, prev_parent_addr);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for MemoryReadsReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (
                proptest::collection::vec(
                    (any::<Felt>(), any::<Felt>(), any::<ContextId>(), any::<RowIndex>()),
                    0..=4,
                ),
                proptest::collection::vec(
                    (any::<Word>(), any::<Felt>(), any::<ContextId>(), any::<RowIndex>()),
                    0..=4,
                ),
            )
                .prop_map(|(element_rows, word_rows)| {
                    let mut replay = Self::default();
                    for (element, addr, ctx, clk) in element_rows {
                        replay.record_read_element(element, addr, ctx, clk);
                    }
                    for (word, addr, ctx, clk) in word_rows {
                        replay.record_read_word(word, addr, ctx, clk);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for MemoryWritesReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (
                proptest::collection::vec(
                    (any::<Felt>(), any::<Felt>(), any::<ContextId>(), any::<RowIndex>()),
                    0..=4,
                ),
                proptest::collection::vec(
                    (any::<Word>(), any::<Felt>(), any::<ContextId>(), any::<RowIndex>()),
                    0..=4,
                ),
            )
                .prop_map(|(element_rows, word_rows)| {
                    let mut replay = Self::default();
                    for (element, addr, ctx, clk) in element_rows {
                        replay.record_write_element(element, addr, ctx, clk);
                    }
                    for (word, addr, ctx, clk) in word_rows {
                        replay.record_write_word(word, addr, ctx, clk);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for AdviceReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (
                proptest::collection::vec(any::<Felt>(), 0..=4),
                proptest::collection::vec(any::<Word>(), 0..=4),
                proptest::collection::vec(any::<[Word; 2]>(), 0..=4),
            )
                .prop_map(|(stack_pops, stack_word_pops, stack_dword_pops)| {
                    let mut replay = Self::default();
                    for value in stack_pops {
                        replay.record_pop_stack(value);
                    }
                    for word in stack_word_pops {
                        replay.record_pop_stack_word(word);
                    }
                    for dword in stack_dword_pops {
                        replay.record_pop_stack_dword(dword);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for BitwiseOp {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            prop_oneof![Just(Self::U32And), Just(Self::U32Xor)].boxed()
        }
    }

    impl Arbitrary for BitwiseReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Populated through the recording API; both op tags are exercised via the mask.
            proptest::collection::vec((any::<BitwiseOp>(), any::<Felt>(), any::<Felt>()), 0..=4)
                .prop_map(|rows| {
                    let mut replay = Self::default();
                    for (op, a, b) in rows {
                        match op {
                            BitwiseOp::U32And => replay.record_u32and(a, b),
                            BitwiseOp::U32Xor => replay.record_u32xor(a, b),
                        }
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for KernelReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            proptest::collection::vec(any::<Word>(), 0..=4)
                .prop_map(|hashes| {
                    let mut replay = Self::default();
                    for proc_hash in hashes {
                        replay.record_kernel_proc_access(proc_hash);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for RangeCheckReplayValues {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            prop_oneof![
                any::<[u16; 2]>().prop_map(Self::Two).boxed(),
                any::<[u16; 4]>().prop_map(Self::Four).boxed(),
                any::<[u16; 5]>().prop_map(Self::Five).boxed(),
            ]
            .boxed()
        }
    }

    impl Arbitrary for RangeCheckerReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Record five-limb rows from canonical field indices so the doubled top limb
            // matches the Merkle-index witness. The other rows accept independent limbs.
            proptest::collection::vec((any::<RangeCheckReplayValues>(), any::<Felt>()), 0..=4)
                .prop_map(|values| {
                    let mut replay = Self::default();
                    for (value, index) in values {
                        match &value {
                            RangeCheckReplayValues::Four(limbs) => {
                                replay.record_range_check_u32(*limbs);
                            },
                            RangeCheckReplayValues::Two(limbs) => {
                                replay.record_u32div_remainder_diff(*limbs);
                            },
                            RangeCheckReplayValues::Five(_) => {
                                replay.record_merkle_index(index);
                            },
                        }
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for BlockAddressReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            proptest::collection::vec(any::<Felt>(), 0..=4)
                .prop_map(|addresses| {
                    let mut replay = Self::default();
                    for addr in addresses {
                        replay.record_block_address(addr);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for CoreTraceState {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Composes three already-covered parts; all fields are pub.
            (any::<SystemState>(), any::<DecoderState>(), any::<StackState>())
                .prop_map(|(system, decoder, stack)| Self { system, decoder, stack })
                .boxed()
        }
    }
}

// ARBITRARY EXTENSIONS (proptest) — replay types with private fields / hasher state
// ================================================================================================

#[cfg(any(test, feature = "arbitrary"))]
mod arbitrary_replays {
    use miden_core::mast::{MastForestId, MastNodeId};
    use proptest::prelude::*;

    use super::{
        AdviceReplay, BlockAddressReplay, BlockStackReplay, CoreTraceFragmentContext,
        CoreTraceState, ExecutionContextReplay, ExecutionReplay, HasherOp, HasherRequestReplay,
        HasherResponseReplay, MastForestResolutionReplay, MemoryReadsReplay, STATE_WIDTH,
        StackOverflowReplay,
    };
    use crate::{ContextId, Felt, RowIndex, Word, crypto::merkle::MerklePath};

    impl Arbitrary for StackOverflowReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (
                proptest::collection::vec((any::<Felt>(), any::<Felt>()), 0..=4),
                proptest::collection::vec((any::<usize>(), any::<Felt>()), 0..=4),
            )
                .prop_map(|(overflow_values, restore_context_info)| {
                    let mut replay = StackOverflowReplay::new();
                    for (value, new_overflow_addr) in overflow_values {
                        replay.record_pop_overflow(value, new_overflow_addr);
                    }
                    for (stack_depth, addr) in restore_context_info {
                        replay.record_restore_context_overflow_addr(stack_depth, addr);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for HasherOp {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Cover all six wire tags; control-block domains must fit in a u32.
            prop_oneof![
                any::<[Felt; STATE_WIDTH]>().prop_map(Self::Compress).boxed(),
                (any::<ContextId>(), any::<RowIndex>(), any::<[Felt; STATE_WIDTH]>())
                    .prop_map(|(ctx, clk, state)| Self::AeadXof(ctx, clk, state))
                    .boxed(),
                (any::<Word>(), any::<Word>(), any::<u32>(), any::<Word>())
                    .prop_map(|(h1, h2, domain, expected_hash)| {
                        Self::HashControlBlock((h1, h2, Felt::from(domain), expected_hash))
                    })
                    .boxed(),
                (any::<MastForestId>(), any::<MastNodeId>(), any::<Word>())
                    .prop_map(|(forest_id, node_id, expected_hash)| {
                        Self::HashBasicBlock((forest_id, node_id, expected_hash))
                    })
                    .boxed(),
                (any::<Word>(), any::<MerklePath>(), any::<Felt>())
                    .prop_map(|(leaf, path, index)| Self::BuildMerkleRoot((leaf, path, index)))
                    .boxed(),
                (any::<Word>(), any::<Word>(), any::<MerklePath>(), any::<Felt>())
                    .prop_map(|(old_root, new_root, path, index)| {
                        Self::UpdateMerkleRoot((old_root, new_root, path, index))
                    })
                    .boxed(),
            ]
            .boxed()
        }
    }

    impl Arbitrary for HasherResponseReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            (
                proptest::collection::vec((any::<Felt>(), any::<[Felt; 12]>()), 0..=4),
                proptest::collection::vec((any::<Felt>(), any::<Word>()), 0..=4),
                proptest::collection::vec((any::<Felt>(), any::<Word>(), any::<Word>()), 0..=4),
            )
                .prop_map(|(compressions, merkle_roots, mrupdates)| {
                    let mut replay = Self::default();
                    for (addr, hashed_state) in compressions {
                        replay.record_compression(addr, hashed_state);
                    }
                    for (addr, computed_root) in merkle_roots {
                        replay.record_build_merkle_root(addr, computed_root);
                    }
                    for (addr, old_root, new_root) in mrupdates {
                        replay.record_update_merkle_root(addr, old_root, new_root);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for HasherRequestReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Buffered-only: the write side panics on Streamed and the reader always yields
            // Buffered, so Default (Buffered) + record is the exact-roundtrip regime.
            proptest::collection::vec(any::<HasherOp>(), 0..=4)
                .prop_map(|ops| {
                    let mut replay = Self::default();
                    for op in ops {
                        replay.record(op);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for MastForestResolutionReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            proptest::collection::vec((any::<MastNodeId>(), any::<MastForestId>()), 0..=4)
                .prop_map(|resolutions| {
                    let mut replay = Self::default();
                    for (node_id, forest_id) in resolutions {
                        replay.record_resolution(node_id, forest_id);
                    }
                    replay
                })
                .boxed()
        }
    }

    impl Arbitrary for ExecutionReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Composes eight already-covered replay parts; all fields are pub.
            (
                any::<BlockStackReplay>(),
                any::<ExecutionContextReplay>(),
                any::<StackOverflowReplay>(),
                any::<MemoryReadsReplay>(),
                any::<AdviceReplay>(),
                any::<HasherResponseReplay>(),
                any::<BlockAddressReplay>(),
                any::<MastForestResolutionReplay>(),
            )
                .prop_map(
                    |(
                        block_stack,
                        execution_context,
                        stack_overflow,
                        memory_reads,
                        advice,
                        hasher,
                        block_address,
                        mast_forest_resolution,
                    )| {
                        Self {
                            block_stack,
                            execution_context,
                            stack_overflow,
                            memory_reads,
                            advice,
                            hasher,
                            block_address,
                            mast_forest_resolution,
                        }
                    },
                )
                .boxed()
        }
    }

    impl Arbitrary for CoreTraceFragmentContext {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Composes four covered parts; all fields are pub.
            (
                any::<CoreTraceState>(),
                any::<ExecutionReplay>(),
                any::<crate::continuation_stack::ContinuationStack<MastForestId>>(),
                any::<MastForestId>(),
            )
                .prop_map(|(state, replay, continuation, initial_mast_forest_id)| Self {
                    state,
                    replay,
                    continuation,
                    initial_mast_forest_id,
                })
                .boxed()
        }
    }
}

#[cfg(any(test, feature = "arbitrary"))]
mod ace_replay_arbitrary {
    use proptest::prelude::*;

    use super::AceReplay;
    use crate::trace::chiplets::CircuitEvaluation;

    impl Arbitrary for AceReplay {
        type Parameters = ();
        type Strategy = BoxedStrategy<Self>;

        fn arbitrary_with(_args: Self::Parameters) -> Self::Strategy {
            // Populated through the recording API; the pair's clk is derived from the evaluation.
            proptest::collection::vec(any::<CircuitEvaluation>(), 0..=3)
                .prop_map(|evaluations| {
                    let mut replay = Self::default();
                    for circuit_eval in evaluations {
                        // The draw is bounded to 3 entries, far below the recording limit, so
                        // this never rejects; ignore the Result to keep the generator infallible.
                        let _ = replay.record_circuit_evaluation(circuit_eval);
                    }
                    replay
                })
                .boxed()
        }
    }
}