feat: GroupV2 through the threaded client, group roster, registry retry

core/conversations/src/conversation.rs

@@ -42,6 +42,10 @@ pub(crate) trait GroupConvo<S: ExternalServices>: Convo<S> + std::fmt::Debug + S
         cx: &mut ServiceContext<S>,
         members: &[IdentIdRef],
     ) -> Result<(), ChatError>;
+
+    /// Each current member's MLS leaf-credential content (hex-encoded), self
+    /// included.
+    fn members(&self) -> Result<Vec<Vec<u8>>, ChatError>;
 }
 
 pub(crate) trait Identified {

core/conversations/src/conversation/group_v1.rs

@@ -341,4 +341,12 @@ impl<S: ExternalServices> GroupConvo<S> for GroupV1Convo {
 
         self.send_payload(cx, commit.to_bytes()?)
     }
+
+    fn members(&self) -> Result<Vec<Vec<u8>>, ChatError> {
+        Ok(self
+            .mls_group
+            .members()
+            .map(|m| m.credential.serialized_content().to_vec())
+            .collect())
+    }
 }

core/conversations/src/conversation/group_v2.rs

@@ -11,8 +11,7 @@ use de_mls::protos::de_mls::messages::v1::{
     AppMessage as AppMessageProto, MemberWelcome, app_message,
 };
 use de_mls::{
-    Conversation, ConversationConfig, ConversationEvent, PeerScoringService, ScoringConfig,
-    default_score_deltas,
+    Conversation, ConversationEvent, PeerScoringService, ScoringConfig, default_score_deltas,
     defaults::{DefaultConsensusPlugin, DefaultPeerScoring, InMemoryPeerScoreStorage},
 };
 use hashgraph_like_consensus::signing::EthereumConsensusSigner;
@@ -22,7 +21,6 @@ use openmls::prelude::{KeyPackageIn, OpenMlsProvider as _, ProtocolVersion};
 use prost::Message;
 use shared_traits::{IdentId, IdentIdRef};
 use std::sync::Arc;
-use std::time::Duration;
 use tracing::{info, instrument, warn};
 
 use crate::IdentityProvider;
@@ -58,21 +56,6 @@ fn make_consensus() -> DefaultConsensusPlugin {
     DefaultConsensusPlugin::new(EthereumConsensusSigner::new(PrivateKeySigner::random()))
 }
 
