diff --git a/akka-actor-typed-tests/src/test/scala/akka/actor/typed/OrElseSpec.scala b/akka-actor-typed-tests/src/test/scala/akka/actor/typed/OrElseSpec.scala
index 472df5720e..fc7da3a659 100644
--- a/akka-actor-typed-tests/src/test/scala/akka/actor/typed/OrElseSpec.scala
+++ b/akka-actor-typed-tests/src/test/scala/akka/actor/typed/OrElseSpec.scala
@@ -6,14 +6,16 @@ package akka.actor.typed
 
 import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.testkit.typed.scaladsl._
-import org.scalatest.Matchers
-import org.scalatest.WordSpec
+import org.scalatest.{ Matchers, WordSpec, WordSpecLike }
+
+object OrElseStubbedSpec {
 
-object OrElseSpec {
   sealed trait Ping
-  case class Ping1(replyTo: ActorRef[Pong]) extends Ping
-  case class Ping2(replyTo: ActorRef[Pong]) extends Ping
-  case class Ping3(replyTo: ActorRef[Pong]) extends Ping
+  final case class Ping1(replyTo: ActorRef[Pong]) extends Ping
+  final case class Ping2(replyTo: ActorRef[Pong]) extends Ping
+  final case class Ping3(replyTo: ActorRef[Pong]) extends Ping
+  final case class PingInfinite(replyTo: ActorRef[Pong]) extends Ping
+
   case class Pong(counter: Int)
 
   def ping(counters: Map[String, Int]): Behavior[Ping] = {
@@ -45,9 +47,9 @@ object OrElseSpec {
 
 }
 
-class OrElseSpec extends WordSpec with Matchers {
+class OrElseStubbedSpec extends WordSpec with Matchers {
 
-  import OrElseSpec._
+  import OrElseStubbedSpec._
 
   "Behavior.orElse" must {
 
@@ -71,6 +73,88 @@ class OrElseSpec extends WordSpec with Matchers {
       testkit.run(Ping1(inbox.ref))
       inbox.receiveMessage() should ===(Pong(3))
     }
+
+  }
+
+}
+
+class OrElseSpec extends ScalaTestWithActorTestKit with WordSpecLike {
+
+  import OrElseStubbedSpec._
+
+  "Behavior.orElse" must {
+    "work for deferred behavior on the left" in {
+      val orElseDeferred = Behaviors.setup[Ping] { _ ⇒
+        Behaviors.receiveMessage { _ ⇒
+          Behaviors.unhandled
+        }
+      }.orElse(ping(Map.empty))
+
+      val p = spawn(orElseDeferred)
+      val probe = TestProbe[Pong]
+      p ! Ping1(probe.ref)
+      probe.expectMessage(Pong(1))
+
+    }
+
+    "work for deferred behavior on the right" in {
+      val orElseDeferred = ping(Map.empty).orElse(Behaviors.setup { _ ⇒
+        Behaviors.receiveMessage {
+          case PingInfinite(replyTo) ⇒
+            replyTo ! Pong(-1)
+            Behaviors.same
+        }
+      })
+
+      val p = spawn(orElseDeferred)
+      val probe = TestProbe[Pong]
+      p ! PingInfinite(probe.ref)
+      probe.expectMessage(Pong(-1))
+    }
+  }
+
+  "handle nested OrElse" in {
+
+    sealed trait Parent
+    final case class Add(o: Any) extends Parent
+    final case class Remove(o: Any) extends Parent
+    final case class Stack(s: ActorRef[Array[StackTraceElement]]) extends Parent
+    final case class Get(s: ActorRef[Set[Any]]) extends Parent
+
+    def dealer(set: Set[Any]): Behavior[Parent] = {
+      val add = Behaviors.receiveMessage[Parent] {
+        case Add(o) ⇒ dealer(set + o)
+        case _      ⇒ Behaviors.unhandled
+      }
+      val remove = Behaviors.receiveMessage[Parent] {
+        case Remove(o) ⇒ dealer(set - o)
+        case _         ⇒ Behaviors.unhandled
+      }
+      val getStack = Behaviors.receiveMessagePartial[Parent] {
+        case Stack(sender) ⇒
+          sender ! Thread.currentThread().getStackTrace
+          Behaviors.same
+      }
+      val getSet = Behaviors.receiveMessagePartial[Parent] {
+        case Get(sender) ⇒
+          sender ! set
+          Behaviors.same
+      }
+      add.orElse(remove).orElse(getStack).orElse(getSet)
+    }
+
+    val y = spawn(dealer(Set.empty))
+
+    (0 to 10000) foreach { i ⇒
+      y ! Add(i)
+    }
+    (0 to 9999) foreach { i ⇒
+      y ! Remove(i)
+    }
+    val probe = TestProbe[Set[Any]]
+    y ! Get(probe.ref)
+    probe.expectMessage(Set[Any](10000))
+
   }
 
 }
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/Behavior.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/Behavior.scala
index 5e6d68ea1c..d6427b48c8 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/Behavior.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/Behavior.scala
@@ -6,8 +6,8 @@ package akka.actor.typed
 
 import akka.actor.InvalidMessageException
 import akka.actor.typed.internal.BehaviorImpl
-import scala.annotation.tailrec
 
+import scala.annotation.tailrec
 import akka.actor.typed.internal.BehaviorImpl.OrElseBehavior
 import akka.actor.typed.internal.WrappingBehavior
 import akka.util.{ LineNumbers, OptionVal }
@@ -51,7 +51,9 @@ sealed abstract class Behavior[T] { behavior ⇒
    *
    *  @param that the fallback `Behavior`
    */
-  final def orElse(that: Behavior[T]): Behavior[T] = new OrElseBehavior[T](this, that)
+  final def orElse(that: Behavior[T]): Behavior[T] = Behavior.DeferredBehavior[T] { ctx ⇒
+    new OrElseBehavior[T](Behavior.start(this, ctx), Behavior.start(that, ctx))
+  }
 }
 
 /**
diff --git a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/BehaviorImpl.scala b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/BehaviorImpl.scala
index bc4777b51a..98a52a6bbc 100644
--- a/akka-actor-typed/src/main/scala/akka/actor/typed/internal/BehaviorImpl.scala
+++ b/akka-actor-typed/src/main/scala/akka/actor/typed/internal/BehaviorImpl.scala
@@ -65,6 +65,7 @@ import akka.actor.typed.scaladsl.{ ActorContext ⇒ SAC }
     InterceptorImpl(interceptor, behavior)
 
   class OrElseBehavior[T](first: Behavior[T], second: Behavior[T]) extends ExtensibleBehavior[T] {
+
     override def receive(ctx: AC[T], msg: T): Behavior[T] = {
       Behavior.interpretMessage(first, ctx, msg) match {
         case _: UnhandledBehavior.type ⇒ Behavior.interpretMessage(second, ctx, msg)
diff --git a/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/scaladsl/ClusterShardingPersistenceSpec.scala b/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/scaladsl/ClusterShardingPersistenceSpec.scala
index 3d83deb629..256d43aadb 100644
--- a/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/scaladsl/ClusterShardingPersistenceSpec.scala
+++ b/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/scaladsl/ClusterShardingPersistenceSpec.scala
@@ -11,6 +11,7 @@ import java.util.concurrent.TimeUnit
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.concurrent.Promise
+import scala.concurrent.duration._
 
 import akka.Done
 import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit