feat(starfish-core,iota-core): emit misbehavior counts to iota-core (#11534)

# Description of change

Stacks on top of #11531. Wires the producer-side misbehavior store
(landed in #10088 + #11531) into the consumer in `iota-core` so that
the `Scoreboard` / `ReportAggregator` (from #10779) finally receives
per-authority misbehavior signal from Starfish.

Each `CommittedSubDag` now carries a per-authority absolute snapshot of
`persisted + in_memory` counts from `MisbehaviorStore` at commit time.
Consumers diff against their own last-seen state if they want deltas —
that responsibility doesn't belong in Starfish, and the sum is invariant
across the eviction-time move between buckets, so the snapshot is
race-free relative to flush.

Producer (`starfish-core`):
- `MisbehaviorStore::snapshot_totals()` — returns the per-authority
  absolute totals.
- `CommittedSubDag` gets a new `misbehavior_counts:
Vec<MisbehaviorCountsV1>`
  field, threaded at all three production `CommittedSubDag::new` call
  sites (`commit_solidifier`, `commit_observer`, `commit_syncer/fast`)
  via the `DagState`-owned `MisbehaviorStore`.
- `CommittedSubDag` is **not** serialized over the wire (local
  `CommitConsumer` channel only), so the added `Vec` is an in-process
  cost only.

Consumer (`iota-core`):
- Overrides `ConsensusOutputAPI::misbehavior_counts()` on
  `starfish_core::CommittedSubDag` to transpose the per-authority
  struct-of-fields snapshot into the four per-field vecs expected by
  `ConsensusOutputMisbehaviorCounts`. Downstream wiring in
  `consensus_handler` and the report aggregator was already in place
  and waiting for non-empty data.

Also folds in a standalone fix: `starfish-core` now declares its
`iota-sdk-types` `serde` feature explicitly. Workspace builds were
masking the missing feature via unification through `iota-types`;
`cargo check -p starfish-core` alone fails without it.

## Links to any relevant issues

fixes iotaledger/iota-private#406
Part of iotaledger/iota-private#173

## How the change has been tested

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [x] Patch-specific tests (correctness, functionality coverage)
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] I have checked that new and existing unit tests pass locally with
my changes

Local verification:
- `cargo ci-clippy` — clean.
- `cargo nextest run -p starfish-core --lib` — new
`test_snapshot_totals_sums_persisted_and_in_memory` passes; existing
misbehavior tests pass.
- `cargo nextest run -p iota-core --lib consensus_output_api::tests` —
new transpose + empty-snapshot tests pass.

Author: piotrm50

crates/iota-core/src/consensus_handler.rs

@@ -883,6 +883,7 @@ mod tests {
             leader_header.timestamp_ms(),
             CommitRef::new(10, CommitDigest::MIN),
             vec![],
+            vec![],
         );
 
         // Test that the consensus handler respects backpressure.

crates/iota-core/src/consensus_types/consensus_output_api.rs

@@ -5,6 +5,7 @@
 use std::{collections::BTreeMap, fmt::Display};
 
 use iota_types::{digests::ConsensusCommitDigest, messages_consensus::ConsensusTransaction};
+use itertools::Itertools as _;
 
 use crate::consensus_types::AuthorityIndex;
 /// A list of tuples of:
@@ -135,4 +136,90 @@ impl ConsensusOutputAPI for starfish_core::CommittedSubDag {
             });
         num_of_committed_headers.into_iter().collect()
     }