-/// TEST-ONLY millisecond timers. de-mls deadlines are real wall-clock, so the
-/// default 60s timers never fire under fast virtual time. Production needs a
-/// real config injected from the caller, not these hardcoded values.
-fn demls_config() -> ConversationConfig {
-    ConversationConfig {
-        commit_inactivity_duration: Duration::from_millis(50),
-        freeze_duration: Duration::from_millis(20),
-        voting_delay: Duration::from_millis(30),
-        election_voting_delay: Duration::from_millis(30),
-        consensus_timeout: Duration::from_millis(150),
-        proposal_expiration: Duration::from_millis(2000),
-        ..ConversationConfig::default()
-    }
-}
-
 pub struct GroupV2Convo {
     convo_id: String,
     conversation: Conversation<DefaultConsensusPlugin, InMemoryPeerScoreStorage>,
@@ -116,7 +99,7 @@ impl GroupV2Convo {
             &make_consensus(),
             make_scoring(),
             rand_app_id(),
-            demls_config(),
+            service_ctx.group_v2_config.clone(),
         )?;
         let convo = GroupV2Convo {
             convo_id,
@@ -147,7 +130,7 @@ impl GroupV2Convo {
             &make_consensus(),
             make_scoring(),
             rand_app_id(),
-            demls_config(),
+            service_ctx.group_v2_config.clone(),
         )?
         else {
             return Err(ChatError::generic("welcome not addressed to this member"));
@@ -277,14 +260,26 @@ where
         service_ctx: &mut ServiceContext<S>,
         members: &[IdentIdRef],
     ) -> Result<(), ChatError> {
-        // Record who WE invited before touching the conversation: after_op
-        // forwards a welcome only to joiners in pending_invites. Members are
-        // signer ids; the de-mls member id must match the id of the
-        // IdentityProvider that generated the key package (its MLS leaf
-        // credential content — de-mls matches members by credential), so it is
-        // read from the fetched key package rather than assumed equal to the
-        // signer id.
-        for member in members {
+        // Dedup the requested signers up front: an account can resolve to the
+        // same signer twice, or a caller can repeat one, and a duplicate would
+        // otherwise cost a redundant key-package fetch here and a second Add
+        // proposal for a member already being added in this batch.
+        let mut seen = std::collections::HashSet::new();
+        let members: Vec<IdentIdRef> = members
+            .iter()
+            .copied()
+            .filter(|m| seen.insert(m.as_str().to_string()))
+            .collect();
+
+        // Fetch and validate every key package before proposing any add, so a
+        // member with no key package fails the call before it opens proposals
+        // for the others. Members are signer ids; the de-mls member id must
+        // match the id of the IdentityProvider that generated the key package
+        // (its MLS leaf credential content — de-mls matches members by
+        // credential), so it is read from the fetched key package rather than
+        // assumed equal to the signer id.
+        let mut invites = Vec::with_capacity(members.len());
+        for member in &members {
             let kp_bytes = service_ctx
                 .registry
                 .retrieve(member.as_str())
@@ -298,17 +293,54 @@ where
                 .credential()
                 .serialized_content()
                 .to_vec();
-            self.pending_invites
-                .push((member_id.clone(), member.to_string()));
-            self.conversation.add_member(
+            invites.push((member_id, member.to_string(), kp_bytes));
+        }
+
+        // pending_invites drives welcome delivery: after_op forwards a welcome
+        // only to a joiner recorded here. Record a member only if de-mls will
+        // actually propose its add — recording one it silently drops strands an
+        // entry that a later re-join can match, firing a spurious duplicate
+        // welcome. de-mls drops self and members already in the group; and since
+        // add_member only opens a proposal, the committed roster won't reflect a
+        // member added earlier in this same loop, so the set tracks those too.
+        // Seed it with the roster and self, insert as we go, and one check
+        // covers all three.
+        let mut roster: std::collections::HashSet<Vec<u8>> =
+            self.conversation.members()?.into_iter().collect();
+        roster.insert(self.conversation.member_id_bytes().to_vec());
+
+        let mut result = Ok(());
+        for (member_id, signer_id, kp_bytes) in invites {
+            if !roster.insert(member_id.clone()) {
+                continue;
+            }
+            self.pending_invites.push((member_id.clone(), signer_id));
+            if let Err(e) = self.conversation.add_member(
                 &service_ctx.mls_provider,
                 &service_ctx.mls_identity,
                 &member_id,
                 &kp_bytes,
-            )?;
+            ) {
+                self.pending_invites.pop();
+                result = Err(e.into());
+                break;
+            }
         }
-        self.after_op(service_ctx)?;
-        Ok(())
+        // Flush even on a mid-loop failure: proposals already opened must be
+        // published and the wakeup re-armed, or they sit dormant until an
+        // unrelated frame drives the conversation.
+        let flushed = self.after_op(service_ctx).map(drop);
+        result.and(flushed)
+    }
+
+    fn members(&self) -> Result<Vec<Vec<u8>>, ChatError> {
+        // Guarantee the local member is listed so callers see the full roster.
+        let mut members = self.conversation.members()?;
+        let self_id = self.conversation.member_id_bytes().to_vec();
+        if !members.contains(&self_id) {
+            members.push(self_id);
+        }
+        Ok(members)
     }
 
     // fn conversation_state(&self) -> Result<ConversationState, ChatError> {

core/conversations/src/core.rs

@@ -3,7 +3,7 @@ use crate::conversation::{
     ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, PrivateV1Convo,
 };
 use crate::service_context::{ExternalServices, ServiceContext};
-use crate::{DeliveryService, IdentityProvider, RegistrationService, WakeupService};
+use crate::{DeliveryService, GroupV2Config, IdentityProvider, RegistrationService, WakeupService};
 use crate::{
     conversation::{Convo, GroupConvo},
     errors::ChatError,
@@ -140,6 +140,7 @@ where
                 causal,
                 identity,
                 wakeup_service,
+                group_v2_config: GroupV2Config::default(),
             },
             inbox,
             pq_inbox,
@@ -174,6 +175,12 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
         self.pq_inbox.register(&mut self.services)
     }
 
+    /// Timing/policy for GroupV2 conversations created or joined after this
+    /// call. Existing conversations keep the config they were built with.
+    pub fn set_group_v2_config(&mut self, config: GroupV2Config) {
+        self.services.group_v2_config = config;
+    }
+
     pub fn installation_name(&self) -> &str {
         self.services.identity.get_name()
     }
@@ -293,6 +300,28 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
         }
     }
 
