newton-chainio 0.5.2

//! AvsWriter

use crate::{
    avs::{
        cancel::{self, CancelOutcome, CancelParams, NonceCanceller},
        errors,
        nonce_allocator::NonceAllocator,
    },
    error::ChainIoError,
};
use alloy::{
    consensus::Transaction as _,
    contract::{CallBuilder, Error as ContractError},
    network::TransactionBuilder,
    primitives::{Address, Bytes, B256, U256},
    providers::{PendingTransactionError, Provider, SendableTx, WalletProvider},
    rpc::types::{TransactionReceipt, TransactionRequest},
    signers::local::PrivateKeySigner,
    sol_types::{SolCall, SolError, SolType},
    transports::{TransportError, TransportErrorKind},
};
use eigensdk::{
    common::{get_signer, SdkSigner},
    types::operator::{QuorumNum, QuorumThresholdPercentage},
};
use hex;
use newton_core::{
    batch_task_manager::{
        BatchTaskManager::BatchTaskManagerInstance,
        INewtonProverTaskManager::{Task as BatchTask, TaskResponse as BatchTaskResponse},
    },
    challenge_verifier::{
        INewtonProverTaskManager::{
            ResponseCertificate as CvResponseCertificate, Task as CvTask, TaskResponse as CvTaskResponse,
        },
        BN254::G1Point as CvG1Point,
    },
    common::task_id,
    config::key::EcdsaKey,
    keys::{error::KeyError, load_ecdsa},
    newton_prover_service_manager::NewtonProverServiceManager,
    newton_prover_task_manager::{
        INewtonProverTaskManager::{ChallengeData, ResponseCertificate, Task, TaskResponse},
        NewtonMessage,
        NewtonProverTaskManager::{
            self,
            // Input validation errors
            BitmapValueTooLarge,
            BytesArrayLengthTooLong,
            BytesArrayNotOrdered,
            // Pause/state errors
            CurrentlyPaused,
            // BLS/Crypto errors
            ECAddFailed,
            ECMulFailed,
            ExpModFailed,
            InputAddressZero,
            InputArrayLengthMismatch,
            InputEmptyQuorumNumbers,
            InputNonSignerLengthMismatch,
            InvalidBLSPairingKey,
            InvalidBLSSignature,
            InvalidNewPausedStatus,
            InvalidQuorumApkHash,
            // Task lifecycle errors
            InvalidReferenceBlocknumber,
            NonSignerPubkeysNotSorted,
            // Access control errors
            OnlyPauser,
            OnlyRegistryCoordinatorOwner,
            OnlyTaskGenerator,
            OnlyUnpauser,
            // Operator errors
            ScalarTooLarge,
            TaskAlreadyExists,
            TaskAlreadyResponded,
            TaskMismatch,
        },
        BN254::G1Point,
    },
    state_commit_registry::{
        IStateRootCommittable::StateCommit,
        StateCommitRegistry::{
            self, CertificateMessageHashMismatch, InvalidNewStateRoot, InvalidPcr0Commitment, InvalidSealedSnapshot,
            SequenceGap, StateCommitRegistryInstance, StateRootMismatch, TimestampRegression,
            UnsupportedStateCommitVersion,
        },
    },
    TaskId,
};
use std::{str::FromStr, sync::Arc, time::Duration};
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};

/// Receipt timeout for `commitStateRoot` transactions.
///
/// Flat 60s across all chains (not chain-adaptive like `batch_receipt_timeout`).
/// Rationale: the protocol commit cadence is fixed at 120s regardless of
/// underlying block time, so a flat half-cadence timeout gives consistent
/// retry behavior on every chain — and ensures a stuck commit never blocks
/// the next aggregator's prepare-phase tick.
///
/// See `docs/PRIVATE_DATA_STORAGE.md` §6.1 for the 120s commit-cadence rationale.
pub const STATE_COMMIT_RECEIPT_TIMEOUT: Duration = Duration::from_secs(60);

/// Outcome of a successful batch broadcast: the transaction hash plus the
/// nonce the signer's filler stack assigned to it.
///
/// Capturing the nonce at send time is the linchpin of nonce-anchored
/// submission: it lets the receipt tracker gas-bump or cancel the exact slot
/// without re-deriving it via `eth_getTransactionByHash` (which returns `None`
/// once the original is evicted from the mempool, orphaning the nonce). See
/// `docs/BATCH_SUBMISSION_NONCE_ANCHORING.md`.
#[derive(Debug, Clone, Copy)]
pub struct BroadcastedTx {
    /// Hash of the broadcast transaction.
    pub tx_hash: B256,
    /// Nonce assigned by the signer's `NonceFiller`.
    pub nonce: u64,
    /// `max_fee_per_gas` the transaction was broadcast with (0 if the node
    /// assigned it and we couldn't read it back — callers floor against this).
    pub max_fee_per_gas: u128,
    /// `max_priority_fee_per_gas` the transaction was broadcast with.
    pub max_priority_fee_per_gas: u128,
}

/// Multiple of `base_fee` reserved as `max_fee` headroom so the full priority
/// tip stays realizable as the base fee rises while a replacement is pending,
/// expressed as `NUM/DEN` (= 2.5×). EIP-1559 caps base-fee growth at 12.5%/block,
/// so 2.5× ≈ 8 blocks of runway (1.125^8 ≈ 2.57) — leans aggressive so the bid
/// stays competitive through a sustained spike. `max_fee` is only a ceiling on
/// what we'd pay, not the actual cost (you pay `base_fee + tip`), so a generous
/// multiple costs nothing when the market is calm.
const BASE_FEE_HEADROOM_NUM: u128 = 5;
const BASE_FEE_HEADROOM_DEN: u128 = 2;

/// Default absolute `max_fee_per_gas` ceiling for escalating replacement/cancel
/// txs: 5000 gwei. A finite default (not `u128::MAX`) so even an `AvsWriter`
/// whose ceiling isn't explicitly configured still has a backstop against a
/// runaway bump/cancel loop — the failure mode that priced a 21k cancel at
/// ~786k gwei. Callers override via [`AvsWriter::with_max_fee_per_gas_ceiling`].
pub const DEFAULT_MAX_FEE_PER_GAS_CEILING_WEI: u128 = 5_000 * 1_000_000_000;

/// Compute escalating replacement fees from a market quote `(base_fee,
/// market_priority)` and the previous attempt's fees. Returns
/// `(max_fee_per_gas, max_priority_fee_per_gas)` satisfying three properties:
///
/// 1. **Competitive, monotonic tip.** `priority = max(market_priority,
///    prev_prio × 1.10)` — tracks the live market tip AND always clears the
///    node's ≥10% replacement floor on the priority field, so the bid never
///    stalls and the replacement is never rejected "underpriced".
///
/// 2. **Realizable tip (the point of this function).** A validator's actual
///    tip is `min(priority, max_fee − base_fee)`, so a high priority bid is
///    wasted unless `max_fee` has headroom above `base_fee`. We size
///    `max_fee ≥ base_fee × 2.5 + priority`, which guarantees `max_fee − base_fee
///    ≥ priority` even as the base fee climbs across the blocks a replacement
///    waits — so the FULL priority reaches the validator, not a throttled
///    fraction. `max_fee` also clears its own ≥10% floor (`prev_max_fee × 1.10`)
///    so the max-fee field never trips the replacement rule either.
///
/// 3. **Bounded.** Both fields are clamped to `ceiling` (absolute `max_fee`
///    backstop, wei), and `priority` is re-clamped to the post-ceiling `max_fee`
///    so `priority ≤ max_fee` always holds. Pass `u128::MAX` to disable.
///
/// When the market quote is unavailable (`None` — fee-history RPC down), fall
/// back to bumping the previous fees by the floor only. The tip still climbs,
/// but without a fresh `base_fee` the headroom guarantee can't be recomputed;
/// the prior `max_fee` (which already carried headroom) is escalated as-is.
/// Pure function for unit testing.
fn escalate_floor(market: Option<(u128, u128)>, prev_max_fee: u128, prev_prio: u128, ceiling: u128) -> (u128, u128) {
    // Priority: track the market tip, floored at a ≥10% bump over the previous.
    let market_prio = market.map(|(_, p)| p).unwrap_or(0);
    let prio = market_prio.max(prev_prio.saturating_mul(110) / 100);

    // Max fee: clear its own ≥10% replacement floor, and — when we have a fresh
    // base fee — reserve enough headroom that the full `prio` is realizable as a
    // tip (`max_fee − base_fee ≥ prio`) even as the base fee rises.
    let floor_max = prev_max_fee.saturating_mul(110) / 100;
    let max_fee = match market {
        Some((base_fee, _)) => {
            let with_headroom = base_fee
                .saturating_mul(BASE_FEE_HEADROOM_NUM)
                .saturating_div(BASE_FEE_HEADROOM_DEN)
                .saturating_add(prio);
            floor_max.max(with_headroom)
        }
        None => floor_max,
    };

    // Apply the absolute ceiling, then keep priority ≤ (clamped) max_fee.
    let max_fee = max_fee.min(ceiling);
    let prio = prio.min(max_fee);
    (max_fee, prio)
}

/// load a signer from an ECDSA key and RPC URL
pub fn load_signer(ecdsa_key: &EcdsaKey, rpc_url: &str) -> eyre::Result<SdkSigner> {
    let signer = load_ecdsa(ecdsa_key)?;
    let signer_private_key = hex::encode(signer.to_field_bytes());
    Ok(get_signer(&signer_private_key, rpc_url))
}

/// AvsWriter struct
#[derive(Debug, Clone)]
pub struct AvsWriter {
    /// task manager
    task_manager_addr: Address,
    /// identity registry
    identity_registry_addr: Address,
    /// state commit registry — anchors the unified per-chain JMT root via
    /// 120s BLS-gated `commitStateRoot` calls. Sourced from the deployment
    /// JSON (`addresses.stateCommitRegistry`) at config load and passed through
    /// `NewtonAvsContractsConfig.state_commit_registry`; callers materialize
    /// `Address::ZERO` via `unwrap_or` when the registry hasn't been deployed
    /// on this chain yet. `commit_state_root` runtime-guards against the zero
    /// address so callers degrade gracefully on chains without PDS wiring.
    state_commit_registry_addr: Address,
    /// Provider used for signing and broadcasting transactions
    signer_provider: SdkSigner,
    /// signer
    pub signer: EcdsaKey,
    /// rpc url
    pub rpc_url: String,
    /// Chain ID — used for chain-adaptive tuning (e.g., receipt timeout)
    pub chain_id: u64,
    /// Absolute `max_fee_per_gas` ceiling (wei) for escalating replacement and
    /// cancel transactions. Defaults to [`DEFAULT_MAX_FEE_PER_GAS_CEILING_WEI`]
    /// (5000 gwei) — a finite backstop so a runaway bump/cancel loop can never
    /// price a tx above the signer's means and lock the slot in an
    /// unaffordable-and-still-climbing state. Only ever consulted on the
    /// escalation path (`gas_bump_batch_tx`/`cancel_nonce`), so it never
    /// interferes with normal node-priced sends. The batch submitter overrides
    /// it from config via [`with_max_fee_per_gas_ceiling`]. See `escalate_floor`.
    ///
    /// [`with_max_fee_per_gas_ceiling`]: AvsWriter::with_max_fee_per_gas_ceiling
    max_fee_per_gas_ceiling: u128,
    /// Optional owned nonce allocator. When `Some`, every broadcast path reserves
    /// its nonce here and sets it explicitly on the request, so alloy's
    /// `CachedNonceManager` never allocates and a failed broadcast can release the
    /// slot (gas-estimate revert) or drive it through confirm-or-cancel (send
    /// failure) instead of orphaning it. MUST be shared (same `Arc`) across every
    /// `AvsWriter` on a given signer — see [`NonceAllocator`] — so the explicit
    /// sequence has a single source of truth. When `None`, nonces fall back to the
    /// implicit filler allocation (legacy behavior for standalone writers).
    nonce_allocator: Option<Arc<NonceAllocator>>,
    /// Optional shutdown signal for the unbounded stuck-nonce cancel loop. When
    /// `Some`, an inline cancel resolution (`resolve_stuck_nonce`) exits promptly
    /// on shutdown instead of retrying forever; when `None` the loop is truly
    /// unbounded (acceptable for short-lived/standalone writers).
    cancel_token: Option<CancellationToken>,
}

/// Parameters for creating a task onchain
/// Note: policyTaskData is no longer sent with createNewTask - operators generate it independently
#[derive(Debug, Clone)]
pub struct SendTaskParams {
    /// Task ID
    pub task_id: TaskId,
    /// Task created block - the offchain estimated block number used as single source of truth
    /// for BLS signature verification. Must be > 0, < current block, and within 256 blocks.
    pub task_created_block: u32,
    /// Task request WASM args for operators to generate policyTaskData
    pub wasm_args: Bytes,
    /// Policy client address
    pub policy_client: Address,
    /// Intent
    pub intent: NewtonMessage::Intent,
    /// Intent signature
    pub intent_signature: Option<Bytes>,
    /// Quorum number
    pub quorum_number: Vec<QuorumNum>,
    /// Quorum threshold percentage
    pub quorum_threshold_percentage: QuorumThresholdPercentage,
    /// timestamp marking the offchain ingestion of the task
    pub initialization_timestamp: u64,
}

/// Parameters for challenging a privacy task with missing or invalid TEE attestation.
#[derive(Clone)]
#[allow(missing_debug_implementations)]
pub struct TeeAttestationChallengeParams {
    /// ChallengeVerifier contract address
    pub challenge_verifier_addr: Address,
    /// The task being challenged
    pub task: Task,
    /// The task response being challenged
    pub task_response: TaskResponse,
    /// Response certificate (for challenge window check)
    pub task_response_metadata: ResponseCertificate,
    /// SP1 attestation proof public values (empty for missing attestation)
    pub attestation_proof_data: Bytes,
    /// SP1 Groth16 proof bytes (empty for missing attestation)
    pub attestation_proof_bytes: Bytes,
    /// BLS G1 pubkeys of non-signing operators
    pub pub_keys_of_non_signing_operators: Vec<G1Point>,
}

