open-standards/joule-code/joule-code-rs/crates/joule-code-collab/src/lib.rs

//! Joule Code federated collaboration (SPEC §7, `spec/collab.md`).
//!
//! **Readiness: Research.** This crate fixes the *interface* for ownerless
//! multi-endpoint editing and ships a **single-node reference cluster** that
//! demonstrates the load-bearing shape: an ordered, de-duplicated, signature-
//! checked log of committed coding operations. It does **not** implement the
//! real binding — lockstep BFT consensus, signed-membership replication, and
//! CommunicationOS (MLS/FROST/MoQT) transport — which is future work
//! (`spec/collab.md` §3). A coding operation is structurally a
//! [Smart Byte](../../../smart-byte/) envelope; here we carry the joule-code
//! [`SignedOp`] directly.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Mutex;

use joule_code_cargo::SignedOp;
use joule_code_schema::ContentId;

pub mod protected;

/// The signed membership of a collaboration cluster. In the full binding this is
/// itself a content-addressed, signed object replicated to every member; here it
/// is the set of member verifying keys an operation's author must belong to.
#[derive(Clone, Default)]
pub struct Membership {
    /// Ed25519 verifying keys of the cluster members.
    pub members: Vec<[u8; 32]>,
}

impl Membership {
    /// A cluster of the given member keys.
    pub fn new(members: Vec<[u8; 32]>) -> Self {
        Self { members }
    }

    /// Whether `key` is a member.
    pub fn contains(&self, key: &[u8; 32]) -> bool {
        self.members.iter().any(|m| m == key)
    }
}

/// An operation committed to the cluster's agreed log, with its commit sequence.
#[derive(Clone, PartialEq, Eq)]
pub struct Committed {
    /// Monotonic position in the agreed order.
    pub sequence: u64,
    /// The committed operation.
    pub op: SignedOp,
}

/// The result of submitting an operation to the cluster.
#[derive(Debug, PartialEq, Eq)]
pub enum CommitOutcome {
    /// Newly committed at this sequence.
    Committed(u64),
    /// Already present (idempotent); the existing sequence is returned.
    Duplicate(u64),
    /// Rejected, with a stable reason token.
    Rejected(&'static str),
}

/// The collaboration surface (SPEC §7). Implementors replicate coding operations
/// to a cluster and expose the agreed, ordered log.
pub trait Collab {
    /// Propose an operation against `prior` (the tree state it was authored on).
    fn submit(&self, op: SignedOp, prior: &ContentId) -> CommitOutcome;
    /// The committed operations in agreed order.
    fn committed(&self) -> Vec<Committed>;
    /// The cluster membership.
    fn membership(&self) -> Membership;
}

/// A **single-node reference cluster** (Research). It accepts operations,
/// verifies each against its stated `prior` and an author within the membership,
/// de-duplicates by `op_id`, and appends to an ordered log with a sequence
/// number. It is deliberately NOT federated — no BFT, no networking, no
/// MLS/FROST — so it proves the API and the log invariants (order, dedup,
/// signature + membership checks) without claiming consensus it does not have.
pub struct LocalCluster {
    membership: Membership,
    log: Mutex<Vec<Committed>>,
}

impl LocalCluster {
    /// A cluster with the given membership and an empty log.
    pub fn new(membership: Membership) -> Self {
        Self {
            membership,
            log: Mutex::new(Vec::new()),
        }
    }

    fn find_seq(log: &[Committed], op_id: &ContentId) -> Option<u64> {
        log.iter()
            .find(|c| c.op.op.op_id().ok().as_ref() == Some(op_id))
            .map(|c| c.sequence)
    }
}

impl Collab for LocalCluster {
    fn submit(&self, op: SignedOp, prior: &ContentId) -> CommitOutcome {
        // The author must be a cluster member.
        if !self.membership.contains(&op.op.author) {
            return CommitOutcome::Rejected("author_not_in_membership");
        }
        // The operation must verify against the stated prior (binds op_id ‖ prior).
        if op.verify(prior).is_err() {
            return CommitOutcome::Rejected("signature_invalid");
        }
        let op_id = match op.op.op_id() {
            Ok(id) => id,
            Err(_) => return CommitOutcome::Rejected("non_canonical_op"),
        };
        let Ok(mut log) = self.log.lock() else {
            return CommitOutcome::Rejected("lock_poisoned");
        };
        if let Some(seq) = Self::find_seq(&log, &op_id) {
            return CommitOutcome::Duplicate(seq);
        }
        let sequence = log.len() as u64;
        log.push(Committed { sequence, op });
        CommitOutcome::Committed(sequence)
    }