+    /// Each member's MLS leaf-credential content (hex-encoded); errors if
+    /// `convo_id` names a direct (non-group) conversation.
+    pub fn group_members(&mut self, convo_id: &str) -> Result<Vec<Vec<u8>>, ChatError> {
+        if self.cached_convos.contains_key(convo_id) {
+            let convo = self
+                .cached_convos
+                .get(convo_id)
+                .ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?;
+
+            match convo {
+                ConvoTypeOwned::Group(group_convo) => group_convo.members(),
+                ConvoTypeOwned::Direct(convo) => Err(ChatError::UnsupportedFunction(
+                    convo.id().into(),
+                    "List Members".into(),
+                )),
+            }
+        } else {
+            let convo = self.load_group_convo(convo_id)?;
+            convo.members()
+        }
+    }
+
     pub fn list_conversations(&self) -> Result<Vec<ConversationId>, ChatError> {
         // Check Legacy load_convo store
         let records = self.services.store.load_conversations()?;
@@ -304,9 +333,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
             convos.push(convo.to_string());
         }
 
-        // Conversations may use both storage mechanisms.
-        // Remove duplicates
-        convos.dedup();
+        // A conversation can live in both the store and the in-memory cache (a
+        // DirectV1 join persists to the store and is also cached), so drop
+        // duplicates across the two. `Vec::dedup` only removes *consecutive*
+        // repeats and `cached_convos` iterates in nondeterministic HashMap
+        // order, so dedup through a set instead.
+        let mut seen = std::collections::HashSet::new();
+        convos.retain(|c| seen.insert(c.clone()));
         Ok(convos)
     }
 
