fix(cgka-engine): phantom committed_from from failed commit staging poisons fork detection

crates/cgka-conformance-simulator/SCENARIOS.md

@@ -190,6 +190,17 @@ These are real simulator scenarios that are still tied to Rust harness details.
   set.
 - Reason: the test inspects recovery observations and exact branch ordering keys.
 
+### `failed_invite_staging_does_not_poison_fork_detection_via_harness`
+
+- Setup: Alice's invite fails during commit staging (duplicate-member KeyPackage). Alice then advances to epoch 2 by
+  settling Bob's commit through convergence, and a partitioned Dave later delivers a sibling commit from the same
+  source epoch.
+- Expected: the sibling commit is adjudicated deterministically (stale losing branch or reorg onto the winner) — never
+  a fail-closed `ForkedEpoch` — and Alice's group stays usable for a follow-up invite.
+- Reason: regression guard for the phantom `committed_from` bookkeeping bug (fixed by recording it only inside
+  `begin_pending`); mirrors the cgka-engine test `failed_invite_staging_does_not_poison_fork_detection` at the
+  multi-client bus level.
+
 ### `convergence-e2e-group-events/v1`
 
 - Setup: Alice creates a four-member group. Alice and Bob race invite commits and each branch has an app payload. Carol

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

@@ -567,15 +567,25 @@ impl HarnessClient {
 
     /// Invite new members to the default group.
     pub async fn invite(&mut self, kps: Vec<KeyPackage>) -> PendingStateRef {
+        self.try_invite(kps).await.expect("send invite")
+    }
+
+    /// Invite new members to the default group, surfacing the engine error
+    /// instead of panicking. Nothing reaches the bus on failure. Used by
+    /// scenarios that deliberately fail an invite during commit staging
+    /// (e.g. the phantom committed-from regression).
+    pub async fn try_invite(
+        &mut self,
+        kps: Vec<KeyPackage>,
+    ) -> Result<PendingStateRef, EngineError> {
         let gid = self.default_group.clone().expect("group");
         let res = self
             .engine
             .send(SendIntent::Invite {
                 group_id: gid.clone(),
                 key_packages: kps,
             })
-            .await
-            .expect("send invite");
+            .await?;
         match res {
             SendResult::GroupEvolution {
                 msg,
@@ -589,9 +599,11 @@ impl HarnessClient {
                     self.bus.send(self.bus_id, w);
                 }
                 self.bus.send(self.bus_id, route(msg, &gid));
-                pending
+                Ok(pending)
             }
-            other => panic!("expected GroupEvolution, got {other:?}"),
+            other => Err(EngineError::Other(format!(
+                "expected GroupEvolution, got {other:?}"
+            ))),
         }
     }
 

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

@@ -2041,3 +2041,121 @@ async fn harness_captures_and_asserts_convergence_decision() {
         "carol should observe the committer-decided convergence decision: {failures:#?}"
     );
 }
+
+#[tokio::test]
+async fn failed_invite_staging_does_not_poison_fork_detection_via_harness() {
+    // Harness mirror of the cgka-engine regression test
+    // `failed_invite_staging_does_not_poison_fork_detection`: a send-path
+    // staging failure must leave no phantom "we committed from this epoch"
+    // bookkeeping behind. Historically it did, and once the client advanced
+    // past that epoch via a PEER's commit (settled through convergence, so
+    // no fork-recovery incumbent of its own exists), a legitimate sibling
+    // commit for the poisoned epoch failed closed with ForkedEpoch and stuck
+    // the group in Recovering.
+    let bus = TransportBus::ordered();
+    let mut alice = ClientBuilder::new(pad32(b"phantom-alice"))
+        .registry(selfremove_registry())
+        .attach(&bus);
+    let mut bob = ClientBuilder::new(pad32(b"phantom-bob"))
+        .registry(selfremove_registry())
+        .attach(&bus);
+    let mut carol = ClientBuilder::new(pad32(b"phantom-carol"))
+        .registry(selfremove_registry())
+        .attach(&bus);
+    let mut dave = ClientBuilder::new(pad32(b"phantom-dave"))
+        .registry(selfremove_registry())
+        .attach(&bus);
+    let mut erin = ClientBuilder::new(pad32(b"phantom-erin"))
+        .registry(selfremove_registry())
+        .attach(&bus);
+
+    // Bob and Dave are admins: each later commits an invite from epoch 1.
+    let bob_kp = bob.fresh_key_package().await;
+    let dave_kp = dave.fresh_key_package().await;
+    let (_group_id, pending) = alice
+        .create_group_with_admins(
+            "phantom-committed-from",
+            vec![bob_kp, dave_kp],
+            vec![],
+            vec![bob.member_id(), dave.member_id()],
+        )
+        .await;
+    alice.confirm(pending).await;
+    bus.deliver_all();
+    bob.tick().await;
+    dave.tick().await;
+    assert_eq!(alice.epoch().0, 1);
+    assert_eq!(bob.epoch().0, 1);
+    assert_eq!(dave.epoch().0, 1);
+
+    // Alice's invite fails DURING commit staging: the duplicate-Bob
+    // KeyPackage carries Bob's existing leaf signature key, which OpenMLS
+    // rejects inside add_members — after capability validation, before the
+    // pending-publish state transition. Nothing reaches the bus.
+    let duplicate_bob_kp = bob.fresh_key_package().await;
+    let failed = alice.try_invite(vec![duplicate_bob_kp]).await;
+    assert!(
+        failed.is_err(),
+        "duplicate-member invite must fail during staging"
+    );
+
+    // Partition Dave away so he stays at epoch 1 and never sees Bob's
+    // commit. Bob invites Carol; Alice advances to epoch 2 purely by
+    // settling Bob's commit through convergence (peer-driven advance — no
+    // own commit, no fork-recovery incumbent at epoch 1).
+    bus.set_partition(Some(vec![alice.bus_id, bob.bus_id, carol.bus_id]));
+    let carol_kp = carol.fresh_key_package().await;
+    let bob_pending = bob.invite(vec![carol_kp]).await;
+    bob.confirm(bob_pending).await;
+    bus.deliver_all();
+    carol.tick().await;
+    let alice_outcomes = alice.tick().await;
+    assert_eq!(
+        alice.epoch().0,
+        2,
+        "alice must settle bob's commit via convergence: {alice_outcomes:?}"
+    );
+
+    // Heal the partition. Dave — still at epoch 1 — commits a sibling
+    // invite from the epoch Alice's failed staging attempt touched.
+    bus.set_partition(None);
+    let erin_kp = erin.fresh_key_package().await;
+    let dave_pending = dave.invite(vec![erin_kp]).await;
+    dave.confirm(dave_pending).await;
+    bus.deliver_all();
+
+    // Alice must classify the sibling (stale losing branch, or a
+    // deterministic reorg onto the winning branch) — never fail closed with
+    // ForkedEpoch, which left the group stuck in Recovering.
+    let alice_outcomes = alice.tick().await;
+    let alice_forked = alice_outcomes
+        .iter()
+        .any(|o| matches!(o, Err(cgka_traits::EngineError::ForkedEpoch { .. })));
+    assert!(
+        !alice_forked,
+        "sibling commit after failed staging must not fail closed: {alice_outcomes:?}"
+    );
+    assert_eq!(alice.epoch().0, 2);
+
+    // Whichever branch won deterministically, alice holds exactly one of
+    // the two invitees and the group remains operational.
+    let members = alice.members();
+    let has_carol = members.iter().any(|m| m.id == carol.member_id());
+    let has_erin = members.iter().any(|m| m.id == erin.member_id());
+    assert_ne!(
+        has_carol, has_erin,
+        "exactly one branch must win: {members:?}"
+    );
+    // Usability probe: a fresh valid invite from Alice must stage normally.
+    // A group stuck in Recovering rejects the send outright.
+    let mut frank = ClientBuilder::new(pad32(b"phantom-frank"))
+        .registry(selfremove_registry())
+        .attach(&bus);
+    let frank_kp = frank.fresh_key_package().await;
+    let probe_pending = alice
+        .try_invite(vec![frank_kp])
+        .await
+        .expect("group must remain usable after failed staging + sibling commit");
+    alice.confirm(probe_pending).await;
+    assert_eq!(alice.epoch().0, 3);
+}

crates/cgka-engine/src/epoch_manager.rs

@@ -243,15 +243,6 @@ impl EpochManager {
         Ok((group_id, prior_epoch))
     }
 
-    /// Record a committed-from epoch outside the begin_pending path (used
-    /// by the auto-committer, which doesn't go through PendingPublish).
-    pub(crate) fn record_committed_from(&mut self, group_id: &GroupId, epoch: EpochId) {
-        self.committed_from
-            .entry(group_id.clone())
-            .or_default()
-            .insert(epoch);
-    }
-
     pub(crate) fn prune_committed_from_before(
         &mut self,
         group_id: &GroupId,

crates/cgka-engine/src/message_processor/send.rs

@@ -124,8 +124,14 @@ impl<S: StorageProvider> Engine<S> {
             parsed_kps.push(parsed);
         }
 
-        // Record the pre-commit epoch so a later same-epoch commit can be
-        // compared against this locally produced branch.
+        // The pre-commit epoch is the commit origin for fork detection.
+        // `committed_from` bookkeeping is deliberately NOT recorded here: it
+        // happens atomically inside `begin_pending` below, after staging
+        // succeeds. Recording it earlier left a phantom "we committed from
+        // this epoch" entry behind when staging failed (the cleanup guard
+        // clears the OpenMLS commit but nothing prunes `committed_from`),
+        // which later mis-routed legitimate sibling commits into
+        // fail-closed fork recovery.
         let pre_commit_epoch = EpochId(mls_group.epoch().as_u64());
         // Arm the cleanup guard before creating the snapshot so the snapshot is
         // released on early return / cancellation even before a pending commit
@@ -142,8 +148,6 @@ impl<S: StorageProvider> Engine<S> {
             pre_commit_epoch,
             "pre_invite_commit",
         );
-        self.epoch_manager
-            .record_committed_from(&group_id, pre_commit_epoch);
         let pre_commit_ctx =
             crate::group_lifecycle::build_group_context_snapshot(&mls_group, &provider)?;
 
@@ -464,8 +468,6 @@ impl<S: StorageProvider> Engine<S> {
         )?;
         let commit_priority =
             crate::app_components::commit_ordering_priority_for_staged(staged_commit);
-        self.epoch_manager
-            .record_committed_from(&group_id, pre_commit_epoch);
 
         let commit_bytes = commit_out
             .tls_serialize_detached()

crates/cgka-engine/src/update_group_data.rs

@@ -210,8 +210,9 @@ impl<S: StorageProvider> Engine<S> {
             pre_commit_epoch,
             "pre_update_group_data_commit",
         );
-        self.epoch_manager
-            .record_committed_from(&group_id, pre_commit_epoch);
+        // `committed_from` is recorded atomically by `begin_pending` after
+        // staging succeeds — never here, where a staging failure would leave
+        // a phantom entry behind (see do_send_invite).
         let pre_commit_ctx =
             crate::group_lifecycle::build_group_context_snapshot(&mls_group, &provider)?;
 

crates/cgka-engine/src/upgrade.rs

@@ -158,8 +158,9 @@ impl<S: StorageProvider> Engine<S> {
             pre_commit_epoch,
             "pre_upgrade_commit",
         );
-        self.epoch_manager
-            .record_committed_from(group_id, pre_commit_epoch);
+        // `committed_from` is recorded atomically by `begin_pending` after
+        // staging succeeds — never here, where a staging failure would leave
+        // a phantom entry behind (see do_send_invite).
         let pre_commit_ctx =
             crate::group_lifecycle::build_group_context_snapshot(&mls_group, &provider)?;
 

crates/cgka-engine/tests/fork_detection.rs

@@ -687,3 +687,131 @@ async fn stale_commit_without_own_commit_is_classified_as_already_at_epoch_not_f
         }
     ));
 }
