meerkat-mob 0.5.2

//! Roster tracking for active meerkats in a mob.
//!
//! The `Roster` is a projection built from `MeerkatSpawned`, `MeerkatRetired`,
//! `PeersWired`, and `PeersUnwired` events.
//!
//! Projection contract:
//! - `Roster` is not canonical transport/comms trust truth.
//! - Canonical wiring side effects (trust edges, notifications, lock discipline)
//!   are performed by runtime orchestration.
//! - `Roster` stores the event/projected peer graph used for read surfaces and
//!   consistency checks.
//! - `wire`/`unwire`/`remove` are projection mutations only.

use crate::event::{MemberRef, MobEvent, MobEventKind};
use crate::ids::{MeerkatId, ProfileName};
use crate::runtime_mode::MobRuntimeMode;
use meerkat_core::comms::TrustedPeerSpec;
use meerkat_core::types::SessionId;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::time::SystemTime;

/// Lifecycle state for a roster member.
///
/// `Retiring` is runtime-only — event projection never produces it
/// (`MeerkatSpawned` creates `Active`; `MeerkatRetired` removes entirely).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum MemberState {
    #[default]
    Active,
    Retiring,
}

/// Resolution state for a member's initial autonomous kickoff turn.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum MobMemberKickoffPhase {
    Pending,
    Starting,
    CallbackPending,
    Started,
    Failed,
    Cancelled,
}

/// Durable projected snapshot of a member's kickoff state.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct MobMemberKickoffSnapshot {
    pub phase: MobMemberKickoffPhase,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    pub updated_at: SystemTime,
}

/// A single meerkat entry in the roster.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RosterEntry {
    /// Unique meerkat identifier.
    pub meerkat_id: MeerkatId,
    /// Profile name this meerkat was spawned from.
    pub profile: ProfileName,
    /// Backend-neutral identity for this meerkat.
    pub member_ref: MemberRef,
    /// Runtime mode for this member.
    #[serde(default)]
    pub runtime_mode: MobRuntimeMode,
    /// Public comms peer identifier when this member exposes a comms runtime.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peer_id: Option<String>,
    /// Lifecycle state (Active or Retiring).
    #[serde(default)]
    pub state: MemberState,
    /// Set of peer meerkat IDs this meerkat is wired to.
    pub wired_to: BTreeSet<MeerkatId>,
    /// Trusted specs for external peers keyed by their projected peer name.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub external_peer_specs: BTreeMap<MeerkatId, TrustedPeerSpec>,
    /// Application-defined labels for this member.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub labels: BTreeMap<String, String>,
    /// Projected kickoff state for autonomous initial turn resolution.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub kickoff: Option<MobMemberKickoffSnapshot>,
    /// Effective profile override from `SpawnTooling::Profile` resolution.
    ///
    /// When a member is spawned with an explicit tooling profile (inline or realm),
    /// the resolved profile is stored here so respawn/restore can use it instead
    /// of re-resolving from the definition. `None` means the definition profile
    /// (keyed by `self.profile`) is authoritative.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub effective_profile_override: Option<crate::profile::Profile>,
}

/// Directed projection presence state for an undirected peer edge.
///
/// This reflects only roster-projected `wired_to` membership; it does not
/// imply comms trust/runtime side effects.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum WiringEdgeState {
    Absent,
    AOnly,
    BOnly,
    Bidirectional,
}

/// Parameters for adding a new member to the roster.
pub struct RosterAddEntry {
    pub meerkat_id: MeerkatId,
    pub profile: ProfileName,
    pub runtime_mode: MobRuntimeMode,
    pub member_ref: MemberRef,
    pub peer_id: Option<String>,
    pub labels: BTreeMap<String, String>,
    /// Effective profile override from `SpawnTooling::Profile` resolution.
    pub effective_profile_override: Option<crate::profile::Profile>,
}

/// Tracks active meerkats and their wiring in a mob.
///
/// Built by replaying events. Shared via `Arc<RwLock<Roster>>` between
/// the actor (writes) and handle (reads).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Roster {
    entries: BTreeMap<MeerkatId, RosterEntry>,
}

impl Roster {
    /// Create an empty roster.
    pub fn new() -> Self {
        Self::default()
    }

    /// Build a roster from a sequence of mob events.
    pub fn project(events: &[MobEvent]) -> Self {
        let mut roster = Self::new();
        for event in events {
            roster.apply(event);
        }
        roster
    }

