fix(tests): stabilize actor benchmark under load

agent/CRATES_ARCHITECTURE.md

@@ -437,6 +437,10 @@ terminating the process. Process-level health supervision would need to detect t
 but the current runtime does not provide that guarantee. A restart, when it occurs, treats the event
 log as authoritative and reconciles missing derived index rows.
 
+History collectors use the same `MAILBOX_LIMIT_LARGE` capacity to reduce mailbox saturation while
+they record bus events for history inspection. The larger mailbox does not make live fanout lossless;
+do not use it as a replacement for acknowledged delivery or replay.
+
 Those replay guarantees bound local replay memory and file-descriptor use, but they do not make the
 whole persistence path synchronously acknowledged. Live publication and the sequencer/store response
 path still contain `do_send` edges. Snapshot replay also forwards with `do_send`, and `BatchRouter`
@@ -568,8 +572,8 @@ sequenceDiagram
     P->>Z: folded/recursive aggregation proof work
     P->>W: aggregated public key
     Chain->>K: ciphertext outputs
-    K->>Z: C6 decryption-share proofs
-    K->>T: share + proof per output and party
+    K->>Z: C6 decryption-share proofs for the canonical H roster
+    K->>T: share + proof per output and honest party
     T->>Z: C7 aggregation proof
     T->>W: plaintext output
 ```

agent/flow-trace/00_INDEX.md

@@ -72,9 +72,9 @@
 12. COMPUTE     Data encrypted with aggregate PK, computation runs
                   → Ciphertext output published on-chain
 
-13. DECRYPT     Committee members produce decryption shares
-                  → C6 proof per share (proves share correctly derived)
-                  → broadcast to all committee members for buffering
+13. DECRYPT     Canonical honest committee members produce decryption shares
+                   → C6 proof per share (proves share correctly derived)
+                   → broadcast to all committee members for buffering
 
 14. AGGREGATE   Active aggregator combines M+1 shares → plaintext
                   → C7 proof (proves reconstruction correct)

agent/flow-trace/04_DKG_AND_COMPUTATION.md

@@ -499,12 +499,15 @@ ThresholdKeyshare receives AllThresholdSharesCollected
 │     │  │  → Stored encrypted locally for later decryption    │
 │     │  └─────────────────────────────────────────────────────┘
 │
-├─ 2b. CANONICAL H ROSTER (when H < N):
+├─ 2b. C4 ROSTER (when H < N):
 │     Before C4 witness layout, merge external honest party_ids with own_party_id,
 │     sort ascending, and keep the lowest H — same rule as PublicKeyAggregator C5 cap
-│     (`e3_zk_helpers::canonical_honest_party_ids_with_own`). Persisted as `honest_parties`.
-│     Parties outside the lowest H still complete KeyshareCreated but are not in the
-│     aggregator's NodeFold / `honest_committee_addresses` roster.
+│     (`e3_zk_helpers::canonical_honest_party_ids_with_own`).
+│
+│     After C5, PublicKeyAggregated carries the authoritative H-address subset. Each
+│     keyshare maps those addresses through the full N committee order and replaces its
+│     persisted `honest_parties` set before C6. Parties outside that global H roster
+│     ignore CiphertextOutputPublished and do not generate C6 work.
 │
 ├─ 3. PUBLISH C4 PROOF REQUESTS:
 │     DecryptionShareProofsPending {
@@ -904,7 +907,11 @@ active E3's deadlines.
 
 ---
 
-## Phase 4: Decryption Share Generation (Each Committee Member, with C6 Proof)
+## Phase 4: Decryption Share Generation (Canonical Honest Members, with C6 Proof)
+
+Only members in the persisted canonical honest roster of size `H` generate a decryption share and
+C6 proof. Other committee members ignore `CiphertextOutputPublished` for this phase. The active
+aggregator verifies and folds the resulting `H` share/proof bundles.
 
 Before proof verification, the BFV wrapper requires every public input to use its canonical BN254
 field representation. Message coefficients must also fit exactly in 64 bits. This second check is

agent/flow-trace/05_FAILURE_REFUND_SLASHING.md

@@ -1136,8 +1136,9 @@ Slash Reasons (derived from ProofType for Lane A):
 │              Complete Proof-to-Slash Pipeline                    │
 ├─────────────────────────────────────────────────────────────────┤
 │                                                                │
-│  1. PROOF GENERATION (each committee member)                   │
+│  1. PROOF GENERATION (per phase-eligible producer)              │
 │     ProofRequestActor generates & signs C0-C7 proofs           │
+│     → C6 generation is limited to the canonical H roster       │
 │     → Broadcasts signed proofs via P2P gossip                  │
 │                                                                │
 │  2. PROOF VERIFICATION (each receiving committee member)       │

circuits/benchmarks/scripts/generate_report.sh

@@ -771,6 +771,12 @@ if [ -n "$INTEGRATION_BLOB" ]; then
             [ -z "$name" ] && continue
             echo "| $name | $(format_s "$avgr") | $runs | $(format_s "$tot") |" >> "$OUTPUT_FILE"
         done < <(jq -r '.operation_timings[]? | [.name, .avg_seconds, .runs, .total_seconds] | @tsv' <<<"$INTEGRATION_BLOB")
+        c6_h=$(jq -r '.benchmark_config.committee_h // empty' <<<"$INTEGRATION_BLOB")
+        c6_n=$(jq -r '.benchmark_config.committee_n // empty' <<<"$INTEGRATION_BLOB")
+        if [ -n "$c6_h" ] && [ -n "$c6_n" ]; then
+            echo "" >> "$OUTPUT_FILE"
+            echo "_Run-count note: DKG preparation rows are N-scoped and remain \`N=${c6_n}\` for this committee. C6 producer rows are H-scoped: \`ZkThresholdShareDecryption\` should run once for each eligible honest party, so its expected request count is \`H=${c6_h}\`. C7 aggregation is one job per ciphertext output and consumes H C6 inputs. The \`runs\` value counts proof-generation or aggregation jobs, as applicable._" >> "$OUTPUT_FILE"
+        fi
         ott=$(jq -r '.operation_timings_total_seconds // empty' <<<"$INTEGRATION_BLOB")
         if [ -n "$ott" ] && [ "$ott" != "null" ]; then
             echo "" >> "$OUTPUT_FILE"

crates/events/src/eventbus.rs

@@ -7,7 +7,7 @@
 use crate::traits::{ErrorEvent, Event};
 use crate::{EventBusBarrier, EventType};
 use actix::prelude::*;
-use e3_utils::{colorize, Color, MAILBOX_LIMIT, MAILBOX_LIMIT_LARGE};
+use e3_utils::{colorize, Color, MAILBOX_LIMIT_LARGE};
 use futures_util::future::join_all;
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::fmt;
@@ -546,7 +546,7 @@ impl<E: Event> HistoryCollector<E> {
 impl<E: Event> Actor for HistoryCollector<E> {
     type Context = Context<Self>;
     fn started(&mut self, ctx: &mut Self::Context) {
-        ctx.set_mailbox_capacity(MAILBOX_LIMIT);
+        ctx.set_mailbox_capacity(MAILBOX_LIMIT_LARGE);
     }
 }
 

crates/keyshare/src/threshold_keyshare/actor.rs

@@ -57,9 +57,10 @@ use crate::actors::threshold_share_collector::{
 use crate::domain::timeout_policy::{resolve_timeout, DkgTimeoutPhase};
 use crate::domain::{
     build_decryption_key_plan, build_shares_generated_plan, generate_bfv_keypair,
-    AggregatingDecryptionKey, BfvKeypairMaterial, CollectingEncryptionKeysData, Decrypting,
-    DecryptionKeyPlan, GeneratingDecryptionProof, GeneratingThresholdShareData, KeyshareState,
-    ProofRequestData, ReadyForDecryption, ReceivedShareProofs, ThresholdKeyshareState,
+    party_ids_for_honest_addresses, AggregatingDecryptionKey, BfvKeypairMaterial,
+    CollectingEncryptionKeysData, Decrypting, DecryptionKeyPlan, GeneratingDecryptionProof,
+    GeneratingThresholdShareData, KeyshareState, ProofRequestData, ReadyForDecryption,
+    ReceivedShareProofs, ThresholdKeyshareState,
 };
 
 #[path = "recovery_state.rs"]

crates/keyshare/src/threshold_keyshare/effects/create_decryption_share.rs

@@ -11,6 +11,23 @@ impl ThresholdKeyshare {
         msg: TypedEvent<CiphertextOutputPublished>,
     ) -> Result<()> {
         let (msg, ec) = msg.into_components();
+        let state = self.state.try_get()?;
+        let Some(honest_parties) = state.honest_parties.as_ref() else {
+            info!(
+                e3_id = %state.e3_id,
+                party_id = state.party_id,
+                "Skipping decryption-share generation because the canonical honest roster is unavailable"
+            );
+            return Ok(());
+        };
+        if !honest_parties.contains(&state.party_id) {
+            info!(
+                e3_id = %state.e3_id,
+                party_id = state.party_id,
+                "Skipping decryption-share generation for a party outside the canonical honest roster"
+            );
+            return Ok(());
+        }
         let ciphertext_output = msg.ciphertext_output;
 
         // If we are already in Decrypting (or beyond), this is a duplicate

crates/keyshare/src/threshold_keyshare/effects/route_events.rs

@@ -16,18 +16,34 @@ impl Handler<InterfoldEvent> for ThresholdKeyshare {
                 self.notify_sync(ctx, TypedEvent::new(data, ec))
             }
             InterfoldEventData::PublicKeyAggregated(data) => {
+                let honest_party_ids = match party_ids_for_honest_addresses(
+                    &data.committee_addresses,
+                    &data.honest_committee_addresses,
+                ) {
+                    Ok(ids) => ids,
+                    Err(err) => {
+                        self.bus.err(EType::KeyGeneration, err);
+                        return;
+                    }
+                };
                 let committee_hash =
                     e3_committee_hash::hash_committee_addresses(&data.committee_addresses);
                 let pk = ArcBytes::from_bytes(&data.pubkey);
                 let _ = self.state.try_mutate(&ec, |mut s| {
                     s.aggregated_pk = Some(pk);
+                    s.honest_parties = Some(honest_party_ids.clone());
                     s.decryption_domain = Some(e3_committee_hash::DecryptionDomainContext {
                         interfold_address: self.interfold_address,
                         committee_hash,
                         committee_public_key: data.pk_commitment.into(),
                     });
                     Ok(s)
                 });
+                info!(
+                    e3_id = %data.e3_id,
+                    honest_party_ids = ?honest_party_ids,
+                    "Stored the canonical honest roster for decryption-share generation"
+                );
             }
             InterfoldEventData::ThresholdShareCreated(data) => {
                 let _ =

crates/keyshare/src/threshold_keyshare/state.rs

@@ -11,6 +11,7 @@
 //! It contains NO actix, persistence, bus or timer dependencies — only plain
 //! synchronous data and transition logic, which makes it directly unit-testable.
 
+use alloy::primitives::Address;
 use anyhow::{anyhow, Result};
 use e3_committee_hash::DecryptionDomainContext;
 use e3_crypto::SensitiveBytes;
@@ -30,6 +31,37 @@ use std::{
 
 use crate::domain::timeout_policy::now_unix_secs;
 
+/// Map the authoritative honest addresses to their party IDs in full committee order.
+pub(crate) fn party_ids_for_honest_addresses(
+    committee_addresses: &[Address],
+    honest_committee_addresses: &[Address],
+) -> Result<BTreeSet<u64>> {
+    if honest_committee_addresses.is_empty() {
+        return Err(anyhow!(
+            "PublicKeyAggregated contained an empty honest committee"
+        ));
+    }
+
+    let mut party_ids = BTreeSet::new();
+    for address in honest_committee_addresses {
+        let index = committee_addresses
+            .iter()
+            .position(|candidate| candidate == address)
+            .ok_or_else(|| {
+                anyhow!("Honest committee address {address} is not in the full committee")
+            })?;
+        let party_id = u64::try_from(index)
+            .map_err(|_| anyhow!("Committee index {index} does not fit in a party ID"))?;
+        if !party_ids.insert(party_id) {
+            return Err(anyhow!(
+                "Honest committee contains duplicate address {address}"
+            ));
+        }
+    }
+
+    Ok(party_ids)
+}
+
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct CollectingEncryptionKeysData {
     pub(crate) sk_bfv: SensitiveBytes,

crates/keyshare/src/threshold_keyshare/state_tests.rs

@@ -5,6 +5,7 @@
 // or FITNESS FOR A PARTICULAR PURPOSE.
 
 use super::*;
+use alloy::primitives::Address;
 use e3_events::E3id;
 
 fn arc(bytes: &[u8]) -> ArcBytes {
@@ -37,6 +38,28 @@ fn new_initialises_defaults_and_records_dkg_start() {
     assert_eq!(s.get_address(), "0xabc");
 }
 
+#[test]
+fn honest_addresses_map_to_full_committee_party_ids() {
+    let committee = vec![
+        Address::from([1u8; 20]),
+        Address::from([2u8; 20]),
+        Address::from([3u8; 20]),
+    ];
+
+    assert_eq!(
+        party_ids_for_honest_addresses(&committee, &[committee[0], committee[2]])
+            .expect("valid honest subset"),
+        BTreeSet::from([0, 2])
+    );
+}
+
+#[test]
+fn honest_address_outside_full_committee_is_rejected() {
+    let committee = vec![Address::from([1u8; 20])];
+
+    assert!(party_ids_for_honest_addresses(&committee, &[Address::from([2u8; 20])]).is_err());
+}
+
 #[test]
 fn same_branch_transition_is_always_valid() {
     // Re-entering the same phase variant must be accepted (idempotent mutations).

crates/keyshare/src/threshold_keyshare/tests.rs

@@ -14,11 +14,11 @@ use e3_data::{AutoPersist, DataStore, InMemStore, Persistable, Repository};
 use e3_events::{
     hlc_factory::HlcFactory, BusHandle, ComputeRequestKind, E3Stage, E3id, EffectsEnabled,
     EventBus, EventBusConfig, EventSource, FailureReason, HistoryCollector, InterfoldEvent,
-    InterfoldEventData, Sequencer, StoreEventRequested, StoreEventResponse, TakeEvents,
-    Unsequenced,
+    InterfoldEventData, PublicKeyAggregated, Sequencer, StoreEventRequested, StoreEventResponse,
+    TakeEvents, Unsequenced,
 };
 use e3_fhe_params::DEFAULT_BFV_PRESET;
-use std::sync::Arc;
+use std::{collections::BTreeSet, sync::Arc};
 
 #[derive(Default)]
 struct TestEventStore {
@@ -109,6 +109,43 @@ async fn start_actor() -> Result<(
     start_actor_with_state(KeyshareState::Init).await
 }
 
+#[actix::test]
+async fn public_key_aggregated_replaces_local_roster_with_global_honest_set() -> Result<()> {
+    let (actor, _history, e3_id, repo) = start_actor().await?;
+    let committee_addresses = vec![
+        Address::from([1u8; 20]),
+        Address::from([2u8; 20]),
+        Address::from([3u8; 20]),
+    ];
+    let event = PublicKeyAggregated {
+        pubkey: ArcBytes::from_bytes(b"public-key"),
+        e3_id,
+        nodes: Default::default(),
+        committee_addresses: committee_addresses.clone(),
+        honest_committee_addresses: vec![committee_addresses[0], committee_addresses[2]],
+        pk_commitment: [0u8; 32],
+        dkg_aggregator_proof: None,
+        dkg_attestation_bundle: None,
+    };
+
+    actor
+        .send(
+            InterfoldEvent::<Unsequenced>::new_with_timestamp(
+                event.into(),
+                None,
+                1,
+                None,
+                EventSource::Local,
+            )
+            .into_sequenced(1),
+        )
+        .await?;
+
+    let state = repo.read().await?.expect("persisted keyshare state");
+    assert_eq!(state.honest_parties, Some(BTreeSet::from([0, 2])));
+    Ok(())
+}
+
 async fn next_event(history: &Addr<HistoryCollector<InterfoldEvent>>) -> Result<InterfoldEvent> {
     let mut result = history.send(TakeEvents::<InterfoldEvent>::new(1)).await?;
     assert!(!result.timed_out, "timed out waiting for an event");