Merge pull request #2110 from vitorpamplona/claude/debug-webrtc-calls-KqELh

Improve group call peer management and signaling robustness

amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt

@@ -349,6 +349,7 @@ class CallController(
         peerPubKey: HexKey,
         sdpAnswer: String,
     ) {
+        Log.d(TAG) { "onCallAnswerReceived: from=${peerPubKey.take(8)}, knownSessions=${peerSessions.keys.map { it.take(8) }}" }
         val ps = peerSessions[peerPubKey]
         if (ps != null) {
             // We have a PeerConnection for this peer — set remote description
@@ -437,10 +438,14 @@ class CallController(
      * peer with the lexicographically lower pubkey initiates.
      */
     fun onNewPeerInGroupCall(peerPubKey: HexKey) {
-        if (peerSessions.containsKey(peerPubKey)) return // already connected/connecting
+        if (peerSessions.containsKey(peerPubKey)) {
+            Log.d(TAG) { "onNewPeerInGroupCall: session already exists for ${peerPubKey.take(8)} — skipping" }
+            return
+        }
 
         scope.launch {
             val myPubKey = signerProvider().pubKey
+            Log.d(TAG) { "onNewPeerInGroupCall: peer=${peerPubKey.take(8)}, myPubKey=${myPubKey.take(8)}, iAmLower=${myPubKey < peerPubKey}" }
             if (myPubKey < peerPubKey) {
                 Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" }
                 createAndOfferToPeer(peerPubKey)
@@ -671,6 +676,7 @@ class CallController(
     // ---- Per-peer PeerConnection creation ----
 
     private fun createPeerSession(peerPubKey: HexKey): PeerSessionState {
+        Log.d(TAG) { "createPeerSession: ${peerPubKey.take(8)}, existing sessions=${peerSessions.keys.map { it.take(8) }}" }
         val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized")
 
         val session =
@@ -722,20 +728,62 @@ class CallController(
     }
 
     private fun onPeerDisconnected(peerPubKey: HexKey) {
-        Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected" }
+        Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected (ICE FAILED)" }
         // If all peers disconnected, hang up
+        val sessionStates = peerSessions.map { (key, ps) -> "${key.take(8)}=${ps.session.getSignalingState()}" }
+        Log.d(TAG) { "onPeerDisconnected: checking remaining sessions: $sessionStates" }
         val allDisconnected =
             peerSessions.keys.all { key ->
                 key == peerPubKey || peerSessions[key]?.session?.getSignalingState() == PeerConnection.SignalingState.CLOSED
             }
         if (allDisconnected) {
+            Log.d(TAG) { "onPeerDisconnected: all peers disconnected, hanging up" }
             scope.launch { callManager.hangup() }
+        } else {
+            Log.d(TAG) { "onPeerDisconnected: other peers still active, continuing call" }
         }
     }
 
+    // ---- Per-peer cleanup ----
+
+    /**
+     * Disposes a single peer's WebRTC session when they leave the call
+     * but the call continues with remaining peers.
+     */
+    fun disposePeerSession(peerPubKey: HexKey) {
+        val ps = peerSessions.remove(peerPubKey)
+        if (ps != null) {
+            Log.d(TAG) { "disposePeerSession: closing session for ${peerPubKey.take(8)}, remaining sessions=${peerSessions.keys.map { it.take(8) }}" }
+            try {
+                ps.session.dispose()
+            } catch (e: Exception) {
+                Log.e(TAG, "disposePeerSession: dispose() failed for ${peerPubKey.take(8)}", e)
+            }
+            // Update remote video tracks
+            val currentTracks = _remoteVideoTracks.value
+            if (peerPubKey in currentTracks) {
+                _remoteVideoTracks.value = currentTracks - peerPubKey
+                // If the disposed peer was the primary remote track, pick a new one
+                if (_remoteVideoTrack.value == currentTracks[peerPubKey]) {
+                    stopRemoteVideoMonitor()
+                    val nextTrack = _remoteVideoTracks.value.values.firstOrNull()
+                    _remoteVideoTrack.value = nextTrack
+                    if (nextTrack != null) {
+                        startRemoteVideoMonitor(nextTrack)
+                    }
+                }
+            }
+        } else {
+            Log.d(TAG) { "disposePeerSession: no session for ${peerPubKey.take(8)} (already disposed or never created)" }
+        }
+        // Clean up any buffered ICE candidates for this peer
+        globalPendingIce.remove(peerPubKey)
+    }
+
     // ---- Cleanup ----
 
     fun cleanup() {
+        Log.d(TAG) { "cleanup: disposing ${peerSessions.size} peer sessions, state=${callManager.state.value::class.simpleName}" }
         // Each block is wrapped individually so that a failure in one
         // (e.g. a WebRTC native crash) does not prevent the rest from
         // running.  Without this, a single exception could leave the

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt

@@ -225,6 +225,7 @@ class AccountViewModel(
         callManager.onIceCandidateReceived = { event -> controller.onIceCandidateReceived(event) }
         callManager.onNewPeerInGroupCall = { peerPubKey -> controller.onNewPeerInGroupCall(peerPubKey) }
         callManager.onMidCallOfferReceived = { peerPubKey, sdpOffer -> controller.onMidCallOfferReceived(peerPubKey, sdpOffer) }
+        callManager.onPeerLeft = { peerPubKey -> controller.disposePeerSession(peerPubKey) }
         callController = controller
 
         // Populate ActiveCallHolder so CallActivity can launch even when the app

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt

@@ -64,10 +64,17 @@ class CallManager(
     /** Called when a mid-call offer is received from another callee in a group call. */
     var onMidCallOfferReceived: ((peerPubKey: HexKey, sdpOffer: String) -> Unit)? = null
 
+    /** Called when a peer leaves the call (hangup) but the call continues with remaining peers. */
+    var onPeerLeft: ((peerPubKey: HexKey) -> Unit)? = null
+
     private var timeoutJob: Job? = null
     private var resetJob: Job? = null
     private val processedEventIds = mutableSetOf<String>()
 
+    /** Peers whose answers we saw while still ringing (IncomingCall).
+     *  After we accept, we trigger callee-to-callee mesh setup with them. */
+    private val discoveredCalleePeers = mutableSetOf<HexKey>()
+
     companion object {
         const val CALL_TIMEOUT_MS = 60_000L // 60 seconds ringing timeout
         const val ENDED_DISPLAY_MS = 2_000L // show "call ended" briefly before resetting
@@ -220,6 +227,17 @@ class CallManager(
         val result = factory.createGroupCallAnswer(sdpAnswer, allRecipients, current.callId, signer)
         result.wraps.forEach { publishEvent(it) }
         Log.d("CallManager") { "acceptCall: answer published, now in Connecting state" }
+
+        // Trigger callee-to-callee mesh connections with peers we discovered
+        // while still ringing (their answers arrived before we accepted).
+        val discovered = discoveredCalleePeers.toSet()
+        discoveredCalleePeers.clear()
+        if (discovered.isNotEmpty()) {
+            Log.d("CallManager") { "acceptCall: triggering mesh setup with ${discovered.size} discovered peers: ${discovered.map { it.take(8) }}" }
+            for (peer in discovered) {
+                onNewPeerInGroupCall?.invoke(peer)
+            }
+        }
     }
 
     suspend fun rejectCall() {
@@ -243,6 +261,18 @@ class CallManager(
                 "currentState=${current::class.simpleName}, sdpAnswerLength=${event.sdpAnswer().length}"
         }
 
+        // Self-answer: only meaningful as "answered elsewhere" in IncomingCall.
+        // In all other states it's our own echo from the relay — ignore it.
+        if (answeringPeer == signer.pubKey) {
+            if (current is CallState.IncomingCall && callId == current.callId) {
+                Log.d("CallManager") { "onCallAnswered: self-answer detected in IncomingCall — answered elsewhere" }
+                transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE)
+            } else {
+                Log.d("CallManager") { "onCallAnswered: ignoring self-answer echo in ${current::class.simpleName}" }
+            }
+            return
+        }
+
         when (current) {
             is CallState.Offering -> {
                 if (callId != current.callId) {
@@ -272,20 +302,17 @@ class CallManager(
                         )
                 }
                 // Forward to CallController — it routes to the correct PeerSession
-                Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding" }
+                // and internally triggers callee-to-callee mesh setup if needed.
+                Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding to CallController" }
                 onAnswerReceived?.invoke(event)
-
-                // Notify callees about new peer for mesh setup
-                onNewPeerInGroupCall?.invoke(answeringPeer)
             }
 
             is CallState.IncomingCall -> {
                 if (callId != current.callId) return
-                if (answeringPeer == signer.pubKey) {
+                // Another group member answered — remember them for mesh setup
-                    transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE)
+                // after we accept the call ourselves.
-                }
+                Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answered while we're still ringing, storing for later mesh" }
-                // Another group member answered — notify for future mesh connections
+                discoveredCalleePeers.add(answeringPeer)
-                // (we'll connect to them after we accept the call ourselves)
             }
 
             is CallState.Connected -> {
@@ -296,13 +323,11 @@ class CallManager(
                             peerPubKeys = current.peerPubKeys + answeringPeer,
                             pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer,
                         )
-                    // Forward to CallController for routing
-                    onAnswerReceived?.invoke(event)
-                    onNewPeerInGroupCall?.invoke(answeringPeer)
-                } else {
-                    // Renegotiation answer
-                    onAnswerReceived?.invoke(event)
                 }
+                // Forward to CallController — it routes to the correct PeerSession
+                // and internally triggers callee-to-callee mesh setup if needed.
+                Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answer in Connected, forwarding to CallController" }
+                onAnswerReceived?.invoke(event)
             }
 
             else -> {
@@ -487,19 +512,24 @@ class CallManager(
         val callId = event.callId() ?: return
         val leavingPeer = event.pubKey
 
+        Log.d("CallManager") { "onPeerHangup: from=${leavingPeer.take(8)}, callId=$callId, state=${current::class.simpleName}" }
+
         when (current) {
             is CallState.Connected -> {
                 if (callId != current.callId) return
                 val connectedRemaining = current.peerPubKeys - leavingPeer
                 val pendingRemaining = current.pendingPeerPubKeys - leavingPeer
                 if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) {
+                    Log.d("CallManager") { "onPeerHangup: last peer left, ending call" }
                     transitionToEnded(callId, current.allPeerPubKeys, EndReason.PEER_HANGUP)
                 } else {
+                    Log.d("CallManager") { "onPeerHangup: ${leavingPeer.take(8)} left, remaining=${connectedRemaining.map { it.take(8) }}, pending=${pendingRemaining.map { it.take(8) }}" }
                     _state.value =
                         current.copy(
                             peerPubKeys = connectedRemaining,
                             pendingPeerPubKeys = pendingRemaining,
                         )
+                    onPeerLeft?.invoke(leavingPeer)
                 }
             }
 
@@ -508,13 +538,16 @@ class CallManager(
                 val connectedRemaining = current.peerPubKeys - leavingPeer
                 val pendingRemaining = current.pendingPeerPubKeys - leavingPeer
                 if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) {
+                    Log.d("CallManager") { "onPeerHangup: last peer left during connecting, ending call" }
                     transitionToEnded(callId, current.peerPubKeys + current.pendingPeerPubKeys, EndReason.PEER_HANGUP)
                 } else {
+                    Log.d("CallManager") { "onPeerHangup: ${leavingPeer.take(8)} left during connecting, remaining=${connectedRemaining.map { it.take(8) }}" }
                     _state.value =
                         current.copy(
                             peerPubKeys = connectedRemaining,
                             pendingPeerPubKeys = pendingRemaining,
                         )
+                    onPeerLeft?.invoke(leavingPeer)
                 }
             }
 
@@ -525,6 +558,7 @@ class CallManager(
                     transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP)
                 } else {
                     _state.value = current.copy(peerPubKeys = remaining)
+                    onPeerLeft?.invoke(leavingPeer)
                 }
             }
 
@@ -555,6 +589,16 @@ class CallManager(
         }
         if (!processedEventIds.add(event.id)) return
 
+        // Filter out our own ICE candidates and hangups echoed back from relays.
+        // These are never useful: ICE candidates are for the remote peer, and
+        // hangups are already handled locally by hangup() → transitionToEnded.
+        // Self-answers and self-rejects are NOT filtered here because they serve
+        // as "answered/rejected elsewhere" signals when in IncomingCall state.
+        if (event.pubKey == signer.pubKey && (event is CallIceCandidateEvent || event is CallHangupEvent)) {
+            Log.d("CallManager") { "Ignoring self-event kind=${event.kind} id=${event.id.take(8)}" }
+            return
+        }
+
         Log.d("CallManager") { "Processing signaling event kind=${event.kind} id=${event.id.take(8)} state=${_state.value::class.simpleName}" }
 
         when (event) {
@@ -604,6 +648,7 @@ class CallManager(
         resetJob?.cancel()
         resetJob = null
         processedEventIds.clear()
+        discoveredCalleePeers.clear()
     }
 
     private fun transitionToEnded(