feat: surface what a GroupV2 conversation reports about itself

de-mls narrates its own commit-and-recovery cycle, and libchat consumes three of the fourteen events it emits and drops the rest. A group that has stopped moving therefore looks like one that is idle: the roster keeps reporting whatever it last committed, and the account of why sits in a log the application cannot reach.

Buffer the phase changes, the commit-round counts and the steps that did not go through the way causal-history observations are buffered, drain them after each drive of the core, and give each a client event. The count is the load-bearing one: a round that ends short of the candidates it expected is a round where members chose from different sets, which is how a group splits.

core/conversations/src/conversation/group_v2.rs

@@ -5,6 +5,7 @@
 use crate::conversation::mls_extensions::{
     ConvoMetaInfo, GROUP_METADATA_EXTENSION_TYPE, capabilities_with_group_metadata,
 };
+use crate::group_v2_status::GroupV2StatusKind;
 use crate::types::{AddressedEncryptedPayload, ConvoMetadata};
 use crate::{Content, WakeupService};
 use alloy::signers::local::PrivateKeySigner;
@@ -469,7 +470,28 @@ impl GroupV2Convo {
             }
         }
 
-        // 2. Publish
+        // 2. Record what the conversation said about running itself, so a
+        //    client can surface a commit round that is missing candidates or a
+        //    step that did not go through.
+        for evt in &events {
+            let kind = match evt {
+                ConversationEvent::PhaseChange(state) => GroupV2StatusKind::Phase(*state),
+                ConversationEvent::CommitRoundProgress { received, expected } => {
+                    GroupV2StatusKind::CommitRound {
+                        received: *received,
+                        expected: *expected,
+                    }
+                }
+                ConversationEvent::Error { operation, message } => GroupV2StatusKind::Failed {
+                    operation: operation.clone(),
+                    message: message.clone(),
+                },
+                _ => continue,
+            };
+            service_ctx.group_v2_status.record(&self.convo_id, kind);
+        }
+
+        // 3. Publish
         for out in outbound {
             let frame = GroupV2Frame {
                 payload: Some(GroupV2Payload::DeMlsWrapper(out.payload.into())),
@@ -489,7 +511,7 @@ impl GroupV2Convo {
                 .map_err(ChatError::generic)?;
         }
 
-        // 3. Re-arm the alarm with the conversation's earliest deadline.
+        // 4. Re-arm the alarm with the conversation's earliest deadline.
         if let Some(d) = wakeup {
             service_ctx
                 .wakeup_service

core/conversations/src/core.rs

@@ -2,6 +2,7 @@ use crate::causal_history::{CausalHistoryStore, DeliveryAck, MissingMessage};
 use crate::conversation::{
     ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, MessageId,
 };
+use crate::group_v2_status::{GroupV2Status, GroupV2StatusStore};
 use crate::service_context::{ExternalServices, ServiceContext};
 use crate::types::ConvoMetadata;
 use crate::{
@@ -146,6 +147,7 @@ where
                 mls_identity,
                 mls_provider,
                 causal,
+                group_v2_status: GroupV2StatusStore::default(),
                 identity,
                 wakeup_service,
                 demls_clock: GroupV2Clock::default(),
@@ -335,6 +337,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
         self.services.causal.take_acks()
     }
 
+    /// Drain what the GroupV2 conversations reported about running themselves
+    /// since the last call: phase changes, commit-round progress, and steps of
+    /// their own that did not go through.
+    pub fn take_group_v2_status(&self) -> Vec<GroupV2Status> {
+        self.services.group_v2_status.take()
+    }
+
     /// Encrypt and publish `content` to an existing conversation, returning the
     /// id assigned to the message so later acknowledgements can be matched to
     /// it.

core/conversations/src/group_v2_status.rs

@@ -0,0 +1,55 @@
+//! What a GroupV2 conversation reports about running itself.
+//!
+//! de-mls narrates its own commit-and-recovery cycle alongside the messages it
+//! decrypts. None of it is content and none of it needs acting on, but it is
+//! the only account of why a group is or is not moving, so it is buffered here
+//! and drained by the client the same way causal-history observations are.
+
+use std::cell::RefCell;
+
+use crate::core::ConversationId;
+
+/// The phase of a GroupV2 conversation's commit-and-recovery cycle.
+pub use de_mls::ConversationState as GroupV2Phase;
+
+/// One report from a GroupV2 conversation.
+#[derive(Debug, Clone)]
+pub struct GroupV2Status {
+    pub convo_id: ConversationId,
+    pub kind: GroupV2StatusKind,
+}
+
+#[derive(Debug, Clone)]
+pub enum GroupV2StatusKind {
+    /// The conversation entered a new phase.
+    Phase(GroupV2Phase),
+    /// `received` of `expected` stewards' commit candidates have arrived for
+    /// the round in progress; reported again whenever the count changes. A
+    /// round that ends with fewer than it expected is one where the members
+    /// chose from different sets of candidates.
+    CommitRound { received: usize, expected: usize },
+    /// A step the conversation was carrying out on its own, such as submitting
+    /// a vote, did not go through. The conversation stays usable.
+    Failed { operation: String, message: String },
+}
+
+/// Session-scoped buffer for [`GroupV2Status`], shared through
+/// `ServiceContext` because conversations are rebuilt from storage on every
+/// inbound payload and cannot hold it themselves.
+#[derive(Debug, Default)]
+pub(crate) struct GroupV2StatusStore {
+    reports: RefCell<Vec<GroupV2Status>>,
+}
+
+impl GroupV2StatusStore {
+    pub(crate) fn record(&self, convo_id: &str, kind: GroupV2StatusKind) {
+        self.reports.borrow_mut().push(GroupV2Status {
+            convo_id: convo_id.to_owned(),
+            kind,
+        });
+    }
+
+    pub(crate) fn take(&self) -> Vec<GroupV2Status> {
+        std::mem::take(&mut self.reports.borrow_mut())
+    }
+}

core/conversations/src/lib.rs

@@ -2,6 +2,7 @@ mod causal_history;
 mod conversation;
 mod core;
 mod errors;
+mod group_v2_status;
 mod inbox_v2;
 mod outcomes;
 mod proto;
@@ -24,6 +25,7 @@ pub use core::{ConversationId, Core};
 pub use de_mls::ConversationConfig as GroupV2Config;
 pub use de_mls::MockClock;
 pub use errors::ChatError;
+pub use group_v2_status::{GroupV2Phase, GroupV2Status, GroupV2StatusKind};
 pub use outcomes::{
     Content, ConversationClass, ConvoOutcome, InboxOutcome, NewConversation, PayloadOutcome,
 };

core/conversations/src/service_context.rs

@@ -6,6 +6,7 @@ use storage::ChatStore;
 use crate::IdentityProvider;
 use crate::causal_history::CausalHistoryStore;
 use crate::conversation::GroupV2Clock;
+use crate::group_v2_status::GroupV2StatusStore;
 use crate::inbox_v2::{MlsEphemeralPqProvider, MlsIdentityProvider};
 use crate::service_traits::WakeupService;
 use crate::{DeliveryService, RegistrationService};
@@ -43,6 +44,7 @@ pub(crate) struct ServiceContext<S: ExternalServices> {
     pub(crate) mls_identity: MlsIdentityProvider<S::IP>,
     pub(crate) mls_provider: MlsEphemeralPqProvider,
     pub(crate) causal: CausalHistoryStore,
+    pub(crate) group_v2_status: GroupV2StatusStore,
     pub(crate) identity: Identity,
     pub(crate) wakeup_service: S::WS,
     /// Time source for GroupV2 (de-mls) conversations.

crates/generic-chat/src/client.rs

@@ -7,8 +7,8 @@ use crossbeam_channel::{Receiver, Sender, select};
 use crypto::Ed25519VerifyingKey;
 use libchat::{
     ConversationId, ConvoMetadata, ConvoOutcome, Core, DeliveryAck, DeliveryService, GroupV2Config,
-    IdentId, IdentIdRef, InboxOutcome, MessageId, MissingMessage, PayloadOutcome,
-    RegistrationService,
+    GroupV2Status, GroupV2StatusKind, IdentId, IdentIdRef, InboxOutcome, MessageId, MissingMessage,
+    PayloadOutcome, RegistrationService,
 };
 use logos_account::{AccountDirectory, resolve_device_ids};
 use parking_lot::Mutex;
@@ -366,6 +366,7 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
                     };
                     events.extend(delivery_ack_events(core.take_acks(), &directory));
                     events.extend(missing_events(core.take_missing_messages(), &directory));
+                    events.extend(group_v2_status_events(core.take_group_v2_status()));
                     events
                 };
                 for event in events {
@@ -390,6 +391,7 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
                     };
                     events.extend(delivery_ack_events(core.take_acks(), &directory));
                     events.extend(missing_events(core.take_missing_messages(), &directory));
+                    events.extend(group_v2_status_events(core.take_group_v2_status()));
                     events
                 };
                 for event in events {
@@ -448,6 +450,38 @@ fn missing_events(missing: Vec<MissingMessage>, directory: &impl AccountDirector
         .collect()
 }
 
+/// Map what the GroupV2 conversations reported about running themselves onto
+/// [`Event::ConversationPhaseChanged`], [`Event::CommitRoundProgress`] and
+/// [`Event::ConversationError`].
+///
+/// Drained after each drive of the core, so these narrate the drive that
+/// produced the batch they arrive with.
+fn group_v2_status_events(reports: Vec<GroupV2Status>) -> Vec<Event> {
+    reports
+        .into_iter()
+        .map(|report| {
+            let convo_id = Arc::from(report.convo_id);
+            match report.kind {
+                GroupV2StatusKind::Phase(phase) => {
+                    Event::ConversationPhaseChanged { convo_id, phase }
+                }
+                GroupV2StatusKind::CommitRound { received, expected } => {
+                    Event::CommitRoundProgress {
+                        convo_id,
+                        received,
+                        expected,
+                    }
+                }
+                GroupV2StatusKind::Failed { operation, message } => Event::ConversationError {
+                    convo_id,
+                    operation,
+                    message,
+                },
+            }
+        })
+        .collect()
+}
+
 /// Resolve a participant a causal-history observation named — the author of a
 /// message we never saw, or the peer acknowledging one of ours.
 ///

crates/generic-chat/src/event.rs

@@ -8,7 +8,7 @@
 
 use std::sync::Arc;
 
-use libchat::{ConversationClass, IdentId};
+use libchat::{ConversationClass, GroupV2Phase, IdentId};
 
 /// The sender of a received message, recovered from its credential.
 ///
@@ -72,6 +72,29 @@ pub enum Event {
     ConversationMembersChanged {
         convo_id: Arc<str>,
     },
+    /// A GroupV2 conversation entered a new phase of its commit-and-recovery
+    /// cycle. Nothing needs acting on, but a conversation parked outside
+    /// `Working` is one that is accepting neither messages nor members.
+    ConversationPhaseChanged {
+        convo_id: Arc<str>,
+        phase: GroupV2Phase,
+    },
+    /// `received` of `expected` stewards' commit candidates have arrived for
+    /// the commit round in progress, reported again whenever the count
+    /// changes. A round that ends short of `expected` is one where members
+    /// chose from different sets of candidates.
+    CommitRoundProgress {
+        convo_id: Arc<str>,
+        received: usize,
+        expected: usize,
+    },
+    /// A step a GroupV2 conversation was carrying out on its own, such as
+    /// submitting a vote, did not go through. The conversation stays usable.
+    ConversationError {
+        convo_id: Arc<str>,
+        operation: String,
+        message: String,
+    },
     InboundError {
         message: String,
     },

crates/generic-chat/src/lib.rs

@@ -15,7 +15,7 @@ pub use event::{Event, MessageSender};
 // Re-export types callers need to interact with ChatClient.
 pub use libchat::{
     AddressedEnvelope, ChatStore, ConversationClass, ConversationId, ConvoMetadata,
-    DeliveryService, GroupV2Config, IdentityProvider, MessageId, RegistrationService,
+    DeliveryService, GroupV2Config, GroupV2Phase, IdentityProvider, MessageId, RegistrationService,
     StorageConfig,
 };
 // The directory trait bounds ChatClient's registry parameter, so callers

crates/generic-chat/tests/group_v2.rs

@@ -12,7 +12,7 @@ use libchat::ChatStorage;
 use logos_account::TestLogosAccount;
 use logos_generic_chat::{
     ChatClient, ChatClientBuilder, ConversationClass, DelegateSigner, Event, GroupMetadata,
-    GroupV2Config, InProcessDelivery, MessageBus,
+    GroupV2Config, GroupV2Phase, InProcessDelivery, MessageBus,
 };
 
 /// Metadata for a group these tests create without a name or description.
@@ -539,3 +539,51 @@ fn a_sent_message_is_acknowledged_by_the_peers_that_reply() {
     expected.sort();
     assert_eq!(holders, expected, "both replying peers should be listed");
 }
+
+/// The commit-and-recovery cycle reaches the application.
+///
+/// A group that stops moving is otherwise silent: the roster keeps reporting
+/// whatever it last committed, and the account of why sits in de-mls's own log.
+/// Adding a member takes the creator through a freeze and a selection, so its
+/// channel has to carry them.
+#[test]
+fn group_v2_phase_changes_reach_the_application() {
+    let bus = MessageBus::default();
+    let reg = EphemeralRegistry::new();
+
+    let (mut saro, saro_events, _saro_addr) = create_test_client(bus.clone(), reg.clone());
+    let (_raya, _raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone());
+
+    // An empty group and then an add, rather than a group created around its
+    // members: only the add runs a commit round.
+    let convo_id = saro
+        .create_group_conversation(&[], unnamed_group())
+        .expect("saro create group");
+    saro.add_group_members(&convo_id, &[&raya_addr])
+        .expect("saro add raya");
+
+    // A conversation opens in `Working`, so the phases worth seeing are the
+    // ones the commit goes through: the freeze that collects candidates, and
+    // the selection that picks one.
+    let mut seen = Vec::new();
+    wait_for_event(
+        &saro_events,
+        "saro selecting a commit candidate",
+        Duration::from_secs(10),
+        |e| match e {
+            Event::ConversationPhaseChanged {
+                convo_id: id,
+                phase,
+            } if id.as_ref() == convo_id => {
+                seen.push(*phase);
+                (*phase == GroupV2Phase::Selection).then_some(())
+            }
+            _ => None,
+        },
+    );
+
+    assert!(
+        seen.contains(&GroupV2Phase::Freezing),
+        "the freeze that minted the commit went unreported :: {seen:?}"
+    );
+}