    /// Apply a single event to update roster state.
    pub fn apply(&mut self, event: &MobEvent) {
        match &event.kind {
            MobEventKind::MeerkatSpawned {
                meerkat_id,
                role,
                runtime_mode,
                member_ref,
                labels,
            } => {
                self.add(RosterAddEntry {
                    meerkat_id: meerkat_id.clone(),
                    profile: role.clone(),
                    runtime_mode: *runtime_mode,
                    member_ref: member_ref.clone(),
                    peer_id: None,
                    labels: labels.clone(),
                    effective_profile_override: None,
                });
            }
            MobEventKind::MeerkatRetired { meerkat_id, .. } => {
                self.remove(meerkat_id);
            }
            MobEventKind::PeersWired { a, b } => {
                self.wire(a, b);
            }
            MobEventKind::ExternalPeerWired { local, spec } => {
                let peer_name = MeerkatId::from(spec.name.clone());
                self.wire_external(local, &peer_name, spec.clone());
            }
            MobEventKind::ExternalPeerUnwired { local, peer_name } => {
                self.unwire_external(local, peer_name);
            }
            MobEventKind::PeersUnwired { a, b } => {
                self.unwire(a, b);
            }
            MobEventKind::MeerkatKickoffUpdated {
                meerkat_id,
                kickoff,
            } => {
                self.set_kickoff(meerkat_id, Some(kickoff.clone()));
            }
            MobEventKind::MobReset => {
                self.entries.clear();
            }
            _ => {}
        }
    }

    /// Add a meerkat to the roster.
    pub fn add(&mut self, entry: RosterAddEntry) -> bool {
        let meerkat_id = entry.meerkat_id.clone();
        self.entries
            .insert(
                meerkat_id,
                RosterEntry {
                    meerkat_id: entry.meerkat_id,
                    profile: entry.profile,
                    member_ref: entry.member_ref,
                    runtime_mode: entry.runtime_mode,
                    peer_id: entry.peer_id,
                    state: MemberState::default(),
                    wired_to: BTreeSet::new(),
                    external_peer_specs: BTreeMap::new(),
                    labels: entry.labels,
                    kickoff: None,
                    effective_profile_override: entry.effective_profile_override,
                },
            )
            .is_none()
    }

    /// Remove a meerkat from the roster projection.
    ///
    /// Projection-only behavior: this prunes `wired_to` references from peers
    /// but does not perform runtime trust-edge teardown.
    pub fn remove(&mut self, meerkat_id: &MeerkatId) {
        if self.entries.remove(meerkat_id).is_some() {
            // Remove this meerkat from all other entries' wired_to sets
            for entry in self.entries.values_mut() {
                entry.wired_to.remove(meerkat_id);
                entry.external_peer_specs.remove(meerkat_id);
            }
        }
    }

    /// Wire two meerkats in the roster projection (bidirectional set update).
    ///
    /// Projection-only behavior: caller is responsible for canonical runtime
    /// trust establishment and lifecycle/event ordering.
    pub fn wire(&mut self, a: &MeerkatId, b: &MeerkatId) {
        if let Some(entry_a) = self.entries.get_mut(a) {
            entry_a.wired_to.insert(b.clone());
        }
        if let Some(entry_b) = self.entries.get_mut(b) {
            entry_b.wired_to.insert(a.clone());
        }
    }

    /// Wire a local meerkat to an external peer name in the roster projection.
    pub fn wire_external(
        &mut self,
        local: &MeerkatId,
        peer_name: &MeerkatId,
        spec: TrustedPeerSpec,
    ) {
        if let Some(entry) = self.entries.get_mut(local) {
            entry.wired_to.insert(peer_name.clone());
            entry.external_peer_specs.insert(peer_name.clone(), spec);
        }
    }

    /// Unwire two meerkats in the roster projection (bidirectional set update).
    ///
    /// Projection-only behavior: caller is responsible for canonical runtime
    /// trust removal and lifecycle/event ordering.
    pub fn unwire(&mut self, a: &MeerkatId, b: &MeerkatId) {
        if let Some(entry_a) = self.entries.get_mut(a) {
            entry_a.wired_to.remove(b);
        }
        if let Some(entry_b) = self.entries.get_mut(b) {
            entry_b.wired_to.remove(a);
        }
    }

    /// Unwire a local meerkat from an external peer in the roster projection.
    pub fn unwire_external(&mut self, local: &MeerkatId, peer_name: &MeerkatId) {
        if let Some(entry) = self.entries.get_mut(local) {
            entry.wired_to.remove(peer_name);
            entry.external_peer_specs.remove(peer_name);
        }
    }

    /// Returns true if both members currently have reciprocal projected edges.
    pub fn has_bidirectional_edge(&self, a: &MeerkatId, b: &MeerkatId) -> bool {
        matches!(self.wiring_edge_state(a, b), WiringEdgeState::Bidirectional)
    }