impl AvsWriter {
    /// new instance
    pub async fn new(
        task_manager_addr: Address,
        identity_registry_addr: Address,
        state_commit_registry_addr: Address,
        rpc_url: String,
        signer: EcdsaKey,
        chain_id: u64,
    ) -> Result<Self, ChainIoError> {
        let signer_provider = load_signer(&signer, &rpc_url).map_err(ChainIoError::SignerError)?;

        Ok(AvsWriter {
            task_manager_addr,
            identity_registry_addr,
            state_commit_registry_addr,
            signer_provider,
            signer,
            rpc_url,
            chain_id,
            max_fee_per_gas_ceiling: DEFAULT_MAX_FEE_PER_GAS_CEILING_WEI,
            nonce_allocator: None,
            cancel_token: None,
        })
    }

    /// Attach a shared owned nonce allocator. Builder-style; without it the writer
    /// uses the implicit filler nonce (legacy). Pass the SAME `Arc` to every
    /// `AvsWriter` sharing a signer so the explicit nonce sequence is coordinated.
    pub fn with_nonce_allocator(mut self, allocator: Arc<NonceAllocator>) -> Self {
        self.nonce_allocator = Some(allocator);
        self
    }

    /// Attach a shutdown signal so the unbounded stuck-nonce cancel loop exits
    /// promptly on shutdown instead of retrying forever. Builder-style; without
    /// it the loop is truly unbounded (fine for short-lived/standalone writers).
    pub fn with_cancel_token(mut self, token: CancellationToken) -> Self {
        self.cancel_token = Some(token);
        self
    }

    /// The shared nonce allocator, if one is attached. Exposed so the batch
    /// submitter can resync it on gap detection and the startup/runtime sweep can
    /// reconcile it with the chain.
    pub fn nonce_allocator(&self) -> Option<&Arc<NonceAllocator>> {
        self.nonce_allocator.as_ref()
    }

    /// Reserve an explicit nonce from the allocator if one is attached. Direct
    /// `.send()` paths (`send_task`, `commit_state_root`) call this and pin the
    /// result on the contract call so the implicit `CachedNonceManager` stays
    /// inert and their nonces stay coordinated with the batch path. Returns `None`
    /// when no allocator is attached (legacy implicit-nonce behavior).
    async fn reserve_nonce(&self) -> Result<Option<u64>, ChainIoError> {
        match self.nonce_allocator.as_ref() {
            Some(alloc) => Ok(Some(alloc.reserve().await?)),
            None => Ok(None),
        }
    }

    /// Override the absolute `max_fee_per_gas` ceiling (wei) for escalating
    /// replacement and cancel transactions. Builder-style; without it the writer
    /// keeps the finite [`DEFAULT_MAX_FEE_PER_GAS_CEILING_WEI`] backstop. A value
    /// of 0 is treated as "explicitly disabled" (`u128::MAX`) so an operator can
    /// opt out via config, while a mis-set/zeroed value can never make every
    /// replacement unpriced.
    pub fn with_max_fee_per_gas_ceiling(mut self, ceiling_wei: u128) -> Self {
        self.max_fee_per_gas_ceiling = if ceiling_wei == 0 { u128::MAX } else { ceiling_wei };
        self
    }

    /// Create a new instance with a shared `SdkSigner` provider.
    ///
    /// Use this when multiple `AvsWriter` instances share the same signing key
    /// so they share a single nonce manager and avoid "nonce too low" races.
    pub fn new_with_provider(
        task_manager_addr: Address,
        identity_registry_addr: Address,
        state_commit_registry_addr: Address,
        rpc_url: String,
        signer: EcdsaKey,
        signer_provider: SdkSigner,
        chain_id: u64,
    ) -> Self {
        // Verify signer key matches the provider's wallet (debug builds only)
        #[cfg(debug_assertions)]
        {
            use alloy::providers::WalletProvider;
            if let Ok(loaded) = load_ecdsa(&signer) {
                debug_assert_eq!(
                    signer_provider.default_signer_address(),
                    loaded.address(),
                    "signer key and provider wallet address must match"
                );
            }
        }
        AvsWriter {
            task_manager_addr,
            identity_registry_addr,
            state_commit_registry_addr,
            signer_provider,
            signer,
            rpc_url,
            chain_id,
            max_fee_per_gas_ceiling: DEFAULT_MAX_FEE_PER_GAS_CEILING_WEI,
            nonce_allocator: None,
            cancel_token: None,
        }
    }

    /// get task manager address
    pub fn task_manager_addr(&self) -> Address {
        self.task_manager_addr
    }

    /// get identity registry address
    pub fn identity_registry_addr(&self) -> Address {
        self.identity_registry_addr
    }

    /// get state commit registry address (`Address::ZERO` if not present in deployment JSON for this chain)
    pub fn state_commit_registry_addr(&self) -> Address {
        self.state_commit_registry_addr
    }

    /// Reference to the underlying RPC provider bound to this writer's chain.
    ///
    /// Exposed for read-only callers that need to issue arbitrary chain queries
    /// against the same provider (e.g., `eth_getTransactionByHash` for replaying
    /// historical `commitStateRoot` calldata in PDS Path A* signed-read).
    pub fn provider(&self) -> &SdkSigner {
        &self.signer_provider
    }

    /// Returns the current block number from the chain's RPC provider.
    pub async fn get_block_number(&self) -> Result<u64, ChainIoError> {
        Ok(self.signer_provider.get_block_number().await?)
    }

    /// Transaction count for the signer address. `pending = true` includes
    /// mempool transactions; `false` counts only mined transactions. The gap
    /// `pending − latest` > 0 indicates a wedged nonce slot (a broadcast TX
    /// that hasn't mined). Used for the `signer_nonce_gap` metric and the
    /// startup gap sweep.
    pub async fn signer_transaction_count(&self, pending: bool) -> Result<u64, ChainIoError> {
        let addr = self.signer_provider.default_signer_address();
        let call = self.signer_provider.get_transaction_count(addr);
        let count = if pending {
            call.pending().await
        } else {
            call.latest().await
        };
        count.map_err(ChainIoError::RpcError)
    }

    /// Broadcast a batch create+respond TX without waiting for the receipt.
    ///
    /// Returns the TX hash on successful broadcast. The caller is responsible
    /// for polling the receipt separately via `get_receipt_by_hash`.
    /// This is the send-only half of the pipelined batch submission path.
    pub async fn send_batch_create_and_respond(
        &self,
        batch_contract_addr: Address,
        tasks: Vec<Task>,
        responses: Vec<TaskResponse>,
        signature_data_array: Vec<Bytes>,
        attestation_data: Vec<Bytes>,
    ) -> Result<BroadcastedTx, ChainIoError> {
        let (batch_tasks, batch_responses, batch_sigs, batch_attestations, batch_size) =
            Self::encode_batch_items(tasks, responses, signature_data_array, attestation_data)?;

        let contract = BatchTaskManagerInstance::new(batch_contract_addr, self.signer_provider.clone());

        info!(
            batch_size,
            batch_contract = %batch_contract_addr,
            "pipelined: submitting batchCreateAndRespondToTasks"
        );

        let call = contract.batchCreateAndRespondToTasks(batch_tasks, batch_responses, batch_sigs, batch_attestations);

        // Fill-then-send so we capture the nonce the filler stack assigns. The
        // wallet filler signs into an `Envelope`; re-sending that envelope via
        // `send_tx_envelope` is a no-op for the fillers (they report `Finished`
        // on a built tx), so the `NonceFiller` increments its cache exactly once.
        self.broadcast_filled(call.into_transaction_request(), batch_size).await
    }

    /// Fill a request through the signer's filler stack, capture the assigned
    /// nonce + fees, then broadcast the resulting envelope. Shared by the
    /// initial send, gas bump, and cancel paths so every broadcast yields a
    /// `BroadcastedTx`.
    ///
    /// When an owned [`NonceAllocator`] is active and the request does not yet
    /// carry a nonce (initial send), this reserves a nonce and pins it on the
    /// request so alloy's `CachedNonceManager` stays inert (it only fills when the
    /// nonce is absent). The reserved nonce's lifecycle is then total:
    ///
    /// - `fill` fails (gas-estimate revert — nothing broadcast) → [`release`] the
    ///   nonce so the next reservation reuses it; the on-chain sequence stays
    ///   gap-free at zero cost.
    /// - `send` fails *after* fill (tx may have propagated) → return
    ///   [`ChainIoError::BroadcastSendFailed`] carrying the reserved nonce so the
    ///   caller resolves the slot via confirm-or-cancel. The nonce is NOT released
    ///   (a propagated tx could still mine).
    ///
    /// A caller-pinned nonce (bump/cancel replacement) is used as-is and never
    /// touches the allocator — the receipt tracker already owns that slot.
    ///
    /// [`release`]: NonceAllocator::release
    async fn broadcast_filled(
        &self,
        request: TransactionRequest,
        batch_size: usize,
    ) -> Result<BroadcastedTx, ChainIoError> {
        let (broadcast, _pending) = self.fill_and_send(request, batch_size).await?;
        Ok(broadcast)
    }

    /// Core of every allocator-aware broadcast: reserve + pin the nonce, run the
    /// filler stack, and broadcast — applying the fill-vs-send failure split that
    /// keeps the nonce sequence gap-free. Returns both the [`BroadcastedTx`]
    /// summary and the live [`PendingTransactionBuilder`] so receipt-waiting
    /// callers (`send_task`, `batch_respond_to_tasks`) reuse the exact same split
    /// instead of a coarse `.send()` that releases a possibly-live nonce.
    ///
    /// Failure handling:
    /// - `fill` fails (gas-estimate revert — nothing broadcast) → [`release`] the
    ///   reserved nonce; the slot is free and reused at zero cost.
    /// - `send` fails after fill (tx may have propagated) → return
    ///   [`ChainIoError::BroadcastSendFailed`] carrying the nonce AND the signed
    ///   fees so the caller resolves the slot via a correctly-priced cancel. The
    ///   nonce is NOT released (a propagated tx could still mine).
    ///
    /// A caller-pinned nonce (bump/cancel replacement) is used as-is and never
    /// touches the allocator — the receipt tracker already owns that slot.
    ///
    /// [`release`]: NonceAllocator::release
    async fn fill_and_send(
        &self,
        mut request: TransactionRequest,
        batch_size: usize,
    ) -> Result<
        (
            BroadcastedTx,
            alloy::providers::PendingTransactionBuilder<alloy::network::Ethereum>,
        ),
        ChainIoError,
    > {
        // Reserve + pin the nonce only for an allocator-owned initial send (no
        // nonce yet). Caller-pinned replacements skip this: their slot is already
        // owned, and reserving again would advance the counter spuriously.
        let reserved = match (request.nonce, self.nonce_allocator.as_ref()) {
            (None, Some(alloc)) => {
                let n = alloc.reserve().await?;
                request.set_nonce(n);
                Some(n)
            }
            _ => None,
        };

        // `fill` runs the gas + nonce fillers; a top-level revert (e.g.
        // `BatchPartialFailure`) surfaces here as `eth_estimateGas` revert data.
        // Nothing was broadcast, so the reserved nonce must not be orphaned:
        // LIFO-reclaim it if it's still the latest reservation (free, gap-less).
        // If a newer nonce was reserved concurrently, `release` returns false —
        // `n` is now a real gap below committed work, so resolve it on-chain via a
        // same-nonce cancel before returning. Then route the error through
        // `classify_send_error` so Tier 1 classification is preserved.
        let sendable = match self.signer_provider.fill(request).await {
            Ok(s) => s,
            Err(e) => {
                if let (Some(n), Some(alloc)) = (reserved, self.nonce_allocator.as_ref()) {
                    if !alloc.release(n).await {
                        self.spawn_resolve_stuck_nonce(n, 0, 0);
                    }
                }
                return Self::classify_send_error(ContractError::TransportError(e), batch_size);
            }
        };

        // After a full fill the tx is a signed `Envelope`; read nonce + fees off it.
        let (nonce, max_fee_per_gas, max_priority_fee_per_gas) = match &sendable {
            SendableTx::Envelope(env) => (
                env.nonce(),
                env.max_fee_per_gas(),
                env.max_priority_fee_per_gas().unwrap_or_default(),
            ),
            SendableTx::Builder(req) => (
                req.nonce().ok_or_else(|| ChainIoError::CreateNewTaskCallFail {
                    reason: "nonce filler did not assign a nonce".into(),
                })?,
                req.max_fee_per_gas().unwrap_or_default(),
                req.max_priority_fee_per_gas().unwrap_or_default(),
            ),
        };

        let send_result = match sendable {
            SendableTx::Envelope(env) => self.signer_provider.send_tx_envelope(env).await,
            SendableTx::Builder(req) => self.signer_provider.send_transaction(req).await,
        };

        match send_result {
            Ok(pending) => {
                let tx_hash = *pending.tx_hash();
                info!(
                    batch_size,
                    %tx_hash, nonce, max_fee_per_gas, max_priority_fee_per_gas,
                    "batch pipelined: TX broadcast, receipt pending"
                );
                Ok((
                    BroadcastedTx {
                        tx_hash,
                        nonce,
                        max_fee_per_gas,
                        max_priority_fee_per_gas,
                    },
                    pending,
                ))
            }
            Err(e) => {
                // The send failed AFTER fill succeeded: the tx was signed at
                // `nonce` and may already have propagated to the mempool. We must
                // NOT release the nonce — a propagated tx could still mine, and
                // reusing the slot would collide. When the allocator owns this
                // send, surface the reserved nonce + signed fees so the caller
                // resolves the slot via a correctly-priced cancel instead of
                // orphaning it.
                if reserved.is_some() {
                    return Err(ChainIoError::BroadcastSendFailed {
                        nonce,
                        max_fee_per_gas,
                        max_priority_fee_per_gas,
                        reason: e.to_string(),
                    });
                }
                Self::classify_send_error(ContractError::TransportError(e), batch_size)
            }
        }
    }

    /// Poll for a transaction receipt by hash with exponential backoff.
    ///
    /// Pass `None` for `timeout` to poll indefinitely (caller manages cancellation).
    /// Pass `Some(duration)` to return `TransactionTimeout` after the deadline.
    pub async fn get_receipt_by_hash(
        &self,
        tx_hash: B256,
        timeout: Option<std::time::Duration>,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let mut backoff = std::time::Duration::from_millis(500);
        let max_backoff = std::time::Duration::from_secs(4);
        let deadline = timeout.map(|t| tokio::time::Instant::now() + t);

        loop {
            match self.signer_provider.get_transaction_receipt(tx_hash).await {
                Ok(Some(receipt)) => return Ok(receipt),
                Ok(None) => {
                    if let Some(dl) = deadline {
                        if tokio::time::Instant::now() >= dl {
                            return Err(ChainIoError::TransactionTimeout {
                                timeout_secs: timeout.unwrap().as_secs(),
                            });
                        }
                    }
                    tokio::time::sleep(backoff).await;
                    backoff = std::cmp::min(backoff * 2, max_backoff);
                }
                Err(e) => return Err(ChainIoError::RpcError(e)),
            }
        }
    }

    /// Quote the current market from a 10-block fee history: the latest block's
    /// base fee and a `bump_percent`-inflated 90th-percentile priority fee
    /// (tip). Returns `(base_fee, bumped_priority)`.
    ///
    /// Returns `None` only when the fee-history RPC fails or carries no base
    /// fee — callers MUST fall back to (and floor against) the previous
    /// attempt's fees so a replacement is never broadcast unpriced (which the
    /// node would reject as "replacement transaction underpriced").
    ///
    /// Unlike before, this returns the RAW base fee rather than a pre-combined
    /// `max_fee`, so [`escalate_floor`] can size `max_fee` to guarantee the tip
    /// is realizable (`max_fee − base_fee ≥ priority`) — see that function.
    async fn market_fees(&self, bump_percent: u32) -> Option<(u128, u128)> {
        let bump_multiplier = (100 + bump_percent) as u128;
        let fee_history = self
            .signer_provider
            .get_fee_history(10, Default::default(), &[90.0])
            .await
            .ok()?;
        let base_fee = fee_history.latest_block_base_fee()?;
        // 90th percentile priority fee across the 10-block window.
        let p90_priority = fee_history
            .reward
            .as_ref()
            .and_then(|rewards| {
                rewards
                    .iter()
                    .filter_map(|block_rewards| block_rewards.first().copied())
                    .max()
            })
            .unwrap_or(2_000_000_000); // fallback: 2 gwei
        let priority = p90_priority.saturating_mul(bump_multiplier) / 100;
        Some((base_fee, priority))
    }

    /// Compute escalating replacement fees: the max of a fresh market quote and
    /// a ≥10% bump over the previous attempt, on BOTH fee fields — and sized so
    /// the priority bid is actually realizable as a validator tip.
    ///
    /// `max(market, prev × 1.10)` guarantees the replacement always clears the
    /// node's EIP-1559 ≥10% replacement floor (so it is never rejected as
    /// underpriced) AND tracks a rising market. With `prev = 0` (first bump
    /// after the original, whose fees we always capture) the market quote wins.
    async fn escalated_fees(&self, prev_max_fee: u128, prev_prio: u128, bump_percent: u32) -> (u128, u128) {
        let market = self.market_fees(bump_percent).await;
        escalate_floor(market, prev_max_fee, prev_prio, self.max_fee_per_gas_ceiling)
    }

    /// Resubmit a batch TX at higher gas to replace a stuck transaction, at the
    /// SAME nonce. Used by the watchdog when a TX has no receipt after the
    /// configured timeout.
    ///
    /// Unlike the previous implementation, this takes the nonce + the previous
    /// attempt's fees directly (captured at broadcast via [`BroadcastedTx`])
    /// rather than re-deriving them from `eth_getTransactionByHash` — which
    /// returns `None` once the original is evicted, orphaning the slot. The
    /// returned [`BroadcastedTx`] carries the (escalated) fees so the caller
    /// can thread them into the next attempt for monotonic escalation.
    #[allow(clippy::too_many_arguments)]
    pub async fn gas_bump_batch_tx(
        &self,
        nonce: u64,
        batch_contract_addr: Address,
        tasks: Vec<Task>,
        responses: Vec<TaskResponse>,
        signature_data_array: Vec<Bytes>,
        attestation_data: Vec<Bytes>,
        prev_max_fee: u128,
        prev_priority_fee: u128,
        bump_percent: u32,
    ) -> Result<BroadcastedTx, ChainIoError> {
        let (batch_tasks, batch_responses, batch_sigs, batch_attestations, batch_size) =
            Self::encode_batch_items(tasks, responses, signature_data_array, attestation_data)?;

        let (max_fee, priority) = self.escalated_fees(prev_max_fee, prev_priority_fee, bump_percent).await;

        let contract = BatchTaskManagerInstance::new(batch_contract_addr, self.signer_provider.clone());
        let request = contract
            .batchCreateAndRespondToTasks(batch_tasks, batch_responses, batch_sigs, batch_attestations)
            .nonce(nonce)
            .max_fee_per_gas(max_fee)
            .max_priority_fee_per_gas(priority)
            .into_transaction_request();

        info!(
            batch_size,
            nonce, max_fee, priority, bump_percent, "batch pipelined: gas bump replacement broadcasting"
        );
        self.broadcast_filled(request, batch_size).await
    }

    /// Broadcast a same-nonce cancel: a 0-value self-send at the stuck `nonce`,
    /// priced to replace the stuck batch (escalated fees). This is the bounded
    /// escape for a structurally un-includable payload — it occupies the nonce
    /// slot cheaply (21k gas) so subsequent nonces can drain, upholding the
    /// "never abandon a consumed nonce" invariant. The intended payload is
    /// sacrificed by the caller (re-queued with a fresh nonce), never dropped.
    pub async fn cancel_nonce(
        &self,
        nonce: u64,
        prev_max_fee: u128,
        prev_priority_fee: u128,
        bump_percent: u32,
    ) -> Result<BroadcastedTx, ChainIoError> {
        let me = self.signer_provider.default_signer_address();
        let (max_fee, priority) = self.escalated_fees(prev_max_fee, prev_priority_fee, bump_percent).await;

        let request = TransactionRequest::default()
            .with_from(me)
            .with_to(me)
            .with_value(U256::ZERO)
            .with_nonce(nonce)
            .with_gas_limit(21_000)
            .with_max_fee_per_gas(max_fee)
            .with_max_priority_fee_per_gas(priority);

        warn!(
            nonce,
            max_fee, priority, "batch pipelined: broadcasting same-nonce cancel to free stuck slot"
        );
        // batch_size 0 — cancel carries no items; classify_send_error only uses it for logging.
        // The cancel pins its own nonce, so `fill_and_send` never reserves/releases.
        self.broadcast_filled(request, 0).await
    }

    /// Resolve a reserved nonce whose on-chain fate is **uncertain** — its tx may
    /// have propagated to the mempool (a post-fill send failure, or a receipt
    /// timeout after a successful broadcast). Releasing it would risk a collision
    /// with a tx that still mines; abandoning it strands every higher nonce
    /// (blocked-from-below). So we resolve it on-chain via the shared
    /// [`cancel::cancel_until_resolved`] loop — the SAME unbounded, escalating,
    /// shutdown-aware logic the pipelined batch path uses, so the "never abandon a
    /// nonce" invariant holds identically here.
    ///
    /// Used by the direct `.send()` / `send_with_retries` paths
    /// (`send_task`, `batch_respond_to_tasks`, `send_aggregated_response`,
    /// `commit_state_root`) which — unlike the pipelined path — have no background
    /// receipt tracker, so the loop runs inline. It is bounded only by the
    /// optional shutdown token (`with_cancel_token`); a wedged signer recovers
    /// automatically once unblocked. No-op when no allocator is attached.
    ///
    /// `seed_*` price the first cancel (pass the stuck tx's fees to out-price it;
    /// 0 falls back to a market quote).
    async fn resolve_stuck_nonce(&self, nonce: u64, seed_max_fee: u128, seed_priority: u128) {
        // Only meaningful when we own the sequence; otherwise the implicit filler
        // manages nonces and there is nothing to coordinate.
        if self.nonce_allocator.is_none() {
            return;
        }
        // Clamp the seed to the fee ceiling. The cancel's per-broadcast price is
        // already ceiling-clamped by `escalate_floor`, but clamping the seed too
        // keeps the starting bid within bounds (a propagated original priced at/above
        // the ceiling can't push the cancel's opening bid past it).
        let seed_max_fee = seed_max_fee.min(self.max_fee_per_gas_ceiling);
        let seed_priority = seed_priority.min(self.max_fee_per_gas_ceiling);
        let params = CancelParams {
            confirm_timeout: self.batch_receipt_timeout(),
            ..CancelParams::default()
        };
        match cancel::cancel_until_resolved(
            self,
            self.chain_id,
            nonce,
            seed_max_fee,
            seed_priority,
            params,
            self.cancel_token.as_ref(),
        )
        .await
        {
            CancelOutcome::Resolved => {
                info!(nonce, "resolve_stuck_nonce: slot resolved on-chain");
            }
            CancelOutcome::Shutdown => {
                warn!(
                    nonce,
                    "resolve_stuck_nonce: shutdown before slot resolved (startup sweep recovers it)"
                );
            }
        }
    }

    /// Whether a failed submission's on-chain outcome is *uncertain* — i.e. a tx
    /// may have propagated and consumed the nonce without us seeing a receipt — so
    /// the slot must be resolved on-chain. A mined revert (`TransactionReverted`,
    /// status=false) is NOT uncertain: the nonce is already consumed, so resolving
    /// it would be a wasted cancel and would muddy the poison signal callers key
    /// on. Fill-stage failures don't reach the resolve sites (the nonce is reclaimed
    /// or never reserved), so this only needs to separate "mined revert" from the
    /// timeout / post-send / retries-exhausted cases.
    fn is_uncertain_outcome(e: &ChainIoError) -> bool {
        !matches!(e, ChainIoError::TransactionReverted(_))
    }

    /// Spawn [`resolve_stuck_nonce`](Self::resolve_stuck_nonce) on a detached task
    /// and return immediately. The direct-send paths (`send_task`,
    /// `batch_respond_to_tasks`, `send_aggregated_response`, `commit_state_root`)
    /// run on serial drivers (the `tx_worker` loop, the 120s commit tick), so
    /// resolving inline would block every queued submission behind an unbounded
    /// cancel whenever the signer is wedged. Spawning matches the pipelined path
    /// (which detaches via `spawn_monitored`) so the driver keeps draining; the
    /// resolver still upholds the never-abandon invariant and exits cleanly on the
    /// shutdown token. Cheap: every `AvsWriter` field is `Arc`-backed or `Copy`.
    fn spawn_resolve_stuck_nonce(&self, nonce: u64, seed_max_fee: u128, seed_priority: u128) {
        // No allocator → nothing to coordinate; don't spawn a task that no-ops.
        if self.nonce_allocator.is_none() {
            return;
        }
        let resolver = self.clone();
        tokio::spawn(async move {
            resolver.resolve_stuck_nonce(nonce, seed_max_fee, seed_priority).await;
        });
    }

    /// Encode task/response/signature/attestation tuples into batch-compatible ABI types.
    /// Shared between `send_batch_create_and_respond` and `gas_bump_batch_tx`.
    #[allow(clippy::result_large_err, clippy::type_complexity)]
    fn encode_batch_items(
        tasks: Vec<Task>,
        responses: Vec<TaskResponse>,
        signature_data_array: Vec<Bytes>,
        attestation_data: Vec<Bytes>,
    ) -> Result<(Vec<BatchTask>, Vec<BatchTaskResponse>, Vec<Bytes>, Vec<Bytes>, usize), ChainIoError> {
        let batch_size = tasks.len();
        let mut batch_tasks: Vec<BatchTask> = Vec::with_capacity(batch_size);
        let mut batch_responses: Vec<BatchTaskResponse> = Vec::with_capacity(batch_size);
        let mut batch_sigs: Vec<Bytes> = Vec::with_capacity(batch_size);
        let mut batch_attestations: Vec<Bytes> = Vec::with_capacity(batch_size);

        for (i, (((t, r), sig), att)) in tasks
            .into_iter()
            .zip(responses)
            .zip(signature_data_array)
            .zip(attestation_data)
            .enumerate()
        {
            let task_encoded = Task::abi_encode(&t);
            let bt = match BatchTask::abi_decode(&task_encoded) {
                Ok(bt) => bt,
                Err(e) => {
                    warn!(error = %e, index = i, "batch ABI-decode task failed, skipping item");
                    continue;
                }
            };

            let resp_encoded = TaskResponse::abi_encode(&r);
            let br = match BatchTaskResponse::abi_decode(&resp_encoded) {
                Ok(br) => br,
                Err(e) => {
                    warn!(error = %e, index = i, "batch ABI-decode response failed, skipping item");
                    continue;
                }
            };

            batch_tasks.push(bt);
            batch_responses.push(br);
            batch_sigs.push(sig);
            batch_attestations.push(att);
        }

        let skipped = batch_size - batch_tasks.len();
        if skipped > 0 {
            warn!(
                batch_size,
                skipped, "batch: {} item(s) skipped due to ABI-decode failure", skipped
            );
        }
        if batch_tasks.is_empty() {
            return Err(ChainIoError::CreateNewTaskCallFail {
                reason: "all task/response items failed ABI-decode; batch is empty".to_string(),
            });
        }

        Ok((batch_tasks, batch_responses, batch_sigs, batch_attestations, batch_size))
    }

    /// Classify an error from `.send().await` into the appropriate `ChainIoError`.
    /// Shared between all batch submission methods.
    #[allow(clippy::result_large_err)]
    fn classify_send_error<T>(e: ContractError, batch_size: usize) -> Result<T, ChainIoError> {
        if let Some(failures) = Self::parse_batch_partial_failure(&e) {
            let fail_count = failures.len();
            info!(batch_size, fail_count, "batch: partial failure from simulation");
            Err(ChainIoError::BatchPartialFailure { failures })
        } else if let Some(revert_data) = e.as_revert_data() {
            let classified = errors::classify_top_level_revert(&revert_data);
            warn!(batch_size, error = %classified, "batch: top-level revert from simulation");
            Err(classified)
        } else {
            Err(ChainIoError::ContractError(e))
        }
    }

    /// Send task. Returns tx receipt and the full Task struct from the on-chain event.
    ///
    /// The returned Task contains the actual `taskCreatedBlock` set by the contract,
    /// which should be used for response submission instead of any pre-estimated value.
    pub async fn send_task(&self, params: SendTaskParams) -> Result<(TransactionReceipt, Task), ChainIoError> {
        let SendTaskParams {
            task_id,
            task_created_block,
            wasm_args,
            policy_client,
            intent,
            intent_signature,
            quorum_number,
            quorum_threshold_percentage,
            initialization_timestamp,
        } = params.clone();

        let task_manager_contract = NewtonProverTaskManager::new(self.task_manager_addr, self.signer_provider.clone());

        // Log task_created_block for diagnostics (tracking InvalidTaskCreatedBlock issue)
        tracing::info!(
            task_id = %newton_core::hex!(task_id),
            task_created_block = task_created_block,
            policy_client = %policy_client,
            "send_task: creating task with taskCreatedBlock"
        );

        // Task is now minimal - policyTaskData is generated by operators independently
        // Use offchain estimated block as single source of truth for BLS verification
        let create_new_task_call = task_manager_contract.createNewTask(Task {
            taskId: task_id,
            policyClient: policy_client,
            taskCreatedBlock: task_created_block,
            quorumThresholdPercentage: quorum_threshold_percentage.into(),
            intent,
            intentSignature: intent_signature.unwrap_or_default(),
            wasmArgs: wasm_args,
            quorumNumbers: quorum_number.into(),
            initializationTimestamp: U256::from(initialization_timestamp),
        });

        // Route through `fill_and_send` so this path gets the same fill-vs-send
        // failure split as the pipelined batch path: a fill-stage failure releases
        // the reserved nonce, while a post-fill send failure surfaces
        // `BroadcastSendFailed` (nonce NOT released — the tx may have propagated).
        // A coarse `.send()` + release-on-any-error here would release a possibly
        // live nonce and re-create the blocked-from-below gap. batch_size 1 — one
        // task; only used for logging/classification.
        let create_new_task_result = self
            .fill_and_send(create_new_task_call.into_transaction_request(), 1)
            .await
            .map(|(_broadcast, pending)| pending);

        match create_new_task_result {
            Ok(create_new_task) => {
                let receipt_result = create_new_task.get_receipt().await;

                match receipt_result {
                    Ok(receipt) => {
                        let log_decoded_option = receipt
                            .inner
                            .logs()
                            .iter()
                            .find_map(|log| log.log_decode::<NewtonProverTaskManager::NewTaskCreated>().ok());

                        if let Some(new_task_created) = log_decoded_option {
                            let data = new_task_created.data();
                            // Return full Task from event with actual taskCreatedBlock
                            let onchain_task = data.task.clone();

                            Ok((receipt, onchain_task))
                        } else {
                            Err(ChainIoError::CreateNewTaskNoEventFound)
                        }
                    }

                    Err(e) => Err(ChainIoError::AlloyProviderError(e)),
                }
            }
            // Post-fill send failure: the nonce may be live on-chain. The caller
            // (tx_worker) only retries/logs — it does NOT resolve the slot — so we
            // must resolve it here, or the reserved nonce is abandoned and strands
            // every higher nonce (the blocked-from-below gap this PR exists to
            // kill). Drive the same-nonce cancel, seeded from the failed tx's fees,
            // then surface a transient error so the caller can re-attempt cleanly.
            Err(ChainIoError::BroadcastSendFailed {
                nonce,
                max_fee_per_gas,
                max_priority_fee_per_gas,
                reason,
            }) => {
                tracing::warn!(
                    task_id = %newton_core::hex!(task_id),
                    nonce,
                    %reason,
                    "send_task: send failed after fill, resolving stuck nonce off the hot path"
                );
                // Spawn the resolution so the serial tx_worker keeps draining
                // queued submissions instead of blocking on the unbounded cancel.
                self.spawn_resolve_stuck_nonce(nonce, max_fee_per_gas, max_priority_fee_per_gas);
                Err(ChainIoError::TransactionTimeout { timeout_secs: 0 })
            }
            Err(e) => {
                // Log the ACTUAL transaction error before simulation
                // This is critical for diagnosis - simulation may mask the real error
                tracing::error!(
                    task_id = %newton_core::hex!(task_id),
                    task_created_block = task_created_block,
                    error = %e,
                    "send_task: createNewTask transaction FAILED (actual error, not simulation)"
                );

                // Log simulation error details when the actual call fails
                // NOTE: Simulation uses hardcoded taskCreatedBlock=1, which will ALWAYS fail
                // with TaskCreatedBlockTooOld on real networks. The simulation is for identifying
                // OTHER errors, not block timing issues.
                if let Some(identified_error) = log_create_task_error(
                    self.task_manager_addr,
                    &self.signer_provider,
                    params.policy_client,
                    params.wasm_args.clone(),
                    &params.intent,
                    &params.intent_signature,
                    params.quorum_number,
                    params.quorum_threshold_percentage,
                    params.initialization_timestamp,
                )
                .await
                {
                    Err(ChainIoError::CreateNewTaskCallFail {
                        reason: identified_error,
                    })
                } else {
                    // `fill_and_send` already classified the fill-stage error into a
                    // typed `ChainIoError`; surface it directly.
                    Err(e)
                }
            }
        }
    }

    /// Raise challenge
    pub async fn raise_challenge(
        &self,
        task: Task,
        task_response: TaskResponse,
        task_response_metadata: ResponseCertificate,
        challenge: ChallengeData,
        pub_keys_of_non_signing_operators: Vec<G1Point>,
    ) -> Result<alloy::rpc::types::TransactionReceipt, ChainIoError> {
        let balance = self
            .signer_provider
            .get_balance(self.signer_provider.default_signer_address())
            .await;
        info!(
            "{} balance: {}",
            self.signer_provider.default_signer_address(),
            balance.unwrap_or_default()
        );
        let task_manager_contract = NewtonProverTaskManager::new(self.task_manager_addr, self.signer_provider.clone());
        let challenge_tx_call = task_manager_contract.raiseAndResolveChallenge(
            task,
            task_response.clone(),
            task_response_metadata,
            challenge,
            pub_keys_of_non_signing_operators,
        );

        match challenge_tx_call.send().await {
            Ok(challenge_tx) => {
                let receipt_result = challenge_tx.get_receipt().await;
                match receipt_result {
                    Ok(receipts) => {
                        info!(
                            "raiseAndResolveChallenge for task_id: {} tx_hash: {}",
                            hex!(task_response.taskId),
                            hex!(receipts.transaction_hash)
                        );
                        Ok(receipts)
                    }
                    Err(e) => Err(ChainIoError::AlloyProviderError(e)),
                }
            }

            Err(e) => Err(ChainIoError::ContractError(e)),
        }
    }

    /// Invalidate a directly-verified attestation whose hashes diverge from the canonical
    /// regular-path hashes (`createNewTask` + `respondToTask`).
    ///
    /// This never slashes. Hash divergence alone is not proof of operator misbehavior —
    /// the direct path may have stored a bad hash while operators signed the correct
    /// response. Slashing for incorrect responses is handled by `raiseAndResolveChallenge`,
    /// which re-executes the policy and produces a ZK proof.
    pub async fn challenge_directly_verified_mismatch(
        &self,
        task: Task,
        task_response: TaskResponse,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let task_manager_contract = NewtonProverTaskManager::new(self.task_manager_addr, self.signer_provider.clone());

        let call = task_manager_contract.challengeDirectlyVerifiedMismatch(task, task_response.clone());

        match call.send().await {
            Ok(pending_tx) => {
                let receipt = pending_tx
                    .get_receipt()
                    .await
                    .map_err(ChainIoError::AlloyProviderError)?;
                info!(
                    "challengeDirectlyVerifiedMismatch for task_id: {} tx_hash: {}",
                    hex!(task_response.taskId),
                    hex!(receipt.transaction_hash)
                );
                Ok(receipt)
            }
            Err(e) => Err(ChainIoError::ContractError(e)),
        }
    }

    /// Challenge a privacy task with missing or invalid TEE attestation.
    ///
    /// Calls ChallengeVerifier.challengeInvalidTeeAttestation directly (not through
    /// TaskManager). For missing attestation, pass empty proof data. For invalid
    /// attestation, pass the SP1 attestation circuit proof.
    pub async fn challenge_invalid_tee_attestation(
        &self,
        params: TeeAttestationChallengeParams,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let TeeAttestationChallengeParams {
            challenge_verifier_addr,
            task,
            task_response,
            task_response_metadata,
            attestation_proof_data,
            attestation_proof_bytes,
            pub_keys_of_non_signing_operators,
        } = params;
        let cv = newton_core::challenge_verifier::ChallengeVerifier::new(
            challenge_verifier_addr,
            self.signer_provider.clone(),
        );

        // ABI round-trip: alloy generates distinct struct types per contract binding
        let cv_task =
            CvTask::abi_decode(&Task::abi_encode(&task)).map_err(|e| ChainIoError::CreateNewTaskCallFail {
                reason: format!("ABI bridge Task failed: {e}"),
            })?;
        let cv_response = CvTaskResponse::abi_decode(&TaskResponse::abi_encode(&task_response)).map_err(|e| {
            ChainIoError::CreateNewTaskCallFail {
                reason: format!("ABI bridge TaskResponse failed: {e}"),
            }
        })?;
        let cv_metadata = CvResponseCertificate::abi_decode(&ResponseCertificate::abi_encode(&task_response_metadata))
            .map_err(|e| ChainIoError::CreateNewTaskCallFail {
                reason: format!("ABI bridge ResponseCertificate failed: {e}"),
            })?;
        let cv_pubkeys: Vec<CvG1Point> = pub_keys_of_non_signing_operators
            .iter()
            .map(|p| CvG1Point { X: p.X, Y: p.Y })
            .collect();

        let call = cv.challengeInvalidTeeAttestation(
            cv_task,
            cv_response,
            cv_metadata,
            attestation_proof_data,
            attestation_proof_bytes,
            cv_pubkeys,
        );

        match call.send().await {
            Ok(pending_tx) => {
                let receipt = pending_tx
                    .get_receipt()
                    .await
                    .map_err(ChainIoError::AlloyProviderError)?;
                info!(
                    "challengeInvalidTeeAttestation for task_id: {} tx_hash: {}",
                    hex!(task_response.taskId),
                    hex!(receipt.transaction_hash)
                );
                Ok(receipt)
            }
            Err(e) => Err(ChainIoError::ContractError(e)),
        }
    }

    /// Send aggregated response
    ///
    /// # Arguments
    ///
    /// * `task` - Task
    /// * `task_response` - Task response
    /// * `signature_data` - Signature data (non-signer stakes and aggregated signature)
    /// * `attestation_data` - Modal keccak256(pcr0) from operator commit responses
    /// * `chain_id` - Target chain ID
    pub async fn send_aggregated_response(
        &self,
        task: Task,
        task_response: TaskResponse,
        signature_data: Bytes,
        attestation_data: Bytes,
        chain_id: u64,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let rpc_url = self.rpc_url.clone();
        let signer_provider = self.signer_provider.clone();
        let task_manager_addr = self.task_manager_addr;
        let receipt_timeout = self.batch_receipt_timeout();
        // Reserve one nonce for the whole retry sequence: every fee-escalation
        // retry replaces the SAME slot, so they all pin this nonce (keeping the
        // implicit filler inert and the sequence coordinated with the batch path).
        let reserved_nonce = self.reserve_nonce().await?;
        // Capture the fees of the LAST attempt so a stuck-nonce cancel can be
        // seeded to out-price a propagated original (see the resolve call below).
        // `send_with_retries` doesn't surface them, so the closure records them.
        let last_fees: Arc<std::sync::Mutex<Option<(u128, u128)>>> = Arc::new(std::sync::Mutex::new(None));
        let last_fees_inner = last_fees.clone();

        let result = crate::tx::send_with_retries(rpc_url.clone(), chain_id, move |fee_override| {
            let task = task.clone();
            let task_for_error_logging = task.clone();
            let task_response = task_response.clone();
            let signature_data = signature_data.clone();
            let attestation_data = attestation_data.clone();
            let signer = signer_provider.clone();
            let last_fees = last_fees_inner.clone();

            async move {
                let task_manager_contract = NewtonProverTaskManager::new(task_manager_addr, signer.clone());

                // DIAGNOSTIC: Log task and response data before submission
                debug!(
                    task_id = %newton_core::hex!(task.taskId),
                    task_created_block = task.taskCreatedBlock,
                    quorum_threshold = task.quorumThresholdPercentage,
                    policy_client = %task.policyClient,
                    task_response_task_id = %newton_core::hex!(task_response.taskId),
                    task_response_policy_id = %newton_core::hex!(task_response.policyId),
                    evaluation_result_len = task_response.evaluationResult.len(),
                    policy_task_data_len = task_response.policyTaskData.policyData.len(),
                    "[DEBUG] respondToTask call parameters"
                );

                // Create the transaction call
                let mut call_builder =
                    task_manager_contract.respondToTask(task, task_response, signature_data, attestation_data);

                // Apply EIP-1559 fee override if provided, recording it so a
                // stuck-nonce cancel can be seeded from the last-used price.
                if let Some(fees) = fee_override {
                    call_builder = call_builder
                        .max_fee_per_gas(fees.max_fee_per_gas)
                        .max_priority_fee_per_gas(fees.max_priority_fee_per_gas);
                    *last_fees.lock().unwrap() = Some((fees.max_fee_per_gas, fees.max_priority_fee_per_gas));
                }
                // Pin the reserved nonce so the implicit filler stays inert and
                // every retry replaces the same slot.
                if let Some(n) = reserved_nonce {
                    call_builder = call_builder.nonce(n);
                }

                let tx_from = signer.default_signer_address();
                let tx_to = task_manager_addr;
                let tx_value = format!("0x{:x}", U256::ZERO);
                let calldata_bytes = call_builder.calldata().to_vec();
                let tx_data = format!("0x{}", hex::encode(&calldata_bytes));

                // Try to send the transaction and get receipt
                match call_builder.send().await {
                    Ok(pending_tx) => {
                        // Wrap get_receipt() in a timeout to prevent indefinite hang
                        // if the tx is dropped from the mempool (same pattern as batch methods).
                        match timeout(receipt_timeout, pending_tx.get_receipt()).await {
                            Ok(Ok(receipt)) => Ok(receipt),
                            Ok(Err(e)) => Err(ChainIoError::AlloyProviderError(e)),
                            Err(_) => Err(ChainIoError::TransactionTimeout {
                                timeout_secs: receipt_timeout.as_secs(),
                            }),
                        }
                    }
                    Err(send_error) => {
                        // Log detailed error information for debugging
                        log_respond_to_task_error(
                            &send_error,
                            &task_for_error_logging,
                            task_manager_addr,
                            tx_from,
                            &calldata_bytes,
                            chain_id,
                        );

                        Err(ChainIoError::ContractErrorWithTx {
                            from: tx_from,
                            to: tx_to,
                            value: tx_value,
                            data: tx_data,
                            source: send_error,
                        })
                    }
                }
            }
        })
        .await;

        // `send_with_retries` reserved one nonce for the whole sequence (replacement
        // semantics). On a final failure whose on-chain outcome is UNCERTAIN (an
        // attempt may have propagated) the reserved nonce must be resolved on-chain
        // rather than left stranded (the next aggregated response reserves above it
        // → blocked-from-below). A mined revert is NOT uncertain — the nonce is
        // already consumed — so we skip the cancel there (see `is_uncertain_outcome`).
        // Spawned off the hot path so the serial caller keeps draining. Seed the
        // cancel from the last attempt's fees; 0 (attempt 1 used node-assigned fees
        // we can't read) falls back to a fresh market quote.
        if let (Err(e), Some(n)) = (&result, reserved_nonce) {
            if Self::is_uncertain_outcome(e) {
                let (seed_max, seed_prio) = last_fees.lock().unwrap().unwrap_or((0, 0));
                self.spawn_resolve_stuck_nonce(n, seed_max, seed_prio);
            }
        }
        result
    }

    /// Submit a unified-tree state root commit to the per-chain `StateCommitRegistry`.
    ///
    /// `commit` is the BLS-signed `StateCommit` payload (version, sequenceNo,
    /// prevStateRoot, newStateRoot, timestamp, daCertHash, pcr0Commitment).
    /// `bls_certificate` is the ABI-encoded `BN254Certificate` covering
    /// `keccak256(abi.encode(commit))`. Both are produced by the aggregator at
    /// the 120s commit cadence.
    ///
    /// Uses `crate::tx::send_with_retries` for transient retry, but every typed
    /// revert classified by `log_commit_state_root_error` is a poison error —
    /// the aggregator must rebuild the commit against the registry's current
    /// `(sequenceNo, stateRoot)` before retrying.
    pub async fn commit_state_root(
        &self,
        commit: StateCommit,
        bls_certificate: Bytes,
    ) -> Result<TransactionReceipt, ChainIoError> {
        if self.state_commit_registry_addr == Address::ZERO {
            return Err(ChainIoError::CommitStateRootCallFail {
                reason: "state_commit_registry address not configured (still Address::ZERO)".to_string(),
            });
        }

        let rpc_url = self.rpc_url.clone();
        let signer_provider = self.signer_provider.clone();
        let registry_addr = self.state_commit_registry_addr;
        let chain_id = self.chain_id;
        // Reserve one nonce for the whole retry sequence (replacement semantics);
        // pinned on every attempt so this path shares the batch path's allocator.
        let reserved_nonce = self.reserve_nonce().await?;
        // Record the last attempt's fees to seed a stuck-nonce cancel (see below).
        let last_fees: Arc<std::sync::Mutex<Option<(u128, u128)>>> = Arc::new(std::sync::Mutex::new(None));
        let last_fees_inner = last_fees.clone();

        let result = crate::tx::send_with_retries(rpc_url, chain_id, move |fee_override| {
            let commit = commit.clone();
            let bls_certificate = bls_certificate.clone();
            let signer = signer_provider.clone();
            let last_fees = last_fees_inner.clone();

            async move {
                let registry = StateCommitRegistryInstance::new(registry_addr, signer.clone());

                debug!(
                    chain_id,
                    sequence_no = commit.sequenceNo,
                    timestamp = commit.timestamp,
                    prev_state_root = %commit.prevStateRoot,
                    new_state_root = %commit.newStateRoot,
                    da_cert_hash = %commit.daCertHash,
                    pcr0_commitment = %commit.pcr0Commitment,
                    cert_len = bls_certificate.len(),
                    "commitStateRoot call parameters"
                );

                let mut call_builder = registry.commitStateRoot(commit.clone(), bls_certificate);

                if let Some(fees) = fee_override {
                    call_builder = call_builder
                        .max_fee_per_gas(fees.max_fee_per_gas)
                        .max_priority_fee_per_gas(fees.max_priority_fee_per_gas);
                    *last_fees.lock().unwrap() = Some((fees.max_fee_per_gas, fees.max_priority_fee_per_gas));
                }
                if let Some(n) = reserved_nonce {
                    call_builder = call_builder.nonce(n);
                }

                let tx_from = signer.default_signer_address();
                let tx_to = registry_addr;
                let tx_value = format!("0x{:x}", U256::ZERO);
                let calldata_bytes = call_builder.calldata().to_vec();
                let tx_data = format!("0x{}", hex::encode(&calldata_bytes));

                match call_builder.send().await {
                    Ok(pending_tx) => match timeout(STATE_COMMIT_RECEIPT_TIMEOUT, pending_tx.get_receipt()).await {
                        Ok(Ok(receipt)) => {
                            let tx_hash = receipt.transaction_hash;
                            if !receipt.status() {
                                warn!(
                                    chain_id,
                                    sequence_no = commit.sequenceNo,
                                    tx_hash = %tx_hash,
                                    "commitStateRoot reverted on-chain (receipt status=false) — \
                                     proposal is poison; rebuild against current registry view before retrying"
                                );
                                return Err(ChainIoError::TransactionReverted(tx_hash));
                            }
                            Ok(receipt)
                        }
                        Ok(Err(e)) => Err(ChainIoError::AlloyProviderError(e)),
                        Err(_) => Err(ChainIoError::TransactionTimeout {
                            timeout_secs: STATE_COMMIT_RECEIPT_TIMEOUT.as_secs(),
                        }),
                    },
                    Err(send_error) => {
                        log_commit_state_root_error(
                            &send_error,
                            &commit,
                            registry_addr,
                            tx_from,
                            &calldata_bytes,
                            chain_id,
                        );

                        Err(ChainIoError::ContractErrorWithTx {
                            from: tx_from,
                            to: tx_to,
                            value: tx_value,
                            data: tx_data,
                            source: send_error,
                        })
                    }
                }
            }
        })
        .await;

        // Resolve the reserved nonce on-chain only when the outcome is UNCERTAIN
        // (timeout / post-send error — an attempt may have propagated). A mined
        // revert (`TransactionReverted`, status=false) already consumed the nonce,
        // so resolving there is a wasted cancel that also muddies the poison signal
        // the aggregator keys on to rebuild — skip it (see `is_uncertain_outcome`).
        // Spawned off the hot path so the 120s commit tick isn't blocked. No-op
        // without an allocator. Dev-stub-gated today (NEWT-1116) but correct the
        // moment the production orchestrator is wired.
        if let (Err(e), Some(n)) = (&result, reserved_nonce) {
            if Self::is_uncertain_outcome(e) {
                let (seed_max, seed_prio) = last_fees.lock().unwrap().unwrap_or((0, 0));
                self.spawn_resolve_stuck_nonce(n, seed_max, seed_prio);
            }
        }
        result
    }

    /// Chain-adaptive timeout for waiting on a batch transaction receipt.
    /// L1 chains (12s block time) need longer timeouts under congestion.
    /// L2 chains (2s block time) can use shorter timeouts.
    fn batch_receipt_timeout(&self) -> Duration {
        match self.chain_id {
            // Ethereum mainnet, Sepolia: 12s/block, 60s ≈ 5 blocks
            1 | 11155111 => Duration::from_secs(60),
            // Local anvil: use shorter timeout for fast test feedback
            31337 | 31338 => Duration::from_secs(30),
            // L2s (Base, Arbitrum, Optimism, Polygon, etc.): 2s/block, 30s ≈ 15 blocks
            _ => Duration::from_secs(30),
        }
    }

    /// Batch-create multiple tasks in a single transaction via BatchTaskManager contract.
    ///
    /// On success (all items pass): returns Ok with receipt. No event parsing needed —
    /// the caller knows all items in the batch succeeded.
    ///
    /// On partial failure: the contract reverts with `BatchPartialFailure` containing
    /// indices and reasons for each failed item. This is returned as
    /// `Err(ChainIoError::BatchPartialFailure)` with the parsed failure details.
    pub async fn batch_create_tasks(
        &self,
        batch_contract_addr: Address,
        tasks: Vec<Task>,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let contract = BatchTaskManagerInstance::new(batch_contract_addr, self.signer_provider.clone());

        let batch_size = tasks.len();
        info!(
            batch_size = batch_size,
            batch_contract = %batch_contract_addr,
            "batch_create_tasks: submitting batch"
        );

        // Convert tasks from newton_prover_task_manager types to batch_task_manager types
        // Both are ABI-identical; round-trip through encoding bridges the type gap.
        let batch_tasks: Vec<BatchTask> = tasks
            .iter()
            .filter_map(|t| {
                let encoded = Task::abi_encode(t);
                match BatchTask::abi_decode(&encoded) {
                    Ok(bt) => Some(bt),
                    Err(e) => {
                        warn!(error = %e, "batch_create_tasks: failed to ABI-decode task, skipping");
                        None
                    }
                }
            })
            .collect();

        let skipped = batch_size - batch_tasks.len();
        if skipped > 0 {
            warn!(
                batch_size = batch_size,
                skipped = skipped,
                "batch_create_tasks: {} item(s) skipped due to ABI-decode failure",
                skipped
            );
        }
        if batch_tasks.is_empty() {
            error!(
                batch_size = batch_size,
                "batch_create_tasks: all items failed ABI-decode, aborting empty batch"
            );
            return Err(ChainIoError::CreateNewTaskCallFail {
                reason: "all tasks failed ABI-decode; batch is empty".to_string(),
            });
        }

        let send_result = contract.batchCreateTasks(batch_tasks).send().await;

        match send_result {
            Ok(pending) => {
                let receipt = match timeout(self.batch_receipt_timeout(), pending.get_receipt()).await {
                    Ok(Ok(r)) => r,
                    Ok(Err(e)) => return Err(ChainIoError::AlloyProviderError(e)),
                    Err(_) => {
                        warn!(
                            batch_size,
                            timeout_secs = self.batch_receipt_timeout().as_secs(),
                            "batch_create_tasks: receipt timed out"
                        );
                        return Err(ChainIoError::TransactionTimeout {
                            timeout_secs: self.batch_receipt_timeout().as_secs(),
                        });
                    }
                };
                let tx_hash = receipt.transaction_hash;
                if !receipt.status() {
                    return Err(ChainIoError::TransactionReverted(tx_hash));
                }
                info!(batch_size, %tx_hash, "batch_create_tasks: all items succeeded");
                Ok(receipt)
            }
            Err(e) => {
                if let Some(failures) = Self::parse_batch_partial_failure(&e) {
                    let fail_count = failures.len();
                    info!(
                        batch_size = batch_size,
                        fail_count = fail_count,
                        "batch_create_tasks: partial failure, {} items failed",
                        fail_count
                    );
                    Err(ChainIoError::BatchPartialFailure { failures })
                } else if let Some(revert_data) = e.as_revert_data() {
                    Err(errors::classify_top_level_revert(&revert_data))
                } else {
                    Err(ChainIoError::ContractError(e))
                }
            }
        }
    }

    /// Batch-respond to multiple tasks in a single transaction via BatchTaskManager contract.
    ///
    /// Same error semantics as `batch_create_tasks`: Ok on full success,
    /// `Err(BatchPartialFailure)` on partial failure with per-item details.
    pub async fn batch_respond_to_tasks(
        &self,
        batch_contract_addr: Address,
        tasks: Vec<Task>,
        responses: Vec<TaskResponse>,
        signature_data_array: Vec<Bytes>,
        attestation_data: Vec<Bytes>,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let contract = BatchTaskManagerInstance::new(batch_contract_addr, self.signer_provider.clone());

        let batch_size = tasks.len();
        info!(
            batch_size = batch_size,
            batch_contract = %batch_contract_addr,
            "batch_respond_to_tasks: submitting batch"
        );

        let mut batch_tasks: Vec<BatchTask> = Vec::with_capacity(batch_size);
        let mut batch_responses: Vec<BatchTaskResponse> = Vec::with_capacity(batch_size);
        let mut batch_sigs: Vec<Bytes> = Vec::with_capacity(batch_size);
        let mut batch_attestations: Vec<Bytes> = Vec::with_capacity(batch_size);

        for (i, (((t, r), sig), att)) in tasks
            .into_iter()
            .zip(responses)
            .zip(signature_data_array)
            .zip(attestation_data)
            .enumerate()
        {
            let task_encoded = Task::abi_encode(&t);
            let bt = match BatchTask::abi_decode(&task_encoded) {
                Ok(bt) => bt,
                Err(e) => {
                    warn!(
                        error = %e,
                        index = i,
                        "batch_respond_to_tasks: failed to ABI-decode task, skipping triple"
                    );
                    continue;
                }
            };

            let resp_encoded = TaskResponse::abi_encode(&r);
            let br = match BatchTaskResponse::abi_decode(&resp_encoded) {
                Ok(br) => br,
                Err(e) => {
                    warn!(
                        error = %e,
                        index = i,
                        "batch_respond_to_tasks: failed to ABI-decode response, skipping triple"
                    );
                    continue;
                }
            };

            batch_tasks.push(bt);
            batch_responses.push(br);
            batch_sigs.push(sig);
            batch_attestations.push(att);
        }

        let skipped = batch_size - batch_tasks.len();
        if skipped > 0 {
            warn!(
                batch_size = batch_size,
                skipped = skipped,
                "batch_respond_to_tasks: {} triple(s) skipped due to ABI-decode failure",
                skipped
            );
        }
        if batch_tasks.is_empty() {
            error!(
                batch_size = batch_size,
                "batch_respond_to_tasks: all items failed ABI-decode, aborting empty batch"
            );
            return Err(ChainIoError::CreateNewTaskCallFail {
                reason: "all task/response triples failed ABI-decode; batch is empty".to_string(),
            });
        }

        // Route through `fill_and_send` so a fill-stage failure releases the
        // reserved nonce while a post-fill send failure surfaces
        // `BroadcastSendFailed` (the tx may have propagated — nonce NOT released).
        let call = contract.batchRespondToTasks(batch_tasks, batch_responses, batch_sigs, batch_attestations);
        let send_result = self.fill_and_send(call.into_transaction_request(), batch_size).await;

        match send_result {
            Ok((broadcast, pending)) => {
                let receipt = match timeout(self.batch_receipt_timeout(), pending.get_receipt()).await {
                    Ok(Ok(r)) => r,
                    Ok(Err(e)) => return Err(ChainIoError::AlloyProviderError(e)),
                    Err(_) => {
                        // Receipt timed out: the tx was accepted but hasn't mined.
                        // Its nonce is consumed-but-uncertain. The caller's retry
                        // loop would otherwise reserve a FRESH nonce and re-broadcast,
                        // stranding this one below it (blocked-from-below). Resolve
                        // the slot on-chain via a bounded same-nonce cancel before
                        // returning, seeded from the broadcast's own fees.
                        warn!(
                            batch_size,
                            nonce = broadcast.nonce,
                            timeout_secs = self.batch_receipt_timeout().as_secs(),
                            "batch_respond_to_tasks: receipt timed out, resolving stuck nonce off the hot path"
                        );
                        self.spawn_resolve_stuck_nonce(
                            broadcast.nonce,
                            broadcast.max_fee_per_gas,
                            broadcast.max_priority_fee_per_gas,
                        );
                        return Err(ChainIoError::TransactionTimeout {
                            timeout_secs: self.batch_receipt_timeout().as_secs(),
                        });
                    }
                };
                let tx_hash = receipt.transaction_hash;
                if !receipt.status() {
                    return Err(ChainIoError::TransactionReverted(tx_hash));
                }
                info!(batch_size, %tx_hash, "batch_respond_to_tasks: all items succeeded");
                Ok(receipt)
            }
            // Post-fill send failure: the nonce may be live. Resolve it on-chain
            // (the caller has no receipt tracker for this path) before returning a
            // transient error so its retry can safely re-reserve.
            Err(ChainIoError::BroadcastSendFailed {
                nonce,
                max_fee_per_gas,
                max_priority_fee_per_gas,
                reason,
            }) => {
                warn!(nonce, %reason, "batch_respond_to_tasks: send failed after fill, resolving stuck nonce off the hot path");
                self.spawn_resolve_stuck_nonce(nonce, max_fee_per_gas, max_priority_fee_per_gas);
                Err(ChainIoError::TransactionTimeout { timeout_secs: 0 })
            }
            // Fill-stage failures are already classified by `fill_and_send` (and
            // the reserved nonce released there).
            Err(e) => Err(e),
        }
    }

