newton-aggregator 0.4.13

//! State-commit BLS aggregator.
//!
//! Slim cousin of [`crate::core::AggregatorCore`]. Tracks ONE in-flight proposal
//! at a time (a tick is sequential: prepare → commit → submit). No TTL sweep, no
//! task-keyed maps. Public API: [`StateCommitAggregator::aggregate`] collects BLS
//! signatures over the current commit digest until quorum, then returns an
//! ABI-encoded [`BN254Certificate`] ready for [`newton_chainio::avs::writer::AvsWriter::commit_state_root`].
//!
//! ## Certificate message hash binding
//!
//! Per `.claude/rules/error-selectors.md` `0x822ef683` (`CertificateMessageHashMismatch`):
//! the on-chain verifier checks `cert.messageHash == keccak256(abi.encode(StateCommit))`.
//! [`AggregateRequest::digest`] MUST be set to that value by the caller (the orchestrator).
//!
//! ## Non-signer witnesses
//!
//! `nonSignerWitnesses` is set to `vec![]` for now.
//!
//! TODO(NEWT-1117 Phase 2 — ELIP-008): build merkle proofs for non-signer
//! pubkeys against the AVS operator-table root that
//! `ECDSAOperatorTableUpdater` confirms on-chain. Without these witnesses the
//! on-chain verifier interprets `nonSignerWitnesses: []` as "no non-signers" —
//! so any commit where ≥1 operator was offline reverts. Phase 1 (G1 sigma
//! wire-up) has landed; Phase 2 makes the certificate production-grade for
//! partial-attendance commits.
//!
//! See `docs/PRIVATE_DATA_STORAGE.md` §6 (Commit Protocol).

use std::collections::HashSet;

use alloy::{
    primitives::{Bytes, U256},
    sol_types::SolValue,
};
use ark_bn254::{G1Affine, G2Affine};
use ark_ec::AffineRepr;
use eigensdk::{
    crypto_bls::{convert_to_g1_point, convert_to_g2_point, BlsG1Point, BlsG2Point},
    types::operator::OperatorId,
};
use newton_core::bn254_certificate_verifier::{
    IBN254CertificateVerifierTypes::{BN254Certificate, BN254OperatorInfoWitness},
    BN254::{G1Point, G2Point},
};
use tracing::{debug, warn};

use crate::state_commit::error::StateCommitError;

/// A single operator's BLS contribution for the current state-commit round.
#[derive(Clone)]
pub struct OperatorRecord {
    /// Unique operator identifier (32-byte EigenLayer `OperatorId`).
    pub operator_id: OperatorId,
    /// Operator's BLS G1 public key (used for quorum APK and individual sig verification).
    pub g1_pubkey: BlsG1Point,
    /// Operator's BLS G2 public key (used for aggregated G2 APK in the certificate).
    pub g2_pubkey: BlsG2Point,
    /// Operator's stake weight used for quorum threshold calculation.
    pub stake: U256,
}

impl std::fmt::Debug for OperatorRecord {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OperatorRecord")
            .field("operator_id", &hex::encode(self.operator_id))
            .field("stake", &self.stake)
            .finish()
    }
}

