Group lifecycle: participation taxonomy + eviction detection, recovery probe, removal notices, quarantine (#727)

crates/cgka-conformance-simulator/src/client.rs

@@ -862,6 +862,9 @@ impl HarnessClient {
                 ..msg.clone()
             }),
             PeeledContent::Welcome { .. } => Err("group peeler returned a welcome".into()),
+            PeeledContent::RemovalNotice { .. } => {
+                Err("group peeler returned a removal notice".into())
+            }
         }
     }
 }

crates/cgka-conformance-simulator/tests/openmls_replay_probe.rs

@@ -1426,6 +1426,8 @@ fn dummy_group(group_id: GroupId) -> Group {
         name: "probe".to_owned(),
         description: String::new(),
         epoch: EpochId(1),
+        participation: Default::default(),
+        membership_intervals: Vec::new(),
         members: vec![Member {
             id: MemberId::new(vec![1]),
             credential: vec![1],

crates/cgka-engine/src/audit_helpers.rs

@@ -77,6 +77,9 @@ pub(crate) fn stale_reason_str(reason: &StaleReason) -> &'static str {
         StaleReason::UnknownGroup => "unknown_group",
         StaleReason::OwnEcho => "own_echo",
         StaleReason::PeelFailed => "peel_failed",
+        StaleReason::Evicted => "evicted",
+        StaleReason::Quarantined => "quarantined",
+        StaleReason::PreMembership => "pre_membership",
     }
 }
 

crates/cgka-engine/src/distributed_convergence.rs

@@ -428,6 +428,7 @@ impl<S: StorageProvider> Engine<S> {
                 selected_tip,
                 origin_commit_id,
                 origin_commit_actor,
+                &observations,
             )?;
         }
         self.emit_application_replay_events(group_id, &observations);