    /// Atomically create and respond to multiple tasks in a single transaction.
    ///
    /// Each item calls `createNewTask` then `respondToTask` within a single EVM execution
    /// context. If either call reverts for an item, the entire item reverts (no zombie tasks).
    /// Same failure semantics as `batch_create_tasks`: `Err(BatchPartialFailure)` on partial failure.
    pub async fn batch_create_and_respond_to_tasks(
        &self,
        batch_contract_addr: Address,
        tasks: Vec<Task>,
        responses: Vec<TaskResponse>,
        signature_data_array: Vec<Bytes>,
        attestation_data: Vec<Bytes>,
    ) -> Result<TransactionReceipt, ChainIoError> {
        let contract = BatchTaskManagerInstance::new(batch_contract_addr, self.signer_provider.clone());

        let batch_size = tasks.len();
        info!(
            batch_size = batch_size,
            batch_contract = %batch_contract_addr,
            "batch_create_and_respond: submitting batch"
        );

        let mut batch_tasks: Vec<BatchTask> = Vec::with_capacity(batch_size);
        let mut batch_responses: Vec<BatchTaskResponse> = Vec::with_capacity(batch_size);
        let mut batch_sigs: Vec<Bytes> = Vec::with_capacity(batch_size);
        let mut batch_attestations: Vec<Bytes> = Vec::with_capacity(batch_size);

        for (i, (((t, r), sig), att)) in tasks
            .into_iter()
            .zip(responses)
            .zip(signature_data_array)
            .zip(attestation_data)
            .enumerate()
        {
            let task_encoded = Task::abi_encode(&t);
            let bt = match BatchTask::abi_decode(&task_encoded) {
                Ok(bt) => bt,
                Err(e) => {
                    warn!(
                        error = %e,
                        index = i,
                        "batch_create_and_respond: failed to ABI-decode task, skipping triple"
                    );
                    continue;
                }
            };

            let resp_encoded = TaskResponse::abi_encode(&r);
            let br = match BatchTaskResponse::abi_decode(&resp_encoded) {
                Ok(br) => br,
                Err(e) => {
                    warn!(
                        error = %e,
                        index = i,
                        "batch_create_and_respond: failed to ABI-decode response, skipping triple"
                    );
                    continue;
                }
            };

            batch_tasks.push(bt);
            batch_responses.push(br);
            batch_sigs.push(sig);
            batch_attestations.push(att);
        }

        let skipped = batch_size - batch_tasks.len();
        if skipped > 0 {
            warn!(
                batch_size = batch_size,
                skipped = skipped,
                "batch_create_and_respond: {} triple(s) skipped due to ABI-decode failure",
                skipped
            );
        }
        if batch_tasks.is_empty() {
            error!(
                batch_size = batch_size,
                "batch_create_and_respond: all items failed ABI-decode, aborting empty batch"
            );
            return Err(ChainIoError::CreateNewTaskCallFail {
                reason: "all task/response triples failed ABI-decode; batch is empty".to_string(),
            });
        }

        let send_result = contract
            .batchCreateAndRespondToTasks(batch_tasks, batch_responses, batch_sigs, batch_attestations)
            .send()
            .await;

        match send_result {
            Ok(pending) => {
                let receipt = timeout(self.batch_receipt_timeout(), pending.get_receipt())
                    .await
                    .map_err(|_| {
                        warn!(
                            batch_size,
                            timeout_secs = self.batch_receipt_timeout().as_secs(),
                            "batch_create_and_respond: receipt timed out"
                        );
                        ChainIoError::TransactionTimeout {
                            timeout_secs: self.batch_receipt_timeout().as_secs(),
                        }
                    })?
                    .map_err(ChainIoError::AlloyProviderError)?;
                let tx_hash = receipt.transaction_hash;
                if !receipt.status() {
                    return Err(ChainIoError::TransactionReverted(tx_hash));
                }
                info!(batch_size, %tx_hash, "batch_create_and_respond: all items succeeded");
                Ok(receipt)
            }
            Err(e) => {
                if let Some(failures) = Self::parse_batch_partial_failure(&e) {
                    let fail_count = failures.len();
                    info!(
                        batch_size = batch_size,
                        fail_count = fail_count,
                        "batch_create_and_respond: partial failure"
                    );
                    Err(ChainIoError::BatchPartialFailure { failures })
                } else if let Some(revert_data) = e.as_revert_data() {
                    let classified = errors::classify_top_level_revert(&revert_data);
                    warn!(
                        batch_size,
                        error = %classified,
                        "batch_create_and_respond: top-level revert"
                    );
                    Err(classified)
                } else {
                    Err(ChainIoError::ContractError(e))
                }
            }
        }
    }

    /// Simulate a batch create+respond via `eth_call` (read-only, no transaction).
    ///
    /// Returns `Ok(())` if all items would succeed on-chain.
    /// Returns `Err(BatchPartialFailure)` if some items would fail (e.g., TaskAlreadyExists).
    /// Returns other `Err` variants for RPC or encoding failures.
    ///
    /// Used after receipt timeouts to determine which items already landed on-chain,
    /// and for poison detection without sending real transactions.
    pub async fn simulate_batch_create_and_respond(
        &self,
        batch_contract_addr: Address,
        tasks: Vec<Task>,
        responses: Vec<TaskResponse>,
        signature_data_array: Vec<Bytes>,
        attestation_data: Vec<Bytes>,
    ) -> Result<(), ChainIoError> {
        let contract = BatchTaskManagerInstance::new(batch_contract_addr, self.signer_provider.clone());

        let batch_size = tasks.len();

        let mut batch_tasks: Vec<BatchTask> = Vec::with_capacity(batch_size);
        let mut batch_responses: Vec<BatchTaskResponse> = Vec::with_capacity(batch_size);
        let mut batch_sigs: Vec<Bytes> = Vec::with_capacity(batch_size);
        let mut batch_attestations: Vec<Bytes> = Vec::with_capacity(batch_size);

        for (i, (((t, r), sig), att)) in tasks
            .into_iter()
            .zip(responses)
            .zip(signature_data_array)
            .zip(attestation_data)
            .enumerate()
        {
            let task_encoded = Task::abi_encode(&t);
            let bt = match BatchTask::abi_decode(&task_encoded) {
                Ok(bt) => bt,
                Err(e) => {
                    warn!(error = %e, index = i, "simulate_batch: ABI-decode task failed, skipping");
                    continue;
                }
            };
            let resp_encoded = TaskResponse::abi_encode(&r);
            let br = match BatchTaskResponse::abi_decode(&resp_encoded) {
                Ok(br) => br,
                Err(e) => {
                    warn!(error = %e, index = i, "simulate_batch: ABI-decode response failed, skipping");
                    continue;
                }
            };
            batch_tasks.push(bt);
            batch_responses.push(br);
            batch_sigs.push(sig);
            batch_attestations.push(att);
        }

        if batch_tasks.is_empty() {
            return Err(ChainIoError::CreateNewTaskCallFail {
                reason: "simulate_batch: all items failed ABI-decode".to_string(),
            });
        }

        // Use .call() (eth_call) instead of .send() — read-only simulation
        contract
            .batchCreateAndRespondToTasks(batch_tasks, batch_responses, batch_sigs, batch_attestations)
            .call()
            .await
            .map(|_| ())
            .map_err(|e| {
                if let Some(failures) = Self::parse_batch_partial_failure(&e) {
                    return ChainIoError::BatchPartialFailure { failures };
                }
                if let Some(revert_data) = e.as_revert_data() {
                    return errors::classify_top_level_revert(&revert_data);
                }
                ChainIoError::ContractError(e)
            })
    }

    /// Parse `BatchPartialFailure` error from a contract revert.
    ///
    /// Uses `as_revert_data()` to extract the raw revert bytes, then ABI-decodes the
    /// `BatchPartialFailure(FailedItem[])` struct where each `FailedItem` is
    /// `(uint256 index, bytes32 taskId, bytes reason)`.
    fn parse_batch_partial_failure(error: &alloy::contract::Error) -> Option<Vec<BatchFailedItem>> {
        use alloy::sol_types::SolType;

        let revert_data = error.as_revert_data()?;
        if revert_data.len() < 4 {
            return None;
        }
        let selector: [u8; 4] = revert_data[..4].try_into().ok()?;
        if selector != errors::selectors::BATCH_PARTIAL_FAILURE {
            return None;
        }

        // ABI layout: BatchPartialFailure(FailedItem[] failures)
        // FailedItem = (uint256 index, bytes32 taskId, bytes reason)
        type FailedItemArray = alloy::sol_types::sol_data::Array<(
            alloy::sol_types::sol_data::Uint<256>,
            alloy::sol_types::sol_data::FixedBytes<32>,
            alloy::sol_types::sol_data::Bytes,
        )>;

        let decoded = match FailedItemArray::abi_decode(&revert_data[4..]) {
            Ok(d) => d,
            Err(e) => {
                warn!(
                    error = %e,
                    revert_len = revert_data.len(),
                    "parse_batch_partial_failure: ABI decode failed"
                );
                return None;
            }
        };

        let items = decoded
            .into_iter()
            .map(|(index, task_id, reason)| BatchFailedItem {
                index: index.to::<usize>(),
                task_id: B256::from(task_id),
                reason,
            })
            .collect();

        Some(items)
    }
}

/// Drives the shared [`cancel::cancel_until_resolved`] loop from an `AvsWriter`
/// (the direct-send paths). Delegates to the writer's own broadcast/poll methods
/// and its attached allocator.
#[async_trait::async_trait]
impl NonceCanceller for AvsWriter {
    async fn broadcast_cancel(
        &self,
        nonce: u64,
        prev_max_fee: u128,
        prev_priority_fee: u128,
        bump_percent: u32,
    ) -> Result<cancel::CancelBroadcast, ChainIoError> {
        let b = self
            .cancel_nonce(nonce, prev_max_fee, prev_priority_fee, bump_percent)
            .await?;
        Ok(cancel::CancelBroadcast {
            tx_hash: b.tx_hash,
            max_fee_per_gas: b.max_fee_per_gas,
            max_priority_fee_per_gas: b.max_priority_fee_per_gas,
        })
    }

    async fn await_receipt(&self, tx_hash: B256, timeout: std::time::Duration) -> Result<(), ChainIoError> {
        self.get_receipt_by_hash(tx_hash, Some(timeout)).await.map(|_| ())
    }

    async fn latest_mined_nonce(&self) -> Result<u64, ChainIoError> {
        self.signer_transaction_count(false).await
    }

    async fn resync_allocator(&self) {
        if let Some(alloc) = self.nonce_allocator.as_ref() {
            let _ = alloc.resync().await;
        }
    }
}

/// A single failed item from a `BatchPartialFailure` revert.
#[derive(Debug, Clone)]
pub struct BatchFailedItem {
    /// Index of the failed item in the original batch array
    pub index: usize,
    /// Task ID of the failed item
    pub task_id: B256,
    /// ABI-encoded revert reason from the underlying call
    pub reason: Bytes,
}