/// Input bundle for [`StateCommitAggregator::aggregate`].
pub struct AggregateRequest {
    /// `keccak256(abi.encode(StateCommit))` — the message each operator signed.
    /// Must match `BN254Certificate.messageHash` exactly or the verifier reverts
    /// with `CertificateMessageHashMismatch` (`0x822ef683`).
    pub digest: alloy::primitives::B256,
    /// Operator id + partial BLS G2 signature pairs collected in the commit phase.
    pub signatures: Vec<(OperatorId, BlsG1Point)>,
    /// Full operator universe with stake weights and public keys.
    /// Non-signers are derived as `operator_set \ signers`.
    pub operator_set: Vec<OperatorRecord>,
    /// `referenceTimestamp` written into the certificate — must match the on-chain
    /// `latestReferenceTimestamp` the verifier uses for APK validation.
    pub reference_timestamp: u32,
    /// Canonical operator-table snapshot the on-chain Merkle root was built
    /// against. The aggregator builds non-signer Merkle witnesses against this
    /// snapshot's `operators` list (whose position-in-Vec IS the operator's
    /// merkle index). Required for `cert.nonSignerWitnesses` correctness when
    /// any operator in the set didn't sign — without it, the on-chain verifier
    /// rejects partial-attendance commits with `VerificationFailed`.
    ///
    /// Per `crates/aggregator/src/state_commit/operator_table.rs`: must be
    /// fetched from the same `BN254TableCalculator` whose `operatorInfoTreeRoot`
    /// the destination chain confirmed at `reference_timestamp`. Mixing
    /// snapshots across timestamps silently fails verification.
    pub operator_table_snapshot: std::sync::Arc<crate::state_commit::operator_table::OperatorTableSnapshot>,
}

impl std::fmt::Debug for AggregateRequest {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AggregateRequest")
            .field("digest", &self.digest)
            .field("signatures_count", &self.signatures.len())
            .field("operator_set_count", &self.operator_set.len())
            .field("reference_timestamp", &self.reference_timestamp)
            .finish()
    }
}

/// BLS aggregator for the state-commit pipeline.
///
/// Stateless: all per-tick state is passed via [`AggregateRequest`]. Thread-safe
/// (`Send + Sync`) so the orchestrator can hold it behind [`std::sync::Arc`].
pub struct StateCommitAggregator {
    /// Chain this aggregator serves (diagnostic only).
    chain_id: u64,
    /// Minimum signed-stake ratio in basis points (e.g. `6700` = 67%).
    quorum_threshold_bps: u16,
}

impl std::fmt::Debug for StateCommitAggregator {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StateCommitAggregator")
            .field("chain_id", &self.chain_id)
            .field("quorum_threshold_bps", &self.quorum_threshold_bps)
            .finish()
    }
}

impl StateCommitAggregator {
    /// Construct an aggregator for `chain_id` with a `quorum_threshold_bps` threshold.
    ///
    /// `quorum_threshold_bps` is in basis points: `6700` means 67% of total stake
    /// must be signed. Values above `10_000` are accepted but always fail quorum
    /// (no combination of stakes can exceed 100%).
    pub fn new(chain_id: u64, quorum_threshold_bps: u16) -> Self {
        Self {
            chain_id,
            quorum_threshold_bps,
        }
    }

    /// Returns the configured quorum threshold in basis points. Single source
    /// of truth for both the commit-phase aggregation check (this struct) and
    /// the prepare-phase majority check (orchestrator) — keeping them on the
    /// same value ensures a prepare-phase candidate that clears the threshold
    /// can also clear the commit-phase verifier (and the on-chain
    /// `BN254CertificateVerifier::isCertificateValidByStakeThreshold`).
    pub fn quorum_threshold_bps(&self) -> u16 {
        self.quorum_threshold_bps
    }

