diff --git a/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/robots/UserRobotCallAsserts.kt b/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/robots/UserRobotCallAsserts.kt
index 4b511ce632c..93e0bd5f7af 100644
--- a/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/robots/UserRobotCallAsserts.kt
+++ b/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/robots/UserRobotCallAsserts.kt
@@ -74,6 +74,16 @@ fun UserRobot.assertThatCallIsEnded(): UserRobot {
     return this
 }
 
+fun UserRobot.assertCallUiIsRendered(): UserRobot {
+    assertTrue(
+        "Call information",
+        CallPage.callInfoView.waitDisplayed(timeOutMillis = 30.seconds),
+    )
+    assertTrue("Hang-up control", CallPage.hangUpButton.waitDisplayed())
+    assertTrue("Video surface", CallPage.videoView.waitDisplayed())
+    return this
+}
+
 fun UserRobot.assertUserMicrophone(isEnabled: Boolean, videoCall: Boolean = true): UserRobot {
     // The participant view icon updates slightly after the control toggle, so both
     // checks poll instead of asserting the icon at the instant the toggle appears.
diff --git a/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/tests/CallLifecycleTests.kt b/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/tests/CallLifecycleTests.kt
index 5f4b2f9ff61..bbf92340b6d 100644
--- a/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/tests/CallLifecycleTests.kt
+++ b/demo-app/src/androidTestE2etestingDebug/kotlin/io/getstream/video/android/tests/CallLifecycleTests.kt
@@ -16,11 +16,18 @@
 
 package io.getstream.video.android.tests
 
+import io.getstream.video.android.core.Call
+import io.getstream.video.android.core.StreamVideo
+import io.getstream.video.android.robots.assertCallUiIsRendered
 import io.getstream.video.android.robots.assertParticipantsCountOnCall
 import io.getstream.video.android.robots.assertThatCallIsEnded
 import io.getstream.video.android.uiautomator.seconds
 import io.qameta.allure.kotlin.Allure.step
 import io.qameta.allure.kotlin.AllureId
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.Assert.assertSame
+import org.junit.Assert.assertTrue
 import org.junit.Test
 import java.util.UUID
 
@@ -105,8 +112,9 @@ class CallLifecycleTests : StreamTestCase() {
 
     @AllureId("7012")
     @Test
-    fun testUserReentersTheCall() {
+    fun testUserReusesTheSameCallObjectAfterLeaving() {
         val participants = 1
+        lateinit var reusableCall: Call
 
         step("GIVEN user starts a call") {
             userRobot.joinCall()
@@ -115,11 +123,44 @@ class CallLifecycleTests : StreamTestCase() {
             participantRobot.joinCall(callId)
             userRobot.waitForParticipantsOnCall(participants)
         }
-        step("WHEN user re-enters the call as the same user") {
-            userRobot.endCall().joinCall(callId)
+        step("AND the call UI is rendered") {
+            userRobot.assertCallUiIsRendered()
+            reusableCall = requireNotNull(StreamVideo.instance().state.activeCall.value) {
+                "Expected the rendered call to be registered as the active call"
+            }
         }
-        step("THEN there is one participant on the call") {
-            userRobot.assertParticipantsCountOnCall(participants)
+        step("WHEN user leaves the call") {
+            userRobot
+                .endCall()
+                .assertThatCallIsEnded()
+        }
+        step("AND the same Call object joins again") {
+            val rejoinResult = runBlocking {
+                withTimeout(60_000) {
+                    reusableCall.join()
+                }
+            }
+
+            assertTrue(
+                "Expected the same Call object to rejoin successfully",
+                rejoinResult.isSuccess,
+            )
+            assertSame(reusableCall, StreamVideo.instance().state.activeCall.value)
+        }
+        step("AND the call screen is opened for the rejoined call") {
+            userRobot.joinCall(callId)
+        }
+        step("THEN the call UI is rendered with the participant") {
+            userRobot
+                .waitForParticipantsOnCall(participants)
+                .assertParticipantsCountOnCall(participants)
+                .assertCallUiIsRendered()
+            assertSame(reusableCall, StreamVideo.instance().state.activeCall.value)
+        }
+        step("AND user can leave the reused call") {
+            userRobot
+                .endCall()
+                .assertThatCallIsEnded()
         }
     }
 
diff --git a/stream-video-android-core/api/stream-video-android-core.api b/stream-video-android-core/api/stream-video-android-core.api
index 7881bd9d182..42afca739c9 100644
--- a/stream-video-android-core/api/stream-video-android-core.api
+++ b/stream-video-android-core/api/stream-video-android-core.api
@@ -9809,7 +9809,6 @@ public abstract interface class io/getstream/video/android/core/CallJoinIntercep
 }
 
 public final class io/getstream/video/android/core/CallState {
-	public fun <init> (Lio/getstream/video/android/core/StreamVideo;Lio/getstream/video/android/core/Call;Lio/getstream/video/android/model/User;Lkotlinx/coroutines/CoroutineScope;)V
 	public final fun clearParticipants ()V
 	public final fun getAcceptedBy ()Lkotlinx/coroutines/flow/StateFlow;
 	public final fun getActiveSpeakers ()Lkotlinx/coroutines/flow/StateFlow;
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt
index 22db5a6cea9..d465cb6678d 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt
@@ -54,6 +54,7 @@ import io.getstream.log.taggedLogger
 import io.getstream.result.Result
 import io.getstream.video.android.core.analytics.call.CallAnalytics
 import io.getstream.video.android.core.analytics.call.observer.model.JoinAnalyticsModel
+import io.getstream.video.android.core.call.Restartable
 import io.getstream.video.android.core.call.RtcSession
 import io.getstream.video.android.core.call.audio.InputAudioFilter
 import io.getstream.video.android.core.call.components.CallApiClient
@@ -65,6 +66,7 @@ import io.getstream.video.android.core.call.components.CallLifecycleManager
 import io.getstream.video.android.core.call.components.CallMediaManager
 import io.getstream.video.android.core.call.components.CallReconnector
 import io.getstream.video.android.core.call.components.CallRenderer
+import io.getstream.video.android.core.call.components.CallRestartCoordinator
 import io.getstream.video.android.core.call.components.CallSessionManager
 import io.getstream.video.android.core.call.components.CallStatsReporter
 import io.getstream.video.android.core.call.components.ClientCallRegistry
@@ -73,7 +75,6 @@ import io.getstream.video.android.core.call.components.NoiseCancellationPolicy
 import io.getstream.video.android.core.call.components.RtcSessionFactory
 import io.getstream.video.android.core.call.components.SessionMonitor
 import io.getstream.video.android.core.call.connection.StreamPeerConnectionFactory
-import io.getstream.video.android.core.call.scope.ScopeProvider
 import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.getstream.video.android.core.call.video.VideoFilter
 import io.getstream.video.android.core.closedcaptions.ClosedCaptionsSettings
@@ -90,8 +91,6 @@ import io.getstream.video.android.core.notifications.internal.telecom.TelecomCal
 import io.getstream.video.android.core.recording.RecordingType
 import io.getstream.video.android.core.ringing.RingJoinSource
 import io.getstream.video.android.core.ringing.RingStatePoller
-import io.getstream.video.android.core.socket.common.scope.ClientScope
-import io.getstream.video.android.core.socket.common.scope.UserScope
 import io.getstream.video.android.core.socket.sfu.state.SfuSocketState
 import io.getstream.video.android.core.trace.PeerConnectionTraceKey
 import io.getstream.video.android.core.utils.SerialProcessor
@@ -103,8 +102,6 @@ import io.getstream.webrtc.android.ui.VideoTextureViewRenderer
 import io.getstream.webrtc.audio.JavaAudioDeviceModule.AudioSamples
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -140,13 +137,19 @@ public class Call(
     val user: User,
 ) {
     internal val clientImpl = client as StreamVideoClient
-    internal val scopeProvider: ScopeProvider = ScopeProviderImpl(clientImpl.scope)
+
+    internal val scopeProvider = ScopeProviderImpl(
+        coroutineScope = clientImpl.scope,
+        callCid = "$type:$id",
+    )
 
     private val logger by taggedLogger("Call:$type:$id")
-    private val supervisorJob = SupervisorJob()
     internal var powerManager: PowerManager? = null
 
-    internal val scope = CoroutineScope(clientImpl.scope.coroutineContext + supervisorJob)
+    internal val scope: CoroutineScope = object : CoroutineScope {
+        override val coroutineContext
+            get() = scopeProvider.callScope.coroutineContext
+    }
 
     /** Delegate that owns the live RTC session state and reconnect bookkeeping. */
     private val sessionManager = CallSessionManager()
@@ -216,6 +219,7 @@ public class Call(
         }
 
         override fun markActive() {
+            clientImpl.onCallJoined(this@Call)
             client.state.setActiveCall(this@Call)
         }
 
@@ -267,7 +271,7 @@ public class Call(
     val events: MutableSharedFlow<VideoEvent> = eventManager.events
 
     /** The call state contains all state such as the participant list, reactions etc */
-    val state = CallState(client, this, user, scope)
+    val state = CallState(client, this, user, scopeProvider)
 
     internal val callAnalytics =
         CallAnalytics(
@@ -370,31 +374,47 @@ public class Call(
         type = type,
         id = id,
         clientImpl = clientImpl,
-        scope = scope,
+        coroutineScope = scope,
         state = state,
         sessionManager = sessionManager,
         eglBase = { eglBase },
         mediaManagerFactory = mediaManagerFactory,
     )
 
+    /** Restores persistent components after the previous Call generation has been cleaned up. */
+    private val restartCoordinator = CallRestartCoordinator(
+        scopeProvider = scopeProvider,
+        restartables = listOf(
+            sessionManager,
+            callAnalytics,
+            statsReporter,
+            Restartable {
+                state.restart()
+                Result.Success(Unit)
+            },
+            media,
+        ),
+        onRestarted = {
+            clientImpl.onCallReused(this@Call)
+            autoOnApplied = false
+            noiseCancellationSignalPending = false
+            desiredNoiseCancellationEnabled = false
+            observeNoiseCancellationSignalTarget()
+            observeNoiseCancellationPolicy()
+        },
+    )
+
     /** Delegate that owns leave / end / cleanup teardown and the destroyed flag. */
     private val lifecycle: CallLifecycleManager = CallLifecycleManager(
         clientImpl = clientImpl,
         sessionManager = sessionManager,
-        scopeProvider = scopeProvider,
         callRegistry = callRegistry,
-        // Lets the REST calls queued before leave finish before the scope is torn down.
         shutDownJobs = {
-            UserScope(ClientScope()).launch {
-                supervisorJob.children.forEach { it.join() }
-                supervisorJob.cancel()
-            }
-            scope.cancel()
-            // Cancelling the scope already reaches the poller, which runs on the call's job.
-            // Stopping it explicitly also clears the job reference, so a call torn down and
-            // re-armed does not see a cancelled job as a running one.
             stopRingStatePolling()
+            _ringStatePoller = null
+            scopeProvider.cleanup()
         },
+        restartCoordinator = restartCoordinator,
         state = state,
         callAnalytics = callAnalytics,
         statsReporter = statsReporter,
@@ -612,6 +632,7 @@ public class Call(
 
     /**
      * Joins the call. Concurrent callers share one in-flight attempt.
+     * The same instance can be joined again after [leave]; pending cleanup finishes first.
      *
      * Cancelling this coroutine does not abort that attempt — only [leave] does.
      *
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallState.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallState.kt
index a8357d9964d..d36e660bb53 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallState.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallState.kt
@@ -85,6 +85,8 @@ import io.getstream.log.taggedLogger
 import io.getstream.result.Result
 import io.getstream.video.android.core.call.CallType
 import io.getstream.video.android.core.call.RtcSession
+import io.getstream.video.android.core.call.scope.RestartableStateFlow
+import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.getstream.video.android.core.closedcaptions.ClosedCaptionManager
 import io.getstream.video.android.core.closedcaptions.ClosedCaptionsSettings
 import io.getstream.video.android.core.dispatchers.DispatcherProvider
@@ -152,7 +154,6 @@ import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.flow.transform
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
@@ -230,16 +231,29 @@ internal typealias IsPinned = Boolean
  *
  */
 @Stable
-public class CallState(
+public class CallState internal constructor(
     private val client: StreamVideo,
     private val call: Call,
     private val user: User,
-    @InternalStreamVideoApi
-    val scope: CoroutineScope,
+    private val scopeProvider: ScopeProviderImpl,
 ) {
 
     private val logger by taggedLogger("CallState")
     private var participantsVisibilityMonitor: Job? = null
+    private var participantVisibilityFlow: Flow<List<String>>? = null
+
+    @InternalStreamVideoApi
+    val scope: CoroutineScope = call.scope
+
+    private fun <T> Flow<T>.stateIn(
+        started: SharingStarted,
+        initialValue: T,
+    ): StateFlow<T> = RestartableStateFlow(
+        upstream = this,
+        scopeProvider = scopeProvider,
+        initialValue = initialValue,
+        started = started,
+    )
 
     // Create a CallActions implementation that delegates to the Call object
     @InternalStreamVideoApi
@@ -360,7 +374,7 @@ public class CallState(
             combined.putAll(server)
             combined.putAll(local) // local overrides server on key collision
             combined.toMap()
-        }.stateIn(scope, SharingStarted.Eagerly, emptyMap())
+        }.stateIn(SharingStarted.Eagerly, emptyMap())
 
     internal val _pinnedParticipants: StateFlow<Map<SessionId, OffsetDateTime>> =
         _pinnedParticipantsDetailed.mapState { detailed ->
@@ -396,7 +410,7 @@ public class CallState(
             val filteredVideo = participants.firstOrNull {
                 it.video.value?.enabled == true
             }?.video?.value
-            scope.launch {
+            launch {
                 if (_backstage.value) {
                     send(null)
                 } else {
@@ -405,7 +419,7 @@ public class CallState(
             }
         }
 
-        scope.launch {
+        launch {
             _participants.collect {
                 logger.v {
                     "[livestreamFlow] #track; participants: ${it.size} =>" + "${it.map { "${it.value.userId.value} - ${it.value.video.value?.enabled}" }}"
@@ -415,7 +429,7 @@ public class CallState(
         }
 
         // The caller i.e. `livestream` is deprecated as well
-        call.subscribe {
+        val subscription = call.subscribe {
             logger.v { "[livestreamFlow] #track; event.type: ${it.getEventType()}" }
             if (it is TrackPublishedEvent) {
                 val participant = getOrCreateParticipant(
@@ -454,7 +468,7 @@ public class CallState(
         logger.d { "[livestreamFlow] #track; no args" }
         emitLivestreamVideo()
 
-        awaitClose { }
+        awaitClose { subscription.dispose() }
     }
 
     @Deprecated(
@@ -471,10 +485,12 @@ public class CallState(
         ),
     )
     val livestream: StateFlow<ParticipantState.Video?> =
-        livestreamFlow.debounce(1000).stateIn(scope, SharingStarted.WhileSubscribed(10_000L), null)
+        livestreamFlow.debounce(
+            1000,
+        ).stateIn(SharingStarted.WhileSubscribed(10_000L), null)
 
     private val _sortedParticipantsState = SortedParticipantsState(
-        scope = scope,
+        scopeProvider = { scopeProvider.callScope },
         call = call,
         participants = _participants,
         pinnedParticipantsDetailed = _pinnedParticipantsDetailed,
@@ -576,11 +592,11 @@ public class CallState(
     /** how long the call has been running, rounded to seconds, null if the call didn't start yet */
     public val duration: StateFlow<Duration?> =
         _durationInMs.transform { emit(((it ?: 0L) / 1000L).toDuration(DurationUnit.SECONDS)) }
-            .stateIn(scope, SharingStarted.WhileSubscribed(10000L), null)
+            .stateIn(SharingStarted.WhileSubscribed(10000L), null)
 
     /** how many milliseconds the call has been running, null if the call didn't start yet */
     public val durationInMs: StateFlow<Long?> =
-        _durationInMs.stateIn(scope, SharingStarted.WhileSubscribed(10000L), null)
+        _durationInMs.stateIn(SharingStarted.WhileSubscribed(10000L), null)
 
     /** how many milliseconds the call has been running in the simple date format. */
     public val durationInDateFormat: StateFlow<String?> = durationInMs.mapState { durationInMs ->
@@ -643,7 +659,7 @@ public class CallState(
                 emit(duration)
             }
         }
-    }.distinctUntilChanged().stateIn(scope, SharingStarted.WhileSubscribed(10000L), null)
+    }.distinctUntilChanged().stateIn(SharingStarted.WhileSubscribed(10000L), null)
 
     /**
      * How long the call has been live for, represented as [Duration], or null if the call hasn't been live yet.
@@ -922,7 +938,7 @@ public class CallState(
             }
 
             is LocalCallMissedEvent -> {
-                scope.launch {
+                scopeProvider.callScope.launch {
                     val newRejectedBySet = _rejectedBy.value.toMutableSet()
                     StreamVideo.instanceOrNull()?.let {
                         newRejectedBySet.add(it.user.id)
@@ -1339,7 +1355,7 @@ public class CallState(
             -> closedCaptionManager.handleEvent(event)
 
             is CallModerationBlurEvent -> {
-                scope.launch {
+                scopeProvider.callScope.launch {
                     val callServiceConfig = StreamVideo.instanceOrNull()?.state?.callConfigRegistry?.get(call.type) ?: CallServiceConfig()
                     if (callServiceConfig.moderationConfig.videoModerationConfig.enable) {
                         call.state.moderationManager.applyVideoModeration()
@@ -1551,7 +1567,7 @@ public class CallState(
         if (!shouldJoin) return false
 
         call.setJoinSource(source)
-        autoJoiningCall = scope.launch {
+        autoJoiningCall = scopeProvider.callScope.launch {
             // errors are handled inside the join function
             val joinResult = call.join()
             autoJoiningCall = null
@@ -1618,7 +1634,7 @@ public class CallState(
     fun markSpeakingAsMuted() {
         _speakingWhileMuted.value = true
         speakingWhileMutedResetJob?.cancel()
-        speakingWhileMutedResetJob = scope.launch {
+        speakingWhileMutedResetJob = scopeProvider.callScope.launch {
             delay(2000)
             _speakingWhileMuted.value = false
         }
@@ -1972,11 +1988,12 @@ public class CallState(
      * @see CallState.updateParticipantVisibility
      */
     fun updateParticipantVisibilityFlow(flow: Flow<List<String>>?) {
+        participantVisibilityFlow = flow
         // Cancel any previous job.
         participantsVisibilityMonitor?.cancel()
 
         if (flow != null) {
-            participantsVisibilityMonitor = scope.launch {
+            participantsVisibilityMonitor = scopeProvider.callScope.launch {
                 flow.collectLatest { visibleParticipantIds ->
                     internalParticipants.forEach {
                         if (visibleParticipantIds.contains(it.key)) {
@@ -2044,7 +2061,7 @@ public class CallState(
     private fun observeTelecomHold(repo: JetpackTelecomRepository) {
         telecomHoldObserverJob?.cancel()
 
-        telecomHoldObserverJob = scope.launch(DispatcherProvider.Default) {
+        telecomHoldObserverJob = scopeProvider.callScope.launch(DispatcherProvider.Default) {
             repo.currentCall
                 .map { (it as? TelecomCall.Registered)?.isOnHold == true }
                 .distinctUntilChanged()
@@ -2060,6 +2077,34 @@ public class CallState(
         }
     }
 
+    internal fun restart() {
+        stats.reset()
+        internalParticipants.clear()
+        _participants.value = emptyMap()
+        _activeSpeakers.value = emptyList()
+        _dominantSpeaker.value = null
+        _screenSharingSession.value = null
+        _speakingWhileMuted.value = false
+        _audioProcessingEnabled.value = false
+        _ringingState.value = RingingState.Idle
+        acceptedOnThisDevice = false
+        isJoinAndRingInProgress.set(false)
+        atomicNotification.set(null)
+        telecomHoldObserverJob = null
+        autoJoiningCall = null
+        speakingWhileMutedResetJob = null
+        participantsVisibilityMonitor = null
+        _notificationIdFlow.value = null
+        _serviceRoute.value = ServiceRoute.UNDECIDED
+        _sortedParticipantsState.restart()
+        restartScopeObservers()
+    }
+
+    private fun restartScopeObservers() {
+        participantVisibilityFlow?.let { updateParticipantVisibilityFlow(it) }
+        jetpackTelecomRepository?.let { observeTelecomHold(it) }
+    }
+
     internal fun cleanup() {
         previousRingingStates.clear()
         activeStateGate.cleanup()
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallStats.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallStats.kt
index 210f19db31f..001f5e7980c 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallStats.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallStats.kt
@@ -22,11 +22,8 @@ import io.getstream.video.android.core.call.stats.model.RtcStatsReport
 import io.getstream.webrtc.CameraEnumerationAndroid
 import io.getstream.webrtc.RTCStats
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.stateIn
 import stream.video.sfu.models.TrackType
 
 data class MediaStatsInfo(
@@ -98,6 +95,20 @@ public class PeerConnectionStats(scope: CoroutineScope) {
     internal var lastAudioBytes: Long? = null
     internal var lastAudioTimestampUs: Double? = null
 
+    internal fun reset() {
+        _latency.value = 0
+        _resolution.value = ""
+        _qualityDropReason.value = ""
+        _jitterInMs.value = 0
+        _bitrateKbps.value = 0F
+        _videoCodec.value = ""
+        _audioBitrateKbps.value = 0F
+        _audioCodec.value = ""
+        _audioTargetBitrateKbps.value = 0F
+        lastAudioBytes = null
+        lastAudioTimestampUs = null
+    }
+
     /**
      * Derives the audio bitrate from the change in [bytes] since the previous poll.
      *
@@ -135,15 +146,17 @@ public data class LocalStats(
 public class CallStats(val call: Call, val callScope: CoroutineScope) {
     private val logger by taggedLogger("CallStats")
 
-    private val supervisorJob = SupervisorJob()
-    private val scope = CoroutineScope(callScope.coroutineContext + supervisorJob)
-    // TODO: cleanup the scope
-
-    val publisher = PeerConnectionStats(scope)
-    val subscriber = PeerConnectionStats(scope)
+    val publisher = PeerConnectionStats(callScope)
+    val subscriber = PeerConnectionStats(callScope)
     val _local = MutableStateFlow<LocalStats?>(null)
     val local: StateFlow<LocalStats?> =
-        _local.stateIn(scope, SharingStarted.WhileSubscribed(), null)
+        _local
+
+    internal fun reset() {
+        publisher.reset()
+        subscriber.reset()
+        _local.value = null
+    }
 
     fun updateFromRTCStats(stats: RtcStatsReport?, isPublisher: Boolean = true) {
         if (stats == null) return
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/MediaManager.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/MediaManager.kt
index dd4dfc39aeb..effe69511a5 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/MediaManager.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/MediaManager.kt
@@ -555,6 +555,7 @@ class MicrophoneManager(
     private val logger by taggedLogger("Media:MicrophoneManager")
 
     private lateinit var audioHandler: AudioSwitchDecorator
+    private var audioHandlerStopped = false
 
     private var setupCompleted = AtomicBoolean(false)
     internal var audioManager: AudioManager? = null
@@ -1233,7 +1234,8 @@ class MicrophoneManager(
     }
 
     fun cleanup() {
-        ifAudioHandlerInitialized { it.stop() }
+        if (!audioHandlerStopped) ifAudioHandlerInitialized { it.stop() }
+        audioHandlerStopped = true
         cleanupUsbDeviceDetection()
         // Dropped with the call that captured it, so a value remembered from a call that ended in
         // music cannot be put back onto the next one.
@@ -1263,7 +1265,7 @@ class MicrophoneManager(
                 setupUsbDeviceDetection()
             }
 
-            if (canHandleDeviceSwitch() && !::audioHandler.isInitialized) {
+            if (canHandleDeviceSwitch() && (!::audioHandler.isInitialized || audioHandlerStopped)) {
                 // Use default priority (Bluetooth -> Wired -> Earpiece -> Speakerphone)
                 // unless preferSpeaker is true, then prioritize speakerphone over earpiece
                 val preferredDeviceList = listOf(
@@ -1315,6 +1317,7 @@ class MicrophoneManager(
                     audioHandler.setCommunicationModeEnabled(false)
                 }
 
+                audioHandlerStopped = false
                 logger.d { "[setup] Calling start on instance $audioHandler" }
                 audioHandler.start()
             } else {
@@ -1711,6 +1714,9 @@ class CameraManager(
     }
 
     fun cleanup() {
+        job?.cancel()
+        job = null
+        if (!setupCompleted) return
         stopCapture()
         if (::videoCapturer.isInitialized) {
             videoCapturer.dispose()
@@ -2042,6 +2048,11 @@ class MediaManagerImpl(
         _screenShareVideoSource = null
     }
 
+    /** Reopens track creation only after the previous call scope and media cleanup have finished. */
+    internal fun prepareForJoin() {
+        synchronized(mediaLock) { released = false }
+    }
+
     fun cleanup() {
         // Terminal teardown: dispose tracks/sources and mark released so the mic/camera mute paths
         // ([runOnAudioTrackIfAvailable]/[runOnVideoTrackIfAvailable]) no-op afterwards instead of
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt
index 95f604197ec..bac99578f34 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/StreamVideoClient.kt
@@ -169,6 +169,7 @@ import okhttp3.Request
 import okhttp3.Response
 import retrofit2.HttpException
 import java.util.*
+import java.util.concurrent.ConcurrentHashMap
 
 internal const val WAIT_FOR_CONNECTION_ID_TIMEOUT = 5000L
 internal const val defaultAudioUsage = AudioAttributes.USAGE_VOICE_COMMUNICATION
@@ -244,7 +245,33 @@ internal class StreamVideoClient internal constructor(
 
     private val logger by taggedLogger("Call:StreamVideo")
     private var subscriptions = mutableSetOf<EventSubscription>()
-    private var calls = mutableMapOf<String, Call>()
+
+    /**
+     * Canonical get-or-create cache used by [call].
+     *
+     * The following APIs can access the cache concurrently:
+     * - `client.call(type, id)` → get or create a call.
+     * - `call.join()` → restore a retained call only when its CID is absent.
+     * - `call.leave()` → remove only the matching call instance.
+     *
+     * [ConcurrentHashMap] coordinates these operations without locking the complete call lifecycle.
+     */
+    private val calls = ConcurrentHashMap<String, Call>()
+
+    /**
+     * Rejoined [Call] instances that are not stored in [calls], but still need coordinator and SFU
+     * events.
+     *
+     * A non-cached call can exist when an app retains `callA`, invokes `callA.leave()` (which
+     * removes it from [calls]), and then obtains `callA1` through [call] for the same CID. If the
+     * app later invokes `callA.join()`, the retained call is allowed to rejoin, but it must not
+     * replace the newer `callA1` in the get-or-create cache.
+     *
+     * Coordinator and SFU events are routed through `calls[cid]`. Because the successfully rejoined
+     * `callA` is no longer that cached instance, it is added here so both event sources are also
+     * propagated to it. The call is removed from this map when it leaves again.
+     */
+    private val reusedCallEventTargets = ConcurrentHashMap<String, MutableSet<Call>>()
     private val destroyedCalls = LruCache<Int, Call>(100)
     internal val callSoundAndVibrationPlayer = CallSoundAndVibrationPlayer(context)
 
@@ -278,12 +305,33 @@ internal class StreamVideoClient internal constructor(
             destroyedCalls.put(call.hashCode(), call)
         }
         logger.d { "[cleanup] Removing call from cache: ${call.cid}" }
-        calls.remove(call.cid)
+        calls.remove(call.cid, call)
+        reusedCallEventTargets.computeIfPresent(call.cid) { _, reusedCalls ->
+            reusedCalls.remove(call)
+            reusedCalls.takeUnless { it.isEmpty() }
+        }
+    }
+
+    internal fun onCallReused(call: Call) {
+        destroyedCalls.remove(call.hashCode())
+        calls.putIfAbsent(call.cid, call)
+    }
+
+    internal fun onCallJoined(call: Call) {
+        val cachedCall = calls.putIfAbsent(call.cid, call)
+        if (cachedCall == null || cachedCall === call) {
+            return
+        }
+
+        reusedCallEventTargets.compute(call.cid) { _, reusedCalls ->
+            (reusedCalls ?: ConcurrentHashMap.newKeySet()).apply { add(call) }
+        }
     }
 
     override fun cleanup() {
         // remove all cached calls
         calls.clear()
+        reusedCallEventTargets.clear()
         destroyedCalls.evictAll()
         // cancel the StreamClient subscription before tearing down the socket
         streamClientSubscription?.cancel()
@@ -698,7 +746,7 @@ internal class StreamVideoClient internal constructor(
     }
 
     internal fun forwardToCallSubscriptions(cid: String, event: VideoEvent) {
-        calls[cid]?.fireEvent(event)
+        forEachCallInstance(cid) { call -> call.fireEvent(event) }
         notifyDestroyedCalls(event)
     }
 
@@ -718,13 +766,21 @@ internal class StreamVideoClient internal constructor(
     }
 
     internal fun propagateEventToCall(cid: String, event: VideoEvent) {
-        calls[cid]?.let { call ->
+        forEachCallInstance(cid) { call ->
             call.state.handleEvent(event)
             call.session.value?.handleEvent(event)
             call.handleEvent(event)
         }
     }
 
+    private inline fun forEachCallInstance(cid: String, action: (Call) -> Unit) {
+        val cachedCall = calls[cid]
+        cachedCall?.let(action)
+        reusedCallEventTargets[cid]?.forEach { reusedCall ->
+            if (reusedCall !== cachedCall) action(reusedCall)
+        }
+    }
+
     internal fun deliverToDestroyedCalls(event: VideoEvent) {
         deliverIntentToDestroyedCalls(event)
     }
@@ -1283,13 +1339,7 @@ internal class StreamVideoClient internal constructor(
         val idOrRandom = id.ifEmpty { UUID.randomUUID().toString() }
 
         val cid = "$type:$idOrRandom"
-        return if (calls.contains(cid)) {
-            calls[cid]!!
-        } else {
-            val call = Call(this, type, idOrRandom, user)
-            calls[cid] = call
-            call
-        }
+        return calls.computeIfAbsent(cid) { Call(this, type, idOrRandom, user) }
     }
 
     @OptIn(InternalCoroutinesApi::class)
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/CallAnalytics.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/CallAnalytics.kt
index ea046203f28..3f4381144c6 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/CallAnalytics.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/CallAnalytics.kt
@@ -18,6 +18,8 @@ package io.getstream.video.android.core.analytics.call
 
 import android.content.Context
 import io.getstream.log.taggedLogger
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
 import io.getstream.video.android.core.CallLeaveReason
 import io.getstream.video.android.core.ParticipantState
 import io.getstream.video.android.core.RealtimeConnection
@@ -34,6 +36,7 @@ import io.getstream.video.android.core.analytics.call.observer.model.Stage
 import io.getstream.video.android.core.analytics.call.observer.toVideoAnalyticsIceState
 import io.getstream.video.android.core.analytics.reporting.ClientEventReporter
 import io.getstream.video.android.core.analytics.reporting.model.AnalyticsCallAbortReason
+import io.getstream.video.android.core.call.Restartable
 import io.getstream.video.android.core.call.RtcSession
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.StateFlow
@@ -47,7 +50,7 @@ internal class CallAnalytics(
     val participants: StateFlow<List<ParticipantState>>,
     val eventReporter: ClientEventReporter,
     val observerScope: CoroutineScope,
-) {
+) : Restartable {
     val logger by taggedLogger("CallAnalyticsHooks")
 
     val joinAnalyticsStateHolder = JoinAnalyticsStateHolder()
@@ -98,6 +101,18 @@ internal class CallAnalytics(
             sfuAnalyticsStateHolder,
         )
 
+    override fun restart(): Result<Unit> {
+        val currentJoinAttemptId = joinAnalyticsStateHolder.state.value.joinStageAttemptId
+        joinAnalyticsStateHolder.reset()
+        joinAnalyticsStateHolder.updateJoinStageAttemptId(currentJoinAttemptId)
+        sfuAnalyticsStateHolder.updateSfuId("")
+        sfuAnalyticsStateHolder.updateStageId("")
+        sfuAnalyticsStateHolder.updateStage(Stage.NOT_STARTED)
+        peerConnectionAnalytics.stateHolder.reset()
+        resetAfterJoinSuccess()
+        return Success(Unit)
+    }
+
     fun resetAfterJoinSuccess() {
         videoAnalytics.reset()
         audioAnalytics.reset()
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalytics.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalytics.kt
index 5c571c9a773..278ba634751 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalytics.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalytics.kt
@@ -128,6 +128,17 @@ internal class JoinAnalytics(
         }
     }
 
+    /** Records a failure that prevents the Call from reaching the Coordinator join request. */
+    fun onJoinPreparationError(failureCode: String, message: String) {
+        resetStage()
+        onJoinRequestStart(JoinReason.FirstAttempt)
+        onJoinRequestPermanentError(
+            retryCount = 0,
+            failureCode = failureCode,
+            message = message,
+        )
+    }
+
     /**
      * Records a terminal join failure after all retry attempts to connect to the Coordinator or
      * SFU have been exhausted.
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/Restartable.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/Restartable.kt
new file mode 100644
index 00000000000..22be312fa74
--- /dev/null
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/Restartable.kt
@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call
+
+import io.getstream.result.Result
+
+/** A component that can restore its own state for another Call lifecycle generation. */
+internal fun interface Restartable {
+    /** Resets owned state, returning a failure when the next lifecycle cannot be prepared. */
+    fun restart(): Result<Unit>
+}
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallCleanupState.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallCleanupState.kt
new file mode 100644
index 00000000000..59bb3313664
--- /dev/null
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallCleanupState.kt
@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.components
+
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
+import kotlinx.coroutines.Job
+
+/** Stores and synchronizes the cleanup state shared by leave, cleanup and the next join. */
+internal class CallCleanupState {
+    private data class PendingJobs(
+        val cleanup: Job?,
+        val cancelledCall: Job?,
+    )
+
+    private val lock = Any()
+    private var cleanupJob: Job? = null
+    private var cancelledCallJob: Job? = null
+    private var cleanedUp = false
+
+    @Volatile
+    private var destroyed = false
+
+    /** True after leave starts and until cleanup completes and the Call is restarted. */
+    val isDestroyed: Boolean get() = destroyed
+
+    /** Runs leave setup once and stores the asynchronous cleanup job before releasing the lock. */
+    fun startLeave(createCleanupJob: () -> Job) {
+        synchronized(lock) {
+            if (destroyed) return
+            destroyed = true
+            cleanupJob = createCleanupJob()
+        }
+    }
+
+    /** Runs terminal cleanup once while keeping its job bookkeeping atomic with joining. */
+    fun cleanupOnce(cleanup: () -> Unit) {
+        synchronized(lock) {
+            if (cleanedUp) return
+            cleanedUp = true
+            destroyed = true
+            cleanup()
+        }
+    }
+
+    /** Records the cancelled Call job while [cleanupOnce] still owns the cleanup lock. */
+    fun recordCancelledCallJob(job: Job?) {
+        synchronized(lock) {
+            cancelledCallJob = job
+        }
+    }
+
+    /** Waits for cleanup and cancellation, then atomically restores the active lifecycle state. */
+    suspend fun prepareForJoin(restart: () -> Result<Unit>): Result<Unit> {
+        while (true) {
+            val pending = synchronized(lock) {
+                PendingJobs(cleanupJob, cancelledCallJob)
+            }
+            pending.cleanup?.join()
+            pending.cancelledCall?.join()
+
+            val restartResult = synchronized(lock) {
+                if (cleanupJob !== pending.cleanup || cancelledCallJob !== pending.cancelledCall) {
+                    return@synchronized null
+                }
+                if (destroyed) {
+                    val result = restart()
+                    if (result !is Success) return@synchronized result
+                    destroyed = false
+                    cleanupJob = null
+                    cancelledCallJob = null
+                    cleanedUp = false
+                }
+                Success(Unit)
+            }
+            if (restartResult != null) return restartResult
+        }
+    }
+}
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt
index 47ea5328afb..52253e208b6 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt
@@ -75,6 +75,7 @@ internal class CallJoinCoordinator(
 ) {
     private companion object {
         const val JOIN_FLIGHT_KEY = "join"
+        const val CALL_RESTART_FAILED = "CALL_RESTART_FAILED"
     }
 
     private val logger by taggedLogger("Call:JoinCoordinator:$type:$id")
@@ -137,6 +138,17 @@ internal class CallJoinCoordinator(
         callJoinInterceptor: CallJoinInterceptor? = null,
     ): Result<RtcSession> {
         callAnalytics.joinAnalytics.onJoinFunctionStart()
+        when (val preparation = lifecycle.prepareForJoin()) {
+            is Failure -> {
+                state._connection.value = RealtimeConnection.Failed(preparation.value)
+                callAnalytics.joinAnalytics.onJoinPreparationError(
+                    failureCode = CALL_RESTART_FAILED,
+                    message = preparation.value.message,
+                )
+                return preparation
+            }
+            is Success -> Unit
+        }
         var coalesced = false
         return joinFlight.run(
             JOIN_FLIGHT_KEY,
@@ -266,7 +278,6 @@ internal class CallJoinCoordinator(
 
         var result: Result<RtcSession>
 
-        lifecycle.resetLeaveGuard()
         while (retryCount < 3) {
             result = joinInternal(
                 create,
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManager.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManager.kt
index ba1a4174e8f..5debc62bd21 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManager.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManager.kt
@@ -24,15 +24,14 @@ import io.getstream.video.android.core.RealtimeConnection
 import io.getstream.video.android.core.SdkCause
 import io.getstream.video.android.core.StreamVideoClient
 import io.getstream.video.android.core.analytics.call.CallAnalytics
-import io.getstream.video.android.core.call.scope.ScopeProvider
-import io.getstream.video.android.core.utils.AtomicUnitCall
 import io.getstream.video.android.core.utils.safeCall
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 
 /**
- * Owns the call's lifecycle / teardown: leaving, ending, the single-shot leave guard
- * ([AtomicUnitCall]), the destroyed flag, and the ordered cleanup of state, session,
- * jobs and media.
+ * Owns leaving, ending, the leave guard and the ordered cleanup of state, session, jobs and media.
+ * A later join waits for cleanup and cancellation before attaching fresh call resources.
  *
  * This component is constructed early in [io.getstream.video.android.core.Call] (the reconnect
  * and event pipeline depend on it), so collaborators created later are injected as lazy
@@ -41,10 +40,10 @@ import kotlinx.coroutines.launch
 internal class CallLifecycleManager(
     private val clientImpl: StreamVideoClient,
     private val sessionManager: CallSessionManager,
-    private val scopeProvider: ScopeProvider,
     private val callRegistry: ClientCallRegistry,
-    /** Cancels the call's supervisor job and scope once in-flight children complete. */
-    private val shutDownJobs: () -> Unit,
+    /** Detaches and cancels the call scope, returning its job so rejoin can await cancellation. */
+    private val shutDownJobs: () -> Job?,
+    private val restartCoordinator: CallRestartCoordinator,
     private val state: CallState,
     private val callAnalytics: CallAnalytics,
     private val statsReporter: CallStatsReporter,
@@ -60,12 +59,10 @@ internal class CallLifecycleManager(
     private val logger by taggedLogger("Call:LifecycleManager:$type:$id")
 
     private val cid = "$type:$id"
+    private val cleanupState = CallCleanupState()
 
-    // Atomic controls
-    private var atomicLeave = AtomicUnitCall()
-
-    /** Call has been left and the object is cleaned up and destroyed. */
-    var isDestroyed = false
+    /** True between leave and the next join, including while cleanup is still running. */
+    val isDestroyed: Boolean get() = cleanupState.isDestroyed
 
     /**
      * Time (in millis) when the full reconnection flow started. Will be null again once
@@ -73,10 +70,11 @@ internal class CallLifecycleManager(
      */
     private var sfuSocketReconnectionTime: Long? = null
 
-    /** Resets the leave guard so a fresh join can run after a previous leave. */
-    fun resetLeaveGuard() {
-        atomicLeave = AtomicUnitCall()
-    }
+    /** Wait outside the call scope, which may already be cancelled by leave. */
+    suspend fun prepareForJoin(): Result<Unit> =
+        cleanupState.prepareForJoin {
+            restartCoordinator.restart()
+        }
 
     fun leave(reason: CallLeaveReason) {
         logger.d { "[leave] #ringing; call_cid:$cid" }
@@ -88,38 +86,37 @@ internal class CallLifecycleManager(
         internalLeave(CallLeaveReason.Custom(reason))
     }
 
-    private fun internalLeave(reason: CallLeaveReason) = atomicLeave {
-        stopConnectionMonitors()
-        callAnalytics.stopObservers()
-        sessionMonitor().cancelSfuObservers()
-        state._connection.value = RealtimeConnection.Disconnected
-        logger.v { "[leave] #ringing; call_id = $id" }
-        if (isDestroyed) {
-            logger.w { "[leave] #ringing; Call already destroyed, ignoring" }
-            return@atomicLeave
-        }
-        isDestroyed = true
-
-        sfuSocketReconnectionTime = null
-
-        /**
-         * TODO Rahul, need to check which call has owned the media at the moment(probably use active call)
-         */
-        media.disableLocalCapture()
-
-        callRegistry.detach()
-
-        clientImpl.scope.launch {
-            val leaveReason = "[reason=${reason::class.simpleName}, message=${reason.message}]"
-            callAnalytics.onCallLeave(sessionManager.session, reason)
-            safeCall {
-                sessionManager.session.value?.sfuTracer?.trace("leave-call", leaveReason)
-                val stats = statsReporter.collectStats()
-                sessionManager.session.value?.sendCallStats(stats)
-            }
-            // Must complete before cleanup() cancels the session's supervisor job.
-            safeCall { sessionManager.session.value?.sendLeaveEvent(leaveReason) }
-            cleanup()
+    private fun internalLeave(reason: CallLeaveReason) {
+        cleanupState.startLeave {
+            stopConnectionMonitors()
+            callAnalytics.stopObservers()
+            sessionMonitor().cancelSfuObservers()
+            state._connection.value = RealtimeConnection.Disconnected
+            logger.v { "[leave] #ringing; call_id = $id" }
+            sfuSocketReconnectionTime = null
+
+            /**
+             * TODO Rahul, need to check which call has owned the media at the moment(probably use active call)
+             */
+            media.disableLocalCapture()
+
+            callRegistry.detach()
+
+            clientImpl.scope.launch(start = CoroutineStart.LAZY) {
+                try {
+                    val leaveReason = "[reason=${reason::class.simpleName}, message=${reason.message}]"
+                    callAnalytics.onCallLeave(sessionManager.session, reason)
+                    safeCall {
+                        sessionManager.session.value?.sfuTracer?.trace("leave-call", leaveReason)
+                        val stats = statsReporter.collectStats()
+                        sessionManager.session.value?.sendCallStats(stats)
+                    }
+                    // Must complete before cleanup() cancels the session's supervisor job.
+                    safeCall { sessionManager.session.value?.sendLeaveEvent(leaveReason) }
+                } finally {
+                    cleanup()
+                }
+            }.also { it.start() }
         }
     }
 
@@ -138,15 +135,20 @@ internal class CallLifecycleManager(
     }
 
     fun cleanup() {
-        // monitor.stop()
-        state.cleanup()
-        sessionManager.session.value?.cleanup()
-        shutDownJobs()
-        statsReporter.stop()
-        media.cleanup() // TODO Rahul, Verify Later: need to check which call has owned the media at the moment(probably use active call)
-        sessionManager.setActiveSession(null)
-        // Cleanup the call's scope provider
-        scopeProvider.cleanup()
+        cleanupState.cleanupOnce {
+            try {
+                state.cleanup()
+                sessionManager.session.value?.cleanup()
+            } finally {
+                cleanupState.recordCancelledCallJob(shutDownJobs())
+                try {
+                    statsReporter.stop()
+                    media.cleanup()
+                } finally {
+                    sessionManager.setActiveSession(null)
+                }
+            }
+        }
     }
 
     /** Stops the ICE and connectivity monitors. */
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallMediaManager.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallMediaManager.kt
index 25781dd514a..af4c5d78c22 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallMediaManager.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallMediaManager.kt
@@ -22,18 +22,22 @@ import io.getstream.android.video.api.generated.models.CallSettingsResponse
 import io.getstream.android.video.api.generated.models.OwnCapability
 import io.getstream.android.video.api.generated.models.VideoSettingsResponse
 import io.getstream.log.taggedLogger
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
 import io.getstream.video.android.core.CallState
 import io.getstream.video.android.core.CameraDirection
 import io.getstream.video.android.core.DeviceStatus
 import io.getstream.video.android.core.MediaManagerImpl
 import io.getstream.video.android.core.StreamVideoClient
 import io.getstream.video.android.core.audio.StreamAudioDevice
+import io.getstream.video.android.core.call.Restartable
 import io.getstream.video.android.core.call.connection.StreamPeerConnectionFactory
 import io.getstream.video.android.core.call.utils.SoundInputProcessor
 import io.getstream.video.android.core.utils.RampValueUpAndDownHelper
 import io.getstream.webrtc.EglBase
 import io.getstream.webrtc.audio.JavaAudioDeviceModule.AudioSamples
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
@@ -53,12 +57,15 @@ internal class CallMediaManager(
     private val type: String,
     private val id: String,
     private val clientImpl: StreamVideoClient,
-    private val scope: CoroutineScope,
+    private val coroutineScope: CoroutineScope,
     private val state: CallState,
     private val sessionManager: CallSessionManager,
     private val eglBase: () -> EglBase,
     private val mediaManagerFactory: MediaManagerFactory,
-) {
+) : Restartable {
+    private var cameraEnabledBeforeLeave: Boolean? = null
+    private var microphoneEnabledBeforeLeave: Boolean? = null
+
     private val logger by taggedLogger("Call:MediaManager:$type:$id")
 
     private val soundInputProcessor = SoundInputProcessor(thresholdCrossedCallback = {
@@ -67,6 +74,7 @@ internal class CallMediaManager(
         }
     })
     private val audioLevelOutputHelper = RampValueUpAndDownHelper()
+    private var audioLevelMonitoringJob: Job? = null
 
     /** Smoothed local microphone volume level (0..1). */
     val localMicrophoneAudioLevel: StateFlow<Float> = audioLevelOutputHelper.currentLevel
@@ -121,15 +129,18 @@ internal class CallMediaManager(
             _peerConnectionFactory = value
         }
 
-    val mediaManager by lazy {
+    private val mediaManagerDelegate = lazy {
         mediaManagerFactory.create(
             audioUsageProvider = { clientImpl.callServiceConfigRegistry.get(type).audioUsage },
         )
     }
 
+    val mediaManager: MediaManagerImpl get() = mediaManagerDelegate.value
+
     /** Starts streaming smoothed microphone audio levels into [localMicrophoneAudioLevel]. */
     fun startAudioLevelMonitoring() {
-        scope.launch {
+        audioLevelMonitoringJob?.cancel()
+        audioLevelMonitoringJob = coroutineScope.launch {
             soundInputProcessor.currentAudioLevel.collect {
                 audioLevelOutputHelper.rampToValue(it)
             }
@@ -246,11 +257,16 @@ internal class CallMediaManager(
             camera.setEnabled(callSettings.video.cameraDefaultOn)
         }
 
+        cameraEnabledBeforeLeave?.let { camera.setEnabled(it) }
+        cameraEnabledBeforeLeave = null
+
         // Mic
         if (microphone.status.value == DeviceStatus.NotSelected) {
             val enabled = callSettings.audio.micDefaultOn
             microphone.setEnabled(enabled)
         }
+        microphoneEnabledBeforeLeave?.let { microphone.setEnabled(it) }
+        microphoneEnabledBeforeLeave = null
     }
 
     private fun monitorHeadset() {
@@ -278,7 +294,7 @@ internal class CallMediaManager(
                     microphone.select(deviceBeforeHeadset)
                 }
             }
-        }.launchIn(scope)
+        }.launchIn(coroutineScope)
     }
 
     fun startScreenSharing(
@@ -403,17 +419,26 @@ internal class CallMediaManager(
 
     /** Disables all local capture devices. Used when leaving the call. */
     fun disableLocalCapture() {
+        if (!mediaManagerDelegate.isInitialized()) return
+        cameraEnabledBeforeLeave = mediaManager.camera.status.value is DeviceStatus.Enabled
+        microphoneEnabledBeforeLeave = mediaManager.microphone.status.value is DeviceStatus.Enabled
         stopScreenSharing()
         mediaManager.camera.disable()
         mediaManager.microphone.disable()
     }
 
+    override fun restart(): Result<Unit> {
+        startAudioLevelMonitoring()
+        if (mediaManagerDelegate.isInitialized()) mediaManager.prepareForJoin()
+        return Success(Unit)
+    }
+
     fun cleanup() {
         // The wanted state must not outlive the call: a reused Call would otherwise re-apply it
         // to the factory built for the next session.
         resetDesiredAudioProcessing()
         resetDesiredHardwareNoiseSuppressor()
         resetDesiredHardwareAcousticEchoCanceler()
-        mediaManager.cleanup()
+        if (mediaManagerDelegate.isInitialized()) mediaManager.cleanup()
     }
 }
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallRestartCoordinator.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallRestartCoordinator.kt
new file mode 100644
index 00000000000..882a0f8b88e
--- /dev/null
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallRestartCoordinator.kt
@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.components
+
+import io.getstream.result.Result
+import io.getstream.result.Result.Failure
+import io.getstream.result.Result.Success
+import io.getstream.video.android.core.call.Restartable
+
+/** Coordinates restoration of a Call after its previous lifecycle generation has ended. */
+internal class CallRestartCoordinator(
+    private val scopeProvider: Restartable,
+    private val restartables: List<Restartable>,
+    private val onRestarted: () -> Unit = {},
+) {
+    /** Restarts generation-owned resources first, then each persistent component in order. */
+    fun restart(): Result<Unit> {
+        val scopeResult = scopeProvider.restart()
+        if (scopeResult is Failure) return scopeResult
+
+        restartables.forEach { restartable ->
+            val result = restartable.restart()
+            if (result is Failure) return result
+        }
+        onRestarted()
+        return Success(Unit)
+    }
+}
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallSessionManager.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallSessionManager.kt
index 472ee2a227b..bb4c3f477a7 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallSessionManager.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallSessionManager.kt
@@ -16,6 +16,9 @@
 
 package io.getstream.video.android.core.call.components
 
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
+import io.getstream.video.android.core.call.Restartable
 import io.getstream.video.android.core.call.RtcSession
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -31,7 +34,7 @@ import java.util.concurrent.ConcurrentHashMap
  * Keeping this state in a single component gives the join, reconnect, connectivity and
  * lifecycle collaborators a single source of truth to depend on.
  */
-internal class CallSessionManager() {
+internal class CallSessionManager() : Restartable {
     private val _session: MutableStateFlow<RtcSession?> = MutableStateFlow(null)
 
     /**
@@ -48,7 +51,20 @@ internal class CallSessionManager() {
 
     var sessionId = UUID.randomUUID().toString()
 
-    val unifiedSessionId = UUID.randomUUID().toString()
+    var unifiedSessionId = UUID.randomUUID().toString()
+        private set
+
+    override fun restart(): Result<Unit> {
+        sessionId = UUID.randomUUID().toString()
+        unifiedSessionId = UUID.randomUUID().toString()
+        nonFastReconnectAttempts = 0
+        connectStartTime = 0L
+        reconnectStartTime = 0L
+        reconnectDeadlineMillis = DEFAULT_RECONNECT_DEADLINE_MILLIS
+        location = null
+        clearFailedSfuIds()
+        return Success(Unit)
+    }
 
     /** Cached SFU location used when (re)joining a call. */
     var location: String? = null
@@ -95,5 +111,9 @@ internal class CallSessionManager() {
      * reconnect loop and the connectivity monitor, so it lives with the other shared
      * reconnect bookkeeping.
      */
-    var reconnectDeadlineMillis: Int = 10_000
+    var reconnectDeadlineMillis: Int = DEFAULT_RECONNECT_DEADLINE_MILLIS
+
+    private companion object {
+        const val DEFAULT_RECONNECT_DEADLINE_MILLIS = 10_000
+    }
 }
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallStatsReporter.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallStatsReporter.kt
index 37b383b3db6..e2ecf99adbe 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallStatsReporter.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallStatsReporter.kt
@@ -17,8 +17,11 @@
 package io.getstream.video.android.core.call.components
 
 import io.getstream.log.taggedLogger
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
 import io.getstream.video.android.core.CallState
 import io.getstream.video.android.core.CallStatsReport
+import io.getstream.video.android.core.call.Restartable
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
@@ -36,7 +39,7 @@ internal class CallStatsReporter(
     private val scope: CoroutineScope,
     private val sessionManager: CallSessionManager,
     private val state: CallState,
-) {
+) : Restartable {
     private val logger by taggedLogger("Call:StatsReporter:$type:$id")
 
     /** Contains stats events for observation. */
@@ -66,6 +69,13 @@ internal class CallStatsReporter(
         callStatsReportingJob?.cancel()
     }
 
+    override fun restart(): Result<Unit> {
+        stop()
+        statsReport.value = null
+        statLatencyHistory.value = listOf(0, 0, 0)
+        return Success(Unit)
+    }
+
     suspend fun collectStats(): CallStatsReport {
         val currentSession = sessionManager.session.value
         val publisherStats = runCatching { currentSession?.getPublisherStats() }.getOrNull()
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/CallScopeGenerationManager.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/CallScopeGenerationManager.kt
new file mode 100644
index 00000000000..faa92f889ec
--- /dev/null
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/CallScopeGenerationManager.kt
@@ -0,0 +1,126 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.scope
+
+import io.getstream.result.Error
+import io.getstream.result.Result
+import io.getstream.result.Result.Failure
+import io.getstream.result.Result.Success
+import kotlinx.coroutines.CompletableJob
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.AtomicReference
+import kotlin.coroutines.CoroutineContext
+
+/** Manages the state and lifecycle transitions of the current Call coroutine scope generation. */
+internal class CallScopeGenerationManager(
+    private val parentCoroutineContext: CoroutineContext,
+) {
+
+    private class GenerationState(
+        val debugInstanceId: Int,
+        val callScope: CoroutineScope,
+        val callJob: CompletableJob,
+    ) {
+        var executor: ExecutorService? = null
+        var isCleanedUp: Boolean = false
+    }
+
+    /** Resources detached from a generation when cleanup changes it to the cleaned state. */
+    internal data class CleanupResult(
+        val debugInstanceId: Int,
+        val callJob: Job,
+        val executor: ExecutorService?,
+        val cleanedUpNow: Boolean,
+    )
+
+    private val lifecycleLock = Any()
+    private val debugInstanceCounter = AtomicInteger(0)
+    private val current = AtomicReference(createGeneration())
+
+    internal val currentCallScope: CoroutineScope
+        get() = current.get().callScope
+
+    internal val currentDebugInstanceId: Int
+        get() = current.get().debugInstanceId
+
+    internal fun requireActive(): Int = synchronized(lifecycleLock) {
+        current.get().also(::checkActive).debugInstanceId
+    }
+
+    internal fun getOrCreateExecutor(
+        callId: String,
+    ): ExecutorService = synchronized(lifecycleLock) {
+        val generation = current.get().also(::checkActive)
+        generation.executor ?: Executors.newSingleThreadExecutor { runnable ->
+            Thread(runnable, "rtc-call-$callId")
+        }.also {
+            generation.executor = it
+        }
+    }
+
+    internal fun cleanup(): CleanupResult = synchronized(lifecycleLock) {
+        val generation = current.get()
+        if (generation.isCleanedUp) {
+            CleanupResult(
+                debugInstanceId = generation.debugInstanceId,
+                callJob = generation.callJob,
+                executor = null,
+                cleanedUpNow = false,
+            )
+        } else {
+            generation.isCleanedUp = true
+            val executor = generation.executor
+            generation.executor = null
+            CleanupResult(
+                debugInstanceId = generation.debugInstanceId,
+                callJob = generation.callJob,
+                executor = executor,
+                cleanedUpNow = true,
+            )
+        }
+    }
+
+    internal fun restart(): Result<Unit> = synchronized(lifecycleLock) {
+        if (!current.get().isCleanedUp) {
+            return@synchronized Failure(
+                Error.GenericError(
+                    "ScopeProvider cannot restart before its current generation is cleaned up",
+                ),
+            )
+        }
+        current.set(createGeneration())
+        Success(Unit)
+    }
+
+    private fun createGeneration(): GenerationState {
+        val callJob = SupervisorJob(parentCoroutineContext[Job])
+        return GenerationState(
+            debugInstanceId = debugInstanceCounter.incrementAndGet(),
+            callScope = CoroutineScope(parentCoroutineContext + callJob),
+            callJob = callJob,
+        )
+    }
+
+    private fun checkActive(generation: GenerationState) {
+        check(!generation.isCleanedUp) { "ScopeProvider has been cleaned up" }
+    }
+}
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/RestartableStateFlow.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/RestartableStateFlow.kt
new file mode 100644
index 00000000000..fb52e0f5c63
--- /dev/null
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/RestartableStateFlow.kt
@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.scope
+
+import kotlinx.coroutines.ExperimentalForInheritanceCoroutinesApi
+import kotlinx.coroutines.InternalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingCommand
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.launch
+
+/**
+ * Keeps state storage and subscriber identity across call scopes. Sharing commands are driven
+ * by the exposed state's subscribers, so WhileSubscribed does not collect without subscribers.
+ */
+@OptIn(InternalCoroutinesApi::class, ExperimentalForInheritanceCoroutinesApi::class)
+internal class RestartableStateFlow<T>(
+    private val upstream: Flow<T>,
+    private val scopeProvider: ScopeProviderImpl,
+    private val initialValue: T,
+    private val started: SharingStarted,
+) : StateFlow<T> {
+    private val state = MutableStateFlow(initialValue)
+    private var producerJob: Job? = null
+
+    init {
+        scopeProvider.onRestart(::restartProducer)
+        restartProducer()
+    }
+
+    private fun restartProducer() {
+        producerJob?.cancel()
+        state.value = initialValue
+        // TODO: Revisit whether this manual SharingStarted implementation can be simplified.
+        //  It currently uses the exposed StateFlow's subscriber count to preserve stateIn's
+        //  WhileSubscribed and replay-reset behavior across scope restarts.
+        producerJob = scopeProvider.callScope.launch {
+            started.command(state.subscriptionCount).distinctUntilChanged().collectLatest { command ->
+                when (command) {
+                    SharingCommand.START -> upstream.collect { state.value = it }
+                    SharingCommand.STOP -> Unit
+                    SharingCommand.STOP_AND_RESET_REPLAY_CACHE -> state.value = initialValue
+                }
+            }
+        }
+    }
+
+    override val value: T get() = state.value
+    override val replayCache: List<T> get() = state.replayCache
+    override suspend fun collect(collector: FlowCollector<T>): Nothing {
+        state.collect(collector)
+    }
+}
+
+internal fun <T> Flow<T>.restartableStateIn(
+    scopeProvider: ScopeProviderImpl,
+    started: SharingStarted,
+    initialValue: T,
+): RestartableStateFlow<T> = RestartableStateFlow(this, scopeProvider, initialValue, started)
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProvider.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProvider.kt
index 28d7b319442..d5530a39286 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProvider.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProvider.kt
@@ -16,14 +16,16 @@
 
 package io.getstream.video.android.core.call.scope
 
+import io.getstream.video.android.core.call.Restartable
 import kotlinx.coroutines.CompletableJob
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+
+/** Owns a Call's coroutine generations and creates the scopes used by its RTC session. */
+internal interface ScopeProvider : Restartable {
+    /** The coroutine scope for the current Call generation. */
+    val callScope: CoroutineScope
 
-/**
- * Provider for coroutine scopes used in RTC sessions.
- * This allows for centralized management and reuse of scopes across SDK instances.
- */
-internal interface ScopeProvider {
     /**
      * Provides the main coroutine scope for RTC session operations.
      * This scope is based on the client's main scope with a supervisor job.
@@ -39,5 +41,5 @@ internal interface ScopeProvider {
     /**
      * Cleans up resources when the provider is no longer needed.
      */
-    fun cleanup()
+    fun cleanup(): Job?
 }
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderImpl.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderImpl.kt
index 50a80c90ee8..0d68110a876 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderImpl.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderImpl.kt
@@ -17,36 +17,45 @@
 package io.getstream.video.android.core.call.scope
 
 import io.getstream.log.taggedLogger
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
 import kotlinx.coroutines.CompletableJob
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.asCoroutineDispatcher
-import java.util.concurrent.ExecutorService
-import java.util.concurrent.Executors
+import java.util.concurrent.CopyOnWriteArrayList
 import kotlin.jvm.Throws
 
 /**
- * Implementation of [ScopeProvider] that manages coroutine scopes for RTC sessions.
+ * Coroutine-scope owner for a reusable Call and its RTC session.
  *
  * This implementation ensures that:
+ * - Each generation has its own call [Job]
  * - Each Call gets its own dedicated single thread executor
  * - RtcSessions use the thread from their parent Call
- * - Executors are created on demand and cleaned up with the Call
- * - Scopes are properly managed and cleaned up
+ * - Cleanup cancels the current generation; a later join restarts this provider
  */
 internal class ScopeProviderImpl(
     private val coroutineScope: CoroutineScope,
+    private val callCid: String,
 ) : ScopeProvider {
 
-    private val logger by taggedLogger("ScopeProviderImpl")
-    private var executor: ExecutorService? = null
-    private var isCleanedUp = false
+    private val logger by taggedLogger("ScopeProvider:$callCid")
+    private val generationManager = CallScopeGenerationManager(
+        parentCoroutineContext = coroutineScope.coroutineContext,
+    )
+    private val restartListeners = CopyOnWriteArrayList<() -> Unit>()
+
+    internal val debugInstanceId: Int get() = generationManager.currentDebugInstanceId
+
+    override val callScope: CoroutineScope
+        get() = generationManager.currentCallScope
 
     @Throws(IllegalStateException::class)
     override fun getCoroutineScope(supervisorJob: CompletableJob): CoroutineScope {
-        // Future: We should not throw exception. Instead return Result.Fail
-        check(!isCleanedUp) { "ScopeProvider has been cleaned up" }
-        logger.d { "Creating coroutine scope for RTC session main" }
+        val debugInstanceId = generationManager.requireActive()
+        logger.d { "${debugInstanceId.logPrefix()} Creating coroutine scope for RTC session main" }
         return CoroutineScope(
             coroutineScope.coroutineContext +
                 supervisorJob +
@@ -56,45 +65,42 @@ internal class ScopeProviderImpl(
 
     @Throws(IllegalStateException::class)
     override fun getRtcSessionScope(supervisorJob: CompletableJob, callId: String): CoroutineScope {
-        // Future: We should not throw exception. Instead return Result.Fail
-        check(!isCleanedUp) { "ScopeProvider has been cleaned up" }
-        logger.d { "Creating RTC session scope for callId: $callId" }
+        val executor = generationManager.getOrCreateExecutor(callId)
+        logger.d {
+            "${generationManager.currentDebugInstanceId.logPrefix()} Creating RTC session scope with dedicated executor for callId: $callId"
+        }
+        return CoroutineScope(
+            coroutineScope.coroutineContext +
+                supervisorJob +
+                CoroutineName("rtc-session-coroutine") +
+                executor.asCoroutineDispatcher(),
+        )
+    }
 
-        // Get or create executor for this call
-        if (executor == null) {
-            logger.d { "Creating new single thread executor for callId: $callId" }
-            executor = Executors.newSingleThreadExecutor { runnable ->
-                Thread(runnable, "rtc-call-$callId")
+    override fun cleanup(): Job? {
+        val cleanupResult = generationManager.cleanup()
+        if (cleanupResult.cleanedUpNow) {
+            cleanupResult.callJob.cancel()
+            logger.d {
+                "${cleanupResult.debugInstanceId.logPrefix()} Cleaning up ScopeProvider and shutting down executor"
             }
+            cleanupResult.executor?.shutdown()
         }
+        return cleanupResult.callJob
+    }
 
-        val currentExecutor = executor
-        return if (currentExecutor != null) {
-            logger.d { "Creating RTC session scope with dedicated executor for callId: $callId" }
-            CoroutineScope(
-                coroutineScope.coroutineContext +
-                    supervisorJob +
-                    CoroutineName("rtc-session-coroutine") +
-                    currentExecutor.asCoroutineDispatcher(),
-            )
-        } else {
-            // Fallback to regular scope without executor if executor is null
-            logger.w { "Executor is null, falling back to regular scope for callId: $callId" }
-            CoroutineScope(
-                coroutineScope.coroutineContext +
-                    supervisorJob +
-                    CoroutineName("rtc-session-coroutine"),
-            )
+    override fun restart(): Result<Unit> = generationManager.restart().also { result ->
+        if (result is Success) {
+            logger.d {
+                "${generationManager.currentDebugInstanceId.logPrefix()} Restarted ScopeProvider"
+            }
+            restartListeners.forEach { it() }
         }
     }
 
-    override fun cleanup() {
-        if (isCleanedUp) return
-        isCleanedUp = true
-        logger.d { "Cleaning up ScopeProvider and shutting down executor" }
-
-        // Shutdown the executor
-        executor?.shutdown()
-        executor = null
+    internal fun onRestart(listener: () -> Unit) {
+        restartListeners += listener
     }
+
+    private fun Int.logPrefix(): String = "[debug-instance-$this]"
 }
diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsState.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsState.kt
index 15fa0b9e672..8780174426d 100644
--- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsState.kt
+++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsState.kt
@@ -17,10 +17,14 @@
 package io.getstream.video.android.core.sorting
 
 import io.getstream.log.taggedLogger
+import io.getstream.result.Result
+import io.getstream.result.Result.Success
 import io.getstream.video.android.core.Call
 import io.getstream.video.android.core.ParticipantState
+import io.getstream.video.android.core.call.Restartable
 import io.getstream.video.android.core.pinning.PinEntry
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.combine
@@ -35,19 +39,19 @@ import kotlinx.coroutines.sync.withLock
  * any collector subscribed (the builder block launched coroutines on the outer scope and
  * returned immediately, so subsequent trySend calls hit a closed channel). The new
  * implementation backs the public list with a [MutableStateFlow] and writes to it from
- * collectors launched on the provided [scope].
+ * collectors launched on the current scope returned by [scopeProvider].
  *
  * Sorting is driven by a [SortPreset] (default [SortPreset.Default]) resolved against the
  * latest pin snapshot before each pass. Consumers can override with a custom
  * [Comparator] via [updateComparator] for ad-hoc orderings.
  */
 internal class SortedParticipantsState(
-    scope: CoroutineScope,
+    private val scopeProvider: () -> CoroutineScope,
     private val call: Call,
     private val participants: StateFlow<Map<String, ParticipantState>>,
     private val pinnedParticipantsDetailed: StateFlow<Map<String, PinEntry>>,
     initialPreset: SortPreset = SortPreset.Default,
-) {
+) : Restartable {
     private val logger by taggedLogger("ParticipantSorting")
 
     private val presetFlow = MutableStateFlow<SortPreset>(initialPreset)
@@ -62,9 +66,20 @@ internal class SortedParticipantsState(
 
     private val _sortedParticipants = MutableStateFlow<List<ParticipantState>>(emptyList())
     val sortedParticipants: StateFlow<List<ParticipantState>> = _sortedParticipants
+    private var inputsJob: Job? = null
+    private var eventsJob: Job? = null
 
     init {
-        scope.launch {
+        restart()
+    }
+
+    override fun restart(): Result<Unit> {
+        inputsJob?.cancel()
+        eventsJob?.cancel()
+        lastSortOrder = emptyList()
+        _sortedParticipants.value = emptyList()
+        val scope = scopeProvider()
+        inputsJob = scope.launch {
             combine(
                 participants,
                 pinnedParticipantsDetailed,
@@ -74,11 +89,12 @@ internal class SortedParticipantsState(
                 SortInputs(p, pins, preset, custom)
             }.collect { resort(it) }
         }
-        scope.launch {
+        eventsJob = scope.launch {
             call.events.collect {
                 resort(currentSortInputs())
             }
         }
+        return Success(Unit)
     }
 
     fun setPreset(preset: SortPreset) {
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallFieldDeclarationOrderTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallFieldDeclarationOrderTest.kt
index a451e605966..15732dd3ff6 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallFieldDeclarationOrderTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallFieldDeclarationOrderTest.kt
@@ -23,7 +23,7 @@ import java.io.File
 /**
  * Structural regression guard for a leaking-this race in [Call]'s primary constructor.
  *
- * `Call` declares `val state = CallState(client, this, user, scope)` which passes
+ * `Call` declares `val state = CallState(client, this, user, scopeProvider)` which passes
  * `this` to [CallState]. CallState constructs [io.getstream.video.android.core.sorting.SortedParticipantsState],
  * whose `init` launches `scope.launch { call.events.collect { ... } }`. The
  * `events` flow is owned by `eventManager` (CallEventManager) and `Call.events`
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateRingJoinTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateRingJoinTest.kt
index cb78783101b..37b6ba0d790 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateRingJoinTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateRingJoinTest.kt
@@ -23,6 +23,7 @@ import io.getstream.result.Result
 import io.getstream.video.android.core.base.TestBase
 import io.getstream.video.android.core.base.toResponse
 import io.getstream.video.android.core.call.RtcSession
+import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.getstream.video.android.core.ringing.RingJoinSource
 import io.getstream.video.android.core.utils.toResponse
 import io.getstream.video.android.model.User
@@ -111,7 +112,7 @@ internal class CallStateRingJoinTest : TestBase() {
     /** An outgoing ring we started, registered as ringing, with nobody having answered yet. */
     private fun outgoingRing(): CallState {
         ringingCall.value = call
-        val callState = CallState(client, call, user, scope)
+        val callState = CallState(client, call, user, ScopeProviderImpl(scope, "default:test-call"))
         callState.updateFromResponse(call.toResponse(user.toResponse()))
         assertThat(callState.ringingState.value)
             .isEqualTo(RingingState.Outgoing(acceptedByCallee = false))
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateTelecomHoldTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateTelecomHoldTest.kt
index 9c9212e3f5f..1d16957b561 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateTelecomHoldTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallStateTelecomHoldTest.kt
@@ -19,6 +19,7 @@ package io.getstream.video.android.core
 import io.getstream.android.video.api.generated.models.VideoEvent
 import io.getstream.video.android.core.base.TestBase
 import io.getstream.video.android.core.base.toResponse
+import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.getstream.video.android.core.notifications.internal.telecom.jetpack.JetpackTelecomRepository
 import io.getstream.video.android.core.notifications.internal.telecom.jetpack.TelecomCall
 import io.getstream.video.android.core.utils.toResponse
@@ -72,7 +73,7 @@ internal class CallStateTelecomHoldTest : TestBase() {
 
     @Test
     fun `putting an active call on hold leaves the call with CALL_ON_HOLD`() {
-        val callState = CallState(client, call, user, scope)
+        val callState = CallState(client, call, user, ScopeProviderImpl(scope, "default:test-call"))
         callState.updateFromResponse(call.toResponse(user.toResponse()))
         activeCall.value = call
         callState.updateRingingState()
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallTestSeams.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallTestSeams.kt
index cfaaae0e23a..b89c14122a2 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallTestSeams.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/CallTestSeams.kt
@@ -75,6 +75,17 @@ internal fun Call.injectLocation(location: String?) {
  */
 internal fun Call.nonFastReconnectAttempts(): Int = sessionManager().nonFastReconnectAttempts
 
+/** Seeds per-session reconnect bookkeeping for reusable Call lifecycle tests. */
+internal fun Call.injectReconnectState(attempts: Int, deadlineMillis: Int) {
+    sessionManager().nonFastReconnectAttempts = attempts
+    sessionManager().reconnectDeadlineMillis = deadlineMillis
+}
+
+/** Seeds a transient ringing state that must not survive leave and rejoin. */
+internal fun CallState.injectRingingState(state: RingingState) {
+    mutableStateField<RingingState>("_ringingState").value = state
+}
+
 /**
  * Reaches the [CallSessionManager] that owns the session and reconnect bookkeeping.
  *
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/MicrophoneManagerTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/MicrophoneManagerTest.kt
index e1affb66668..74017e180da 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/MicrophoneManagerTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/MicrophoneManagerTest.kt
@@ -30,6 +30,7 @@ import io.getstream.android.video.api.generated.models.AudioSettingsResponse
 import io.getstream.android.video.api.generated.models.CallSettingsResponse
 import io.getstream.android.video.api.generated.models.OwnCapability
 import io.getstream.video.android.core.audio.AudioHandler
+import io.getstream.video.android.core.audio.AudioSwitchDecorator
 import io.getstream.video.android.core.audio.StreamAudioDevice
 import io.getstream.video.android.core.audio.UsbAudioInputDevice
 import io.getstream.video.android.core.call.RtcSession
@@ -41,8 +42,10 @@ import io.getstream.webrtc.AudioTrack
 import io.mockk.clearMocks
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.mockkConstructor
 import io.mockk.slot
 import io.mockk.spyk
+import io.mockk.unmockkConstructor
 import io.mockk.verify
 import io.mockk.verifyOrder
 import kotlinx.coroutines.channels.Channel
@@ -139,6 +142,29 @@ class MicrophoneManagerTest {
         }
     }
 
+    @Test
+    fun `setup starts fresh audio routing after cleanup`() {
+        mockkConstructor(AudioSwitchDecorator::class)
+        try {
+            every { anyConstructed<AudioSwitchDecorator>().start() } returns Unit
+            every { anyConstructed<AudioSwitchDecorator>().stop() } returns Unit
+            val mediaManager = mockk<MediaManagerImpl>(relaxed = true)
+            val context = mockk<Context>(relaxed = true)
+            every { mediaManager.context } returns context
+            val audioManager = mockk<AudioManager>(relaxed = true)
+            every { context.getSystemService(Context.AUDIO_SERVICE) } returns audioManager
+            every { context.getSystemService(AudioManager::class.java) } returns audioManager
+            val microphone = MicrophoneManager(mediaManager, audioUsage, audioUsageProvider)
+            microphone.setup()
+            microphone.cleanup()
+            microphone.setup()
+            verify(exactly = 2) { anyConstructed<AudioSwitchDecorator>().start() }
+            verify(exactly = 1) { anyConstructed<AudioSwitchDecorator>().stop() }
+        } finally {
+            unmockkConstructor(AudioSwitchDecorator::class)
+        }
+    }
+
     @Test
     fun `resume calls setup again after cleanup`() {
         val mediaManager = mockk<MediaManagerImpl>(relaxed = true)
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/ReusableCallTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/ReusableCallTest.kt
new file mode 100644
index 00000000000..1d553dd373b
--- /dev/null
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/ReusableCallTest.kt
@@ -0,0 +1,382 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core
+
+import android.content.Context
+import androidx.lifecycle.Lifecycle
+import androidx.test.core.app.ApplicationProvider
+import com.google.common.truth.Truth.assertThat
+import com.jakewharton.threetenabp.AndroidThreeTen
+import io.getstream.android.core.api.StreamClient
+import io.getstream.android.core.api.model.connection.StreamConnectionState
+import io.getstream.android.core.api.subscribe.StreamSubscription
+import io.getstream.android.video.api.generated.models.Credentials
+import io.getstream.android.video.api.generated.models.JoinCallResponse
+import io.getstream.android.video.api.generated.models.SFUResponse
+import io.getstream.android.video.api.generated.models.StatsOptions
+import io.getstream.android.video.api.generated.models.UserResponse
+import io.getstream.android.video.api.generated.models.VideoEvent
+import io.getstream.result.Result.Success
+import io.getstream.video.android.core.analytics.AnalyticsModule
+import io.getstream.video.android.core.analytics.reporting.ClientEventReporter
+import io.getstream.video.android.core.base.DispatcherRule
+import io.getstream.video.android.core.base.toResponse
+import io.getstream.video.android.core.call.RtcSession
+import io.getstream.video.android.core.call.SfuConnectionResult
+import io.getstream.video.android.core.call.connection.Publisher
+import io.getstream.video.android.core.call.connection.Subscriber
+import io.getstream.video.android.core.events.ParticipantMigrationCompleteEvent
+import io.getstream.video.android.core.internal.module.CoordinatorConnectionModule
+import io.getstream.video.android.core.internal.network.NetworkStateProvider
+import io.getstream.video.android.core.model.AudioTrack
+import io.getstream.video.android.core.model.VideoTrack
+import io.getstream.video.android.core.notifications.internal.StreamNotificationManager
+import io.getstream.video.android.core.notifications.internal.service.CallServiceConfigRegistry
+import io.getstream.video.android.core.permission.android.StreamPermissionCheck
+import io.getstream.video.android.core.socket.common.token.TokenRepository
+import io.getstream.video.android.core.sounds.RingingCallVibrationConfig
+import io.getstream.video.android.core.sounds.Sounds
+import io.getstream.video.android.model.User
+import io.mockk.coEvery
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.spyk
+import io.mockk.unmockkAll
+import io.mockk.verify
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.After
+import org.junit.Before
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import org.threeten.bp.OffsetDateTime
+import stream.video.sfu.models.Participant
+import stream.video.sfu.models.TrackType
+
+@RunWith(RobolectricTestRunner::class)
+class ReusableCallTest {
+
+    @get:Rule
+    val dispatcherRule = DispatcherRule()
+
+    @Before
+    fun setUp() {
+        AndroidThreeTen.init(ApplicationProvider.getApplicationContext())
+    }
+
+    @After
+    fun tearDown() {
+        Call.testInstanceProvider.mediaManagerCreator = null
+        unmockkAll()
+    }
+
+    @Test
+    fun `same Call joins leaves and joins again with fresh session state`() = runTest {
+        val mediaManager = reusableMediaManager()
+        Call.testInstanceProvider.mediaManagerCreator = { mediaManager }
+        val client = reusableClient(backgroundScope)
+        val call = client.call("default", "reusable-call")
+        val events = call.events
+
+        assertThat(client.call("default", "reusable-call")).isSameInstanceAs(call)
+
+        val firstSession = reusableSession()
+        val secondSession = reusableSession()
+        val thirdSession = reusableSession()
+        val sessions = ArrayDeque(listOf(firstSession, secondSession, thirdSession))
+        var secondGenerationSnapshot: ResetSnapshot? = null
+        val firstSessionId = call.sessionId
+        call.unitTestRtcSessionFactory = {
+            if (sessions.size == 2) {
+                secondGenerationSnapshot = ResetSnapshot(
+                    sessionWasCleared = call.session.value == null,
+                    participantsWereCleared = call.state.participants.value.isEmpty(),
+                    statsReportWasCleared = call.statsReport.value == null,
+                    statsHistory = call.statLatencyHistory.value,
+                    publisherLatency = call.state.stats.publisher.latency.value,
+                    reconnectAttempts = call.nonFastReconnectAttempts(),
+                    reconnectDeadlineMillis = call.reconnectDeadlineMillis,
+                    ringingState = call.state.ringingState.value,
+                    sessionId = call.sessionId,
+                )
+            }
+            sessions.removeFirst()
+        }
+        var joinNumber = 0
+        coEvery { client.getCachedLocation() } returns Success("test-location")
+        coEvery {
+            client.joinCall(
+                any(), any(), any(), any(), any(), any(), any(), any(),
+                any(), any(), any(), any(), any(), any(), any(),
+            )
+        } coAnswers {
+            joinNumber++
+            Success(joinResponse(call, "session-$joinNumber"))
+        }
+        coEvery { firstSession.connectInternal(any(), any()) } returns SfuConnectionResult.Success
+        coEvery { thirdSession.connectInternal(any(), any()) } returns SfuConnectionResult.Success
+
+        val secondVideoTrack = VideoTrack(
+            streamId = "second-video",
+            video = mockk(relaxed = true),
+        )
+        val secondAudioTrack = AudioTrack(
+            streamId = "second-audio",
+            audio = mockk(relaxed = true),
+        )
+        coEvery { secondSession.connectInternal(any(), any()) } coAnswers {
+            val participant = call.state.getOrCreateParticipant(
+                Participant(
+                    user_id = "second-user",
+                    session_id = "second-participant",
+                    published_tracks = listOf(
+                        TrackType.TRACK_TYPE_VIDEO,
+                        TrackType.TRACK_TYPE_AUDIO,
+                    ),
+                    track_lookup_prefix = "second",
+                    name = "Second participant",
+                    image = "",
+                ),
+            )
+            participant.setVideoTrack(secondVideoTrack)
+            participant._audioTrack.value = secondAudioTrack
+            call.state.replaceParticipants(listOf(participant))
+            SfuConnectionResult.Success
+        }
+
+        val participantEmissions = mutableListOf<List<String>>()
+        val participantCollector = backgroundScope.launch {
+            call.state.participants.collect { participants ->
+                participantEmissions += participants.map(ParticipantState::sessionId)
+            }
+        }
+        val receivedEvents = mutableListOf<VideoEvent>()
+        val eventCollector = backgroundScope.launch { events.collect(receivedEvents::add) }
+        runCurrent()
+
+        assertThat(call.join()).isEqualTo(Success(firstSession))
+
+        val oldVideoTrack = VideoTrack(
+            streamId = "old-video",
+            video = mockk(relaxed = true),
+        )
+        val oldAudioTrack = AudioTrack(
+            streamId = "old-audio",
+            audio = mockk(relaxed = true),
+        )
+        val oldParticipant = call.state.getOrCreateParticipant(
+            Participant(
+                user_id = "old-user",
+                session_id = "old-participant",
+                published_tracks = listOf(
+                    TrackType.TRACK_TYPE_VIDEO,
+                    TrackType.TRACK_TYPE_AUDIO,
+                ),
+                track_lookup_prefix = "old",
+                name = "Old participant",
+                image = "",
+            ),
+        )
+        oldParticipant.setVideoTrack(oldVideoTrack)
+        oldParticipant._audioTrack.value = oldAudioTrack
+        call.state.replaceParticipants(listOf(oldParticipant))
+        call.statsReport.value = mockk(relaxed = true)
+        call.statLatencyHistory.value = listOf(17, 23)
+        call.state.stats.publisher._latency.value = 91
+        call.injectReconnectState(attempts = 4, deadlineMillis = 42_000)
+        call.state.injectRingingState(RingingState.Active)
+
+        call.leave()
+        assertThat(call.join()).isEqualTo(Success(secondSession))
+        runCurrent()
+
+        val snapshot = checkNotNull(secondGenerationSnapshot)
+        assertThat(snapshot.sessionWasCleared).isTrue()
+        assertThat(snapshot.participantsWereCleared).isTrue()
+        assertThat(snapshot.statsReportWasCleared).isTrue()
+        assertThat(snapshot.statsHistory).containsExactly(0, 0, 0).inOrder()
+        assertThat(snapshot.publisherLatency).isEqualTo(0)
+        assertThat(snapshot.reconnectAttempts).isEqualTo(0)
+        assertThat(snapshot.reconnectDeadlineMillis).isEqualTo(10_000)
+        assertThat(snapshot.ringingState).isEqualTo(RingingState.Idle)
+        assertThat(snapshot.sessionId).isNotEqualTo(firstSessionId)
+
+        assertThat(call.session.value).isSameInstanceAs(secondSession)
+        assertThat(call.state.getParticipantBySessionId("old-participant")).isNull()
+        val secondParticipant = call.state.getParticipantBySessionId("second-participant")
+        assertThat(secondParticipant).isNotNull()
+        assertThat(secondParticipant!!.videoTrack.value).isSameInstanceAs(secondVideoTrack)
+        assertThat(secondParticipant.audioTrack.value).isSameInstanceAs(secondAudioTrack)
+        assertThat(secondParticipant.videoEnabled.value).isTrue()
+        assertThat(secondParticipant.audioEnabled.value).isTrue()
+        assertThat(participantCollector.isActive).isTrue()
+        assertThat(participantEmissions).contains(listOf("second-participant"))
+
+        assertThat(call.events).isSameInstanceAs(events)
+        val event = mockk<VideoEvent>(relaxed = true)
+        call.fireEvent(event)
+        runCurrent()
+        assertThat(receivedEvents).contains(event)
+        assertThat(eventCollector.isActive).isTrue()
+        assertThat(client.call("default", "reusable-call")).isSameInstanceAs(call)
+
+        verify(atLeast = 1) { mediaManager.cleanup() }
+        verify(atLeast = 1) { mediaManager.prepareForJoin() }
+
+        call.leave()
+        runCurrent()
+
+        val replacementCall = client.call("default", "reusable-call")
+        assertThat(replacementCall).isNotSameInstanceAs(call)
+
+        assertThat(call.join()).isEqualTo(Success(thirdSession))
+        assertThat(client.call("default", "reusable-call")).isSameInstanceAs(replacementCall)
+
+        val routedEvent = mockk<VideoEvent>(relaxed = true)
+        client.fireEvent(routedEvent, call.cid)
+        runCurrent()
+        assertThat(receivedEvents).contains(routedEvent)
+
+        val sfuEvent = ParticipantMigrationCompleteEvent
+        client.fireEvent(sfuEvent, call.cid)
+        runCurrent()
+        assertThat(receivedEvents).contains(sfuEvent)
+
+        call.leave()
+        runCurrent()
+        assertThat(client.call("default", "reusable-call")).isSameInstanceAs(replacementCall)
+    }
+
+    private data class ResetSnapshot(
+        val sessionWasCleared: Boolean,
+        val participantsWereCleared: Boolean,
+        val statsReportWasCleared: Boolean,
+        val statsHistory: List<Int>,
+        val publisherLatency: Int,
+        val reconnectAttempts: Int,
+        val reconnectDeadlineMillis: Int,
+        val ringingState: RingingState,
+        val sessionId: String,
+    )
+
+    private fun reusableClient(scope: CoroutineScope): StreamVideoClient {
+        val streamClient = mockk<StreamClient>(relaxed = true)
+        every { streamClient.subscribe(any()) } returns Result.success(
+            mockk<StreamSubscription>(relaxed = true),
+        )
+        every { streamClient.connectionState } returns
+            MutableStateFlow(StreamConnectionState.Idle)
+
+        val network = mockk<NetworkStateProvider>(relaxed = true) {
+            every { isConnected() } returns true
+        }
+        val coordinator = mockk<CoordinatorConnectionModule>(relaxed = true) {
+            every { networkStateProvider } returns network
+        }
+        val analytics = mockk<AnalyticsModule>(relaxed = true) {
+            every { clientEventReporter } returns mockk<ClientEventReporter>(relaxed = true)
+        }
+        val permissions = mockk<StreamPermissionCheck> {
+            every { checkAndroidPermissionsGroup(any(), any()) } returns (true to emptySet())
+        }
+        val callConfigs = CallServiceConfigRegistry().apply {
+            register("default") { setRunCallServiceInForeground(false) }
+        }
+
+        return spyk(
+            StreamVideoClient(
+                context = ApplicationProvider.getApplicationContext<Context>(),
+                scope = scope,
+                initialUser = User(id = "test-user", role = "user"),
+                apiKey = "test-api-key",
+                token = "test-token",
+                lifecycle = mockk<Lifecycle>(relaxed = true),
+                coordinatorConnectionModule = coordinator,
+                streamClient = streamClient,
+                tokenRepository = mockk<TokenRepository>(relaxed = true),
+                streamNotificationManager = mockk<StreamNotificationManager>(relaxed = true),
+                enableCallNotificationUpdates = false,
+                callServiceConfigRegistry = callConfigs,
+                sounds = mockk<Sounds>(relaxed = true),
+                vibrationConfig = mockk<RingingCallVibrationConfig>(relaxed = true),
+                permissionCheck = permissions,
+                analytics = analytics,
+            ),
+        )
+    }
+
+    private fun reusableMediaManager(): MediaManagerImpl {
+        val camera = mockk<CameraManager>(relaxed = true) {
+            every { status } returns MutableStateFlow(DeviceStatus.NotSelected)
+        }
+        val microphone = mockk<MicrophoneManager>(relaxed = true) {
+            every { status } returns MutableStateFlow(DeviceStatus.NotSelected)
+            every { isEnabled } returns MutableStateFlow(false)
+            every { devices } returns MutableStateFlow(emptyList())
+        }
+        val speaker = mockk<SpeakerManager>(relaxed = true) {
+            every { status } returns MutableStateFlow(DeviceStatus.NotSelected)
+        }
+        return mockk(relaxed = true) {
+            every { this@mockk.camera } returns camera
+            every { this@mockk.microphone } returns microphone
+            every { this@mockk.speaker } returns speaker
+            every { screenShare } returns mockk(relaxed = true)
+        }
+    }
+
+    private fun reusableSession(): RtcSession = mockk(relaxed = true) {
+        every { publisher } returns MutableStateFlow<Publisher?>(null)
+        every { subscriber } returns MutableStateFlow<Subscriber?>(null)
+        every { socket.events() } returns MutableSharedFlow()
+    }
+
+    private fun joinResponse(call: Call, sessionId: String): JoinCallResponse {
+        val now = OffsetDateTime.parse("2026-01-01T00:00:00Z")
+        val createdBy = UserResponse(
+            createdAt = now,
+            id = "test-user",
+            language = "en",
+            role = "user",
+            updatedAt = now,
+        )
+        return JoinCallResponse(
+            created = false,
+            duration = "0",
+            call = call.toResponse(createdBy).copy(currentSessionId = sessionId),
+            credentials = Credentials(
+                token = "sfu-token",
+                server = SFUResponse(
+                    edgeName = "test-edge",
+                    url = "https://test-sfu",
+                    wsEndpoint = "wss://test-sfu",
+                ),
+            ),
+            statsOptions = StatsOptions(
+                enableRtcStats = false,
+                reportingIntervalMs = 60_000,
+            ),
+        )
+    }
+}
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/RingingStateJoinAndRingTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/RingingStateJoinAndRingTest.kt
index 04b93ac9661..dd4399ee6ec 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/RingingStateJoinAndRingTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/RingingStateJoinAndRingTest.kt
@@ -19,6 +19,7 @@ package io.getstream.video.android.core
 import io.getstream.android.video.api.generated.models.VideoEvent
 import io.getstream.video.android.core.base.TestBase
 import io.getstream.video.android.core.base.toResponse
+import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.getstream.video.android.core.events.JoinCallResponseEvent
 import io.getstream.video.android.core.events.ParticipantCount
 import io.getstream.video.android.core.utils.toResponse
@@ -91,7 +92,7 @@ internal class RingingStateJoinAndRingTest : TestBase() {
     }
 
     private fun callStateInJoinAndRingWindow(): CallState {
-        val callState = CallState(client, call, user, scope)
+        val callState = CallState(client, call, user, ScopeProviderImpl(scope, "default:test-call"))
         // The call was created by us; nobody accepted or rejected yet.
         callState.updateFromResponse(call.toResponse(user.toResponse()))
         // joinAndRing() toggled the flag and the join completed (active call registered),
@@ -114,7 +115,7 @@ internal class RingingStateJoinAndRingTest : TestBase() {
 
     @Test
     fun `a recompute before the SFU join sets Outgoing still yields Idle`() {
-        val callState = CallState(client, call, user, scope)
+        val callState = CallState(client, call, user, ScopeProviderImpl(scope, "default:test-call"))
         callState.updateFromResponse(call.toResponse(user.toResponse()))
         callState.toggleJoinAndRingProgress(true)
         activeCall.value = call
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt
index 01a3898d48e..7e9917f3463 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/StreamVideoClientTest.kt
@@ -60,9 +60,14 @@ import kotlinx.coroutines.test.runTest
 import org.junit.After
 import org.junit.Before
 import org.threeten.bp.OffsetDateTime
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
+import kotlin.test.assertNotSame
+import kotlin.test.assertSame
 import kotlin.test.assertTrue
 
 class StreamVideoClientTest {
@@ -132,6 +137,48 @@ class StreamVideoClientTest {
         return ClientHarness(client, streamClientMock, tokenRepo)
     }
 
+    @Test
+    fun `concurrent call requests return the same cached instance`() {
+        val requestCount = 16
+        val startRequests = CountDownLatch(1)
+        val executor = Executors.newFixedThreadPool(requestCount)
+
+        try {
+            val requests = List(requestCount) {
+                executor.submit<Call> {
+                    startRequests.await()
+                    client.call("default", "concurrent-call")
+                }
+            }
+
+            startRequests.countDown()
+            val calls = requests.map { it.get(10, TimeUnit.SECONDS) }
+
+            assertTrue(calls.all { it === calls.first() })
+        } finally {
+            executor.shutdownNow()
+        }
+    }
+
+    @Test
+    fun `reused call fills only an empty cache and old cleanup preserves its replacement`() {
+        val originalCall = client.call("default", "reusable-call")
+
+        client.onCallCleanUp(originalCall)
+        client.onCallReused(originalCall)
+        assertSame(originalCall, client.call("default", "reusable-call"))
+
+        client.onCallCleanUp(originalCall)
+        val replacementCall = client.call("default", "reusable-call")
+        assertNotSame(originalCall, replacementCall)
+
+        client.onCallReused(originalCall)
+        assertSame(replacementCall, client.call("default", "reusable-call"))
+
+        client.onCallCleanUp(originalCall)
+        assertSame(replacementCall, client.call("default", "reusable-call"))
+    }
+
     @Test
     fun `resolveSelectedCid returns explicit cid when provided`() {
         val event = mockk<VideoEvent>()
@@ -236,10 +283,12 @@ class StreamVideoClientTest {
             every { session } returns sessionFlow
         }
 
-        client::class.java.getDeclaredField("calls").apply {
+        @Suppress("UNCHECKED_CAST")
+        val calls = client::class.java.getDeclaredField("calls").run {
             isAccessible = true
-            set(client, mutableMapOf("video:123" to call))
+            get(client) as MutableMap<String, Call>
         }
+        calls["video:123"] = call
 
         client.propagateEventToCall("video:123", event)
 
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/CallAnalyticsTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/CallAnalyticsTest.kt
index a16a14d1478..0767b7dc975 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/CallAnalyticsTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/CallAnalyticsTest.kt
@@ -202,4 +202,19 @@ class CallAnalyticsTest {
 
         assertEquals("", analytics.videoAnalytics.stageId.value)
     }
+
+    @Test
+    fun `restart preserves the join attempt started before lifecycle preparation`() = runTest {
+        val analytics = callAnalytics(backgroundScope)
+        analytics.joinAnalyticsStateHolder.updateJoinStageAttemptId("current-attempt")
+        analytics.joinAnalyticsStateHolder.updateStage(Stage.COMPLETED)
+
+        analytics.restart()
+
+        assertEquals(
+            "current-attempt",
+            analytics.joinAnalyticsStateHolder.state.value.joinStageAttemptId,
+        )
+        assertEquals(Stage.NOT_STARTED, analytics.joinAnalyticsStateHolder.state.value.joinStage)
+    }
 }
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalyticsTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalyticsTest.kt
index aaabe5c5bc7..74ae6ecd2f1 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalyticsTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/analytics/call/observer/JoinAnalyticsTest.kt
@@ -172,6 +172,28 @@ class JoinAnalyticsTest {
         assertEquals(0, joinSuccessCount)
     }
 
+    @Test
+    fun `onJoinPreparationError reports the failure message`() {
+        joinAnalytics.onJoinFunctionStart()
+
+        joinAnalytics.onJoinPreparationError(
+            failureCode = "CALL_RESTART_FAILED",
+            message = "scope restart failed",
+        )
+
+        verify(exactly = 1) {
+            reporter.reportCoordinatorJoinCompleted(
+                "stage-1",
+                false,
+                0,
+                "scope restart failed",
+                "CALL_RESTART_FAILED",
+                null,
+            )
+        }
+        assertEquals(Stage.COMPLETED, stateHolder.state.value.joinStage)
+    }
+
     @Test
     fun `onJoinRequestRetryExhausted behaves like a permanent error`() {
         joinAnalytics.onJoinRequestStart(JoinReason.FirstAttempt)
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallCleanupStateTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallCleanupStateTest.kt
new file mode 100644
index 00000000000..db5fbe154c7
--- /dev/null
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallCleanupStateTest.kt
@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.components
+
+import io.getstream.result.Error
+import io.getstream.result.Result.Failure
+import io.getstream.result.Result.Success
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class CallCleanupStateTest {
+
+    @Test
+    fun `failed restart keeps the call destroyed so a later join can retry`() = runTest {
+        val cleanupState = CallCleanupState()
+        cleanupState.startLeave {
+            Job().apply { complete() }
+        }
+        val failure = Failure(Error.GenericError("scope restart failed"))
+
+        assertEquals(failure, cleanupState.prepareForJoin { failure })
+        assertTrue(cleanupState.isDestroyed)
+
+        assertEquals(Success(Unit), cleanupState.prepareForJoin { Success(Unit) })
+        assertFalse(cleanupState.isDestroyed)
+    }
+}
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt
index c216deab323..5d559cdc9a5 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt
@@ -36,6 +36,7 @@ import io.getstream.video.android.core.base.DispatcherRule
 import io.getstream.video.android.core.call.RtcSession
 import io.getstream.video.android.core.call.SfuConnectFailureCause
 import io.getstream.video.android.core.call.SfuConnectionResult
+import io.getstream.video.android.core.call.scope.ScopeProviderImpl
 import io.mockk.coEvery
 import io.mockk.coVerify
 import io.mockk.every
@@ -44,6 +45,9 @@ import io.mockk.unmockkAll
 import io.mockk.verify
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.cancel
@@ -51,6 +55,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.junit.After
 import org.junit.Before
@@ -113,6 +118,7 @@ class CallJoinCoordinatorTest {
         every { state.connection } returns connectionFlow
         every { state.settings } returns MutableStateFlow<CallSettingsResponse?>(null)
         every { state.e2eeEnabled } returns MutableStateFlow(false)
+        coEvery { lifecycle.prepareForJoin() } returns Success(Unit)
         coEvery { clientImpl.getCachedLocation() } returns Success("test-location")
     }
 
@@ -121,15 +127,20 @@ class CallJoinCoordinatorTest {
         unmockkAll()
     }
 
-    private fun coordinator() = CallJoinCoordinator(
+    private fun coordinator(
+        scope: CoroutineScope = testScope,
+        lifecycle: CallLifecycleManager = this.lifecycle,
+        sessionFactory: RtcSessionFactory =
+            RtcSessionFactory { _, _, _, _, _, _, _ -> mockSession },
+    ) = CallJoinCoordinator(
         clientImpl = clientImpl,
         state = state,
         callAnalytics = callAnalytics,
         type = "default",
         id = "test-call",
-        scope = testScope,
+        scope = scope,
         sessionManager = sessionManager,
-        sessionFactory = RtcSessionFactory { _, _, _, _, _, _, _ -> mockSession },
+        sessionFactory = sessionFactory,
         media = media,
         lifecycle = lifecycle,
         apiClient = apiClient,
@@ -147,6 +158,64 @@ class CallJoinCoordinatorTest {
         } returns result
     }
 
+    @Test
+    fun `join succeeds repeatedly after leave with a restarted scope provider`() = runTest(
+        testDispatcher,
+    ) {
+        every { clientImpl.scope } returns backgroundScope
+        val provider = ScopeProviderImpl(backgroundScope, "default:test-call")
+        val actualLifecycle = CallLifecycleManager(
+            clientImpl = clientImpl,
+            sessionManager = sessionManager,
+            callRegistry = callRegistry,
+            shutDownJobs = { provider.cleanup() },
+            restartCoordinator = CallRestartCoordinator(
+                scopeProvider = provider,
+                restartables = emptyList(),
+            ),
+            state = state,
+            callAnalytics = callAnalytics,
+            statsReporter = mockk(relaxed = true),
+            media = media,
+            iceMonitor = mockk(relaxed = true),
+            sessionMonitor = { sessionMonitor },
+            connectivityMonitor = { mockk(relaxed = true) },
+            type = "default",
+            id = "test-call",
+        )
+        val providerJobs = mutableListOf<Job>()
+        val coordinator = coordinator(
+            scope = backgroundScope,
+            lifecycle = actualLifecycle,
+            sessionFactory = RtcSessionFactory { _, _, _, _, _, _, _ ->
+                // RtcSession needs this provider on every join, including after cleanup.
+                providerJobs += provider.callScope.coroutineContext[Job]!!
+                val job = SupervisorJob()
+                provider.getRtcSessionScope(job, "test-call")
+                job.cancel()
+                mockSession
+            },
+        )
+        stubJoinCall(Success(mockJoinResponse))
+        coEvery { mockSession.connectInternal() } returns SfuConnectionResult.Success
+        repeat(3) {
+            assertThat(coordinator.join()).isInstanceOf(Success::class.java)
+            val gate = CompletableDeferred<Unit>()
+            coEvery { mockSession.sendLeaveEvent(any()) } coAnswers { gate.await() }
+            actualLifecycle.leave()
+            val nextJoin = async { coordinator.join() }
+            runCurrent()
+            assertThat(nextJoin.isCompleted).isFalse()
+            gate.complete(Unit)
+            assertThat(nextJoin.await()).isInstanceOf(Success::class.java)
+        }
+        actualLifecycle.leave()
+        runCurrent()
+        coVerify(exactly = 4) { mockSession.connectInternal() }
+        assertThat(providerJobs.distinct()).hasSize(4)
+        assertThat(providerJobs.all { it.isCancelled }).isTrue()
+    }
+
     @Test
     fun `join succeeds and returns the connected session`() = runTest(testDispatcher) {
         stubJoinCall(Success(mockJoinResponse))
@@ -160,6 +229,29 @@ class CallJoinCoordinatorTest {
         verify { sessionMonitor.monitorSession(mockJoinResponse) }
     }
 
+    @Test
+    fun `join returns lifecycle preparation failure and reports its message`() = runTest(
+        testDispatcher,
+    ) {
+        val error = Error.GenericError("scope restart failed")
+        coEvery { lifecycle.prepareForJoin() } returns Failure(error)
+
+        val result = coordinator().join()
+
+        assertThat(result).isEqualTo(Failure(error))
+        assertThat(connectionFlow.value).isEqualTo(RealtimeConnection.Failed(error))
+        verify(exactly = 1) { callAnalytics.joinAnalytics.onJoinFunctionStart() }
+        verify(exactly = 1) {
+            callAnalytics.joinAnalytics.onJoinPreparationError(
+                failureCode = "CALL_RESTART_FAILED",
+                message = "scope restart failed",
+            )
+        }
+        coVerify(exactly = 0) {
+            apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any(), any())
+        }
+    }
+
     @Test
     fun `first join passes e2ee from call state`() = runTest(testDispatcher) {
         every { state.e2eeEnabled } returns MutableStateFlow(true)
@@ -379,7 +471,7 @@ class CallJoinCoordinatorTest {
         val joinAnalytics = callAnalytics.joinAnalytics
         verify(exactly = 2) { joinAnalytics.onJoinFunctionStart() }
         verify(exactly = 1) { callAnalytics.mediaPermissionObserver.mediaPermissionStatus() }
-        verify(exactly = 1) { lifecycle.resetLeaveGuard() }
+        coVerify(exactly = 2) { lifecycle.prepareForJoin() }
         verify { state.callJoinInterceptor = interceptor }
         verify(exactly = 0) { state.callJoinInterceptor = null }
     }
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManagerTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManagerTest.kt
new file mode 100644
index 00000000000..928016149fe
--- /dev/null
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallLifecycleManagerTest.kt
@@ -0,0 +1,155 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.components
+
+import io.getstream.video.android.core.CallState
+import io.getstream.video.android.core.RealtimeConnection
+import io.getstream.video.android.core.StreamVideoClient
+import io.getstream.video.android.core.call.Restartable
+import io.getstream.video.android.core.call.RtcSession
+import io.getstream.video.android.core.call.scope.ScopeProviderImpl
+import io.mockk.coEvery
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class CallLifecycleManagerTest {
+    private class Fixture(test: TestScope) {
+        val session = mockk<RtcSession>(relaxed = true)
+        val sessions = CallSessionManager().apply { setActiveSession(this@Fixture.session) }
+        val state = mockk<CallState>(relaxed = true)
+        var restarts = 0
+        var shutdowns = 0
+        private val parent = test.backgroundScope
+        val scopeProvider = ScopeProviderImpl(parent, "default:test-call")
+        val scope: CoroutineScope get() = scopeProvider.callScope
+        init {
+            every { state._connection } returns MutableStateFlow(RealtimeConnection.PreJoin)
+        }
+        private val client = mockk<StreamVideoClient>(relaxed = true).also {
+            every { it.scope } returns parent
+        }
+        val lifecycle = CallLifecycleManager(
+            clientImpl = client,
+            sessionManager = sessions,
+            callRegistry = mockk(relaxed = true),
+            shutDownJobs = {
+                shutdowns++
+                scopeProvider.cleanup()
+            },
+            restartCoordinator = CallRestartCoordinator(
+                scopeProvider = Restartable {
+                    restarts++
+                    scopeProvider.restart()
+                },
+                restartables = emptyList(),
+            ),
+            state = state,
+            callAnalytics = mockk(relaxed = true),
+            statsReporter = mockk(relaxed = true),
+            media = mockk(relaxed = true),
+            iceMonitor = mockk(relaxed = true),
+            sessionMonitor = { mockk(relaxed = true) },
+            connectivityMonitor = { mockk(relaxed = true) },
+            type = "default",
+            id = "test-call",
+        )
+    }
+
+    @Test
+    fun `immediate concurrent rejoins wait for cleanup and restart only once`() = runTest {
+        val fixture = Fixture(this)
+        val leaveEvent = CompletableDeferred<Unit>()
+        coEvery { fixture.session.sendLeaveEvent(any()) } coAnswers { leaveEvent.await() }
+        fixture.lifecycle.leave()
+        fixture.lifecycle.leave()
+        val first = async { fixture.lifecycle.prepareForJoin() }
+        val second = async { fixture.lifecycle.prepareForJoin() }
+        runCurrent()
+        assertFalse(first.isCompleted)
+        assertFalse(second.isCompleted)
+        assertEquals(0, fixture.restarts)
+        leaveEvent.complete(Unit)
+        first.await()
+        second.await()
+        assertEquals(1, fixture.shutdowns)
+        assertEquals(1, fixture.restarts)
+        assertFalse(fixture.lifecycle.isDestroyed)
+    }
+
+    @Test
+    fun `rejoin waits for the cancelled call producers to finish their finalizers`() = runTest {
+        val fixture = Fixture(this)
+        val finished = CompletableDeferred<Unit>()
+        fixture.scope.launch {
+            try {
+                awaitCancellation()
+            } finally {
+                withContext(NonCancellable) { finished.await() }
+            }
+        }
+        runCurrent()
+        fixture.lifecycle.leave()
+        val rejoin = async { fixture.lifecycle.prepareForJoin() }
+        runCurrent()
+        assertFalse(rejoin.isCompleted)
+        assertEquals(0, fixture.restarts)
+        finished.complete(Unit)
+        rejoin.await()
+        assertEquals(1, fixture.restarts)
+    }
+
+    @Test
+    fun `repeated leave join cycles clean up each generation`() = runTest {
+        val fixture = Fixture(this)
+        repeat(3) { cycle ->
+            fixture.sessions.setActiveSession(fixture.session)
+            fixture.lifecycle.leave()
+            fixture.lifecycle.prepareForJoin()
+            assertEquals(cycle + 1, fixture.shutdowns)
+            assertEquals(cycle + 1, fixture.restarts)
+            assertTrue(fixture.scope.coroutineContext[Job]!!.isActive)
+        }
+    }
+
+    @Test
+    fun `leave without another join releases resources without restarting producers`() = runTest {
+        val fixture = Fixture(this)
+        fixture.lifecycle.leave()
+        runCurrent()
+        assertEquals(1, fixture.shutdowns)
+        assertEquals(0, fixture.restarts)
+        assertTrue(fixture.scope.coroutineContext[Job]!!.isCancelled)
+        verify { fixture.state.cleanup() }
+    }
+}
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallMediaManagerTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallMediaManagerTest.kt
index 50a2870a594..b14c8863cce 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallMediaManagerTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallMediaManagerTest.kt
@@ -70,7 +70,7 @@ class CallMediaManagerTest {
         type = "default",
         id = "call-id",
         clientImpl = clientImpl,
-        scope = testScope,
+        coroutineScope = testScope,
         state = state,
         sessionManager = sessionManager,
         eglBase = { mockk(relaxed = true) },
@@ -101,9 +101,38 @@ class CallMediaManagerTest {
         verify { mediaManager.screenShare.disable(fromUser = true) }
     }
 
+    @Test
+    fun `rejoin restores the camera and microphone choices from before leave`() {
+        val cameraStatus = MutableStateFlow<DeviceStatus>(DeviceStatus.Enabled)
+        val microphoneStatus = MutableStateFlow<DeviceStatus>(DeviceStatus.Disabled)
+        every { mediaManager.camera.status } returns cameraStatus
+        every { mediaManager.microphone.status } returns microphoneStatus
+        every { mediaManager.camera.disable(any()) } answers { cameraStatus.value = DeviceStatus.Disabled }
+        val manager = manager()
+        manager.mediaManager
+        manager.disableLocalCapture()
+        manager.cleanup()
+        manager.restart()
+        manager.updateMediaManagerFromSettings(mockk(relaxed = true))
+        verify { mediaManager.prepareForJoin() }
+        verify { mediaManager.camera.setEnabled(true) }
+        verify { mediaManager.microphone.setEnabled(false) }
+    }
+
+    @Test
+    fun `leaving an unused call does not create media resources`() {
+        val manager = manager()
+        manager.disableLocalCapture()
+        manager.cleanup()
+        verify(exactly = 0) { mediaManager.cleanup() }
+        verify(exactly = 0) { mediaManager.camera }
+    }
+
     @Test
     fun `cleanup delegates to the media manager`() {
-        manager().cleanup()
+        val manager = manager()
+        manager.mediaManager
+        manager.cleanup()
         verify { mediaManager.cleanup() }
     }
 
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallRestartCoordinatorTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallRestartCoordinatorTest.kt
new file mode 100644
index 00000000000..8d40923d3dc
--- /dev/null
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallRestartCoordinatorTest.kt
@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.components
+
+import io.getstream.result.Error
+import io.getstream.result.Result.Failure
+import io.getstream.result.Result.Success
+import io.getstream.video.android.core.call.Restartable
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertSame
+
+class CallRestartCoordinatorTest {
+
+    @Test
+    fun `restart recreates scope before restarting components and completing bookkeeping`() {
+        val events = mutableListOf<String>()
+        val coordinator = CallRestartCoordinator(
+            scopeProvider = Restartable {
+                events += "scope"
+                Success(Unit)
+            },
+            restartables = listOf(
+                Restartable {
+                    events += "state"
+                    Success(Unit)
+                },
+                Restartable {
+                    events += "media"
+                    Success(Unit)
+                },
+            ),
+            onRestarted = { events += "bookkeeping" },
+        )
+
+        assertEquals(Success(Unit), coordinator.restart())
+
+        assertEquals(listOf("scope", "state", "media", "bookkeeping"), events)
+    }
+
+    @Test
+    fun `restart returns scope failure without restarting persistent components`() {
+        val failure = Failure(Error.GenericError("scope restart failed"))
+        val events = mutableListOf<String>()
+        val coordinator = CallRestartCoordinator(
+            scopeProvider = Restartable { failure },
+            restartables = listOf(
+                Restartable {
+                    events += "state"
+                    Success(Unit)
+                },
+            ),
+            onRestarted = { events += "bookkeeping" },
+        )
+
+        val result = coordinator.restart()
+
+        assertSame(failure, result)
+        assertEquals(emptyList(), events)
+    }
+}
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/RestartableStateFlowTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/RestartableStateFlowTest.kt
new file mode 100644
index 00000000000..6b8fc41966d
--- /dev/null
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/RestartableStateFlowTest.kt
@@ -0,0 +1,107 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core.call.scope
+
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+
+class RestartableStateFlowTest {
+    @Test
+    fun `provider restart moves production without replacing collectors`() = runTest {
+        val scopeProvider = ScopeProviderImpl(backgroundScope, "default:test-call")
+        val upstream = MutableStateFlow(1)
+        val state = RestartableStateFlow(
+            upstream = upstream,
+            scopeProvider = scopeProvider,
+            initialValue = 0,
+            started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 0),
+        )
+        val values = mutableListOf<Int>()
+        val collector = backgroundScope.launch { state.collect(values::add) }
+        runCurrent()
+        assertEquals(1, state.value)
+
+        scopeProvider.cleanup()?.join()
+        upstream.value = 2
+        runCurrent()
+        assertEquals(1, state.value)
+
+        scopeProvider.restart()
+        runCurrent()
+
+        assertEquals(2, state.value)
+        assertFalse(collector.isCancelled)
+        assertEquals(listOf(0, 1, 0, 2), values)
+    }
+
+    @Test
+    fun `while subscribed starts and stops upstream using actual subscriber count`() = runTest {
+        val scope = ScopeProviderImpl(backgroundScope, "default:test-call")
+        var active = 0
+        val state = flow {
+            active++
+            try {
+                emit(1)
+                awaitCancellation()
+            } finally { active-- }
+        }.restartableStateIn(scope, SharingStarted.WhileSubscribed(stopTimeoutMillis = 0), 0)
+        runCurrent()
+        assertEquals(0, active)
+        val collector = backgroundScope.launch { state.collect {} }
+        runCurrent()
+        assertEquals(1, active)
+        collector.cancelAndJoin()
+        runCurrent()
+        assertEquals(0, active)
+    }
+
+    @Test
+    fun `detached launches cannot run without an owner`() = runTest {
+        val scope = ScopeProviderImpl(backgroundScope, "default:test-call")
+        scope.cleanup()?.join()
+        var ran = false
+        val job = scope.callScope.launch { ran = true }
+        job.join()
+        assertFalse(ran)
+    }
+
+    @Test
+    fun `replay expiration restores initial value`() = runTest {
+        val scope = ScopeProviderImpl(backgroundScope, "default:test-call")
+        val upstream = MutableStateFlow(1)
+        val state = upstream.restartableStateIn(
+            scope,
+            SharingStarted.WhileSubscribed(replayExpirationMillis = 0),
+            0,
+        )
+        val collector = backgroundScope.launch { state.collect {} }
+        runCurrent()
+        assertEquals(1, state.value)
+        collector.cancelAndJoin()
+        runCurrent()
+        assertEquals(0, state.value)
+    }
+}
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderTest.kt
index 74b7e00b852..2ad5aa481fe 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/scope/ScopeProviderTest.kt
@@ -16,14 +16,20 @@
 
 package io.getstream.video.android.core.call.scope
 
+import io.getstream.result.Result.Failure
+import io.getstream.result.Result.Success
 import io.getstream.video.android.core.socket.common.scope.ClientScope
 import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.test.runTest
 import org.junit.After
 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotEquals
 import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNotSame
+import org.junit.Assert.assertSame
 import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
@@ -31,12 +37,12 @@ import org.junit.Test
 class ScopeProviderTest {
 
     private lateinit var clientScope: ClientScope
-    private lateinit var scopeProvider: ScopeProvider
+    private lateinit var scopeProvider: ScopeProviderImpl
 
     @Before
     fun setup() {
         clientScope = ClientScope()
-        scopeProvider = ScopeProviderImpl(clientScope)
+        scopeProvider = ScopeProviderImpl(clientScope, "default:test-call")
     }
 
     @After
@@ -124,4 +130,39 @@ class ScopeProviderTest {
         assertNotNull(scope1)
         assertNotNull(scope2)
     }
+
+    @Test
+    fun `restart creates a new call scope and allows new RTC scopes`() = runTest {
+        val firstCallScope = scopeProvider.callScope
+        val firstCallJob = firstCallScope.coroutineContext[Job]!!
+        scopeProvider.getRtcSessionScope(SupervisorJob(), "test-call-1")
+
+        scopeProvider.cleanup()?.join()
+        assertTrue(firstCallJob.isCancelled)
+
+        assertEquals(Success(Unit), scopeProvider.restart())
+        val secondCallScope = scopeProvider.callScope
+        val secondCallJob = secondCallScope.coroutineContext[Job]!!
+
+        assertEquals(2, scopeProvider.debugInstanceId)
+        assertNotSame(firstCallScope, secondCallScope)
+        assertNotEquals(firstCallJob, secondCallJob)
+        assertTrue(secondCallJob.isActive)
+        assertNotNull(scopeProvider.getRtcSessionScope(SupervisorJob(), "test-call-1"))
+    }
+
+    @Test
+    fun `restart before cleanup returns a failure and keeps the current scope`() = runTest {
+        val firstCallScope = scopeProvider.callScope
+
+        val result = scopeProvider.restart()
+
+        assertTrue(result is Failure)
+        assertEquals(
+            "ScopeProvider cannot restart before its current generation is cleaned up",
+            (result as Failure).value.message,
+        )
+        assertSame(firstCallScope, scopeProvider.callScope)
+        assertEquals(1, scopeProvider.debugInstanceId)
+    }
 }
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/e2ee/CallE2EETest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/e2ee/CallE2EETest.kt
index 3b10b2079b7..c6dc72e36fe 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/e2ee/CallE2EETest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/e2ee/CallE2EETest.kt
@@ -40,6 +40,7 @@ import io.mockk.impl.annotations.RelaxedMockK
 import io.mockk.mockk
 import io.mockk.unmockkAll
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
@@ -93,6 +94,8 @@ class CallE2EETest {
             }
         every { mockClientImpl.scope } returns testScope as CoroutineScope
         every { mockClientImpl.apiKey } returns "test-api-key"
+        every { mockClientImpl.state.activeCall } returns MutableStateFlow<Call?>(null)
+        every { mockClientImpl.state.ringingCall } returns MutableStateFlow<Call?>(null)
         // Keeps joinRequest off the success path, which would try to apply a mocked response.
         coEvery {
             mockClientImpl.joinCall(
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/ReconnectEscalationTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/ReconnectEscalationTest.kt
index 92d64363040..39caad0cb4c 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/ReconnectEscalationTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/ReconnectEscalationTest.kt
@@ -89,6 +89,8 @@ class ReconnectEscalationTest {
         every { mockClientImpl.scope } returns testScope as CoroutineScope
         every { mockClientImpl.leaveAfterDisconnectSeconds } returns 120L
         every { mockClientImpl.apiKey } returns "test-api-key"
+        every { mockClientImpl.state.activeCall } returns MutableStateFlow<Call?>(null)
+        every { mockClientImpl.state.ringingCall } returns MutableStateFlow<Call?>(null)
 
         val user = User(id = "test-user", role = "user")
 
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsChannelFlowRegressionTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsChannelFlowRegressionTest.kt
index 5203001154f..9c2d81a46f3 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsChannelFlowRegressionTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsChannelFlowRegressionTest.kt
@@ -71,7 +71,7 @@ class SortedParticipantsChannelFlowRegressionTest {
             val pinned: StateFlow<Map<String, PinEntry>> =
                 MutableStateFlow(emptyMap())
 
-            val sut = SortedParticipantsState(scope, call, participants, pinned)
+            val sut = SortedParticipantsState({ scope }, call, participants, pinned)
 
             val sortedFlow = sut.sortedParticipants
 
diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsStateTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsStateTest.kt
index ea71b70cfb5..6aecdc8d5e1 100644
--- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsStateTest.kt
+++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/sorting/SortedParticipantsStateTest.kt
@@ -251,7 +251,7 @@ class SortedParticipantsStateTest {
             MutableStateFlow<Map<String, ParticipantState>>(emptyMap())
         val pinned: MutableStateFlow<Map<String, PinEntry>> =
             MutableStateFlow(emptyMap())
-        val state = SortedParticipantsState(scope, call, participants, pinned)
+        val state = SortedParticipantsState({ scope }, call, participants, pinned)
         return SutHandle(
             scope = scope,
             callActions = callActions,