/// Log detailed error information for createNewTask call failures.
/// This function performs a simulation call and logs detailed error information.
/// Returns the identified error message if one was identified, None otherwise.
#[allow(clippy::too_many_arguments)]
pub async fn log_create_task_error(
    task_manager_addr: Address,
    signer: &SdkSigner,
    policy_client: Address,
    wasm_args: Bytes,
    intent: &NewtonMessage::Intent,
    intent_signature: &Option<Bytes>,
    quorum_number: Vec<QuorumNum>,
    quorum_threshold_percentage: QuorumThresholdPercentage,
    initialization_timestamp: u64,
) -> Option<String> {
    info!("Simulating createNewTask call to identify exact revert location");

    let task_manager_contract = NewtonProverTaskManager::new(task_manager_addr, signer.clone());
    // Use a placeholder block for simulation - actual block validation is done in contract
    // NOTE: simulation uses hardcoded taskCreatedBlock=1, which will trigger InvalidTaskCreatedBlock
    // if the contract requires taskCreatedBlock > 0 and < block.number within buffer window.
    // This simulation is for identifying OTHER errors, not block validation issues.
    info!(
        simulation_task_created_block = 1,
        "log_create_task_error: simulation using hardcoded taskCreatedBlock=1"
    );
    let task = Task {
        taskId: task_id(None),
        policyClient: policy_client,
        taskCreatedBlock: 1,
        quorumThresholdPercentage: quorum_threshold_percentage.into(),
        intent: intent.clone(),
        intentSignature: intent_signature.clone().unwrap_or_default(),
        wasmArgs: wasm_args.clone(),
        quorumNumbers: quorum_number.clone().into(),
        initializationTimestamp: U256::from(initialization_timestamp),
    };

    let create_new_task_call = task_manager_contract.createNewTask(task.clone());

    let simulation_result = create_new_task_call.call().await;

    if let Err(error) = simulation_result {
        error!("createNewTask simulation error: {:?}", error);

        let error_str = format!("{:?}", error);
        if !error_str.contains("execution reverted") {
            return None;
        }

        // Check for empty revert data (0x)
        if error_str.contains("data: Some(RawValue(\"0x\"))") {
            use newton_core::newton_prover_task_manager::NewtonProverTaskManager::createNewTaskCall;
            let call_obj = createNewTaskCall { task };
            let call_data = call_obj.abi_encode();
            let signer_address = signer.default_signer_address();

            error!(
                task_manager = %task_manager_addr,
                signer = %signer_address,
                calldata_hex = %hex::encode(&call_data),
                "Empty revert (0x) in createNewTask"
            );
            return Some("Empty revert (0x) in createNewTask".to_string());
        }

        // Try to extract error selector from revert data
        if let Some(start_idx) = error_str.find("RawValue(\"0x") {
            if let Some(end_idx) = error_str[start_idx..].find("\")") {
                let hex_data = &error_str[start_idx + 11..start_idx + end_idx];
                if hex_data.len() >= 8 && hex_data != "0x" {
                    if let Ok(bytes) = Bytes::from_str(hex_data) {
                        if bytes.len() >= 4 {
                            let selector: [u8; 4] = bytes[..4].try_into().unwrap_or([0; 4]);
                            let selector_hex = hex::encode(selector);

                            if let Some(decoded_msg) = decode_error_selector(selector) {
                                error!(
                                    error_selector = %format!("0x{}", selector_hex),
                                    error_message = %decoded_msg,
                                    "createNewTask error identified"
                                );
                                return Some(decoded_msg);
                            }

                            // Unknown selector
                            error!(
                                error_selector = %format!("0x{}", selector_hex),
                                "Unknown error selector in createNewTask"
                            );
                            return Some(format!("Unknown error selector: 0x{}", selector_hex));
                        }
                    }
                }
            }
        }
    }
    None
}

/// Decode an error selector to a human-readable message.
/// Uses generated bindings for Newton errors and the external registry for library errors.
fn decode_error_selector(selector: [u8; 4]) -> Option<String> {
    // Task lifecycle errors
    if selector == TaskMismatch::SELECTOR {
        return Some(format!(
            "{} - task hash differs from stored (check taskCreatedBlock)",
            TaskMismatch::SIGNATURE
        ));
    }
    if selector == TaskAlreadyResponded::SELECTOR {
        return Some(format!(
            "{} - task already has a response (existing hash in revert data)",
            TaskAlreadyResponded::SIGNATURE
        ));
    }
    if selector == TaskAlreadyExists::SELECTOR {
        return Some(format!(
            "{} - task with this ID already exists (existing hash in revert data)",
            TaskAlreadyExists::SIGNATURE
        ));
    }
    if selector == InvalidReferenceBlocknumber::SELECTOR {
        return Some(format!(
            "{} - taskCreatedBlock must be < current block",
            InvalidReferenceBlocknumber::SIGNATURE
        ));
    }

    // BLS/Crypto errors
    if selector == InvalidBLSPairingKey::SELECTOR {
        return Some(format!(
            "{} - BLS pairing verification failed",
            InvalidBLSPairingKey::SIGNATURE
        ));
    }
    if selector == InvalidBLSSignature::SELECTOR {
        return Some(format!(
            "{} - BLS signature verification failed",
            InvalidBLSSignature::SIGNATURE
        ));
    }
    if selector == InvalidQuorumApkHash::SELECTOR {
        return Some(format!(
            "{} - quorum APK hash mismatch",
            InvalidQuorumApkHash::SIGNATURE
        ));
    }
    if selector == NonSignerPubkeysNotSorted::SELECTOR {
        return Some(format!(
            "{} - non-signer public keys must be sorted",
            NonSignerPubkeysNotSorted::SIGNATURE
        ));
    }
    if selector == ECAddFailed::SELECTOR {
        return Some(format!(
            "{} - elliptic curve point addition failed",
            ECAddFailed::SIGNATURE
        ));
    }
    if selector == ECMulFailed::SELECTOR {
        return Some(format!(
            "{} - elliptic curve scalar multiplication failed",
            ECMulFailed::SIGNATURE
        ));
    }
    if selector == ExpModFailed::SELECTOR {
        return Some(format!("{} - modular exponentiation failed", ExpModFailed::SIGNATURE));
    }
    if selector == ScalarTooLarge::SELECTOR {
        return Some(format!(
            "{} - scalar value exceeds field modulus",
            ScalarTooLarge::SIGNATURE
        ));
    }

    // Access control errors
    if selector == OnlyTaskGenerator::SELECTOR {
        return Some(format!(
            "{} - caller not registered as task generator",
            OnlyTaskGenerator::SIGNATURE
        ));
    }
    if selector == OnlyPauser::SELECTOR {
        return Some(format!("{} - caller is not authorized pauser", OnlyPauser::SIGNATURE));
    }
    if selector == OnlyUnpauser::SELECTOR {
        return Some(format!(
            "{} - caller is not authorized unpauser",
            OnlyUnpauser::SIGNATURE
        ));
    }
    if selector == OnlyRegistryCoordinatorOwner::SELECTOR {
        return Some(format!(
            "{} - caller is not registry coordinator owner",
            OnlyRegistryCoordinatorOwner::SIGNATURE
        ));
    }

    // Input validation errors
    if selector == InputAddressZero::SELECTOR {
        return Some(format!(
            "{} - address parameter cannot be zero",
            InputAddressZero::SIGNATURE
        ));
    }
    if selector == InputArrayLengthMismatch::SELECTOR {
        return Some(format!(
            "{} - array lengths do not match",
            InputArrayLengthMismatch::SIGNATURE
        ));
    }
    if selector == InputEmptyQuorumNumbers::SELECTOR {
        return Some(format!(
            "{} - quorum numbers array is empty",
            InputEmptyQuorumNumbers::SIGNATURE
        ));
    }
    if selector == InputNonSignerLengthMismatch::SELECTOR {
        return Some(format!(
            "{} - non-signer arrays have mismatched lengths",
            InputNonSignerLengthMismatch::SIGNATURE
        ));
    }
    if selector == BitmapValueTooLarge::SELECTOR {
        return Some(format!(
            "{} - bitmap value exceeds maximum",
            BitmapValueTooLarge::SIGNATURE
        ));
    }
    if selector == BytesArrayLengthTooLong::SELECTOR {
        return Some(format!(
            "{} - bytes array exceeds maximum length",
            BytesArrayLengthTooLong::SIGNATURE
        ));
    }
    if selector == BytesArrayNotOrdered::SELECTOR {
        return Some(format!(
            "{} - bytes array elements not in required order",
            BytesArrayNotOrdered::SIGNATURE
        ));
    }

    // Pause/state errors
    if selector == CurrentlyPaused::SELECTOR {
        return Some(format!("{} - contract is currently paused", CurrentlyPaused::SIGNATURE));
    }
    if selector == InvalidNewPausedStatus::SELECTOR {
        return Some(format!(
            "{} - invalid pause status transition",
            InvalidNewPausedStatus::SIGNATURE
        ));
    }

    // PDS state-commit errors (StateCommitRegistry / IStateRootCommittable)
    if selector == SequenceGap::SELECTOR {
        return Some(format!(
            "{} - submitted sequenceNo != lastCommittedSequenceNo + 1 (out-of-order commit)",
            SequenceGap::SIGNATURE
        ));
    }
    if selector == StateRootMismatch::SELECTOR {
        return Some(format!(
            "{} - prevStateRoot != lastCommittedStateRoot (state divergence or stale view)",
            StateRootMismatch::SIGNATURE
        ));
    }
    if selector == TimestampRegression::SELECTOR {
        return Some(format!(
            "{} - commit.timestamp <= lastCommittedTimestamp (replay or reorg)",
            TimestampRegression::SIGNATURE
        ));
    }
    if selector == InvalidPcr0Commitment::SELECTOR {
        return Some(format!(
            "{} - pcr0Commitment is bytes32(0) (missing enclave attestation)",
            InvalidPcr0Commitment::SIGNATURE
        ));
    }
    if selector == InvalidNewStateRoot::SELECTOR {
        return Some(format!(
            "{} - newStateRoot is bytes32(0) (cannot represent any legitimate JMT root)",
            InvalidNewStateRoot::SIGNATURE
        ));
    }
    if selector == UnsupportedStateCommitVersion::SELECTOR {
        return Some(format!(
            "{} - StateCommit.version != STATE_COMMIT_V1 (schema mismatch)",
            UnsupportedStateCommitVersion::SIGNATURE
        ));
    }
    if selector == InvalidSealedSnapshot::SELECTOR {
        return Some(format!(
            "{} - sealed snapshot payload malformed or signature empty",
            InvalidSealedSnapshot::SIGNATURE
        ));
    }
    if selector == CertificateMessageHashMismatch::SELECTOR {
        return Some(format!(
            "{} - cert.messageHash != keccak256(abi.encode(StateCommit)) (cert reuse / digest drift)",
            CertificateMessageHashMismatch::SIGNATURE
        ));
    }

    // External errors from registry (library/policy errors not directly in TaskManager bindings)
    if let Some(ext_error) = errors::lookup_external_error(&selector) {
        return Some(format!("{} - {}", ext_error.name, ext_error.description));
    }

    None
}

/// Log detailed error information for respondToTask call failures.
/// This function analyzes the error to identify the cause and logs diagnostic information.
/// Returns the identified error message if one was identified, None otherwise.
pub fn log_respond_to_task_error(
    error: &alloy::contract::Error,
    task: &Task,
    task_manager_addr: Address,
    signer_address: Address,
    calldata: &[u8],
    chain_id: u64,
) -> Option<String> {
    info!("Analyzing respondToTask error to identify failure cause");

    // Use alloy's as_revert_data() method to extract revert data properly
    // This is more reliable than parsing Debug output strings
    if let Some(revert_data) = error.as_revert_data() {
        // Check for empty revert data (0x) - indicates BN254 EC operation failure or other assembly revert
        if revert_data.is_empty() {
            error!(
                chain_id,
                task_id = %newton_core::hex!(task.taskId),
                task_created_block = task.taskCreatedBlock,
                task_manager = %task_manager_addr,
                signer = %signer_address,
                calldata_len = calldata.len(),
                "[DEBUG] Empty revert (0x): BN254 EC operation failure"
            );

            // [DEBUG] Provide additional context for BN254 failure root cause analysis
            error!(
                chain_id,
                task_id = %newton_core::hex!(task.taskId),
                "[DEBUG] BN254 EC Operation Failure - Potential causes:\n\
                 1. Signers APK G2 is point at infinity (no valid signers aggregated)\n\
                 2. Aggregated signature is point at infinity (no valid signatures)\n\
                 3. Non-signer public keys contain invalid/malformed G1 points\n\
                 4. Quorum APK contains invalid G1 points\n\
                 5. Pairing check failed due to signature/message mismatch\n\
                 Check the [DEBUG] logs above for 'point at infinity' warnings."
            );

            return Some("Empty revert (0x): BN254 EC operation failure".to_string());
        }

        // Extract 4-byte selector from revert data
        if revert_data.len() >= 4 {
            let selector: [u8; 4] = revert_data[..4].try_into().unwrap_or([0; 4]);
            let selector_hex = hex::encode(selector);

            if let Some(decoded_msg) = decode_error_selector(selector) {
                error!(
                    chain_id,
                    task_id = %newton_core::hex!(task.taskId),
                    task_created_block = task.taskCreatedBlock,
                    error_selector = %format!("0x{}", selector_hex),
                    error_message = %decoded_msg,
                    task_manager = %task_manager_addr,
                    signer = %signer_address,
                    "respondToTask error identified"
                );
                // Log debug hint for transaction investigation
                error!("Debug hint: Run `cast run --trace <tx_hash> --rpc-url <rpc_url>` to investigate");
                return Some(decoded_msg);
            }

            // Unknown selector - log full revert data for debugging
            error!(
                chain_id,
                task_id = %newton_core::hex!(task.taskId),
                task_created_block = task.taskCreatedBlock,
                error_selector = %format!("0x{}", selector_hex),
                revert_data_hex = %format!("0x{}", hex::encode(&revert_data)),
                task_manager = %task_manager_addr,
                signer = %signer_address,
                "Unknown error selector in respondToTask"
            );
            error!("Debug hint: Run `cast run --trace <tx_hash> --rpc-url <rpc_url>` to investigate");
            return Some(format!("Unknown error selector: 0x{}", selector_hex));
        }
    }

    // Fallback: try to extract from error string for non-revert errors
    let error_str = format!("{:?}", error);
    error!(
        chain_id,
        task_id = %newton_core::hex!(task.taskId),
        task_created_block = task.taskCreatedBlock,
        task_manager = %task_manager_addr,
        signer = %signer_address,
        error = %error_str,
        "Failed to extract revert data from respondToTask error"
    );
    None
}