@@ -500,6 +501,7 @@ impl<S: StorageProvider> Engine<S> {
         selected_tip: EpochId,
         origin_commit_id: Option<MessageId>,
         origin_commit_actor: Option<MemberId>,
+        observations: &[OpenMlsReplayObservation],
     ) -> Result<(), OpenMlsProjectionError> {
         if previous_tip != selected_tip {
             self.events_buf.push_back(GroupEvent::EpochChanged {
@@ -593,18 +595,57 @@ impl<S: StorageProvider> Engine<S> {
                 );
             }
         }
+        // SelfRemove senders consumed by the replayed commits, so departures
+        // attribute to the leaver here exactly like the direct-ingest seam
+        // (`MemberLeft` vs `MemberRemoved` is no longer path-dependent).
+        let self_removed: HashSet<MemberId> = observations
+            .iter()
+            .flat_map(|observation| match observation {
+                OpenMlsReplayObservation::CommitStaged {
+                    self_remove_senders,
+                    ..
+                } => self_remove_senders.as_slice(),
+                _ => &[],
+            })
+            .map(|identity| MemberId::new(identity.clone()))
+            .collect();
         for member_id in previous_ids.difference(&current_ids) {
             if member_id == self.identity.self_id() {
                 self.clear_leave_request_state(group_id)
                     .map_err(|e| OpenMlsProjectionError::Storage(format!("{e:?}")))?;
+                // Applied removal via the convergence path: the same
+                // authoritative participation transition as the direct seam
+                // (spec group-state.md, "Reaching a non-member state").
+                let participation = if self_removed.contains(self.identity.self_id()) {
+                    cgka_traits::GroupParticipation::Left
+                } else {
+                    cgka_traits::GroupParticipation::Evicted
+                };
+                self.set_group_participation(group_id, participation)
+                    .map_err(|e| OpenMlsProjectionError::Storage(format!("{e:?}")))?;
             }
+            let (change, actor) = if self_removed.contains(member_id) {
+                // A leave is attributed to the leaver, not the member that
+                // sequenced the auto-commit.
+                (
+                    GroupStateChange::MemberLeft {
+                        member: member_id.clone(),
+                    },
+                    Some(member_id.clone()),
+                )
+            } else {
+                (
+                    GroupStateChange::MemberRemoved {
+                        member: member_id.clone(),
+                    },
+                    None,
+                )
+            };
             self.push_group_state_change(
                 group_id,
                 selected_tip,
-                None,
-                GroupStateChange::MemberRemoved {
-                    member: member_id.clone(),
-                },
+                actor,
+                change,
                 origin_commit_id.clone(),
             );
         }

crates/cgka-engine/src/engine.rs

@@ -136,6 +136,12 @@ pub struct Engine<S: StorageProvider> {
     /// Fast runtime gate for groups where the local member must not produce
     /// further outbound group traffic while a leave request is outstanding.
     pub(crate) leaving_groups: HashSet<GroupId>,
+    /// Groups mid-`resolve_group_quarantine`: the explicit recovery pass
+    /// re-runs withheld input through ordinary ingest, so the quarantine
+    /// withhold guard stands down for exactly these groups for exactly that
+    /// pass — the mechanism behind "no silent re-activation" (spec
+    /// group-state.md, "Quarantine").
+    pub(crate) quarantine_resolutions: HashSet<GroupId>,
 
     /// Delayed SelfRemove auto-commit attempts keyed by the standalone
     /// proposal's content-derived message id. These are re-checked against live
@@ -339,6 +345,7 @@ impl<S: StorageProvider> EngineBuilder<S> {
             sent_message_ids: BoundedIdSet::with_capacity(DEDUP_CACHE_CAPACITY),
             leave_requests: HashMap::new(),
             leaving_groups: HashSet::new(),
+            quarantine_resolutions: HashSet::new(),
             scheduled_self_remove_auto_commits: HashMap::new(),
             pending_convergence_groups: HashSet::new(),
             convergence_policy: crate::canonicalization::CanonicalizationPolicy::default(),
@@ -1001,6 +1008,76 @@ impl<S: StorageProvider> Engine<S> {
         Ok(())
     }
 
+    /// Transition the local identity's durable participation record for a
+    /// group and surface the change as `GroupEvent::ParticipationChanged`.
+    /// Participation lives on the durable Marmot group record
+    /// (`spec/protocol-core/group-state.md`, "Participation"), so it survives
+    /// live-OpenMLS-state teardown and rolls back together with the record
+    /// when fork recovery restores a pre-commit snapshot. Idempotent: writing
+    /// the already-current state emits nothing.
+    pub(crate) fn set_group_participation(
+        &mut self,
+        group_id: &GroupId,
+        participation: cgka_traits::GroupParticipation,
+    ) -> Result<(), EngineError> {
+        let mut group = match self.storage.get_group(group_id) {
+            Ok(group) => group,
+            // No durable record: nothing to transition. Participation exists
+            // only for groups the engine holds a record of; "no such group"
+            // stays distinguishable at the accessor.
+            Err(cgka_traits::StorageError::NotFound) => return Ok(()),
+            Err(e) => return Err(EngineError::Backend(format!("get_group: {e:?}"))),
+        };
+        if group.participation == participation {
+            return Ok(());
+        }
+        group.participation = participation;
+        // Membership intervals ride the same transition (errors.md,
+        // `PreMembership`): ending membership closes the open interval at the
+        // epoch the removing commit reached; a restored `Member` (rejoin or
+        // quarantine resolution) opens a new one. Quarantine holds change no
+        // interval — they assert nothing about membership.
+        match participation {
+            cgka_traits::GroupParticipation::Left | cgka_traits::GroupParticipation::Evicted => {
+                if let Some(open) = group
+                    .membership_intervals
+                    .iter_mut()
+                    .find(|interval| interval.ended_at.is_none())
+                {
+                    // The removing commit advanced the record to `epoch`; the
+                    // last epoch this identity held keys for is the commit's
+                    // SOURCE epoch (`epoch - 1`) — post-removal epochs were
+                    // never readable, so they stay outside the interval.
+                    open.ended_at = Some(EpochId(group.epoch.0.saturating_sub(1)));
+                }
+            }
+            cgka_traits::GroupParticipation::Member => {
+                if !group
+                    .membership_intervals
+                    .iter()
+                    .any(|interval| interval.ended_at.is_none())
+                {
+                    group
+                        .membership_intervals
+                        .push(cgka_traits::MembershipInterval {
+                            joined_at: group.epoch,
+                            ended_at: None,
+                        });
+                }
+            }
+            cgka_traits::GroupParticipation::Quarantined { .. } => {}
+        }
+        self.storage
+            .put_group(&group)
+            .map_err(|e| EngineError::Backend(format!("put_group: {e:?}")))?;
+        self.events_buf
+            .push_back(cgka_traits::engine::GroupEvent::ParticipationChanged {
+                group_id: group_id.clone(),
+                participation,
+            });
+        Ok(())
+    }
+
     fn sent_self_remove_leaving_gate_should_restore(
         &self,
         group_id: &GroupId,
@@ -1522,11 +1599,114 @@ impl<S: StorageProvider + 'static> CgkaEngine for Engine<S> {
         &mut self,
         group_id: &GroupId,
     ) -> Result<Vec<SendResult>, EngineError> {
+        // A withheld group is excluded from live convergence (spec
+        // group-state.md, "Quarantine"); only the explicit
+        // `resolve_group_quarantine` transition may process its stored input.
+        match self.storage.get_group(group_id).map(|g| g.participation) {
+            Ok(cgka_traits::GroupParticipation::Quarantined { .. }) => {
+                return Ok(Vec::new());
+            }
+            Ok(_) | Err(cgka_traits::StorageError::NotFound) => {}
+            Err(e) => return Err(EngineError::Backend(format!("get_group: {e:?}"))),
+        }
         let now_ms = self.convergence_now_ms();
         self.converge_and_drain_queued_outbound_intents(group_id, now_ms)
             .await
     }
 
+    async fn quarantine_group(
+        &mut self,
+        group_id: &GroupId,
+        reason: cgka_traits::QuarantineReason,
+    ) -> Result<(), EngineError> {
+        let group = match self.storage.get_group(group_id) {
+            Ok(group) => group,
+            Err(cgka_traits::StorageError::NotFound) => {
+                return Err(EngineError::UnknownGroup(group_id.clone()));
+            }
+            Err(e) => return Err(EngineError::Backend(format!("get_group: {e:?}"))),
+        };
+        // Quarantine is a hold, not a verdict: never downgrade an
+        // authoritative non-member state to "withheld".
+        if matches!(
+            group.participation,
+            cgka_traits::GroupParticipation::Left | cgka_traits::GroupParticipation::Evicted
+        ) {
+            return Ok(());
+        }
+        self.set_group_participation(
+            group_id,
+            cgka_traits::GroupParticipation::Quarantined { reason },
+        )
+    }
+
+    async fn resolve_group_quarantine(
+        &mut self,
+        group_id: &GroupId,
+    ) -> Result<cgka_traits::GroupParticipation, EngineError> {
+        let current = self
+            .participation(group_id)?
+            .ok_or_else(|| EngineError::UnknownGroup(group_id.clone()))?;
+        if !matches!(current, cgka_traits::GroupParticipation::Quarantined { .. }) {
+            return Ok(current);
+        }
+        // One deliberate convergence pass over the withheld/stored input. The
+        // ordinary apply seams decide the outcome: a stored removal commit
+        // resolves participation to Left/Evicted through the same code path
+        // as live traffic. The withhold guard stands down for this group for
+        // exactly this pass, and the pass runs past the settlement quiescence
+        // window: resolution is an explicit decision over a closed input set,
+        // not live traffic waiting for stragglers.
+        self.quarantine_resolutions.insert(group_id.clone());
+        let quiescence_ms = self
+            .convergence_policy_for_group(group_id)
+            .map(|policy| policy.settlement_quiescence_ms)
+            .unwrap_or_default();
+        // The pass re-ingests withheld rows, stamping fresh receive times
+        // *after* this snapshot; a bare `window + 1` margin would leave them
+        // inside quiescence forever. One minute of slack makes the closed
+        // input set decisively quiescent without touching wall-clock state.
+        let now_ms = self
+            .convergence_now_ms()
+            .saturating_add(quiescence_ms.saturating_add(60_000));
+        let pass = self
+            .advance_convergence_inputs_until_settled(group_id, now_ms)
+            .await;
+        self.quarantine_resolutions.remove(group_id);
+        let _ = pass?;
+        let after = self.participation(group_id)?.unwrap_or(current);
+        if !matches!(after, cgka_traits::GroupParticipation::Quarantined { .. }) {
+            return Ok(after);
+        }
+        // No authoritative outcome from the pass. Restore Member only when the
+        // live MLS state is active and the canonical roster still contains the
+        // local identity — otherwise the hold stands.
+        let provider = crate::provider::EngineOpenMlsProvider::<S>::new(
+            &self.crypto,
+            self.storage.mls_storage(),
+        );
+        let mls_gid = openmls::group::GroupId::from_slice(group_id.as_slice());
+        let storage = <crate::provider::EngineOpenMlsProvider<'_, S> as openmls_traits::OpenMlsProvider>::storage(
+            &provider,
+        );
+        let live_group = openmls::group::MlsGroup::load(storage, &mls_gid)
+            .map_err(|e| EngineError::Backend(format!("load live group: {e:?}")))?;
+        let self_in_roster = self
+            .storage
+            .get_group(group_id)
+            .map_err(|e| EngineError::Backend(format!("get_group: {e:?}")))?
+            .members
+            .iter()
+            .any(|member| &member.id == self.identity.self_id());
+        let live_and_member =
+            live_group.is_some_and(|mls_group| mls_group.is_active()) && self_in_roster;
+        if live_and_member {
+            self.set_group_participation(group_id, cgka_traits::GroupParticipation::Member)?;
+            return Ok(cgka_traits::GroupParticipation::Member);
+        }
+        Ok(after)
+    }
+
     async fn confirm_published(
         &mut self,
         pending: PendingStateRef,
@@ -1731,6 +1911,28 @@ impl<S: StorageProvider + 'static> CgkaEngine for Engine<S> {
         self.do_members(group_id)
     }
 
+    async fn wrap_removal_notice(
+        &self,
+        commit: &TransportMessage,
+        recipient: &MemberId,
+    ) -> Result<Option<TransportMessage>, EngineError> {
+        self.peeler
+            .wrap_removal_notice(commit, recipient)
+            .await
+            .map_err(EngineError::Peeler)
+    }
+
+    fn participation(
+        &self,
+        group_id: &GroupId,
+    ) -> Result<Option<cgka_traits::GroupParticipation>, EngineError> {
+        match self.storage.get_group(group_id) {
+            Ok(group) => Ok(Some(group.participation)),
+            Err(cgka_traits::StorageError::NotFound) => Ok(None),
+            Err(e) => Err(EngineError::Backend(format!("get_group: {e:?}"))),
+        }
+    }
+
     fn epoch(&self, group_id: &GroupId) -> Result<EpochId, EngineError> {
         self.epoch_manager
             .epoch(group_id)