    fn committed(&self) -> Vec<Committed> {
        self.log.lock().map(|log| log.clone()).unwrap_or_default()
    }

    fn membership(&self) -> Membership {
        self.membership.clone()
    }
}

/// A **multi-node lockstep BFT cluster** (Research) — the substrate's consensus
/// model: *"lockstep deterministic simulation with BFT supermajority commit"*
/// (SPEC §7, `spec/collab.md`). `n` nodes step in rounds; a coding operation
/// commits only when at least a Byzantine supermajority (`2f+1` of `n = 3f+1`)
/// of honest nodes independently validate it (signature, membership, prior-bind,
/// no replay). Because validation is deterministic and committed ops are ordered
/// by `op_id`, every honest node derives a **byte-identical log** — safety holds
/// with up to `f` Byzantine nodes; liveness is lost (nothing commits) only when
/// more than `f` are faulty.
///
/// This is the genuine consensus core, in-process and deterministic — it does
/// not include the layers above it: MLS group keying, FROST threshold signatures
/// for protected operations, and CommunicationOS (MoQT) network transport. Those
/// are the remaining federation work; the agreement protocol is here and tested.
pub struct LockstepCluster {
    membership: Membership,
    n: usize,
    faulty: BTreeSet<usize>,
    /// Per-node committed logs. Honest nodes stay byte-identical by construction.
    logs: Vec<Vec<Committed>>,
}

impl LockstepCluster {
    /// A cluster of `n` nodes (`n = 3f+1` recommended) with the given Byzantine
    /// node indices and signed membership.
    pub fn new(membership: Membership, n: usize, faulty: impl IntoIterator<Item = usize>) -> Self {
        Self {
            membership,
            n,
            faulty: faulty.into_iter().filter(|&i| i < n).collect(),
            logs: vec![Vec::new(); n],
        }
    }

    /// The Byzantine fault tolerance `f = floor((n-1)/3)`.
    pub fn f(&self) -> usize {
        (self.n.saturating_sub(1)) / 3
    }

    /// The commit threshold `2f+1`.
    pub fn supermajority(&self) -> usize {
        2 * self.f() + 1
    }

    fn first_honest(&self) -> Option<usize> {
        (0..self.n).find(|i| !self.faulty.contains(i))
    }

    /// The agreed log (any honest node's; all honest nodes are identical).
    pub fn agreed_log(&self) -> &[Committed] {
        match self.first_honest() {
            Some(i) => &self.logs[i],
            None => &[],
        }
    }

    /// Whether every honest node holds an identical log (the safety invariant).
    pub fn honest_logs_agree(&self) -> bool {
        let honest: Vec<&Vec<Committed>> = (0..self.n)
            .filter(|i| !self.faulty.contains(i))
            .map(|i| &self.logs[i])
            .collect();
        honest.windows(2).all(|w| w[0] == w[1])
    }

    /// Run one lockstep round over a batch of proposed operations. Each honest
    /// node independently validates every proposal; an op commits iff at least
    /// `2f+1` honest nodes accept it. Committed ops are appended in `op_id` order
    /// to every honest node's log with consecutive sequences. Returns the
    /// per-proposal outcome.
    pub fn submit_round(&mut self, proposals: Vec<(SignedOp, ContentId)>) -> Vec<CommitOutcome> {
        let honest_count = self.n - self.faulty.len();
        let threshold = self.supermajority();

        // op_ids already committed (honest log is authoritative).
        let mut seen: BTreeSet<String> = self
            .agreed_log()
            .iter()
            .filter_map(|c| c.op.op.op_id().ok().map(|id| id.0))
            .collect();

        // Decide each proposal in order; record op_ids of those that will commit.
        enum Decision {
            Rejected(&'static str),
            Duplicate,
            Commit(ContentId, SignedOp),
        }
        let mut decisions = Vec::with_capacity(proposals.len());
        for (op, prior) in proposals {
            // Honest nodes vote only for ops that pass deterministic validation:
            // author in membership AND signature binds op_id ‖ prior.
            let valid = self.membership.contains(&op.op.author) && op.verify(&prior).is_ok();
            let op_id = match op.op.op_id() {
                Ok(id) if valid => id,
                _ => {
                    decisions.push(Decision::Rejected("invalid"));
                    continue;
                }
            };
            if !seen.insert(op_id.0.clone()) {
                decisions.push(Decision::Duplicate);
                continue;
            }
            // Byzantine nodes cannot manufacture a commit for an invalid op
            // (honest nodes never vote for one); they can only withhold votes,
            // so an op commits exactly when the honest set meets 2f+1.
            if honest_count >= threshold {
                decisions.push(Decision::Commit(op_id, op));
            } else {
                seen.remove(&op_id.0);
                decisions.push(Decision::Rejected("no_supermajority"));
            }
        }

        // Append committing ops in deterministic op_id order; map id -> sequence.
        let mut committing: Vec<(ContentId, SignedOp)> = decisions
            .iter()
            .filter_map(|d| match d {
                Decision::Commit(id, op) => Some((id.clone(), op.clone())),
                _ => None,
            })
            .collect();
        committing.sort_by(|a, b| a.0 .0.cmp(&b.0 .0));
        let first_sequence = self.agreed_log().len() as u64;
        let mut assigned: std::collections::BTreeMap<String, u64> = Default::default();
        for (sequence, (id, op)) in (first_sequence..).zip(committing) {
            for i in 0..self.n {
                if !self.faulty.contains(&i) {
                    self.logs[i].push(Committed {
                        sequence,
                        op: op.clone(),
                    });
                }
            }
            assigned.insert(id.0, sequence);
        }

        // Outcomes aligned 1:1 with the proposals.
        decisions
            .into_iter()
            .map(|d| match d {
                Decision::Rejected(reason) => CommitOutcome::Rejected(reason),
                Decision::Duplicate => CommitOutcome::Duplicate(0),
                Decision::Commit(id, _) => CommitOutcome::Committed(assigned[&id.0]),
            })
            .collect()
    }
}

/// A single node's view of one lockstep BFT round, for a **real distributed**
/// binding over a network transport.
///
/// Unlike [`LockstepCluster`] — which simulates every node inside one process
/// and holds all their logs — a `RoundNode` holds only *this* node's state and
/// decides purely from the messages it receives over the wire. The safety
/// invariant is structural, not coordinated: validation is deterministic and
/// commits are applied in `op_id` order, so any two honest nodes that received
/// the same proposal set and the same per-node vote sets compute a
/// **byte-identical** log — with no leader, no view-change, no agreement round
/// beyond a single all-to-all proposal + vote exchange.
///
/// The transport (e.g. `joule-moq` over QUIC) is responsible only for reliably
/// broadcasting two messages per round — this node's proposals and its
/// [`vote`](RoundNode::vote) set — and for the round barrier; the commit
/// decision is local and deterministic.
pub struct RoundNode {
    index: usize,
    membership: Membership,
    n: usize,
    log: Vec<Committed>,
}

impl RoundNode {
    /// A node `index` of `n` over the signed `membership`.
    pub fn new(index: usize, membership: Membership, n: usize) -> Self {
        Self { index, membership, n, log: Vec::new() }
    }

    pub fn index(&self) -> usize {
        self.index
    }

    /// The Byzantine fault tolerance `f = floor((n-1)/3)`.
    pub fn f(&self) -> usize {
        self.n.saturating_sub(1) / 3
    }

    /// The commit threshold `2f+1`.
    pub fn supermajority(&self) -> usize {
        2 * self.f() + 1
    }

    /// This node's committed log so far.
    pub fn log(&self) -> &[Committed] {
        &self.log
    }

    /// Deterministic validation, applied identically by every honest node: the
    /// author is a member and the signature binds `op_id ‖ prior`.
    pub fn validates(&self, op: &SignedOp, prior: &ContentId) -> bool {
        self.membership.contains(&op.op.author) && op.verify(prior).is_ok()
    }

    fn committed_ids(&self) -> BTreeSet<String> {
        self.log.iter().filter_map(|c| c.op.op.op_id().ok().map(|id| id.0)).collect()
    }