/// Log detailed error information for `commitStateRoot` call failures.
///
/// Follows the same pattern as `log_respond_to_task_error` — extract revert data,
/// route through `decode_error_selector`, fall back to raw revert hex on unknown
/// selectors. Log fields are commit-specific (sequenceNo, prev/newStateRoot,
/// daCertHash, pcr0Commitment) and the empty-revert cause list is specific to
/// the `ViewBN254CertificateVerifier` + `StateCommitRegistry` path.
///
/// Returns the identified error message if one was decoded, None otherwise.
pub fn log_commit_state_root_error(
    error: &alloy::contract::Error,
    commit: &StateCommit,
    registry_addr: Address,
    signer_address: Address,
    calldata: &[u8],
    chain_id: u64,
) -> Option<String> {
    info!("Analyzing commitStateRoot error to identify failure cause");

    if let Some(revert_data) = error.as_revert_data() {
        // Empty 0x revert: BLS pairing failure or assembly-level revert in
        // the on-chain certificate verifier
        if revert_data.is_empty() {
            error!(
                chain_id,
                sequence_no = commit.sequenceNo,
                timestamp = commit.timestamp,
                prev_state_root = %commit.prevStateRoot,
                new_state_root = %commit.newStateRoot,
                registry = %registry_addr,
                signer = %signer_address,
                calldata_len = calldata.len(),
                "[DEBUG] Empty revert (0x): possible BLS pairing failure or assembly revert in cert verifier"
            );

            // [DEBUG] Provide additional context for state-commit empty-revert root cause analysis
            error!(
                chain_id,
                sequence_no = commit.sequenceNo,
                "[DEBUG] State Commit Empty Revert - Potential causes:\n\
                 1. ViewBN254CertificateVerifier BLS pairing math failed: signers APK at infinity, quorum APK invalid, or quorum stake below threshold. Note: cert.messageHash drift fires typed CertificateMessageHashMismatch (0x822ef683), NOT empty revert.\n\
                 2. Operator set table not yet populated on this chain (source: BN254TableCalculator unseeded after redeployment; destination: ECDSAOperatorTableUpdater behind transporter sync)\n\
                 3. Out-of-gas during BN254 EC operations in cert verification\n\
                 4. Assembly-level revert in StateCommitRegistry (check redeployment / storage-layout integrity)\n\
                 Verify cert with `cast call` on viewBN254CertificateVerifier; check operator table population on this chain."
            );

            return Some("Empty revert (0x): BLS verification or assembly revert".to_string());
        }

        if revert_data.len() >= 4 {
            let selector: [u8; 4] = revert_data[..4].try_into().unwrap_or([0; 4]);
            let selector_hex = hex::encode(selector);

            if let Some(decoded_msg) = decode_error_selector(selector) {
                error!(
                    chain_id,
                    sequence_no = commit.sequenceNo,
                    timestamp = commit.timestamp,
                    prev_state_root = %commit.prevStateRoot,
                    new_state_root = %commit.newStateRoot,
                    da_cert_hash = %commit.daCertHash,
                    pcr0_commitment = %commit.pcr0Commitment,
                    error_selector = %format!("0x{}", selector_hex),
                    error_message = %decoded_msg,
                    registry = %registry_addr,
                    signer = %signer_address,
                    "commitStateRoot error identified"
                );
                error!("Debug hint: Run `cast run --trace <tx_hash> --rpc-url <rpc_url>` to investigate");
                return Some(decoded_msg);
            }

            error!(
                chain_id,
                sequence_no = commit.sequenceNo,
                timestamp = commit.timestamp,
                prev_state_root = %commit.prevStateRoot,
                new_state_root = %commit.newStateRoot,
                error_selector = %format!("0x{}", selector_hex),
                revert_data_hex = %format!("0x{}", hex::encode(&revert_data)),
                registry = %registry_addr,
                signer = %signer_address,
                "Unknown error selector in commitStateRoot"
            );
            error!("Debug hint: Run `cast run --trace <tx_hash> --rpc-url <rpc_url>` to investigate");
            return Some(format!("Unknown error selector: 0x{}", selector_hex));
        }
    }

    let error_str = format!("{:?}", error);
    error!(
        chain_id,
        sequence_no = commit.sequenceNo,
        timestamp = commit.timestamp,
        prev_state_root = %commit.prevStateRoot,
        new_state_root = %commit.newStateRoot,
        registry = %registry_addr,
        signer = %signer_address,
        error = %error_str,
        "Failed to extract revert data from commitStateRoot error"
    );
    None
}

#[cfg(test)]
mod tests {
    use super::*;

    // ----- STATE_COMMIT_RECEIPT_TIMEOUT --------------------------------------
    //
    // The 60s receipt timeout is half of the protocol's 120s commit cadence
    // (see PRIVATE_DATA_STORAGE.md §6.1). A drift here means a stuck commit
    // could block the next aggregator's prepare-phase tick. Pin the constant
    // so refactors that change the cadence have to update this test too.

    #[test]
    fn test_state_commit_receipt_timeout_is_60s() {
        assert_eq!(STATE_COMMIT_RECEIPT_TIMEOUT, Duration::from_secs(60));
    }

    // ----- escalate_floor — gas-bump replacement pricing ---------------------
    //
    // The bump must always beat the replaced TX by >=10% on BOTH fee fields
    // (EIP-1559 replacement rule) AND track a rising market AND keep the tip
    // realizable. `market` is now `(base_fee, market_priority)`; the function
    // sizes max_fee = max(prev × 1.10, base_fee × 2.5 + priority) so the validator
    // tip `min(priority, max_fee − base_fee)` always equals the full priority.
    // These pin the floor logic that prevents "replacement transaction
    // underpriced" (see docs/BATCH_SUBMISSION_NONCE_ANCHORING.md).

    // Sentinel for "no ceiling" in the floor tests.
    const NO_CEIL: u128 = u128::MAX;

    /// The realized validator tip from a 1559 tx: `min(priority, max_fee − base)`.
    fn realized_tip(max_fee: u128, prio: u128, base_fee: u128) -> u128 {
        prio.min(max_fee.saturating_sub(base_fee))
    }

    #[test]
    fn escalate_floor_market_priority_wins_and_is_fully_realizable() {
        // Hot market: priority tracks the market tip (100 > prev×1.10=55), and
        // max_fee carries enough headroom that the FULL tip reaches the validator.
        let base_fee = 1_000;
        let (max_fee, prio) = escalate_floor(Some((base_fee, 100)), 500, 50, NO_CEIL);
        assert_eq!(prio, 100, "priority tracks the market tip");
        assert_eq!(
            max_fee,
            base_fee * 5 / 2 + 100,
            "max_fee = base×2.5 + priority headroom"
        );
        assert_eq!(
            realized_tip(max_fee, prio, base_fee),
            prio,
            "the full priority must be realizable as a tip"
        );
    }

    #[test]
    fn escalate_floor_applies_10pct_floor_when_market_cooled() {
        // Cooled market (low base, low tip): the ≥10% floor over prev wins on
        // both fields, else the replacement is rejected as underpriced.
        let (max_fee, prio) = escalate_floor(Some((100, 40)), 500, 50, NO_CEIL);
        assert_eq!(max_fee, 550, "prev_max_fee × 1.10 floor wins over base×2.5+prio=305");
        assert_eq!(prio, 55, "prev_prio × 1.10 floor wins over market tip 40");
    }

    #[test]
    fn escalate_floor_market_unavailable_still_escalates() {
        // fee-history failed (None): never broadcast at/below prev — apply the
        // floor so the replacement still clears the >=10% rule on both fields.
        let (max_fee, prio) = escalate_floor(None, 500, 50, NO_CEIL);
        assert_eq!(max_fee, 550);
        assert_eq!(prio, 55, "priority still climbs even with no market quote");
    }

    #[test]
    fn escalate_floor_tip_realizable_as_base_fee_rises() {
        // The core of Fix: the base fee climbs sharply between attempts. The
        // headroom (base × 2.5) must keep the full priority tip realizable at the
        // NEW base fee, not a fraction throttled by a stale max_fee.
        let prev_base = 100;
        let (prev_max, prev_prio) = escalate_floor(Some((prev_base, 30)), 0, 0, NO_CEIL);
        // Base fee doubles (≈6 blocks of max EIP-1559 growth) by the next attempt.
        let new_base = 200;
        let (max_fee, prio) = escalate_floor(Some((new_base, 30)), prev_max, prev_prio, NO_CEIL);
        assert_eq!(
            realized_tip(max_fee, prio, new_base),
            prio,
            "full tip must survive a doubling base fee: max_fee {max_fee}, base {new_base}, prio {prio}"
        );
    }

    #[test]
    fn escalate_floor_is_monotonic_across_attempts() {
        // 3 attempts in a cooled market (None each): each must strictly exceed
        // the previous on both fields, so the node never rejects the replacement.
        let mut max_fee = 1_000u128;
        let mut prio = 100u128;
        for _ in 0..3 {
            let (nm, np) = escalate_floor(None, max_fee, prio, NO_CEIL);
            assert!(nm > max_fee, "max_fee must climb: {nm} !> {max_fee}");
            assert!(np > prio, "priority must climb: {np} !> {prio}");
            max_fee = nm;
            prio = np;
        }
    }

    #[test]
    fn escalate_floor_saturates_without_overflow() {
        // Near u128::MAX the ×1.10 / ×2 must saturate, not panic or wrap to a
        // small value. (Saturating-mul-then-divide lands below u128::MAX — that's
        // fine; the only real invariants are "didn't wrap" and "priority ≤ max".)
        // Saturating ×110 then ÷100 lands near u128::MAX/100; assert we stay far
        // above any "wrapped to small" value without pinning the exact figure.
        let huge = u128::MAX / 1000;
        for market in [Some((u128::MAX, u128::MAX)), None] {
            let (max_fee, prio) = escalate_floor(market, u128::MAX, u128::MAX, NO_CEIL);
            assert!(max_fee > huge, "max_fee must stay huge (no wrap): {max_fee}");
            assert!(prio > huge, "priority must stay huge (no wrap): {prio}");
            assert!(prio <= max_fee, "priority {prio} must not exceed max_fee {max_fee}");
        }
    }

    #[test]
    fn escalate_floor_clamps_to_ceiling() {
        // A hot market quote above the ceiling must be clamped down. The runaway
        // bug priced a 21k cancel at ~786k gwei; a sane ceiling caps it.
        let ceiling = 1_000u128;
        let (max_fee, prio) = escalate_floor(Some((5_000, 4_000)), 500, 50, ceiling);
        assert_eq!(max_fee, 1_000, "max_fee clamped to ceiling");
        assert_eq!(
            prio, 1_000,
            "priority clamped to the post-ceiling max_fee, never above it"
        );
    }

    #[test]
    fn escalate_floor_ceiling_breaks_runaway_escalation() {
        // The runaway regression: without a ceiling, repeated cooled-market floor
        // bumps compound geometrically. With one, fees plateau at the ceiling
        // instead of climbing until the signer can't pay.
        let ceiling = 10_000u128;
        let mut max_fee = 1_000u128;
        let mut prio = 100u128;
        for _ in 0..50 {
            let (nm, np) = escalate_floor(None, max_fee, prio, ceiling);
            max_fee = nm;
            prio = np;
        }
        assert_eq!(max_fee, ceiling, "max_fee must plateau at the ceiling, not run away");
        assert!(prio <= max_fee, "priority must never exceed max_fee");
    }

    #[test]
    fn escalate_floor_keeps_priority_below_clamped_max_fee() {
        // Even when prev priority was already near the ceiling, clamping max_fee
        // down must drag priority with it so priority ≤ max_fee holds.
        let ceiling = 2_000u128;
        let (max_fee, prio) = escalate_floor(None, 10_000, 9_000, ceiling);
        assert_eq!(max_fee, 2_000);
        assert!(
            prio <= max_fee,
            "priority {prio} must not exceed clamped max_fee {max_fee}"
        );
    }

    // ----- decode_error_selector — PDS state-commit selectors ----------------
    //
    // Selector hex values are pinned per `.claude/rules/error-selectors.md` to
    // catch silent ABI drift in `StateCommitRegistry.sol` / `IStateRootCommittable.sol`.
    // A renamed field or reordered struct in the Solidity source would shift the
    // keccak256-derived selector, which would route to the unknown-selector
    // branch (raw hex in logs) instead of the typed diagnostic path.

    #[test]
    fn test_decode_sequence_gap() {
        let selector = [0x10, 0x4d, 0x00, 0x50];
        assert_eq!(SequenceGap::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("SequenceGap"));
        assert!(msg.contains("out-of-order commit"));
    }

    #[test]
    fn test_decode_state_root_mismatch() {
        let selector = [0x37, 0xf0, 0x4d, 0x41];
        assert_eq!(StateRootMismatch::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("StateRootMismatch"));
        assert!(msg.contains("state divergence"));
    }

    #[test]
    fn test_decode_timestamp_regression() {
        let selector = [0x5a, 0x61, 0x2e, 0x4c];
        assert_eq!(TimestampRegression::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("TimestampRegression"));
        assert!(msg.contains("replay or reorg"));
    }

    #[test]
    fn test_decode_invalid_pcr0_commitment() {
        let selector = [0x6d, 0xfb, 0xfc, 0x74];
        assert_eq!(InvalidPcr0Commitment::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("InvalidPcr0Commitment"));
        assert!(msg.contains("missing enclave attestation"));
    }

    #[test]
    fn test_decode_invalid_new_state_root() {
        let selector = [0x5b, 0xf0, 0xf7, 0x68];
        assert_eq!(InvalidNewStateRoot::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("InvalidNewStateRoot"));
        assert!(msg.contains("legitimate JMT root"));
    }

    #[test]
    fn test_decode_unsupported_state_commit_version() {
        let selector = [0xb6, 0x81, 0x66, 0x8e];
        assert_eq!(UnsupportedStateCommitVersion::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("UnsupportedStateCommitVersion"));
        assert!(msg.contains("schema mismatch"));
    }

    #[test]
    fn test_decode_invalid_sealed_snapshot() {
        let selector = [0xdc, 0x4e, 0x1d, 0x57];
        assert_eq!(InvalidSealedSnapshot::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("InvalidSealedSnapshot"));
        assert!(msg.contains("malformed"));
    }

    #[test]
    fn test_decode_certificate_message_hash_mismatch() {
        let selector = [0x82, 0x2e, 0xf6, 0x83];
        assert_eq!(CertificateMessageHashMismatch::SELECTOR, selector);
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_some());
        let msg = decoded.unwrap();
        assert!(msg.contains("CertificateMessageHashMismatch"));
        assert!(msg.contains("cert reuse"));
    }

    #[test]
    fn test_decode_unknown_selector_returns_none() {
        // 0xfefdfcfb is a synthetic value not present in any Newton or external
        // error registry. If this test starts failing, audit the new selector
        // before adjusting — a recognized selector should add a typed test above.
        let selector = [0xfe, 0xfd, 0xfc, 0xfb];
        let decoded = decode_error_selector(selector);
        assert!(decoded.is_none());
    }
}