    /// Returns the directed projection presence state for edge `(a, b)`.
    pub fn wiring_edge_state(&self, a: &MeerkatId, b: &MeerkatId) -> WiringEdgeState {
        let a_has_b = self
            .entries
            .get(a)
            .is_some_and(|entry| entry.wired_to.contains(b));
        let b_has_a = self
            .entries
            .get(b)
            .is_some_and(|entry| entry.wired_to.contains(a));
        match (a_has_b, b_has_a) {
            (false, false) => WiringEdgeState::Absent,
            (true, false) => WiringEdgeState::AOnly,
            (false, true) => WiringEdgeState::BOnly,
            (true, true) => WiringEdgeState::Bidirectional,
        }
    }

    /// Returns `true` when every projected edge is reciprocal and endpoint-present.
    pub fn is_wiring_projection_consistent(&self) -> bool {
        self.wiring_projection_inconsistencies().is_empty()
    }

    /// Returns canonicalized endpoint pairs with projection inconsistencies.
    ///
    /// An inconsistency means a directed edge to another local member is not
    /// reciprocated. Missing peers are treated as external projection targets.
    pub fn wiring_projection_inconsistencies(&self) -> Vec<(MeerkatId, MeerkatId)> {
        let mut inconsistencies = BTreeSet::<(MeerkatId, MeerkatId)>::new();
        for (a_id, a_entry) in &self.entries {
            for b_id in &a_entry.wired_to {
                if let Some(b_entry) = self.entries.get(b_id)
                    && !b_entry.wired_to.contains(a_id)
                {
                    let pair = if a_id <= b_id {
                        (a_id.clone(), b_id.clone())
                    } else {
                        (b_id.clone(), a_id.clone())
                    };
                    inconsistencies.insert(pair);
                }
            }
        }
        inconsistencies.into_iter().collect()
    }

    /// Debug-only assertion helper for projection consistency checks.
    pub fn debug_assert_wiring_projection_consistent(&self) {
        let _ = self;
        #[cfg(debug_assertions)]
        {
            let inconsistencies = self.wiring_projection_inconsistencies();
            debug_assert!(
                inconsistencies.is_empty(),
                "roster wiring projection is inconsistent: {inconsistencies:?}"
            );
        }
    }

    /// Get a roster entry by meerkat ID.
    pub fn get(&self, meerkat_id: &MeerkatId) -> Option<&RosterEntry> {
        self.entries.get(meerkat_id)
    }

    /// Update the member reference for an existing meerkat.
    pub fn set_member_ref(&mut self, meerkat_id: &MeerkatId, member_ref: MemberRef) -> bool {
        if let Some(entry) = self.entries.get_mut(meerkat_id) {
            entry.member_ref = member_ref;
            return true;
        }
        false
    }

    /// Update the bridge session ID while preserving backend-specific identity.
    pub fn set_session_id(&mut self, meerkat_id: &MeerkatId, session_id: SessionId) -> bool {
        if let Some(entry) = self.entries.get_mut(meerkat_id) {
            entry.member_ref = match &entry.member_ref {
                MemberRef::Session { .. } => MemberRef::Session { session_id },
                MemberRef::BackendPeer {
                    peer_id, address, ..
                } => MemberRef::BackendPeer {
                    peer_id: peer_id.clone(),
                    address: address.clone(),
                    session_id: Some(session_id),
                },
            };
            return true;
        }
        false
    }

    /// Update the resolved comms peer id for an existing meerkat.
    pub fn set_peer_id(&mut self, meerkat_id: &MeerkatId, peer_id: Option<String>) -> bool {
        if let Some(entry) = self.entries.get_mut(meerkat_id) {
            entry.peer_id = peer_id;
            return true;
        }
        false
    }

    /// Update the projected kickoff state for an existing meerkat.
    pub fn set_kickoff(
        &mut self,
        meerkat_id: &MeerkatId,
        kickoff: Option<MobMemberKickoffSnapshot>,
    ) -> bool {
        if let Some(entry) = self.entries.get_mut(meerkat_id) {
            entry.kickoff = kickoff;
            return true;
        }
        false
    }

    /// List active roster entries (excludes `Retiring`).
    pub fn list(&self) -> impl Iterator<Item = &RosterEntry> {
        self.entries
            .values()
            .filter(|e| e.state == MemberState::Active)
    }

    /// List all roster entries including `Retiring` members.
    pub fn list_all(&self) -> impl Iterator<Item = &RosterEntry> {
        self.entries.values()
    }

    /// List only `Retiring` members.
    pub fn list_retiring(&self) -> impl Iterator<Item = &RosterEntry> {
        self.entries
            .values()
            .filter(|e| e.state == MemberState::Retiring)
    }

