Fix join flow: skip stale groups in discovery + add sync fallback

.gitignore

@@ -129,3 +129,4 @@ qa/cxdb/qa.sqlite
 qa/cxdb/qa.sqlite-wal
 qa/cxdb/qa.sqlite-shm
 .DS_Store
+.mote-preview.html

ConvosCore/Sources/ConvosCore/Inboxes/ConversationStateMachine.swift

@@ -95,6 +95,9 @@ public actor ConversationStateMachine {
     // Database observation task for tracking conversation join
     private var observationTask: Task<String, Error>?
 
+    // Discovery loop task for syncing welcomes during join flow
+    private var discoveryTask: Task<Void, Never>?
+
     // MARK: - State Observation
 
     private var stateContinuations: [AsyncStream<State>.Continuation] = []
@@ -169,6 +172,7 @@ public actor ConversationStateMachine {
         messageStreamContinuation?.finish()
         messageProcessingTask?.cancel()
         observationTask?.cancel()
+        discoveryTask?.cancel()
     }
 
     private func setupMessageStream() {
@@ -668,12 +672,35 @@ public actor ConversationStateMachine {
             inviteTag: invite.invitePayload.tag
         )
 
+        // The XMTP conversation stream does not call sync_welcomes() on startup,
+        // so it cannot discover groups the client is added to after subscription.
+        // Run a discovery loop that periodically syncs welcomes and processes any
+        // newly discovered groups. The DB observation above fires as soon as the
+        // group is written to the database by discoverNewConversations.
+        discoveryTask = Task { [weak self] in
+            var interval: Duration = .seconds(3)
+            let maxInterval: Duration = .seconds(15)
+            while !Task.isCancelled {
+                do {
+                    try await Task.sleep(for: interval)
+                    try Task.checkCancellation()
+                    Log.info("Join flow: triggering discovery sync...")
+                    await self?.inboxStateManager.requestDiscovery()
+                    interval = min(interval * 2, maxInterval)
+                } catch {
+                    break
+                }
+            }
+        }
+
         do {
             guard let task = observationTask else {
                 throw ConversationStateMachineError.timedOut
             }
             let conversationId = try await task.value
             observationTask = nil
+            discoveryTask?.cancel()
+            discoveryTask = nil
 
             Log.info("Conversation joined successfully: \(conversationId)")
             QAEvent.emit(.conversation, "joined", ["id": conversationId])
@@ -691,6 +718,8 @@ public actor ConversationStateMachine {
             )))
         } catch is CancellationError {
             observationTask = nil
+            discoveryTask?.cancel()
+            discoveryTask = nil
             await clearInviteJoinErrorHandler()
             Log.debug("Conversation join observation cancelled")
             guard case .joinFailed = _state else {
@@ -699,6 +728,8 @@ public actor ConversationStateMachine {
             Log.debug("Already in joinFailed state, not propagating cancellation")
         } catch {
             observationTask = nil
+            discoveryTask?.cancel()
+            discoveryTask = nil
             await clearInviteJoinErrorHandler()
             Log.error("Error waiting for conversation to join: \(error)")
             guard case .joinFailed = _state else {
@@ -757,6 +788,8 @@ extension ConversationStateMachine {
         messageProcessingTask?.cancel()
         observationTask?.cancel()
         observationTask = nil
+        discoveryTask?.cancel()
+        discoveryTask = nil
 
         if let conversationId {
             let inboxReady = try await inboxStateManager.waitForInboxReadyResult()
@@ -863,6 +896,8 @@ extension ConversationStateMachine {
         messageProcessingTask?.cancel()
         observationTask?.cancel()
         observationTask = nil
+        discoveryTask?.cancel()
+        discoveryTask = nil
         await clearInviteJoinErrorHandler()
         emitStateChange(.uninitialized)
     }

ConvosCore/Sources/ConvosCore/Inboxes/InboxLifecycleManager.swift

@@ -60,7 +60,8 @@ public protocol InboxLifecycleManagerProtocol: Actor {
     func getOrCreateService(clientId: String, inboxId: String) -> any MessagingServiceProtocol
 
     func getOrWake(clientId: String, inboxId: String) async throws -> any MessagingServiceProtocol
-    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async
+    @discardableResult
+    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool
     func isAwake(clientId: String) -> Bool
     func isSleeping(clientId: String) -> Bool
     func rebalance() async
@@ -352,10 +353,11 @@ public actor InboxLifecycleManager: InboxLifecycleManagerProtocol {
         return try await wake(clientId: clientId, inboxId: inboxId, reason: .userInteraction)
     }
 
-    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async {
+    @discardableResult
+    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool {
         guard awakeInboxes[clientId] == nil else {
             Log.debug("Inbox already tracked, skipping external registration: \(clientId)")
-            return
+            return false
         }
         if awakeInboxes.count >= maxAwakeInboxes {
             let freed = await sleepLeastRecentlyUsed(excluding: [clientId])
@@ -365,12 +367,13 @@ public actor InboxLifecycleManager: InboxLifecycleManagerProtocol {
         }
         guard awakeInboxes[clientId] == nil else {
             Log.debug("Inbox registered by another task during eviction, skipping: \(clientId)")
-            return
+            return false
         }
         awakeInboxes[clientId] = service
         _sleepingClientIds.remove(clientId)
         _sleepTimes.removeValue(forKey: clientId)
         Log.debug("Registered external service: \(clientId), total awake: \(awakeInboxes.count)")
+        return true
     }
 
     public func isAwake(clientId: String) -> Bool {

ConvosCore/Sources/ConvosCore/Inboxes/InboxStateMachine.swift

@@ -309,6 +309,11 @@ public actor InboxStateMachine {
         await syncingManager.setInviteJoinErrorHandler(handler)
     }
 
+    func requestDiscovery() async {
+        guard let syncingManager else { return }
+        await syncingManager.requestDiscovery()
+    }
+
     // MARK: - Private
 
     private func enqueueAction(_ action: Action) {

ConvosCore/Sources/ConvosCore/Inboxes/InboxStateManager.swift

@@ -15,6 +15,8 @@ public protocol InboxStateManagerProtocol: AnyObject, Sendable {
     func waitForDeletionComplete() async
     func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async throws
 
+    func requestDiscovery() async
+
     func addObserver(_ observer: InboxStateObserver)
     func removeObserver(_ observer: InboxStateObserver)
 
@@ -163,6 +165,11 @@ public final class InboxStateManager: InboxStateManagerProtocol, @unchecked Send
         await stateMachine.setInviteJoinErrorHandler(handler)
     }
 
+    public func requestDiscovery() async {
+        guard let stateMachine else { return }
+        await stateMachine.requestDiscovery()
+    }
+
     public func reauthorize(inboxId: String, clientId: String) async throws -> InboxReadyResult {
         guard let stateMachine = stateMachine else {
             throw InboxStateError.inboxNotReady

ConvosCore/Sources/ConvosCore/Mocks/MockInboxLifecycleManager.swift

@@ -95,10 +95,12 @@ public actor MockInboxLifecycleManager: InboxLifecycleManagerProtocol {
         _sleepingClientIds.removeAll()
     }
 
-    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async {
+    @discardableResult
+    public func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool {
         mockServices[clientId] = service
         _awakeClientIds.insert(clientId)
         _sleepingClientIds.remove(clientId)
+        return true
     }
 
     public func prepareUnusedConversationIfNeeded() async {}

ConvosCore/Sources/ConvosCore/Mocks/MockInboxStateManager.swift

@@ -45,6 +45,9 @@ public final class MockInboxStateManager: InboxStateManagerProtocol, @unchecked
     public func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async throws {
     }
 
+    public func requestDiscovery() async {
+    }
+
     public func addObserver(_ observer: any InboxStateObserver) {
         observers.removeAll { $0.observer == nil }
         observers.append(WeakStateObserver(observer: observer))

ConvosCore/Sources/ConvosCore/Sessions/SessionManager.swift

@@ -284,7 +284,11 @@ public final class SessionManager: SessionManagerProtocol, @unchecked Sendable {
             apiClient: apiClient
         )
         Task { [lifecycleManager] in
-            await lifecycleManager.registerExternalService(service, clientId: clientId)
+            let registered = await lifecycleManager.registerExternalService(service, clientId: clientId)
+            if !registered {
+                Log.warning("Stopping duplicate MessagingService for \(clientId)")
+                await service.stop()
+            }
         }
         return service
     }

ConvosCore/Sources/ConvosCore/Syncing/SyncingManager.swift

@@ -10,6 +10,7 @@ public protocol SyncingManagerProtocol: Actor {
     func stop() async
     func pause() async
     func resume() async
+    func requestDiscovery() async
     func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async
 }
 
@@ -440,6 +441,12 @@ actor SyncingManager: SyncingManagerProtocol {
             var discoveredCount: Int = 0
             for group in groups where !existingIds.contains(group.id) {
                 do {
+                    let creatorInboxId = try await group.creatorInboxId()
+                    let memberCount = try await group.members.count
+                    if creatorInboxId == params.client.inboxId && memberCount <= 1 {
+                        Log.debug("Skipping self-created single-member group: \(group.id)")
+                        continue
+                    }
                     try await streamProcessor.processConversation(group, params: params)
                     discoveredCount += 1
                 } catch {
@@ -610,6 +617,22 @@ actor SyncingManager: SyncingManagerProtocol {
         await processJoinRequestsAfterSync(params: params)
     }
 
+    func requestDiscovery() async {
+        guard case .ready(let params) = _state else {
+            Log.debug("requestDiscovery ignored - not in ready state (\(_state))")
+            return
+        }
+        do {
+            let syncStart = CFAbsoluteTimeGetCurrent()
+            _ = try await params.client.conversationsProvider.syncAllConversations(consentStates: params.consentStates)
+            let syncElapsed = String(format: "%.0f", (CFAbsoluteTimeGetCurrent() - syncStart) * 1000)
+            Log.info("[PERF] sync.requestDiscovery: \(syncElapsed)ms")
+            await discoverNewConversations(params: params)
+        } catch {
+            Log.error("requestDiscovery failed: \(error)")
+        }
+    }
+
     private func handleStop() async throws {
         Log.info("Stopping sync...")
         emitStateChange(.stopping)

ConvosCore/Tests/ConvosCoreTests/SleepingInboxMessageCheckerTests.swift

@@ -649,9 +649,11 @@ actor TestableInboxLifecycleManager: InboxLifecycleManagerProtocol {
         fatalError("Not implemented for tests")
     }
 
-    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async {
+    @discardableResult
+    func registerExternalService(_ service: any MessagingServiceProtocol, clientId: String) async -> Bool {
         _awakeClientIds.insert(clientId)
         _sleepingClientIds.remove(clientId)
+        return true
     }
 
     func prepareUnusedConversationIfNeeded() async {}

ConvosCore/Tests/ConvosCoreTests/TestHelpers.swift

@@ -213,6 +213,9 @@ actor MockSyncingManager: SyncingManagerProtocol {
 
     func setInviteJoinErrorHandler(_ handler: (any InviteJoinErrorHandler)?) async {
     }
+
+    func requestDiscovery() async {
+    }
 }
 
 /// Mock implementation of NetworkMonitor for testing

qa/tests/23-scheduled-explode.md

@@ -0,0 +1,68 @@
+# Test: Scheduled Explode (In-App Expiration)
+
+Verify that a conversation with a scheduled explosion is automatically cleaned up when the timer expires while the app is in the foreground. This tests the scenario where another device schedules the explosion and the app must detect the expiration in real time.
+
+## Prerequisites
+
+- The app is running and past onboarding.
+- The convos CLI is initialized for the dev environment.
+
+## Setup
+
+1. Create a conversation from the CLI (so the CLI user is the super admin with explode permission).
+
+```bash
+CONV=$(convos conversations create --name "Boom Test" --profile-name "Bomber" --json)
+CONV_ID=$(echo "$CONV" | python3 -c "import json,sys; print(json.load(sys.stdin)['conversationId'])")
+```
+
+2. Generate an invite and have the app join via deep link.
+3. Process the join request from the CLI (per invite ordering rules in RULES.md).
+4. Send a few messages from both sides so the conversation has content and is clearly active.
+
+## Steps
+
+### Verify conversation is active
+
+1. The app should be viewing the conversation (or navigate to it from the conversations list).
+2. Verify the conversation name "Boom Test" is visible.
+3. Verify messages are visible in the conversation.
+
+### Schedule explosion from CLI
+
+4. From the CLI, schedule an explosion ~15 seconds in the future. Compute the ISO8601 timestamp dynamically:
+
+```bash
+EXPIRES_AT=$(python3 -c "from datetime import datetime, timedelta, timezone; print((datetime.now(timezone.utc) + timedelta(seconds=15)).strftime('%Y-%m-%dT%H:%M:%SZ'))")
+convos conversation explode "$CONV_ID" --scheduled "$EXPIRES_AT" --force
+```
+
+### Verify scheduled state in app
+
+5. The app should receive the ExplodeSettings message via the XMTP stream within a few seconds.
+6. The explode button in the conversation toolbar should change to show a countdown (e.g., "Explodes in 0:12"). Look for the `explode-button` accessibility identifier with "Scheduled to explode" label.
+
+### Wait for expiration
+
+7. Keep the app in the foreground. Do not navigate away or background the app.
+8. Wait for the scheduled time to pass (~15 seconds from when the explosion was scheduled).
+
+### Verify automatic cleanup
+
+9. The conversation should be automatically cleaned up when the timer expires. The app should navigate back to the conversations list (or show an exploded state).
+10. Verify the conversation no longer appears in the conversations list. Search for the conversation name "Boom Test" — it should not be found.
+
+## Teardown
+
+No cleanup needed — the conversation is destroyed by the explosion.
+
+## Pass/Fail Criteria
+
+- [ ] Conversation is active and showing messages before explosion is scheduled
+- [ ] App receives the scheduled explosion and shows countdown in the explode button
+- [ ] Conversation is automatically cleaned up when the timer expires while the app is foregrounded
+- [ ] Conversation no longer appears in the conversations list after expiration
+
+## Accessibility Improvements Needed
+
+(To be filled in during test execution if any elements are hard to find.)