refactor: replace CallManager mutable callbacks with SharedFlow

Convert all 5 mutable `var` callback fields on CallManager
(onAnswerReceived, onIceCandidateReceived, onNewPeerInGroupCall,
onMidCallOfferReceived, onPeerLeft) into a single
`sessionEvents: SharedFlow<CallSessionEvent>`.

Benefits:
- Eliminates the stale-callback window between Activity destroy and
  recreate. No more nulling callbacks in onDestroy — the `closed`
  flag on CallSession is sufficient.
- No mutable state shared between Activity and background code.
- CallActivity no longer wires or tears down callbacks — CallSession
  subscribes to the flow in its init block.
- Type-safe sealed interface (CallSessionEvent) replaces 5 separate
  lambda types.

The renegotiationEvents SharedFlow remains separate because
renegotiation has its own glare-resolution logic in CallSession.

Updated all 11 test sites in CallManagerTest to collect from
sessionEvents instead of setting callback vars.

https://claude.ai/code/session_019yNnDjGKmJb19gadmojq54

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/CallActivity.kt

@@ -126,12 +126,9 @@ class CallActivity : AppCompatActivity() {
             )
         session = callSession
 
-        // Wire CallManager callbacks to this session's handlers.
-        callManager.onAnswerReceived = { event -> callSession.onCallAnswerReceived(event.pubKey, event.sdpAnswer()) }
-        callManager.onIceCandidateReceived = { event -> callSession.onIceCandidateReceived(event) }
-        callManager.onNewPeerInGroupCall = { peerPubKey -> callSession.onNewPeerInGroupCall(peerPubKey) }
-        callManager.onMidCallOfferReceived = { peerPubKey, sdpOffer -> callSession.onMidCallOfferReceived(peerPubKey, sdpOffer) }
-        callManager.onPeerLeft = { peerPubKey -> callSession.disposePeerSession(peerPubKey) }
+        // No callback wiring needed — CallSession collects from
+        // callManager.sessionEvents and renegotiationEvents SharedFlows
+        // in its init block.
 
         registerPipReceiver()
         observeVideoStateForPip(callSession)