    /// Find active meerkats with a given profile name.
    pub fn by_profile(&self, profile: &ProfileName) -> impl Iterator<Item = &RosterEntry> {
        self.entries
            .values()
            .filter(move |e| e.profile == *profile && e.state == MemberState::Active)
    }

    /// Find the first active member matching a label key-value pair.
    pub fn find_by_label(&self, key: &str, value: &str) -> Option<&RosterEntry> {
        self.entries.values().find(|e| {
            e.state == MemberState::Active && e.labels.get(key).is_some_and(|v| v == value)
        })
    }

    /// Find all active members matching a label key-value pair.
    pub fn find_all_by_label<'a>(
        &'a self,
        key: &'a str,
        value: &'a str,
    ) -> impl Iterator<Item = &'a RosterEntry> {
        self.entries.values().filter(move |e| {
            e.state == MemberState::Active && e.labels.get(key).is_some_and(|v| v == value)
        })
    }

    /// Look up the session ID for a meerkat by its ID.
    ///
    /// Returns `Some(&SessionId)` for `Session` members and `BackendPeer`
    /// members with a bridge session. Returns `None` if the meerkat is not
    /// in the roster or its member ref has no session bridge.
    pub fn session_id(&self, meerkat_id: &MeerkatId) -> Option<&SessionId> {
        self.entries.get(meerkat_id)?.member_ref.session_id()
    }

    /// Get the set of peer meerkat IDs wired to a given meerkat.
    pub fn wired_peers_of(&self, meerkat_id: &MeerkatId) -> Option<&BTreeSet<MeerkatId>> {
        self.entries.get(meerkat_id).map(|e| &e.wired_to)
    }

    /// Number of active meerkats in the roster.
    pub fn len(&self) -> usize {
        self.entries
            .values()
            .filter(|e| e.state == MemberState::Active)
            .count()
    }

    /// Whether the roster has no active meerkats.
    pub fn is_empty(&self) -> bool {
        !self
            .entries
            .values()
            .any(|e| e.state == MemberState::Active)
    }

    /// Mark a member as `Retiring`. Returns `true` if the member was found and
    /// transitioned (i.e. it was `Active`); `false` otherwise.
    pub fn mark_retiring(&mut self, meerkat_id: &MeerkatId) -> bool {
        if let Some(entry) = self.entries.get_mut(meerkat_id)
            && entry.state == MemberState::Active
        {
            entry.state = MemberState::Retiring;
            return true;
        }
        false
    }
}