    /// This node's vote for the round: the `op_id`s of the proposals it accepts
    /// (valid and not already committed). Broadcast verbatim to every peer.
    pub fn vote(&self, proposals: &[(SignedOp, ContentId)]) -> BTreeSet<String> {
        let seen = self.committed_ids();
        let mut accepted = BTreeSet::new();
        for (op, prior) in proposals {
            if let Ok(id) = op.op.op_id() {
                if !seen.contains(&id.0) && self.validates(op, prior) {
                    accepted.insert(id.0);
                }
            }
        }
        accepted
    }

    /// Commit the round from the full proposal set and **every** node's vote set
    /// (this node's included). An op commits iff at least `2f+1` distinct nodes
    /// voted to accept it — Byzantine nodes can only withhold votes, never
    /// manufacture one, since this node independently re-validates and never
    /// commits an op it deems invalid. Committed ops are appended in `op_id`
    /// order with consecutive sequences; returns those newly committed.
    pub fn commit_round(
        &mut self,
        proposals: &[(SignedOp, ContentId)],
        votes: &[BTreeSet<String>],
    ) -> Vec<Committed> {
        let threshold = self.supermajority();
        let seen = self.committed_ids();

        // Tally accept votes per op_id across all nodes.
        let mut tally: BTreeMap<String, usize> = BTreeMap::new();
        for vote_set in votes {
            for id in vote_set {
                *tally.entry(id.clone()).or_default() += 1;
            }
        }

        // Dedup proposals by op_id (first wins; honest proposals of the same op
        // are byte-identical anyway).
        let mut by_id: BTreeMap<String, (SignedOp, ContentId)> = BTreeMap::new();
        for (op, prior) in proposals {
            if let Ok(id) = op.op.op_id() {
                by_id.entry(id.0).or_insert_with(|| (op.clone(), prior.clone()));
            }
        }

        // Committing ids: unseen, locally valid, and at/above the vote threshold.
        let mut committing: Vec<(String, SignedOp)> = by_id
            .into_iter()
            .filter(|(id, (op, prior))| {
                !seen.contains(id)
                    && self.validates(op, prior)
                    && tally.get(id).copied().unwrap_or(0) >= threshold
            })
            .map(|(id, (op, _prior))| (id, op))
            .collect();
        committing.sort_by(|a, b| a.0.cmp(&b.0));

        let first_sequence = self.log.len() as u64;
        let mut newly = Vec::new();
        for (sequence, (_id, op)) in (first_sequence..).zip(committing) {
            let committed = Committed { sequence, op };
            self.log.push(committed.clone());
            newly.push(committed);
        }
        newly
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use ed25519_dalek::SigningKey;
    use joule_code_cargo::{Anchor, CodingOp, Edit};
    use joule_code_schema::{Tier, ALG_ED25519};

    fn signed_op(sk: &SigningKey, prior: &str, intent: &str) -> (SignedOp, ContentId) {
        let prior = ContentId(prior.into());
        let op = CodingOp {
            alg: ALG_ED25519,
            prior: prior.clone(),
            edits: vec![Edit {
                path: "src/lib.rs".into(),
                anchor: Anchor::ByteRange { start: 0, end: 0 },
                replacement: intent.as_bytes().to_vec(),
            }],
            author: sk.verifying_key().to_bytes(),
            intent: intent.into(),
            tier: Tier::L4,
        };
        (op.sign(sk).expect("sign"), prior)
    }

    #[test]
    fn commits_in_order_and_dedups() {
        let sk = SigningKey::from_bytes(&[3u8; 32]);
        let cluster = LocalCluster::new(Membership::new(vec![sk.verifying_key().to_bytes()]));

        let (op_a, prior_a) = signed_op(&sk, "b3:AAAA", "add a");
        let (op_b, prior_b) = signed_op(&sk, "b3:AAAA", "add b");

        assert_eq!(cluster.submit(op_a.clone(), &prior_a), CommitOutcome::Committed(0));
        assert_eq!(cluster.submit(op_b, &prior_b), CommitOutcome::Committed(1));
        // Re-submitting an identical op is idempotent.
        assert_eq!(cluster.submit(op_a, &prior_a), CommitOutcome::Duplicate(0));

        let log = cluster.committed();
        assert_eq!(log.len(), 2);
        assert_eq!(log[0].sequence, 0);
        assert_eq!(log[1].sequence, 1);
    }

    #[test]
    fn rejects_non_member_author() {
        let member = SigningKey::from_bytes(&[3u8; 32]);
        let outsider = SigningKey::from_bytes(&[9u8; 32]);
        let cluster = LocalCluster::new(Membership::new(vec![member.verifying_key().to_bytes()]));

        let (op, prior) = signed_op(&outsider, "b3:AAAA", "sneak");
        assert_eq!(
            cluster.submit(op, &prior),
            CommitOutcome::Rejected("author_not_in_membership")
        );
        assert!(cluster.committed().is_empty());
    }

    #[test]
    fn rejects_op_replayed_onto_different_prior() {
        let sk = SigningKey::from_bytes(&[3u8; 32]);
        let cluster = LocalCluster::new(Membership::new(vec![sk.verifying_key().to_bytes()]));
        let (op, _prior) = signed_op(&sk, "b3:AAAA", "x");
        // Same signed op, but presented against a different base: must reject.
        let other_prior = ContentId("b3:CCCC".into());
        assert_eq!(
            cluster.submit(op, &other_prior),
            CommitOutcome::Rejected("signature_invalid")
        );
    }

    // ---- LockstepCluster (BFT consensus) ----

    #[test]
    fn lockstep_commits_and_all_honest_nodes_agree() {
        let sk = SigningKey::from_bytes(&[5u8; 32]);
        let mut cluster =
            LockstepCluster::new(Membership::new(vec![sk.verifying_key().to_bytes()]), 4, []);
        assert_eq!(cluster.f(), 1);
        assert_eq!(cluster.supermajority(), 3);

        let (op, prior) = signed_op(&sk, "b3:AAAA", "x");
        let out = cluster.submit_round(vec![(op, prior)]);
        assert_eq!(out, vec![CommitOutcome::Committed(0)]);
        assert_eq!(cluster.agreed_log().len(), 1);
        assert!(cluster.honest_logs_agree());
    }

    #[test]
    fn lockstep_tolerates_one_byzantine_of_four() {
        let sk = SigningKey::from_bytes(&[5u8; 32]);
        // n=4, f=1: 3 honest == 2f+1, so a valid op still commits.
        let mut cluster =
            LockstepCluster::new(Membership::new(vec![sk.verifying_key().to_bytes()]), 4, [3]);
        let (op, prior) = signed_op(&sk, "b3:AAAA", "x");
        assert_eq!(cluster.submit_round(vec![(op, prior)]), vec![CommitOutcome::Committed(0)]);
        assert!(cluster.honest_logs_agree());
        assert_eq!(cluster.agreed_log().len(), 1);
    }

    #[test]
    fn lockstep_loses_liveness_beyond_f_faulty() {
        let sk = SigningKey::from_bytes(&[5u8; 32]);
        // n=7, f=2, 2f+1=5; with 3 faulty only 4 honest remain (< 5) → no commit.
        let mut cluster = LockstepCluster::new(
            Membership::new(vec![sk.verifying_key().to_bytes()]),
            7,
            [4, 5, 6],
        );
        let (op, prior) = signed_op(&sk, "b3:AAAA", "x");
        assert_eq!(
            cluster.submit_round(vec![(op, prior)]),
            vec![CommitOutcome::Rejected("no_supermajority")]
        );
        assert!(cluster.agreed_log().is_empty());
        assert!(cluster.honest_logs_agree());
    }

    #[test]
    fn lockstep_rejects_non_member_even_with_quorum() {
        let member = SigningKey::from_bytes(&[5u8; 32]);
        let outsider = SigningKey::from_bytes(&[9u8; 32]);
        let mut cluster =
            LockstepCluster::new(Membership::new(vec![member.verifying_key().to_bytes()]), 4, []);
        let (op, prior) = signed_op(&outsider, "b3:AAAA", "sneak");
        assert_eq!(
            cluster.submit_round(vec![(op, prior)]),
            vec![CommitOutcome::Rejected("invalid")]
        );
        assert!(cluster.agreed_log().is_empty());
    }

    #[test]
    fn lockstep_orders_a_round_deterministically_across_nodes() {
        let sk = SigningKey::from_bytes(&[5u8; 32]);
        let mut cluster =
            LockstepCluster::new(Membership::new(vec![sk.verifying_key().to_bytes()]), 4, [2]);
        let (op_a, prior_a) = signed_op(&sk, "b3:AAAA", "alpha");
        let (op_b, prior_b) = signed_op(&sk, "b3:AAAA", "beta");
        let out = cluster.submit_round(vec![(op_a, prior_a), (op_b, prior_b)]);
        // Both commit; sequences are 0 and 1 (assignment order is op_id-sorted,
        // but each proposal gets its own sequence).
        assert!(matches!(out[0], CommitOutcome::Committed(_)));
        assert!(matches!(out[1], CommitOutcome::Committed(_)));
        assert_eq!(cluster.agreed_log().len(), 2);
        assert_eq!(cluster.agreed_log()[0].sequence, 0);
        assert_eq!(cluster.agreed_log()[1].sequence, 1);
        // The committed op_ids are in ascending order (deterministic).
        let id0 = cluster.agreed_log()[0].op.op.op_id().unwrap().0;
        let id1 = cluster.agreed_log()[1].op.op.op_id().unwrap().0;
        assert!(id0 < id1);
        assert!(cluster.honest_logs_agree());
    }

    // ---- RoundNode (distributed per-node BFT round) ----

    /// Drive one round across **independent** node states (not the all-knowing
    /// sim): each node validates and votes on its own, the votes are exchanged,
    /// and every honest node commits locally. A Byzantine node that withholds
    /// its votes does not prevent commitment, and all honest logs are identical.
    #[test]
    fn round_nodes_reach_byte_identical_logs_with_a_byzantine_node() {
        // n = 4, f = 1, threshold = 3. Members 0..4; node 3 is Byzantine.
        let keys: Vec<SigningKey> = (0..4).map(|i| SigningKey::from_bytes(&[10 + i as u8; 32])).collect();
        let membership = Membership::new(keys.iter().map(|k| k.verifying_key().to_bytes()).collect());
        let n = 4;

        // Two valid proposals (authored by members) plus one from an outsider
        // that every honest node must refuse to vote for.
        let outsider = SigningKey::from_bytes(&[200u8; 32]);
        let (op_a, prior_a) = signed_op(&keys[0], "b3:AAAA", "alpha");
        let (op_b, prior_b) = signed_op(&keys[1], "b3:AAAA", "beta");
        let (op_bad, prior_bad) = signed_op(&outsider, "b3:AAAA", "intrusion");
        let proposals = vec![
            (op_a.clone(), prior_a),
            (op_b.clone(), prior_b),
            (op_bad, prior_bad),
        ];

        let mut nodes: Vec<RoundNode> =
            (0..n).map(|i| RoundNode::new(i, membership.clone(), n)).collect();
        assert_eq!(nodes[0].supermajority(), 3);

        // Each node votes independently; the Byzantine node (3) withholds.
        let votes: Vec<BTreeSet<String>> = (0..n)
            .map(|i| {
                if i == 3 {
                    BTreeSet::new()
                } else {
                    nodes[i].vote(&proposals)
                }
            })
            .collect();

        // Honest nodes vote for the two valid ops only (the outsider's is refused).
        let bad_id = op_bad_id(&outsider);
        for vote in votes.iter().take(3) {
            assert_eq!(vote.len(), 2, "honest nodes accept exactly the two valid ops");
            assert!(!vote.contains(&bad_id), "no honest node votes for a non-member's op");
        }

        // Every node commits from the same proposals + the same vote sets.
        for node in nodes.iter_mut() {
            node.commit_round(&proposals, &votes);
        }

        // The two valid ops reached 3 accept votes (= threshold) and committed;
        // the outsider's op got 0 and did not.
        let honest_logs: Vec<Vec<(u64, String)>> = nodes
            .iter()
            .take(3)
            .map(|node| {
                node.log()
                    .iter()
                    .map(|c| (c.sequence, c.op.op.op_id().unwrap().0))
                    .collect()
            })
            .collect();
        assert_eq!(honest_logs[0].len(), 2, "two ops committed");
        assert!(honest_logs.windows(2).all(|w| w[0] == w[1]), "honest logs are byte-identical");
        let committed_ids: BTreeSet<&String> =
            honest_logs[0].iter().map(|(_, id)| id).collect();
        assert!(!committed_ids.contains(&bad_id), "the non-member's op never commits");

        // Idempotent: re-running the same round commits nothing new.
        let again = nodes[0].commit_round(&proposals, &votes);
        assert!(again.is_empty(), "a replayed round is a no-op");
    }

    fn op_bad_id(outsider: &SigningKey) -> String {
        let (op, _) = signed_op(outsider, "b3:AAAA", "intrusion");
        op.op.op_id().unwrap().0
    }
}