@@ -240,15 +237,8 @@ class CallActivity : AppCompatActivity() {
         session?.close()
         session = null
 
-        // Null out callbacks so signaling events arriving after this
-        // Activity is gone don't route to a dead CallSession.
-        if (manager != null) {
-            manager.onAnswerReceived = null
-            manager.onIceCandidateReceived = null
-            manager.onNewPeerInGroupCall = null
-            manager.onMidCallOfferReceived = null
-            manager.onPeerLeft = null
-        }
+        // No callback nulling needed — CallSession collects from
+        // SharedFlows; the `closed` flag prevents processing after close().
 
         // Publish reject/hangup so the remote side stops ringing.
         // Use goAsync-style: the hangup is best-effort. If the process

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/call/session/CallSession.kt

@@ -141,9 +141,28 @@ class CallSession(
     // ---- Initialization ----
 
     init {
-        // Subscribe to renegotiation events via SharedFlow instead of a
-        // mutable callback, so we never miss an event across Activity
-        // restarts.
+        // Collect all session-level events (answers, ICE candidates,
+        // peer joins/leaves, mid-call offers) from the single SharedFlow.
+        scope.launch {
+            callManager.sessionEvents.collect { event ->
+                if (closed) return@collect
+                when (event) {
+                    is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.AnswerReceived ->
+                        onCallAnswerReceived(event.event.pubKey, event.event.sdpAnswer())
+                    is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.IceCandidateReceived ->
+                        onIceCandidateReceived(event.event)
+                    is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.NewPeerInGroupCall ->
+                        onNewPeerInGroupCall(event.peerPubKey)
+                    is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.MidCallOfferReceived ->
+                        onMidCallOfferReceived(event.peerPubKey, event.sdpOffer)
+                    is com.vitorpamplona.amethyst.commons.call.CallSessionEvent.PeerLeft ->
+                        disposePeerSession(event.peerPubKey)
+                }
+            }
+        }
+
+        // Renegotiation events have their own flow because they go
+        // through dedicated glare-resolution logic.
         scope.launch {
             callManager.renegotiationEvents.collect { event ->
                 if (!closed) onRenegotiationOfferReceived(event)

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

@@ -76,29 +76,30 @@ class CallManager(
     private val _state = MutableStateFlow<CallState>(CallState.Idle)
     val state: StateFlow<CallState> = _state.asStateFlow()
 
-    /** Called for every answer that should be applied to a PeerConnection (includes peer pubkey). */
-    var onAnswerReceived: ((CallAnswerEvent) -> Unit)? = null
-    var onIceCandidateReceived: ((CallIceCandidateEvent) -> Unit)? = null
+    /**
+     * All events that the Activity-scoped `CallSession` must handle are
+     * emitted through this single flow. Using a SharedFlow instead of
+     * mutable `var` callbacks means:
+     *  - No stale references to a dead CallSession between Activity
+     *    destroy and recreate.
+     *  - The collector can (re)subscribe via `repeatOnLifecycle` without
+     *    missing buffered events.
+     *  - No null-check on every emit site.
+     *
+     * Buffer size 16 covers bursts of ICE candidates + answers arriving
+     * in quick succession during call setup.
+     */
+    private val _sessionEvents = MutableSharedFlow<CallSessionEvent>(extraBufferCapacity = 16)
+    val sessionEvents: SharedFlow<CallSessionEvent> = _sessionEvents.asSharedFlow()
 
     /**
-     * Renegotiation offers from remote peers are emitted as flow events so the
-     * Activity-scoped `CallSession` can (re)subscribe via `repeatOnLifecycle`
-     * without losing messages across Activity restarts. Buffered so a short
-     * gap between Activity teardown and re-collection cannot drop a
-     * renegotiation.
+     * Renegotiation offers from remote peers. Separate from [sessionEvents]
+     * because renegotiation has its own glare-resolution logic in
+     * CallSession and benefits from a dedicated collector.
      */
     private val _renegotiationEvents = MutableSharedFlow<CallRenegotiateEvent>(extraBufferCapacity = 8)
     val renegotiationEvents: SharedFlow<CallRenegotiateEvent> = _renegotiationEvents.asSharedFlow()
 
-    /** Called when a new peer joins the group call (callee-to-callee mesh setup). */
-    var onNewPeerInGroupCall: ((peerPubKey: HexKey) -> Unit)? = null
-
-    /** 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 = LinkedHashSet<String>()
@@ -349,7 +350,7 @@ class CallManager(
 
                     else -> {}
                 }
-                onMidCallOfferReceived?.invoke(callerPubKey, event.sdpOffer())
+                _sessionEvents.tryEmit(CallSessionEvent.MidCallOfferReceived(callerPubKey, event.sdpOffer()))
                 return
             }
         }
@@ -450,7 +451,7 @@ class CallManager(
         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)
+                _sessionEvents.tryEmit(CallSessionEvent.NewPeerInGroupCall(peer))
             }
         }
     }
@@ -517,7 +518,7 @@ class CallManager(
                 disarmRingingWatchdog()
                 armConnectingWatchdog(current.callId)
                 Log.d("CallManager") { "onCallAnswered: Offering -> Connecting, forwarding answer to CallSession" }
-                onAnswerReceived?.invoke(event)
+                _sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event))
             }
 
             is CallState.Connecting -> {
@@ -546,7 +547,7 @@ class CallManager(
                 // Forward to CallController — it routes to the correct PeerSession
                 // 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)
+                _sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event))
             }
 
             is CallState.IncomingCall -> {
@@ -585,7 +586,7 @@ class CallManager(
                 // 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)
+                _sessionEvents.tryEmit(CallSessionEvent.AnswerReceived(event))
             }
 
             else -> {
@@ -603,7 +604,7 @@ class CallManager(
         // still ringing (IncomingCall).  In every other state it is our own
         // echo from a relay after a sibling device rejected a call that we
         // have already accepted and are now actively in — treating it as a
-        // peer rejection would drop ourselves from the call (onPeerLeft
+        // peer rejection would drop ourselves from the call (PeerLeft
         // would dispose our own PeerConnection, killing the live audio on
         // the device that picked up).  Just drop the echo.
         if (rejectingPeer == signer.pubKey) {
@@ -625,7 +626,7 @@ class CallManager(
                     transitionToEnded(current.callId, current.peerPubKeys, EndReason.PEER_REJECTED)
                 } else {
                     _state.value = current.copy(peerPubKeys = remaining)
-                    onPeerLeft?.invoke(rejectingPeer)
+                    _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer))
                 }
             }
 
@@ -634,7 +635,7 @@ class CallManager(
                 cancelPeerTimeout(rejectingPeer)
                 _state.value =
                     current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer)
-                onPeerLeft?.invoke(rejectingPeer)
+                _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer))
             }
 
             is CallState.Connected -> {
@@ -642,7 +643,7 @@ class CallManager(
                 cancelPeerTimeout(rejectingPeer)
                 _state.value =
                     current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer)
-                onPeerLeft?.invoke(rejectingPeer)
+                _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(rejectingPeer))
             }
 
             is CallState.IncomingCall -> {
@@ -669,7 +670,7 @@ class CallManager(
     }
 
     private fun onIceCandidate(event: CallIceCandidateEvent) {
-        onIceCandidateReceived?.invoke(event)
+        _sessionEvents.tryEmit(CallSessionEvent.IceCandidateReceived(event))
     }
 
     private fun onRenegotiate(event: CallRenegotiateEvent) {
@@ -831,7 +832,7 @@ class CallManager(
                             peerPubKeys = connectedRemaining,
                             pendingPeerPubKeys = pendingRemaining,
                         )
-                    onPeerLeft?.invoke(leavingPeer)
+                    _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer))
                 }
             }
 
@@ -856,7 +857,7 @@ class CallManager(
                             peerPubKeys = connectedRemaining,
                             pendingPeerPubKeys = pendingRemaining,
                         )
-                    onPeerLeft?.invoke(leavingPeer)
+                    _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer))
                 }
             }
 
@@ -868,7 +869,7 @@ class CallManager(
                     transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP)
                 } else {
                     _state.value = current.copy(peerPubKeys = remaining)
-                    onPeerLeft?.invoke(leavingPeer)
+                    _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(leavingPeer))
                 }
             }
 
@@ -1160,7 +1161,7 @@ class CallManager(
      *   at least one other peer is connected by definition, so the call
      *   always continues.
      *
-     * Fires [onPeerLeft] so the CallController disposes the per-peer
+     * Emits [CallSessionEvent.PeerLeft] so the CallSession disposes the per-peer
      * PeerConnection (and any pending ICE buffers) for the dropped peer.
      */
     private suspend fun handlePeerTimeout(
@@ -1186,7 +1187,7 @@ class CallManager(
                         transitionToEnded(current.callId, current.peerPubKeys, EndReason.TIMEOUT)
                     } else {
                         _state.value = current.copy(peerPubKeys = remaining)
-                        onPeerLeft?.invoke(peerPubKey)
+                        _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey))
                     }
                 }
 
@@ -1204,7 +1205,7 @@ class CallManager(
                         )
                     } else {
                         _state.value = current.copy(pendingPeerPubKeys = newPending)
-                        onPeerLeft?.invoke(peerPubKey)
+                        _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey))
                     }
                 }
 