impl RosterEntry {
    pub fn session_id(&self) -> Option<&SessionId> {
        self.member_ref.session_id()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ids::MobId;
    use chrono::Utc;
    use meerkat_core::comms::TrustedPeerSpec;
    use uuid::Uuid;

    fn session_id() -> SessionId {
        SessionId::from_uuid(Uuid::new_v4())
    }

    /// Test helper: converts old-style 4-arg add into RosterAddEntry with empty labels.
    fn add_member(
        roster: &mut Roster,
        meerkat_id: MeerkatId,
        profile: ProfileName,
        runtime_mode: MobRuntimeMode,
        member_ref: MemberRef,
    ) -> bool {
        roster.add(RosterAddEntry {
            meerkat_id,
            profile,
            runtime_mode,
            member_ref,
            peer_id: None,
            labels: BTreeMap::new(),
            effective_profile_override: None,
        })
    }

    fn make_event(cursor: u64, kind: MobEventKind) -> MobEvent {
        MobEvent {
            cursor,
            timestamp: Utc::now(),
            mob_id: MobId::from("test-mob"),
            kind,
        }
    }

    #[test]
    fn test_roster_add_and_get() {
        let mut roster = Roster::new();
        let sid = session_id();
        add_member(
            &mut roster,
            MeerkatId::from("agent-1"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(sid.clone()),
        );
        assert_eq!(roster.len(), 1);
        let entry = roster.get(&MeerkatId::from("agent-1")).unwrap();
        assert_eq!(entry.profile.as_str(), "worker");
        assert_eq!(entry.session_id(), Some(&sid));
        assert!(entry.wired_to.is_empty());
    }

    #[test]
    fn test_roster_remove() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("agent-1"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("agent-2"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        roster.wire(&MeerkatId::from("agent-1"), &MeerkatId::from("agent-2"));
        roster.remove(&MeerkatId::from("agent-1"));

        assert_eq!(roster.len(), 1);
        assert!(roster.get(&MeerkatId::from("agent-1")).is_none());
        // agent-2 should no longer have agent-1 in wired_to
        let entry2 = roster.get(&MeerkatId::from("agent-2")).unwrap();
        assert!(entry2.wired_to.is_empty());
    }

    #[test]
    fn test_roster_remove_nonexistent_is_noop() {
        let mut roster = Roster::new();
        roster.remove(&MeerkatId::from("nonexistent"));
        assert!(roster.is_empty());
    }

    #[test]
    fn test_set_session_id_preserves_backend_member_ref_identity() {
        let mut roster = Roster::new();
        let old_sid = session_id();
        add_member(
            &mut roster,
            MeerkatId::from("ext-1"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::BackendPeer {
                peer_id: "peer-ext-1".to_string(),
                address: "https://backend.example.invalid/mesh/ext-1".to_string(),
                session_id: Some(old_sid),
            },
        );

        let new_sid = session_id();
        assert!(roster.set_session_id(&MeerkatId::from("ext-1"), new_sid.clone()));
        let entry = roster
            .get(&MeerkatId::from("ext-1"))
            .expect("entry should remain present");
        match &entry.member_ref {
            MemberRef::BackendPeer {
                peer_id,
                address,
                session_id,
            } => {
                assert_eq!(peer_id, "peer-ext-1");
                assert_eq!(address, "https://backend.example.invalid/mesh/ext-1");
                assert_eq!(session_id.as_ref(), Some(&new_sid));
            }
            other => panic!("expected backend peer member ref, got {other:?}"),
        }
    }

    #[test]
    fn test_roster_wire_and_unwire() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("b"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );

        roster.wire(&MeerkatId::from("a"), &MeerkatId::from("b"));

        let peers_a = roster.wired_peers_of(&MeerkatId::from("a")).unwrap();
        assert!(peers_a.contains(&MeerkatId::from("b")));
        let peers_b = roster.wired_peers_of(&MeerkatId::from("b")).unwrap();
        assert!(peers_b.contains(&MeerkatId::from("a")));

        roster.unwire(&MeerkatId::from("a"), &MeerkatId::from("b"));

        let peers_a = roster.wired_peers_of(&MeerkatId::from("a")).unwrap();
        assert!(peers_a.is_empty());
        let peers_b = roster.wired_peers_of(&MeerkatId::from("b")).unwrap();
        assert!(peers_b.is_empty());
    }

    #[test]
    fn test_roster_wire_idempotent() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("b"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );

        roster.wire(&MeerkatId::from("a"), &MeerkatId::from("b"));
        roster.wire(&MeerkatId::from("a"), &MeerkatId::from("b"));

        let peers_a = roster.wired_peers_of(&MeerkatId::from("a")).unwrap();
        assert_eq!(peers_a.len(), 1); // No duplicates (BTreeSet)
    }

    #[test]
    fn test_roster_wire_external_treats_missing_peer_as_external() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );

        roster.wire_external(
            &MeerkatId::from("a"),
            &MeerkatId::from("remote-mob/worker/agent-b"),
            TrustedPeerSpec::new(
                "remote-mob/worker/agent-b",
                "ed25519:remote-b",
                "inproc://remote-mob/worker/agent-b",
            )
            .expect("valid trusted peer spec"),
        );

        let peers_a = roster.wired_peers_of(&MeerkatId::from("a")).unwrap();
        assert!(peers_a.contains(&MeerkatId::from("remote-mob/worker/agent-b")));
        assert!(
            roster
                .get(&MeerkatId::from("a"))
                .expect("entry should exist")
                .external_peer_specs
                .contains_key(&MeerkatId::from("remote-mob/worker/agent-b"))
        );
        assert!(
            roster.is_wiring_projection_consistent(),
            "missing local peer should be treated as an external projection target"
        );
    }

    #[test]
    fn test_roster_by_profile() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("w1"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("w2"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("lead"),
            ProfileName::from("orchestrator"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );

        let workers: Vec<_> = roster.by_profile(&ProfileName::from("worker")).collect();
        assert_eq!(workers.len(), 2);

        let orchestrators: Vec<_> = roster
            .by_profile(&ProfileName::from("orchestrator"))
            .collect();
        assert_eq!(orchestrators.len(), 1);
    }

    #[test]
    fn test_roster_list() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("b"),
            ProfileName::from("lead"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );

        let all: Vec<_> = roster.list().collect();
        assert_eq!(all.len(), 2);
    }