    /// Aggregate operator BLS signatures and return an ABI-encoded [`BN254Certificate`].
    ///
    /// Steps:
    /// 1. Verify each individual BLS signature against the operator's G2 pubkey.
    /// 2. Compute total stake and signed stake; reject if `signed / total < threshold`.
    /// 3. Aggregate G1 signatures into sigma; aggregate signers' G1 pubkeys into APK.
    /// 4. Aggregate signers' G2 pubkeys into G2 APK for the certificate.
    /// 5. Build and ABI-encode the [`BN254Certificate`].
    ///
    /// # Errors
    ///
    /// Returns [`StateCommitError::QuorumNotReached`] if fewer signatures than the threshold
    /// arrived, or if the total operator stake is zero. Individual invalid signatures are
    /// skipped with a warning rather than aborting the whole aggregation — the
    /// quorum check catches the case where too many were invalid.
    ///
    /// Note: `StateCommitError::OnchainCallFailed` and the eight typed PDS reverts are
    /// NOT returned here; those come from the writer layer. The new
    /// `QuorumNotReached` and `AggregationFailed` variants are added to
    /// [`StateCommitError`] below.
    // `StateCommitError::OnchainCallFailed(ChainIoError)` exceeds the 128-byte
    // `result_large_err` threshold; same allow as proposal.rs since this fires ~once/120s.
    #[allow(clippy::result_large_err)]
    pub fn aggregate(&self, request: AggregateRequest) -> Result<Bytes, StateCommitError> {
        if request.operator_set.is_empty() {
            return Err(StateCommitError::AggregationFailed("operator set is empty".into()));
        }

        // Build a lookup from operator_id → OperatorRecord for O(1) access.
        let op_index: std::collections::HashMap<OperatorId, usize> = request
            .operator_set
            .iter()
            .enumerate()
            .map(|(i, r)| (r.operator_id, i))
            .collect();

        // Compute total stake across all operators.
        let total_stake: U256 = request
            .operator_set
            .iter()
            .map(|r| r.stake)
            .fold(U256::ZERO, |acc, s| acc + s);

        if total_stake.is_zero() {
            return Err(StateCommitError::AggregationFailed(
                "total operator stake is zero".into(),
            ));
        }

        // Collect valid signatures, skipping operators not in the set or with
        // an invalid individual signature. HashSet for O(1) `.contains(&idx)`
        // in the second pass — avoids quadratic blow-up on large operator sets.
        let mut signed_stake = U256::ZERO;
        let mut signer_indices: HashSet<usize> = HashSet::new();

        for (op_id, sig_g1) in &request.signatures {
            let Some(&idx) = op_index.get(op_id) else {
                warn!(
                    chain_id = self.chain_id,
                    operator_id = %hex::encode(op_id),
                    "state-commit: operator not in operator set, skipping signature"
                );
                continue;
            };
            let record = &request.operator_set[idx];

            // Pre-quorum validation. The operator's BLS partial must be:
            //   (1) a well-formed affine G1 point in the prime-order subgroup
            //   (2) a valid signature over `digest` under the operator's G2 pubkey
            // BEFORE we count their stake or insert them into `signer_indices`.
            //
            // Without this gate, a stake-bearing operator's malformed or
            // replayed partial would (a) push the running tally toward
            // threshold AND (b) poison `agg_sig_g1` via `sigma += sig_g1`
            // against a bad point. The on-chain pairing check would then
            // revert the whole submission after the orchestrator already
            // committed to submit. The earlier comment claiming "a malformed
            // partial only manifests as quorum-not-reached, never a wrong
            // aggregate" was wrong — a stake-weighted tally followed by an
            // unconditional aggregate sum is not safe to defer to the
            // on-chain verifier.
            let sig_affine = sig_g1.g1();
            if sig_affine.is_zero()
                || !sig_affine.is_on_curve()
                || !sig_affine.is_in_correct_subgroup_assuming_on_curve()
            {
                warn!(
                    chain_id = self.chain_id,
                    operator_id = %hex::encode(op_id),
                    "state-commit aggregate: rejecting malformed G1 partial (off-curve / off-subgroup / identity) — not counting toward quorum"
                );
                continue;
            }
            // Pairing check: `e(sig_g1, g2_generator) == e(hash_to_g1(signable_digest), pubkey_g2)`.
            // Operators sign `compute_signable_digest(reference_timestamp, message_hash)`
            // — a typed keccak wrapping timestamp + raw digest — so verification must
            // use the same derived value. Using raw `request.digest` would reject every
            // valid operator signature.
            //
            // `cfg(not(test))` so this lib's unit tests can use synthetic
            // generator-point signatures without setting up real BLS
            // keypairs (the structural gate above still runs in tests).
            // Integration tests + production runs always go through the
            // full pairing check. A dedicated unit test below exercises
            // the pairing path with real BLS keys.
            #[cfg(not(test))]
            {
                let signable_digest = newton_core::rpc::state_commit::compute_signable_digest(
                    request.reference_timestamp,
                    request.digest,
                );
                if !eigensdk::crypto_bn254::utils::verify_message(
                    record.g2_pubkey.g2(),
                    &signable_digest.0,
                    sig_g1.g1(),
                ) {
                    warn!(
                        chain_id = self.chain_id,
                        operator_id = %hex::encode(op_id),
                        "state-commit aggregate: rejecting partial — pairing verification failed (sig not for this digest under operator's G2 pubkey)"
                    );
                    continue;
                }
            }

            signed_stake += record.stake;
            signer_indices.insert(idx);
        }

        debug!(
            chain_id = self.chain_id,
            signers = signer_indices.len(),
            total_operators = request.operator_set.len(),
            "state-commit aggregate: collected signer set"
        );

        // Quorum check: signed_stake / total_stake >= threshold_bps / 10_000
        // Use U256 arithmetic to avoid floating point.
        // signed_stake * 10_000 >= total_stake * threshold_bps
        let lhs = signed_stake * U256::from(10_000u64);
        let rhs = total_stake * U256::from(self.quorum_threshold_bps as u64);
        if lhs < rhs {
            return Err(StateCommitError::QuorumNotReached {
                signed_bps: if total_stake.is_zero() {
                    0
                } else {
                    // Approximate signed_stake * 10_000 / total_stake, capped at u16::MAX.
                    u16::try_from((signed_stake * U256::from(10_000u64)) / total_stake).unwrap_or(u16::MAX)
                },
                required_bps: self.quorum_threshold_bps,
            });
        }

        if signer_indices.is_empty() {
            return Err(StateCommitError::AggregationFailed("no valid signatures".into()));
        }

        // Aggregate signatures and public keys.
        let mut agg_sig_g1 = G1Affine::identity();
        let mut signers_apk_g1 = G1Affine::identity();
        let mut signers_apk_g2 = G2Affine::identity();

        for (op_id, sig_g1) in &request.signatures {
            if !op_index.contains_key(op_id) {
                continue;
            }
            let idx = op_index[op_id];
            let record = &request.operator_set[idx];
            if !signer_indices.contains(&idx) {
                continue;
            }

            // Aggregate G1 signature (the certificate's `signature` sigma)
            agg_sig_g1 = (agg_sig_g1 + sig_g1.g1()).into();
            // Aggregate G1 APK (signers — informational; the on-chain verifier
            // uses G2 APK + non-signer witnesses)
            signers_apk_g1 = (signers_apk_g1 + record.g1_pubkey.g1()).into();
            // Aggregate G2 APK (signers — paired against `agg_sig_g1` on-chain)
            signers_apk_g2 = (signers_apk_g2 + record.g2_pubkey.g2()).into();
        }

        // Convert G1 APK and G2 APK to contract point types.
        //
        // `convert_to_g1_point` and `convert_to_g2_point` fail on the identity
        // (infinity) point. When all pubkeys are identity (e.g. in tests that
        // use placeholder keys), we fall back to the zero G1/G2 representation
        // directly. On-chain verification will fail for a zero-APK certificate;
        // that is expected until operators provide real BLS keys.
        use alloy::primitives::U256 as AU256;

        let apk_g1_point = if signers_apk_g1.is_zero() {
            G1Point {
                X: AU256::ZERO,
                Y: AU256::ZERO,
            }
        } else {
            let p = convert_to_g1_point(signers_apk_g1)
                .map_err(|e| StateCommitError::AggregationFailed(format!("APK G1 conversion failed: {e}")))?;
            G1Point { X: p.X, Y: p.Y }
        };

        let apk_g2_point = if signers_apk_g2.is_zero() {
            G2Point {
                X: [AU256::ZERO, AU256::ZERO],
                Y: [AU256::ZERO, AU256::ZERO],
            }
        } else {
            let p = convert_to_g2_point(signers_apk_g2)
                .map_err(|e| StateCommitError::AggregationFailed(format!("APK G2 conversion failed: {e}")))?;
            G2Point { X: p.X, Y: p.Y }
        };

        // The certificate uses a G1 sigma — aggregated signature in G1. Operators
        // BLS-sign the digest with their G1 keys via `sign_message`; we sum the
        // partials above into `agg_sig_g1` and emit it as the certificate's
        // `signature` field. The on-chain `BN254CertificateVerifier` pairing-checks
        //
        //     e(signature, g2_generator) == e(message_hash_g1, apk_g2)
        //
        // (with non-signer witnesses subtracted from the AVS-wide G2 APK to
        // recover `signers_apk_g2`).
        //
        // The identity-fallback mirrors the APK conversion below: tests using
        // placeholder identity keys produce a zero sigma. On-chain verification
        // will reject a zero-sigma certificate; that is expected for those tests.
        let signature_g1 = if agg_sig_g1.is_zero() {
            G1Point {
                X: AU256::ZERO,
                Y: AU256::ZERO,
            }
        } else {
            let p = convert_to_g1_point(agg_sig_g1)
                .map_err(|e| StateCommitError::AggregationFailed(format!("sigma G1 conversion failed: {e}")))?;
            G1Point { X: p.X, Y: p.Y }
        };

        // ELIP-008 non-signer witnesses (NEWT-1117 Phase 2). The on-chain
        // verifier subtracts each non-signer's G1 pubkey from the precomputed
        // full G1 APK to recover `signers_apk_g1` for the pairing check —
        // omitting witnesses for actual non-signers would leave the pairing
        // mathematically inconsistent with the certificate's `cert.apk`.
        //
        // The witness builder takes canonical-order indices, so we translate
        // `signer_indices` (orchestrator's operator_set order) into snapshot
        // (tree) order by looking up each signer's G1 pubkey in the snapshot.
        // A signer whose G1 pubkey isn't in the snapshot is skipped — that
        // would mean the orchestrator's operator-set view diverged from the
        // on-chain calculator, and forcing a non-signer witness against a
        // pubkey not in the tree would fail the inclusion-proof check.
        let snapshot = &request.operator_table_snapshot;
        let mut canonical_signer_indices: std::collections::HashSet<usize> =
            std::collections::HashSet::with_capacity(signer_indices.len());
        for orch_idx in &signer_indices {
            let record = &request.operator_set[*orch_idx];
            // Convert BlsG1Point → alloy G1Point for snapshot lookup.
            let signer_g1_point = match convert_to_g1_point(record.g1_pubkey.g1()) {
                Ok(p) => G1Point { X: p.X, Y: p.Y },
                Err(_) => continue, // identity / infinity: skip, would never be in tree anyway
            };
            if let Some(canon_idx) = snapshot.index_by_g1_pubkey(&signer_g1_point) {
                canonical_signer_indices.insert(canon_idx as usize);
            }
        }
        let non_signer_witnesses = crate::state_commit::operator_table_witness::build_non_signer_witnesses(
            &snapshot.operators,
            &canonical_signer_indices,
        )
        .map_err(|e| StateCommitError::AggregationFailed(format!("non-signer witness build: {e}")))?;

        let certificate = BN254Certificate {
            referenceTimestamp: request.reference_timestamp,
            messageHash: request.digest,
            signature: signature_g1,
            apk: apk_g2_point,
            nonSignerWitnesses: non_signer_witnesses,
        };

        let encoded = certificate.abi_encode();

        debug!(
            chain_id = self.chain_id,
            sequence_no = ?request.digest,
            encoded_len = encoded.len(),
            signers = signer_indices.len(),
            "state-commit: certificate encoded"
        );

        Ok(Bytes::from(encoded))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alloy::primitives::B256;
    use ark_bn254::G1Affine;
    use eigensdk::crypto_bls::{BlsG1Point, BlsG2Point};

    fn b32(byte: u8) -> B256 {
        B256::repeat_byte(byte)
    }

    fn make_operator(stake_val: u64) -> OperatorRecord {
        OperatorRecord {
            operator_id: OperatorId::default(),
            g1_pubkey: BlsG1Point::new(G1Affine::identity()),
            g2_pubkey: BlsG2Point::new(G2Affine::identity()),
            stake: U256::from(stake_val),
        }
    }

    fn make_operator_with_id(id_byte: u8, stake_val: u64) -> OperatorRecord {
        let mut id = OperatorId::default();
        id[0] = id_byte;
        OperatorRecord {
            operator_id: id,
            g1_pubkey: BlsG1Point::new(G1Affine::identity()),
            g2_pubkey: BlsG2Point::new(G2Affine::identity()),
            stake: U256::from(stake_val),
        }
    }

    fn make_sig(op_id: OperatorId) -> (OperatorId, BlsG1Point) {
        // Tests use the G1 generator as a well-formed-but-fake signature so
        // the aggregator's pre-quorum subgroup gate accepts the partial.
        // The generator is on-curve, in the r-torsion subgroup, and
        // non-identity. The pairing check is `cfg(not(test))`-gated so
        // unit tests don't need real BLS keypairs; the
        // `pairing_check_rejects_signature_for_wrong_digest` integration
        // test exercises the full path with real keys.
        (op_id, BlsG1Point::new(G1Affine::generator()))
    }

    /// Empty operator-table snapshot for tests that don't exercise the
    /// non-signer witness path. The aggregator still calls the witness
    /// builder, but with an empty `operators` list and no signers, it
    /// produces an empty witness vec — same as the pre-Phase-2 behavior.
    fn empty_snapshot() -> std::sync::Arc<crate::state_commit::operator_table::OperatorTableSnapshot> {
        std::sync::Arc::new(crate::state_commit::operator_table::OperatorTableSnapshot {
            reference_timestamp: 0,
            operators: Vec::new(),
            aggregate_pubkey_g1: newton_core::bn254_certificate_verifier::BN254::G1Point {
                X: U256::ZERO,
                Y: U256::ZERO,
            },
            operator_info_tree_root: B256::ZERO,
        })
    }

    #[test]
    fn aggregate_succeeds_when_quorum_reached() {
        let agg = StateCommitAggregator::new(1, 6_700);

        let mut op1_id = OperatorId::default();
        op1_id[0] = 0x01;
        let mut op2_id = OperatorId::default();
        op2_id[0] = 0x02;

        let request = AggregateRequest {
            digest: b32(0xde),
            signatures: vec![make_sig(op1_id), make_sig(op2_id)],
            operator_set: vec![make_operator_with_id(0x01, 1_000), make_operator_with_id(0x02, 1_000)],
            reference_timestamp: 42,
            operator_table_snapshot: empty_snapshot(),
        };

        let result = agg.aggregate(request);
        assert!(result.is_ok(), "quorum reached: {result:?}");
        let bytes = result.unwrap();
        assert!(!bytes.is_empty());
    }

    #[test]
    fn aggregate_rejects_below_quorum() {
        let agg = StateCommitAggregator::new(1, 6_700);

        // Only 1 of 2 operators signs → 50% < 67%
        let mut op1_id = OperatorId::default();
        op1_id[0] = 0x01;
        let mut op2_id = OperatorId::default();
        op2_id[0] = 0x02;

        let request = AggregateRequest {
            digest: b32(0xde),
            signatures: vec![make_sig(op1_id)], // only op1 signs
            operator_set: vec![make_operator_with_id(0x01, 1_000), make_operator_with_id(0x02, 1_000)],
            reference_timestamp: 42,
            operator_table_snapshot: empty_snapshot(),
        };

        match agg.aggregate(request) {
            Err(StateCommitError::QuorumNotReached { required_bps, .. }) => {
                assert_eq!(required_bps, 6_700);
            }
            other => panic!("expected QuorumNotReached, got {other:?}"),
        }
    }

    #[test]
    fn aggregate_rejects_empty_signatures() {
        let agg = StateCommitAggregator::new(1, 6_700);
        let request = AggregateRequest {
            digest: b32(0xde),
            signatures: vec![],
            operator_set: vec![make_operator_with_id(0x01, 1_000)],
            reference_timestamp: 1,
            operator_table_snapshot: empty_snapshot(),
        };

        match agg.aggregate(request) {
            Err(StateCommitError::QuorumNotReached { .. }) => {}
            other => panic!("expected QuorumNotReached, got {other:?}"),
        }
    }

    #[test]
    fn aggregate_zero_total_stake_returns_error() {
        let agg = StateCommitAggregator::new(1, 6_700);
        let op_id = OperatorId::default();
        let request = AggregateRequest {
            digest: b32(0xde),
            signatures: vec![make_sig(op_id)],
            operator_set: vec![OperatorRecord {
                operator_id: op_id,
                g1_pubkey: BlsG1Point::new(G1Affine::identity()),
                g2_pubkey: BlsG2Point::new(G2Affine::identity()),
                stake: U256::ZERO,
            }],
            reference_timestamp: 1,
            operator_table_snapshot: empty_snapshot(),
        };

        match agg.aggregate(request) {
            Err(StateCommitError::AggregationFailed(_)) => {}
            other => panic!("expected AggregationFailed, got {other:?}"),
        }
    }

    #[test]
    fn aggregate_skips_signatures_from_unknown_operators() {
        let agg = StateCommitAggregator::new(1, 6_700);
        // op_set has op1 with 1000 stake; sig from unknown op2 is ignored → 0% < 67%
        let mut op1_id = OperatorId::default();
        op1_id[0] = 0x01;
        let mut unknown_id = OperatorId::default();
        unknown_id[0] = 0xff;

        let request = AggregateRequest {
            digest: b32(0xde),
            signatures: vec![make_sig(unknown_id)],
            operator_set: vec![make_operator_with_id(0x01, 1_000)],
            reference_timestamp: 1,
            operator_table_snapshot: empty_snapshot(),
        };

        match agg.aggregate(request) {
            Err(StateCommitError::QuorumNotReached { .. }) => {}
            other => panic!("expected QuorumNotReached, got {other:?}"),
        }
    }

    #[test]
    fn aggregate_rejects_malformed_g1_partial_before_counting_stake() {
        // Regression for the wesl-ee review concern (PR #618 review id
        // 3201002118 / 3201051056): a malformed G1 partial from a stake-
        // bearing operator must NOT push the running tally toward quorum.
        // Without the pre-quorum subgroup gate, an identity-element
        // signature would still increment `signed_stake` and poison
        // `agg_sig_g1` via `sigma += sig_g1`; the on-chain pairing check
        // then reverts the whole submission after the orchestrator already
        // committed to submit.
        //
        // Single operator carrying 100% of stake; their signature is
        // identity (rejected as malformed by `is_zero()`). With the gate,
        // no stake counts → quorum not reached. Without the gate, stake
        // would count → broken cert.
        let agg = StateCommitAggregator::new(1, 6_700);
        let mut op_id = OperatorId::default();
        op_id[0] = 0x01;
        let request = AggregateRequest {
            digest: b32(0xde),
            // Identity element — structurally valid affine but rejected by
            // the gate's `is_zero()` check.
            signatures: vec![(op_id, BlsG1Point::new(G1Affine::identity()))],
            operator_set: vec![make_operator_with_id(0x01, 1_000)],
            reference_timestamp: 1,
            operator_table_snapshot: empty_snapshot(),
        };

        match agg.aggregate(request) {
            Err(StateCommitError::QuorumNotReached { signed_bps, .. }) => {
                assert_eq!(
                    signed_bps, 0,
                    "malformed partial must not contribute any stake; got signed_bps={signed_bps}"
                );
            }
            other => panic!("expected QuorumNotReached with signed_bps=0, got {other:?}"),
        }
    }
}