+
+#[tokio::test]
+async fn failed_invite_staging_does_not_poison_fork_detection() {
+    // Regression: `do_send_invite` used to record `committed_from` BEFORE
+    // staging the commit. When staging failed, the cleanup guard cleared the
+    // OpenMLS pending commit but nothing pruned the phantom "we committed
+    // from this epoch" entry. Once Alice later advanced past that epoch via a
+    // PEER's commit (settled through convergence — so no fork-recovery
+    // incumbent of her own exists), a legitimate same-epoch sibling commit
+    // was mis-routed: the phantom blocked convergence entry, the WrongEpoch
+    // fork branch found no recovery snapshot, and ingest failed closed with
+    // ForkedEpoch, sticking the group in Recovering. `committed_from` is now
+    // recorded only inside `begin_pending`, atomically with the transition.
+    let (mut alice, _alice_storage) = build_client_with_storage(b"phantom-alice");
+    let mut bob = build_client(b"phantom-bob");
+    let mut carol = build_client(b"phantom-carol");
+    let (mut dave, dave_storage) = build_client_with_storage(b"phantom-dave");
+    let dave_id = MemberId::new(pad32(b"phantom-dave"));
+
+    let bob_kp = bob.fresh_key_package().await.unwrap();
+    let dave_kp = dave.fresh_key_package().await.unwrap();
+    let (group_id, create) = alice
+        .create_group(CreateGroupRequest {
+            name: "".into(),
+            description: "".into(),
+            members: vec![bob_kp, dave_kp],
+            required_features: vec![],
+            app_components: vec![],
+            // Bob must be an admin so his sibling-epoch invite below passes
+            // the MIP-03 committer guard.
+            initial_admins: vec![MemberId::new(pad32(b"phantom-bob"))],
+        })
+        .await
+        .unwrap();
+    let (bob_welcome, dave_welcome) = match create {
+        SendResult::GroupCreated {
+            pending,
+            mut welcomes,
+        } => {
+            alice.confirm_published(pending).await.unwrap();
+            let dave_welcome = welcomes.remove(1);
+            (welcomes.remove(0), dave_welcome)
+        }
+        _ => unreachable!(),
+    };
+    bob.join_welcome(bob_welcome).await.unwrap();
+    dave.join_welcome(dave_welcome).await.unwrap();
+
+    // Dave produces (but never publishes/applies) a sibling commit from the
+    // current epoch — the legitimate same-epoch race Alice must classify
+    // later.
+    let dave_sibling_commit = raw_self_update_commit(&dave_storage, &dave_id, &group_id);
+
+    // Alice's invite fails DURING staging: the duplicate-Bob KeyPackage is
+    // signed with Bob's existing leaf signature key, which OpenMLS rejects
+    // inside `add_members` — after capability validation, before
+    // `begin_pending`.
+    let duplicate_bob_kp = bob.fresh_key_package().await.unwrap();
+    let failed = alice
+        .send(SendIntent::Invite {
+            group_id: group_id.clone(),
+            key_packages: vec![duplicate_bob_kp],
+        })
+        .await;
+    assert!(
+        failed.is_err(),
+        "duplicate-member invite must fail during staging"
+    );
+
+    // Bob commits from the same epoch (invites Carol). Alice advances by
+    // settling Bob's commit through convergence — a peer-driven advance, so
+    // she has no fork-recovery incumbent of her own for the source epoch.
+    let carol_kp = carol.fresh_key_package().await.unwrap();
+    let invite = bob
+        .send(SendIntent::Invite {
+            group_id: group_id.clone(),
+            key_packages: vec![carol_kp],
+        })
+        .await
+        .unwrap();
+    let (bob_commit, pending) = match invite {
+        SendResult::GroupEvolution { msg, pending, .. } => (msg, pending),
+        _ => unreachable!(),
+    };
+    bob.confirm_published(pending).await.unwrap();
+    let routed_bob_commit = TransportMessage {
+        envelope: TransportEnvelope::GroupMessage {
+            transport_group_id: group_id.as_slice().to_vec(),
+        },
+        ..bob_commit
+    };
+    alice
+        .buffer_openmls_convergence_message(&group_id, routed_bob_commit, 1_000)
+        .expect("bob commit buffered");
+    alice
+        .converge_stored_openmls_messages(&group_id, 3_000)
+        .expect("bob commit settles");
+    assert_eq!(alice.epoch(&group_id).unwrap(), EpochId(2));
+
+    // Dave's sibling commit for the now-past epoch arrives. Alice never
+    // published a commit from that epoch, so ingest must classify it (stale
+    // losing branch, or a reorg onto the winning branch) — NOT take the
+    // fork-recovery path, which with the phantom entry and no tracked
+    // snapshot failed closed with ForkedEpoch and left the group stuck in
+    // Recovering.
+    alice
+        .ingest(dave_sibling_commit)
+        .await
+        .expect("sibling commit after failed staging must not fail closed");
+
+    // The group is still operational: a fresh (valid) invite from Alice is
+    // accepted — staged immediately or queued behind unresolved convergence
+    // input. In Recovering it would error instead (`begin_pending` rejects
+    // non-Stable states).
+    let mut erin = build_client(b"phantom-erin");
+    let erin_kp = erin.fresh_key_package().await.unwrap();
+    let recovery_probe = alice
+        .send(SendIntent::Invite {
+            group_id: group_id.clone(),
+            key_packages: vec![erin_kp],
+        })
+        .await
+        .expect("group must remain usable after failed staging + sibling commit");
+    assert!(matches!(
+        recovery_probe,
+        SendResult::GroupEvolution { .. } | SendResult::Queued { .. }
+    ));
+}