    #[test]
    fn test_roster_project_from_events() {
        let sid1 = session_id();
        let sid2 = session_id();
        let events = vec![
            make_event(
                1,
                MobEventKind::MeerkatSpawned {
                    meerkat_id: MeerkatId::from("a"),
                    role: ProfileName::from("worker"),
                    runtime_mode: MobRuntimeMode::AutonomousHost,
                    member_ref: MemberRef::from_session_id(sid1),
                    labels: BTreeMap::new(),
                },
            ),
            make_event(
                2,
                MobEventKind::MeerkatSpawned {
                    meerkat_id: MeerkatId::from("b"),
                    role: ProfileName::from("worker"),
                    runtime_mode: MobRuntimeMode::AutonomousHost,
                    member_ref: MemberRef::from_session_id(sid2),
                    labels: BTreeMap::new(),
                },
            ),
            make_event(
                3,
                MobEventKind::PeersWired {
                    a: MeerkatId::from("a"),
                    b: MeerkatId::from("b"),
                },
            ),
        ];
        let roster = Roster::project(&events);
        assert_eq!(roster.len(), 2);
        let peers_a = roster.wired_peers_of(&MeerkatId::from("a")).unwrap();
        assert!(peers_a.contains(&MeerkatId::from("b")));
    }

    #[test]
    fn test_roster_project_with_retire() {
        let sid1 = session_id();
        let sid2 = session_id();
        let events = vec![
            make_event(
                1,
                MobEventKind::MeerkatSpawned {
                    meerkat_id: MeerkatId::from("a"),
                    role: ProfileName::from("worker"),
                    runtime_mode: MobRuntimeMode::AutonomousHost,
                    member_ref: MemberRef::from_session_id(sid1.clone()),
                    labels: BTreeMap::new(),
                },
            ),
            make_event(
                2,
                MobEventKind::MeerkatSpawned {
                    meerkat_id: MeerkatId::from("b"),
                    role: ProfileName::from("worker"),
                    runtime_mode: MobRuntimeMode::AutonomousHost,
                    member_ref: MemberRef::from_session_id(sid2.clone()),
                    labels: BTreeMap::new(),
                },
            ),
            make_event(
                3,
                MobEventKind::PeersWired {
                    a: MeerkatId::from("a"),
                    b: MeerkatId::from("b"),
                },
            ),
            make_event(
                4,
                MobEventKind::MeerkatRetired {
                    meerkat_id: MeerkatId::from("a"),
                    role: ProfileName::from("worker"),
                    member_ref: MemberRef::from_session_id(sid1),
                },
            ),
        ];
        let roster = Roster::project(&events);
        assert_eq!(roster.len(), 1);
        assert!(roster.get(&MeerkatId::from("a")).is_none());
        let peers_b = roster.wired_peers_of(&MeerkatId::from("b")).unwrap();
        assert!(peers_b.is_empty());
    }

    #[test]
    fn test_roster_project_idempotent() {
        let sid = session_id();
        let events = vec![make_event(
            1,
            MobEventKind::MeerkatSpawned {
                meerkat_id: MeerkatId::from("a"),
                role: ProfileName::from("worker"),
                runtime_mode: MobRuntimeMode::AutonomousHost,
                member_ref: MemberRef::from_session_id(sid),
                labels: BTreeMap::new(),
            },
        )];
        let roster1 = Roster::project(&events);
        let roster2 = Roster::project(&events);
        assert_eq!(roster1.len(), roster2.len());
        assert_eq!(
            roster1.get(&MeerkatId::from("a")).unwrap().profile,
            roster2.get(&MeerkatId::from("a")).unwrap().profile,
        );
    }