+
+    fn misbehavior_counts(&self) -> ConsensusOutputMisbehaviorCounts {
+        let (faulty_blocks_provable, faulty_blocks_unprovable, missing_proposals, equivocations) =
+            self.misbehavior_counts
+                .iter()
+                .map(|counts| match counts {
+                    starfish_core::MisbehaviorCounts::V1(v1) => (
+                        v1.faulty_blocks_provable,
+                        v1.faulty_blocks_unprovable,
+                        v1.missing_proposals,
+                        v1.equivocations,
+                    ),
+                })
+                .multiunzip();
+        ConsensusOutputMisbehaviorCounts {
+            faulty_blocks_provable,
+            faulty_blocks_unprovable,
+            missing_proposals,
+            equivocations,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use starfish_core::{
+        BlockRef, CommitDigest, CommitRef, CommittedSubDag, MisbehaviorCounts, MisbehaviorCountsV1,
+        VerifiedBlockHeader,
+    };
+
+    use super::*;
+
+    #[test]
+    fn test_misbehavior_counts_transposes_per_authority_to_per_field_vecs() {
+        let counts = vec![
+            MisbehaviorCounts::V1(MisbehaviorCountsV1 {
+                faulty_blocks_provable: 1,
+                faulty_blocks_unprovable: 2,
+                missing_proposals: 3,
+                equivocations: 4,
+            }),
+            MisbehaviorCounts::V1(MisbehaviorCountsV1 {
+                faulty_blocks_provable: 10,
+                faulty_blocks_unprovable: 20,
+                missing_proposals: 30,
+                equivocations: 40,
+            }),
+            MisbehaviorCounts::default(),
+        ];
+
+        let subdag = CommittedSubDag::new(
+            BlockRef::MIN,
+            Vec::<VerifiedBlockHeader>::new(),
+            vec![],
+            vec![],
+            0,
+            CommitRef::new(1, CommitDigest::MIN),
+            vec![],
+            counts,
+        );
+
+        let out = subdag.misbehavior_counts();
+        assert_eq!(out.faulty_blocks_provable, vec![1, 10, 0]);
+        assert_eq!(out.faulty_blocks_unprovable, vec![2, 20, 0]);
+        assert_eq!(out.missing_proposals, vec![3, 30, 0]);
+        assert_eq!(out.equivocations, vec![4, 40, 0]);
+    }
+
+    #[test]
+    fn test_misbehavior_counts_empty_snapshot_produces_empty_vecs() {
+        let subdag = CommittedSubDag::new(
+            BlockRef::MIN,
+            Vec::<VerifiedBlockHeader>::new(),
+            vec![],
+            vec![],
+            0,
+            CommitRef::new(1, CommitDigest::MIN),
+            vec![],
+            vec![],
+        );
+        let out = subdag.misbehavior_counts();
+        assert!(out.faulty_blocks_provable.is_empty());
+        assert!(out.faulty_blocks_unprovable.is_empty());
+        assert!(out.missing_proposals.is_empty());
+        assert!(out.equivocations.is_empty());
+    }
 }

crates/iota-core/src/unit_tests/starfish_manager_tests.rs

@@ -201,6 +201,7 @@ async fn test_starfish_consensus_handler_handles_older_commits() {
                 1000 + commit_idx * 1000,
                 StarfishCommitRef::new(commit_idx as u32, StarfishCommitDigest::MIN),
                 vec![],
+                vec![],
             )
         })
         .collect();

crates/starfish/core/src/commit.rs

@@ -25,6 +25,7 @@ use crate::{
     },
     context::Context,
     leader_scoring::ReputationScores,
+    misbehavior_store::MisbehaviorCounts,
     storage::Store,
     transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI as _, TransactionRef},
 };