@@ -362,16 +395,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
 
     // Dispatch encrypted payload to the post-quantum inbox.
     fn dispatch_to_inbox2(&mut self, payload: &[u8]) -> Result<PayloadOutcome, ChatError> {
-        if let Some(convo) = self.pq_inbox.handle_frame(&mut self.services, payload)? {
+        if let Some((convo, class)) = self.pq_inbox.handle_frame(&mut self.services, payload)? {
             let convo_id = convo.id().to_string();
             // Cache convos created by InboxV2
             self.register_convo(ConvoTypeOwned::Group(convo))?;
 
             Ok(PayloadOutcome::Inbox(InboxOutcome {
-                new_conversation: crate::NewConversation {
-                    convo_id,
-                    class: crate::ConversationClass::Group,
-                },
+                new_conversation: crate::NewConversation { convo_id, class },
                 initial: None,
             }))
         } else {

core/conversations/src/inbox_v2.rs

@@ -20,6 +20,7 @@ use crate::conversation::GroupConvo;
 use crate::conversation::GroupV1Convo;
 use crate::conversation::GroupV2Convo;
 use crate::conversation::Identified as _;
+use crate::outcomes::ConversationClass;
 use crate::service_context::{ExternalServices, ServiceContext};
 use crate::utils::{blake2b_hex, hash_size};
 use crate::{AddressedEnvelope, IdentId, IdentIdRef, IdentityProvider};
@@ -70,6 +71,10 @@ pub fn invite_user_v2<DS: DeliveryService>(
     .map_err(ChatError::generic)
 }
 
+/// A convo built from an InboxV2 invite, paired with the display class its
+/// invite type implies.
+type ClassifiedConvo<S> = (Box<dyn GroupConvo<S>>, ConversationClass);
+
 /// A PQ focused Conversation initializer.
 /// InboxV2 is signer-scoped: it receives invites under this installation's
 /// signer id (the hex of the signer's verifying key), supporting PQ based
@@ -95,8 +100,11 @@ impl InboxV2 {
     ) -> Result<(), ChatError> {
         let keypackage_bytes = Self::create_keypackage(cx)?.tls_serialize_detached()?;
 
-        // TODO: (P3) Each keypackage can only be used once either enable...
-        // "LastResort" package or publish multiple
+        // TODO: publishes a single key package per installation. The intended
+        // design is a pool of one-time key packages (the registry pops one per
+        // fetch, the client replenishes) with the last-resort key package as the
+        // exhaustion fallback rather than the primary; that needs pop/claim
+        // semantics in the registry service. Tracked in #169.
         cx.registry
             .register(&cx.mls_identity, keypackage_bytes)
             .map_err(ChatError::generic)?;
@@ -112,12 +120,15 @@ impl InboxV2 {
         conversation_id_for(&self.ident_id)
     }
 
+    /// The convo built from an invite, paired with the display class its invite
+    /// type implies: `InviteType::GroupV1` carries the pairwise DirectV1 welcome,
+    /// so it is `Private`; `InviteType::GroupV2` is a real group.
     #[instrument(name = "inboxV2.handle_frame", skip_all, fields(user_id = %service_ctx.mls_identity.display_name()))]
     pub fn handle_frame<S: ExternalServices>(
         &self,
         service_ctx: &mut ServiceContext<S>,
         payload_bytes: &[u8],
-    ) -> Result<Option<Box<dyn GroupConvo<S>>>, ChatError> {
+    ) -> Result<Option<ClassifiedConvo<S>>, ChatError> {
         // On a broadcast transport the inbox address also receives traffic
         // that isn't an invite (or that prost decodes into an empty frame).
         // Treat anything we can't interpret as "not for us" and skip it,
@@ -131,14 +142,15 @@ impl InboxV2 {
 
         match payload {
             InviteType::GroupV1(inv) => {
-                Ok(Some(Box::new(self.handle_heavy_invite(service_ctx, inv)?)))
+                let convo = self.handle_heavy_invite(service_ctx, inv)?;
+                Ok(Some((Box::new(convo), ConversationClass::Private)))
             }
             InviteType::GroupV2(welcome_bytes) => {
                 info!("Process WelcomeMessage");
                 let mw =
                     MemberWelcome::decode(welcome_bytes.as_slice()).map_err(ChatError::generic)?;
                 let convo = GroupV2Convo::new_from_welcome(service_ctx, &mw)?;
-                Ok(Some(Box::new(convo)))
+                Ok(Some((Box::new(convo), ConversationClass::Group)))
             }
         }
     }
@@ -183,11 +195,22 @@ impl InboxV2 {
     fn create_keypackage<S: ExternalServices>(
         cx: &ServiceContext<S>,
     ) -> Result<KeyPackage, ChatError> {
+        // Last-resort key package. openmls consumes (deletes) a normal key
+        // package's init key on the first welcome that uses it; since each
+        // installation publishes just one, a second group inviting it would find
+        // no matching key package and reject the welcome ("welcome not addressed
+        // to this member"). Last-resort key packages are retained, so one admits
+        // the installation to any number of groups. Every key-package extension
+        // must be advertised in the leaf capabilities, hence LastResort there.
         let capabilities = Capabilities::builder()
             .ciphersuites(vec![CIPHER_SUITE])
-            .extensions(vec![ExtensionType::ApplicationId])
+            .extensions(vec![
+                ExtensionType::ApplicationId,
+                ExtensionType::LastResort,
+            ])
             .build();
         let a = KeyPackage::builder()
+            .mark_as_last_resort()
             .leaf_node_capabilities(capabilities)
             .build(
                 CIPHER_SUITE,

core/conversations/src/lib.rs

@@ -16,6 +16,13 @@ pub use causal_history::{Frontier, MissingMessage};
 pub use chat_sqlite::ChatStorage;
 pub use chat_sqlite::StorageConfig;
 pub use core::{ConversationId, Core, Introduction};
+/// Timing/policy for GroupV2 conversations (de-mls's per-conversation config).
+/// Defaults to the de-mls library defaults; inject via
+/// [`Core::set_group_v2_config`]. The creator's phase durations (commit
+/// inactivity, freeze, recovery, voting inactivity, proposal expiration,
+/// consensus timeout) travel to joiners with the welcome and overwrite
+/// theirs; vote delays and the policy fields stay local to each member.
+pub use de_mls::ConversationConfig as GroupV2Config;
 pub use errors::ChatError;
 pub use outcomes::{
     Content, ConversationClass, ConvoOutcome, InboxOutcome, NewConversation, PayloadOutcome,

core/conversations/src/service_context.rs

@@ -44,6 +44,11 @@ pub(crate) struct ServiceContext<S: ExternalServices> {
     pub(crate) causal: CausalHistoryStore,
     pub(crate) identity: Identity,
     pub(crate) wakeup_service: S::WS,
+    /// Timing/policy applied to GroupV2 conversations created or joined by
+    /// this core. Read at conversation construction; a joiner's phase
+    /// durations are then overwritten by the creator's, carried with the
+    /// welcome (vote delays and policy fields stay local).
+    pub(crate) group_v2_config: de_mls::ConversationConfig,
 }
 
 #[cfg(test)]
@@ -110,6 +115,7 @@ mod test_support {
                 causal: CausalHistoryStore::new(),
                 identity: Identity::new(name),
                 wakeup_service: NoopWakeups {},
+                group_v2_config: de_mls::ConversationConfig::default(),
             })
         }
     }