    #[test]
    fn test_roster_serde_entry_roundtrip() {
        let entry = RosterEntry {
            meerkat_id: MeerkatId::from("test"),
            profile: ProfileName::from("worker"),
            member_ref: MemberRef::from_session_id(session_id()),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            peer_id: None,
            state: MemberState::default(),
            wired_to: {
                let mut s = BTreeSet::new();
                s.insert(MeerkatId::from("peer-1"));
                s
            },
            external_peer_specs: BTreeMap::new(),
            labels: BTreeMap::new(),
            kickoff: None,
            effective_profile_override: None,
        };
        let json = serde_json::to_string(&entry).unwrap();
        let parsed: RosterEntry = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.meerkat_id, entry.meerkat_id);
        assert_eq!(parsed.wired_to.len(), 1);
    }

    #[test]
    fn test_mark_retiring() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        assert!(roster.mark_retiring(&MeerkatId::from("a")));
        // Second call returns false (already Retiring)
        assert!(!roster.mark_retiring(&MeerkatId::from("a")));
        // Nonexistent returns false
        assert!(!roster.mark_retiring(&MeerkatId::from("nope")));
    }

    #[test]
    fn test_list_excludes_retiring() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("b"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        roster.mark_retiring(&MeerkatId::from("a"));

        let active: Vec<_> = roster.list().collect();
        assert_eq!(active.len(), 1);
        assert_eq!(active[0].meerkat_id, MeerkatId::from("b"));
    }

    #[test]
    fn test_list_all_includes_retiring() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("b"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        roster.mark_retiring(&MeerkatId::from("a"));

        let all: Vec<_> = roster.list_all().collect();
        assert_eq!(all.len(), 2);
    }

    #[test]
    fn test_list_retiring_only() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("b"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        roster.mark_retiring(&MeerkatId::from("a"));

        let retiring: Vec<_> = roster.list_retiring().collect();
        assert_eq!(retiring.len(), 1);
        assert_eq!(retiring[0].meerkat_id, MeerkatId::from("a"));
    }

    #[test]
    fn test_len_and_is_empty_count_active_only() {
        let mut roster = Roster::new();
        assert!(roster.is_empty());
        assert_eq!(roster.len(), 0);

        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        assert_eq!(roster.len(), 1);
        assert!(!roster.is_empty());

        roster.mark_retiring(&MeerkatId::from("a"));
        assert_eq!(roster.len(), 0);
        assert!(roster.is_empty());
    }

    #[test]
    fn test_by_profile_excludes_retiring() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("w1"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        add_member(
            &mut roster,
            MeerkatId::from("w2"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        roster.mark_retiring(&MeerkatId::from("w1"));

        let workers: Vec<_> = roster.by_profile(&ProfileName::from("worker")).collect();
        assert_eq!(workers.len(), 1);
        assert_eq!(workers[0].meerkat_id, MeerkatId::from("w2"));
    }

    #[test]
    fn test_get_returns_retiring() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(session_id()),
        );
        roster.mark_retiring(&MeerkatId::from("a"));

        let entry = roster.get(&MeerkatId::from("a"));
        assert!(entry.is_some());
        assert_eq!(entry.unwrap().state, MemberState::Retiring);
    }

    #[test]
    fn test_serde_roundtrip_with_state_field() {
        let entry = RosterEntry {
            meerkat_id: MeerkatId::from("test"),
            profile: ProfileName::from("worker"),
            member_ref: MemberRef::from_session_id(session_id()),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            peer_id: None,
            state: MemberState::Active,
            wired_to: BTreeSet::new(),
            external_peer_specs: BTreeMap::new(),
            labels: BTreeMap::new(),
            kickoff: None,
            effective_profile_override: None,
        };
        let json = serde_json::to_string(&entry).unwrap();
        let parsed: RosterEntry = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.state, MemberState::Active);
    }

    #[test]
    fn test_session_id_convenience_session_member() {
        let mut roster = Roster::new();
        let sid = session_id();
        add_member(
            &mut roster,
            MeerkatId::from("a"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::from_session_id(sid.clone()),
        );
        assert_eq!(roster.session_id(&MeerkatId::from("a")), Some(&sid));
    }

    #[test]
    fn test_session_id_convenience_backend_peer_with_bridge() {
        let mut roster = Roster::new();
        let sid = session_id();
        add_member(
            &mut roster,
            MeerkatId::from("ext-1"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::BackendPeer {
                peer_id: "peer-ext-1".to_string(),
                address: "https://backend.example.invalid/mesh/ext-1".to_string(),
                session_id: Some(sid.clone()),
            },
        );
        assert_eq!(roster.session_id(&MeerkatId::from("ext-1")), Some(&sid));
    }

    #[test]
    fn test_session_id_convenience_backend_peer_no_bridge() {
        let mut roster = Roster::new();
        add_member(
            &mut roster,
            MeerkatId::from("ext-2"),
            ProfileName::from("worker"),
            MobRuntimeMode::AutonomousHost,
            MemberRef::BackendPeer {
                peer_id: "peer-ext-2".to_string(),
                address: "https://backend.example.invalid/mesh/ext-2".to_string(),
                session_id: None,
            },
        );
        assert_eq!(roster.session_id(&MeerkatId::from("ext-2")), None);
    }

    #[test]
    fn test_session_id_convenience_not_found() {
        let roster = Roster::new();
        assert_eq!(roster.session_id(&MeerkatId::from("nonexistent")), None);
    }

    #[test]
    fn test_serde_roundtrip_missing_state_defaults_to_active() {
        // Simulate old serialized data without the state field
        let json = r#"{"meerkat_id":"old","profile":"worker","member_ref":{"kind":"session","session_id":"00000000-0000-0000-0000-000000000001"},"runtime_mode":"autonomous_host","wired_to":[]}"#;
        let parsed: RosterEntry = serde_json::from_str(json).unwrap();
        assert_eq!(parsed.state, MemberState::Active);
    }

    #[test]
    fn test_project_never_produces_retiring() {
        let sid = session_id();
        let events = vec![make_event(
            1,
            MobEventKind::MeerkatSpawned {
                meerkat_id: MeerkatId::from("a"),
                role: ProfileName::from("worker"),
                runtime_mode: MobRuntimeMode::AutonomousHost,
                member_ref: MemberRef::from_session_id(sid),
                labels: BTreeMap::new(),
            },
        )];
        let roster = Roster::project(&events);
        let entry = roster.get(&MeerkatId::from("a")).unwrap();
        assert_eq!(entry.state, MemberState::Active);
    }

    #[test]
    fn test_roster_labels_populated_from_event() {
        let sid = session_id();
        let mut labels = BTreeMap::new();
        labels.insert("faction".to_string(), "north".to_string());
        labels.insert("tier".to_string(), "1".to_string());
        let events = vec![make_event(
            1,
            MobEventKind::MeerkatSpawned {
                meerkat_id: MeerkatId::from("a"),
                role: ProfileName::from("worker"),
                runtime_mode: MobRuntimeMode::AutonomousHost,
                member_ref: MemberRef::from_session_id(sid),
                labels: labels.clone(),
            },
        )];
        let roster = Roster::project(&events);
        let entry = roster.get(&MeerkatId::from("a")).unwrap();
        assert_eq!(entry.labels, labels);
    }

    #[test]
    fn test_find_by_label_returns_active_member() {
        let mut roster = Roster::new();
        roster.add(RosterAddEntry {
            meerkat_id: MeerkatId::from("a"),
            profile: ProfileName::from("worker"),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            member_ref: MemberRef::from_session_id(session_id()),
            peer_id: None,
            labels: {
                let mut m = BTreeMap::new();
                m.insert("faction".to_string(), "north".to_string());
                m
            },
            effective_profile_override: None,
        });
        roster.add(RosterAddEntry {
            meerkat_id: MeerkatId::from("b"),
            profile: ProfileName::from("worker"),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            member_ref: MemberRef::from_session_id(session_id()),
            peer_id: None,
            labels: {
                let mut m = BTreeMap::new();
                m.insert("faction".to_string(), "south".to_string());
                m
            },
            effective_profile_override: None,
        });
        let found = roster.find_by_label("faction", "north");
        assert!(found.is_some());
        assert_eq!(found.unwrap().meerkat_id, MeerkatId::from("a"));
    }

    #[test]
    fn test_find_all_by_label_returns_all_matching() {
        let mut roster = Roster::new();
        roster.add(RosterAddEntry {
            meerkat_id: MeerkatId::from("a"),
            profile: ProfileName::from("worker"),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            member_ref: MemberRef::from_session_id(session_id()),
            peer_id: None,
            labels: {
                let mut m = BTreeMap::new();
                m.insert("tier".to_string(), "1".to_string());
                m
            },
            effective_profile_override: None,
        });
        roster.add(RosterAddEntry {
            meerkat_id: MeerkatId::from("b"),
            profile: ProfileName::from("worker"),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            member_ref: MemberRef::from_session_id(session_id()),
            peer_id: None,
            labels: {
                let mut m = BTreeMap::new();
                m.insert("tier".to_string(), "1".to_string());
                m
            },
            effective_profile_override: None,
        });
        roster.add(RosterAddEntry {
            meerkat_id: MeerkatId::from("c"),
            profile: ProfileName::from("worker"),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            member_ref: MemberRef::from_session_id(session_id()),
            peer_id: None,
            labels: {
                let mut m = BTreeMap::new();
                m.insert("tier".to_string(), "2".to_string());
                m
            },
            effective_profile_override: None,
        });
        let found: Vec<_> = roster.find_all_by_label("tier", "1").collect();
        assert_eq!(found.len(), 2);
    }

    #[test]
    fn test_find_by_label_excludes_retiring() {
        let mut roster = Roster::new();
        roster.add(RosterAddEntry {
            meerkat_id: MeerkatId::from("a"),
            profile: ProfileName::from("worker"),
            runtime_mode: MobRuntimeMode::AutonomousHost,
            member_ref: MemberRef::from_session_id(session_id()),
            peer_id: None,
            labels: {
                let mut m = BTreeMap::new();
                m.insert("faction".to_string(), "north".to_string());
                m
            },
            effective_profile_override: None,
        });
        roster.mark_retiring(&MeerkatId::from("a"));
        assert!(roster.find_by_label("faction", "north").is_none());
        assert_eq!(roster.find_all_by_label("faction", "north").count(), 0);
    }

    #[test]
    fn test_roster_labels_empty_backward_compat() {
        // Old serialized data without labels field should default to empty
        let json = r#"{"meerkat_id":"old","profile":"worker","member_ref":{"kind":"session","session_id":"00000000-0000-0000-0000-000000000001"},"runtime_mode":"autonomous_host","wired_to":[]}"#;
        let parsed: RosterEntry = serde_json::from_str(json).unwrap();
        assert!(parsed.labels.is_empty());
    }
}