@@ -578,6 +579,10 @@ pub struct CommittedSubDag {
     pub base: SubDagBase,
     /// All the committed blocks that are part of this sub-dag
     pub transactions: Vec<VerifiedTransactions>,
+    /// Absolute per-authority misbehavior counts (`persisted + in_memory`)
+    /// snapshotted from `MisbehaviorStore` at emission. Indexed by
+    /// `AuthorityIndex`; consumers diff for deltas.
+    pub misbehavior_counts: Vec<MisbehaviorCounts>,
 }
 
 impl CommittedSubDag {
@@ -590,6 +595,7 @@ impl CommittedSubDag {
         timestamp_ms: BlockTimestampMs,
         commit_ref: CommitRef,
         reputation_scores_desc: Vec<(AuthorityIndex, u64)>,
+        misbehavior_counts: Vec<MisbehaviorCounts>,
     ) -> Self {
         Self {
             base: SubDagBase {
@@ -601,6 +607,7 @@ impl CommittedSubDag {
                 reputation_scores_desc,
             },
             transactions,
+            misbehavior_counts,
         }
     }
 

crates/starfish/core/src/commit_observer.rs

@@ -247,6 +247,7 @@ impl CommitObserver {
             commit.timestamp_ms(),
             commit.reference(),
             reputation_scores,
+            self.dag_state.read().misbehavior_store().snapshot_totals(),
         ))
     }
 

crates/starfish/core/src/commit_solidifier.rs

@@ -198,6 +198,9 @@ impl CommitSolidifier {
                 .map(|tx| tx.expect("Transaction must exist since we checked"))
                 .collect();
 
+            // Delayed subdags see current store state, not state at leader-round
+            // commit time. Safe: counts are absolute and consumers merge-max.
+            let misbehavior_counts = dag_state.misbehavior_store().snapshot_totals();
             Ok(CommittedSubDag::new(
                 subdag.leader,
                 subdag.base.headers.clone(),
@@ -206,6 +209,7 @@ impl CommitSolidifier {
                 subdag.timestamp_ms,
                 subdag.commit_ref,
                 subdag.reputation_scores_desc.clone(),
+                misbehavior_counts,
             ))
         } else {
             Err(missing)

crates/starfish/core/src/commit_syncer/fast.rs

@@ -662,6 +662,10 @@ impl<C: NetworkClient> FastCommitSyncer<C> {
         // For fast commit sync, we use block headers refs and reputation scores from
         // the commit.
         let mut committed_subdags = Vec::new();
+        // Replayed commits share one snapshot of current store state; downstream
+        // consumers merge-max against their last-seen, so repeating absolute
+        // totals across the batch is a no-op after the first.
+        let misbehavior_counts = inner.dag_state.read().misbehavior_store().snapshot_totals();
         for commit in &commits {
             // Get block headers from the commit
             let committed_header_refs = commit.block_headers().to_vec();
@@ -684,6 +688,7 @@ impl<C: NetworkClient> FastCommitSyncer<C> {
                 commit.timestamp_ms(),
                 commit.reference(),
                 reputation_scores,
+                misbehavior_counts.clone(),
             ));
         }
 

crates/starfish/core/src/core.rs

@@ -2678,7 +2678,12 @@ mod test {
         // Record traversed headers and sequenced transactions
         while let Some(sub_dag) = commit_receiver_own.recv().await {
             let sub_dag_leader_round = sub_dag.leader.round;
-            let CommittedSubDag { base, transactions } = sub_dag;
+            let CommittedSubDag {
+                base,
+                transactions,
+                misbehavior_counts,
+            } = sub_dag;
+            assert_eq!(misbehavior_counts.len(), committee_size);
 
             for block_ref in &base.committed_header_refs {
                 existing_headers.insert(*block_ref);

crates/starfish/core/src/dag_state.rs

@@ -2503,7 +2503,6 @@ impl DagState {
         self.pending_acknowledgments = acknowledgments.into_iter().collect::<BTreeSet<_>>();
     }
 
-    #[cfg(test)]
     pub(crate) fn misbehavior_store(&self) -> &MisbehaviorStore {
         &self.misbehavior_store
     }

crates/starfish/core/src/leader_schedule.rs

@@ -841,6 +841,7 @@ mod tests {
                 context.clock.timestamp_utc_ms(),
                 CommitRef::new(1, CommitDigest::MIN),
                 vec![],
+                vec![],
             )
             .base,
         ];
@@ -944,6 +945,7 @@ mod tests {
                 context.clock.timestamp_utc_ms(),
                 last_commit.reference(),
                 vec![],
+                vec![],
             )
             .base,
         ];