@@ -1214,7 +1215,7 @@ class CallManager(
                     Log.d("CallManager") { "Per-peer timeout: dropping ${peerPubKey.take(8)} from Connected (invitedByUs=$wasInvitedByUs)" }
                     shouldPublishHangup = wasInvitedByUs
                     _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - peerPubKey)
-                    onPeerLeft?.invoke(peerPubKey)
+                    _sessionEvents.tryEmit(CallSessionEvent.PeerLeft(peerPubKey))
                 }
 
                 else -> {

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.call
+
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallAnswerEvent
+import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallIceCandidateEvent
+
+/**
+ * Events emitted by [CallManager] that the Activity-scoped `CallSession`
+ * must handle. Delivered via `SharedFlow` so the collector can
+ * (re)subscribe across Activity restarts without missing buffered events.
+ */
+sealed interface CallSessionEvent {
+    /** A peer answered our offer (or we received an answer in a group call). */
+    data class AnswerReceived(val event: CallAnswerEvent) : CallSessionEvent
+
+    /** An ICE candidate arrived from a peer. */
+    data class IceCandidateReceived(val event: CallIceCandidateEvent) : CallSessionEvent
+
+    /** A new peer joined the group call and needs callee-to-callee mesh setup. */
+    data class NewPeerInGroupCall(val peerPubKey: HexKey) : CallSessionEvent
+
+    /** A mid-call offer arrived from another callee in a group call. */
+    data class MidCallOfferReceived(val peerPubKey: HexKey, val sdpOffer: String) : CallSessionEvent
+
+    /** A peer left the call (hangup/reject/timeout) but the call continues. */
+    data class PeerLeft(val peerPubKey: HexKey) : CallSessionEvent
+}

commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/CallManagerTest.kt

@@ -33,6 +33,8 @@ import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallRenegotiateEvent
 import com.vitorpamplona.quartz.nipACWebRtcCalls.tags.CallType
 import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.advanceTimeBy
 import kotlinx.coroutines.test.advanceUntilIdle
@@ -100,6 +102,16 @@ class CallManagerTest {
         return manager to published
     }
 
+    /**
+     * Starts collecting [CallManager.sessionEvents] into a list.
+     * Returns (events, job) — cancel the job when done.
+     */
+    private fun TestScope.collectSessionEvents(manager: CallManager): Pair<MutableList<CallSessionEvent>, Job> {
+        val events = mutableListOf<CallSessionEvent>()
+        val job = launch { manager.sessionEvents.collect { events.add(it) } }
+        return events to job
+    }
+
     // ---- Event construction helpers ----
     // These use the real event builders from quartz so that tag structures
     // stay in sync with the production code.  The builder returns an
@@ -323,15 +335,15 @@ class CallManagerTest {
             manager.initiateCall(bob, CallType.VOICE, callId, sdpOffer)
             assertIs<CallState.Offering>(manager.state.value)
 
-            var answerReceived = false
-            manager.onAnswerReceived = { answerReceived = true }
+            val (events, job) = collectSessionEvents(manager)
 
             val answer = makeAnswer(from = bob, to = alice)
             manager.onSignalingEvent(answer)
 
             val state = manager.state.value
             assertIs<CallState.Connecting>(state)
-            assertTrue(answerReceived, "onAnswerReceived callback should fire")
+            assertTrue(events.any { it is CallSessionEvent.AnswerReceived }, "AnswerReceived event should be emitted")
+            job.cancel()
         }
 
     // ========================================================================
@@ -461,14 +473,14 @@ class CallManagerTest {
             manager.onSignalingEvent(makeOffer(from = alice, to = bob))
             manager.acceptCall(sdpAnswer)
 
-            var iceCalled = false
-            manager.onIceCandidateReceived = { iceCalled = true }
+            val (events, job) = collectSessionEvents(manager)
 
             // ICE candidate from self (echoed back by relay)
             val selfIce = makeIceCandidate(from = bob, to = alice)
             manager.onSignalingEvent(selfIce)
 
-            assertTrue(!iceCalled, "Self ICE candidates MUST be ignored")
+            assertTrue(events.none { it is CallSessionEvent.IceCandidateReceived }, "Self ICE candidates MUST be ignored")
+            job.cancel()
         }
 
     @Test
@@ -533,13 +545,14 @@ class CallManagerTest {
             manager.onSignalingEvent(makeOffer(from = alice, to = bob))
             manager.acceptCall(sdpAnswer)
 
-            var receivedIce: CallIceCandidateEvent? = null
-            manager.onIceCandidateReceived = { receivedIce = it }
+            val (events, job) = collectSessionEvents(manager)
 
             val ice = makeIceCandidate(from = alice, to = bob)
             manager.onSignalingEvent(ice)
 
-            assertNotNull(receivedIce, "ICE candidate should be forwarded via callback")
+            val iceEvent = events.filterIsInstance<CallSessionEvent.IceCandidateReceived>().firstOrNull()
+            assertNotNull(iceEvent, "ICE candidate should be emitted via sessionEvents")
+            job.cancel()
         }
 
     // ========================================================================
@@ -794,15 +807,16 @@ class CallManagerTest {
             // Bob still ringing
             assertIs<CallState.IncomingCall>(manager.state.value)
 
-            // Track mesh setup callback
-            val newPeers = mutableListOf<HexKey>()
-            manager.onNewPeerInGroupCall = { newPeers.add(it) }
+            // Track mesh setup events
+            val (events, job) = collectSessionEvents(manager)
 
             // Bob accepts
             manager.acceptCall(sdpAnswer)
 
             // Should trigger callee-to-callee mesh setup with carol
+            val newPeers = events.filterIsInstance<CallSessionEvent.NewPeerInGroupCall>().map { it.peerPubKey }
             assertTrue(carol in newPeers, "Should discover carol for mesh setup after accepting")
+            job.cancel()
         }
 
     // ========================================================================
@@ -819,19 +833,17 @@ class CallManagerTest {
             manager.onPeerConnected()
             assertIs<CallState.Connected>(manager.state.value)
 
-            var midCallPeer: HexKey? = null
-            var midCallSdp: String? = null
-            manager.onMidCallOfferReceived = { peer, sdp ->
-                midCallPeer = peer
-                midCallSdp = sdp
-            }
+            val (events, job) = collectSessionEvents(manager)
 
             // Carol sends a mid-call offer (callee-to-callee mesh)
             val carolOffer = makeOffer(from = carol, to = bob, callId = callId, sdp = "carol-sdp")
             manager.onSignalingEvent(carolOffer)
 
-            assertEquals(carol, midCallPeer)
-            assertEquals("carol-sdp", midCallSdp)
+            val midCallEvent = events.filterIsInstance<CallSessionEvent.MidCallOfferReceived>().firstOrNull()
+            assertNotNull(midCallEvent)
+            assertEquals(carol, midCallEvent.peerPubKey)
+            assertEquals("carol-sdp", midCallEvent.sdpOffer)
+            job.cancel()
         }
 
     // ========================================================================
@@ -882,12 +894,13 @@ class CallManagerTest {
             manager.onPeerConnected()
             manager.onSignalingEvent(makeAnswer(from = carol, to = alice))
 
-            val leftPeers = mutableListOf<HexKey>()
-            manager.onPeerLeft = { leftPeers.add(it) }
+            val (events, job) = collectSessionEvents(manager)
 
             manager.onSignalingEvent(makeHangup(from = bob, to = alice))
 
-            assertTrue(bob in leftPeers, "onPeerLeft should fire when a peer hangs up")
+            val leftPeers = events.filterIsInstance<CallSessionEvent.PeerLeft>().map { it.peerPubKey }
+            assertTrue(bob in leftPeers, "PeerLeft should be emitted when a peer hangs up")
+            job.cancel()
         }
 
     // ========================================================================
@@ -1128,8 +1141,8 @@ class CallManagerTest {
      * Bob is already in a Connected group call with Alice. Alice invites Carol.
      * Carol's broadcast CallAnswer reaches Bob. Bob's state must expand to
      * include Carol in [CallState.Connected.peerPubKeys] and the answer must
-     * still be forwarded via [CallManager.onAnswerReceived] so the caller-side
-     * [CallController] can unconditionally initiate a mesh offer to Carol.
+     * still be emitted via [CallManager.sessionEvents] so the caller-side
+     * CallSession can unconditionally initiate a mesh offer to Carol.
      */
     @Test
     fun midCallInviteAnswerFromUnknownPeerInConnectedExpandsMembership() =
@@ -1142,8 +1155,7 @@ class CallManagerTest {
             manager.onPeerConnected()
             assertIs<CallState.Connected>(manager.state.value)
 
-            var forwardedPeer: HexKey? = null
-            manager.onAnswerReceived = { event -> forwardedPeer = event.pubKey }
+            val (events, job) = collectSessionEvents(manager)
 
             // Alice invited Carol mid-call; Carol broadcasts her acceptance.
             // Bob sees a CallAnswer from Carol (unknown peer, same call-id)
@@ -1155,7 +1167,9 @@ class CallManagerTest {
             assertIs<CallState.Connected>(state)
             assertTrue(carol in state.peerPubKeys, "Mid-call joiner must be added to peerPubKeys")
             assertTrue(alice in state.peerPubKeys, "Existing peer must still be present")
-            assertEquals(carol, forwardedPeer, "Answer must still be forwarded to CallController")
+            val forwardedPeer = events.filterIsInstance<CallSessionEvent.AnswerReceived>().firstOrNull()?.event?.pubKey
+            assertEquals(carol, forwardedPeer, "Answer must still be emitted to CallSession")
+            job.cancel()
         }
 
     /**
@@ -1296,16 +1310,17 @@ class CallManagerTest {
 
             // Step 6: Bob receives Carol's answer broadcast. Bob's state must
             // expand to include Carol (mid-call join), and the answer must be
-            // forwarded so Bob's CallController can initiate a mesh offer.
-            var bobForwardedAnswer: HexKey? = null
-            bobManager.onAnswerReceived = { event -> bobForwardedAnswer = event.pubKey }
+            // emitted so Bob's CallSession can initiate a mesh offer.
+            val (bobEvents, bobJob) = collectSessionEvents(bobManager)
             bobManager.onSignalingEvent(
                 makeGroupAnswer(from = carol, members = setOf(alice, bob, carol), sdp = "carol-answer-sdp"),
             )
             val bobAfterCarolAnswer = bobManager.state.value
             assertIs<CallState.Connected>(bobAfterCarolAnswer)
             assertTrue(carol in bobAfterCarolAnswer.peerPubKeys, "Bob should add Carol to his membership")
-            assertEquals(carol, bobForwardedAnswer, "Bob must forward Carol's answer to his CallController")
+            val bobForwardedAnswer = bobEvents.filterIsInstance<CallSessionEvent.AnswerReceived>().firstOrNull()?.event?.pubKey
+            assertEquals(carol, bobForwardedAnswer, "Bob must emit Carol's answer to his CallSession")
+            bobJob.cancel()
         }
 
     // ========================================================================
@@ -1528,15 +1543,16 @@ class CallManagerTest {
             assertTrue(carol in connecting.pendingPeerPubKeys)
 
             // Carol publishes a mesh offer to Bob (same call-id).
-            var midCallOfferPeer: HexKey? = null
-            manager.onMidCallOfferReceived = { peer, _ -> midCallOfferPeer = peer }
+            val (events, job) = collectSessionEvents(manager)
             manager.onSignalingEvent(makeGroupOffer(from = carol, members = setOf(alice, bob, carol)))
 
             val after = manager.state.value
             assertIs<CallState.Connecting>(after)
             assertTrue(carol in after.peerPubKeys, "Carol should move into peerPubKeys")
             assertTrue(carol !in after.pendingPeerPubKeys, "Carol should leave pending")
-            assertEquals(carol, midCallOfferPeer, "CallController must still receive the mid-call offer")
+            val midCallOfferPeer = events.filterIsInstance<CallSessionEvent.MidCallOfferReceived>().firstOrNull()?.peerPubKey
+            assertEquals(carol, midCallOfferPeer, "CallSession must still receive the mid-call offer")
+            job.cancel()
 
             // And her watchdog should be cancelled — advancing past the
             // 30 s budget must not further alter state.
@@ -1929,7 +1945,7 @@ class CallManagerTest {
 
     /**
      * Regression for the subtle bug where a self-reject echo would trigger
-     * onPeerLeft(signer.pubKey) on a device that had already accepted,
+     * PeerLeft(signer.pubKey) on a device that had already accepted,
      * disposing its own PeerSession and killing the live call.
      */
     @Test
@@ -1937,8 +1953,7 @@ class CallManagerTest {
         runTest {
             val (manager, _) = createManager(localPubKey = bob, followedKeys = setOf(alice))
 
-            var peerLeftCalled = false
-            manager.onPeerLeft = { peerLeftCalled = true }
+            val (events, job) = collectSessionEvents(manager)
 
             manager.onSignalingEvent(makeOffer(from = alice, to = bob))
             manager.acceptCall(sdpAnswer)
@@ -1950,29 +1965,29 @@ class CallManagerTest {
             val siblingReject = makeReject(from = bob, to = alice)
             manager.onSignalingEvent(siblingReject)
 
-            // Our live call must survive — the echo must not fire
-            // onPeerLeft (which would dispose our own PeerSession) and must
+            // Our live call must survive — the echo must not emit
+            // PeerLeft (which would dispose our own PeerSession) and must
             // not change state.
             assertIs<CallState.Connected>(manager.state.value)
             assertTrue(
-                !peerLeftCalled,
-                "onPeerLeft must not fire for a self-reject echo — the UI would " +
+                events.none { it is CallSessionEvent.PeerLeft },
+                "PeerLeft must not be emitted for a self-reject echo — the UI would " +
                     "tear down our own PeerSession, killing the live call audio.",
             )
+            job.cancel()
         }
 
     /**
      * Same guard for the Connecting state: a self-reject arriving during
      * the connection handshake must not drop us from our own pending set
-     * and must not invoke onPeerLeft(self).
+     * and must not emit PeerLeft(self).
      */
     @Test
     fun selfRejectEchoDuringConnectingDoesNotFireOnPeerLeft() =
         runTest {
             val (manager, _) = createManager(localPubKey = bob, followedKeys = setOf(alice))
 
-            var peerLeftCalled = false
-            manager.onPeerLeft = { peerLeftCalled = true }
+            val (events, job) = collectSessionEvents(manager)
 
             manager.onSignalingEvent(makeOffer(from = alice, to = bob))
             manager.acceptCall(sdpAnswer)
@@ -1982,7 +1997,8 @@ class CallManagerTest {
             manager.onSignalingEvent(siblingReject)
 
             assertIs<CallState.Connecting>(manager.state.value)
-            assertTrue(!peerLeftCalled, "onPeerLeft must not fire for a self-reject echo during Connecting")
+            assertTrue(events.none { it is CallSessionEvent.PeerLeft }, "PeerLeft must not be emitted for a self-reject echo during Connecting")
+            job.cancel()
         }
 
     /**