diff --git a/src/core/Akka.Remote.Tests/Serialization/Bugfix3903Spec.cs b/src/core/Akka.Remote.Tests/Serialization/Bugfix3903Spec.cs
new file mode 100644
index 00000000000..65811962fa2
--- /dev/null
+++ b/src/core/Akka.Remote.Tests/Serialization/Bugfix3903Spec.cs
@@ -0,0 +1,124 @@
+//-----------------------------------------------------------------------
+// <copyright file="Bugfix3903Specs.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.TestKit;
+using Akka.Util.Internal;
+using Xunit;
+using Xunit.Abstractions;
+using FluentAssertions;
+
+namespace Akka.Remote.Tests.Serialization
+{
+    public class Bugfix3903Spec : AkkaSpec
+    {
+        // hocon config enabling akka.remote
+        private static readonly Config Config = @"akka.actor.provider = remote
+                                                    akka.remote.dot-netty.tcp.hostname = localhost
+                                                    akka.remote.dot-netty.tcp.port = 0";
+
+        public Bugfix3903Spec(ITestOutputHelper outputHelper) : base(Config, outputHelper)
+        {
+        }
+
+        #region Internal Types
+
+        // parent actor type that will remotely deploy a child actor type onto a specific address
+        private class ParentActor : ReceiveActor
+        {
+            // message type that includes an Address
+            public class DeployChild
+            {
+                public DeployChild(Address address)
+                {
+                    Address = address;
+                }
+
+                public Address Address { get; }
+            }
+
+            public ParentActor()
+            {
+                Receive<DeployChild>(s =>
+                {
+                    // props to deploy an EchoActor at the address specified in DeployChild
+                    var props = Props.Create<EchoActor>().WithDeploy(new Deploy(new RemoteScope(s.Address)));
+                    var child = Context.ActorOf(props, "child");
+                    Sender.Tell(child);
+                });
+            }
+        }
+
+        internal class EchoActor : ReceiveActor
+        {
+            public class Fail
+            {
+                public static readonly Fail Instance = new Fail();
+                private Fail(){}
+            }
+            
+            public EchoActor()
+            {
+                // receive message that will cause this actor to fail
+                Receive<Fail>(s =>
+                {
+                    throw new ApplicationException("fail");
+                });
+                ReceiveAny(o => Sender.Tell(o));
+            }
+        }
+
+        #endregion
+
+        // a test where Sys starts a ParentActor and has it remotely deploy an EchoActor onto a second ActorSystem
+        [Fact]
+        public async Task ParentActor_should_be_able_to_deploy_EchoActor_to_remote_system()
+        {
+            // create a second ActorSystem
+            var system2 = ActorSystem.Create(Sys.Name, Sys.Settings.Config);
+            InitializeLogger(system2);
+            try
+            {
+                // create a supervision strategy that will send a message to the TestActor including the exception of the child that failed
+                var strategy = new OneForOneStrategy(ex =>
+                {
+                    TestActor.Tell(ex);
+                    return Directive.Stop;
+                });
+
+                // create a ParentActor in the first ActorSystem
+                var parent = Sys.ActorOf(Props.Create<ParentActor>().WithSupervisorStrategy(strategy), "parent");
+
+                // have the ParentActor remotely deploy an EchoActor onto the second ActorSystem
+                var child = await parent
+                    .Ask<IActorRef>(new ParentActor.DeployChild(
+                        system2.AsInstanceOf<ExtendedActorSystem>().Provider.DefaultAddress), RemainingOrDefault).ConfigureAwait(false);
+
+                // assert that Child is a remote actor reference
+                child.Should().BeOfType<RemoteActorRef>();
+                Watch(child);
+                
+                // send a message to the EchoActor and verify that it is received
+                (await child.Ask<string>("hello", RemainingOrDefault).ConfigureAwait(false)).Should().Be("hello");
+                
+                // cause the child to crash
+                child.Tell(EchoActor.Fail.Instance);
+                var exception = ExpectMsg<ApplicationException>();
+                exception.Message.Should().Be("fail");
+                ExpectTerminated(child);
+            }
+            finally
+            {
+                // shut down the second ActorSystem
+                Shutdown(system2);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs b/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs
index 7065028ed99..33e708691a4 100644
--- a/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs
+++ b/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs
@@ -76,6 +76,30 @@ public void Can_serialize_IdentifyWithNull()
             AssertEqual(identify);
         }
 
+        [Theory]
+        [InlineData(null)]
+        [InlineData(1)]
+        [InlineData("hi")]
+        public void Can_serialize_StatusSuccess(object payload)
+        {
+            var success = new Status.Success(payload);
+            AssertEqual(success);
+        }
+        
+        [Theory]
+        [InlineData(null)]
+        [InlineData(1)]
+        [InlineData("hi")]
+        public void Can_serialize_StatusFailure(object payload)
+        {
+            var success = new Status.Failure(new ApplicationException("foo"),payload);
+            // can't use AssertEqual here since the Exception data isn't 100% identical after round-trip serialization
+            var deserialized = AssertAndReturn(success);
+            deserialized.State.Should().BeEquivalentTo(success.State);
+            deserialized.Cause.Message.Should().BeEquivalentTo(success.Cause.Message);
+            deserialized.Cause.Should().BeOfType(success.Cause.GetType());
+        }
+
         [Fact]
         public void Can_serialize_ActorIdentity()
         {
@@ -372,7 +396,7 @@ private T AssertAndReturn<T>(T message)
         private void AssertEqual<T>(T message)
         {
             var deserialized = AssertAndReturn(message);
-            Assert.Equal(message, deserialized);
+            deserialized.Should().BeEquivalentTo(message);
         }
     }
 }
diff --git a/src/core/Akka.Remote/Configuration/Remote.conf b/src/core/Akka.Remote/Configuration/Remote.conf
index 3b593c2f75a..5f1441256ee 100644
--- a/src/core/Akka.Remote/Configuration/Remote.conf
+++ b/src/core/Akka.Remote/Configuration/Remote.conf
@@ -39,6 +39,8 @@ akka {
       "Akka.Actor.PoisonPill, Akka" = akka-misc
       "Akka.Actor.Kill, Akka" = akka-misc
       "Akka.Actor.PoisonPill, Akka" = akka-misc
+      "Akka.Actor.Status+Failure, Akka" = akka-misc
+      "Akka.Actor.Status+Success, Akka" = akka-misc
       #"Akka.Actor.LocalScope, Akka" = akka-misc
       "Akka.Actor.RemoteScope, Akka" = akka-misc
       "Akka.Routing.FromConfig, Akka" = akka-misc
diff --git a/src/core/Akka.Remote/Serialization/ExceptionSupport.cs b/src/core/Akka.Remote/Serialization/ExceptionSupport.cs
index ba304ceaf00..2b09bde70d9 100644
--- a/src/core/Akka.Remote/Serialization/ExceptionSupport.cs
+++ b/src/core/Akka.Remote/Serialization/ExceptionSupport.cs
@@ -150,7 +150,7 @@ public Exception ExceptionFromProtoNet(Proto.Msg.ExceptionData proto)
             return obj;
         }
 
-        private string ValueOrNull(string value)
+        private static string ValueOrNull(string value)
             => string.IsNullOrEmpty(value) ? null : value;
     }
 }
diff --git a/src/core/Akka.Remote/Serialization/MessageContainerSerializer.cs b/src/core/Akka.Remote/Serialization/MessageContainerSerializer.cs
index 087501bd2ef..0c140482d2f 100644
--- a/src/core/Akka.Remote/Serialization/MessageContainerSerializer.cs
+++ b/src/core/Akka.Remote/Serialization/MessageContainerSerializer.cs
@@ -9,7 +9,6 @@
 using System.Linq;
 using System.Runtime.Serialization;
 using Akka.Actor;
-using Akka.Remote.Serialization.Proto.Msg;
 using Akka.Serialization;
 using Akka.Util;
 using Google.Protobuf;
@@ -108,7 +107,7 @@ public override object FromBinary(byte[] bytes, Type type)
             return new ActorSelectionMessage(message, elements);
         }
 
-        private Proto.Msg.Selection BuildPattern(string matcher, Proto.Msg.Selection.Types.PatternType tpe)
+        private static Proto.Msg.Selection BuildPattern(string matcher, Proto.Msg.Selection.Types.PatternType tpe)
         {
             var selection = new Proto.Msg.Selection { Type = tpe };
             if (matcher != null)
diff --git a/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs b/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs
index d3e54e48e67..df5b70ae016 100644
--- a/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs
+++ b/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs
@@ -42,10 +42,13 @@ public sealed class MiscMessageSerializer : SerializerWithStringManifest
         private const string TailChoppingPoolManifest = "ROTCP";
         private const string ConsistentHashingPoolManifest = "ROCHP";
         private const string RemoteRouterConfigManifest = "RORRC";
+        private const string StatusFailureManifest = "STF";
+        private const string StatusSuccessManifest = "STS";
 
-        private static readonly byte[] EmptyBytes = {};
+        private static readonly byte[] EmptyBytes = Array.Empty<byte>();
 
         private readonly WrappedPayloadSupport _payloadSupport;
+        private readonly ExceptionSupport _exceptionSupport;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="MiscMessageSerializer" /> class.
@@ -54,84 +57,161 @@ public sealed class MiscMessageSerializer : SerializerWithStringManifest
         public MiscMessageSerializer(ExtendedActorSystem system) : base(system)
         {
             _payloadSupport = new WrappedPayloadSupport(system);
+            _exceptionSupport = new ExceptionSupport(system);
         }
 
         /// <inheritdoc />
         public override byte[] ToBinary(object obj)
         {
-            if (obj is Identify identify) return IdentifyToProto(identify);
-            if (obj is ActorIdentity actorIdentity) return ActorIdentityToProto(actorIdentity);
-            if (obj is IActorRef actorRef) return ActorRefToProto(actorRef);
-            if (obj is PoisonPill) return EmptyBytes;
-            if (obj is Kill) return EmptyBytes;
-            if (obj is RemoteWatcher.Heartbeat) return EmptyBytes;
-            if (obj is RemoteWatcher.HeartbeatRsp heartbeatRsp) return HeartbeatRspToProto(heartbeatRsp);
-            if (obj is LocalScope) return EmptyBytes;
-            if (obj is RemoteScope remoteScope) return RemoteScopeToProto(remoteScope);
-            if (obj is Config config) return ConfigToProto(config);
-            if (obj is FromConfig fromConfig) return FromConfigToProto(fromConfig);
-            if (obj is DefaultResizer defaultResizer) return DefaultResizerToProto(defaultResizer);
-            if (obj is RoundRobinPool roundRobinPool) return RoundRobinPoolToProto(roundRobinPool);
-            if (obj is BroadcastPool broadcastPool) return BroadcastPoolToProto(broadcastPool);
-            if (obj is RandomPool randomPool) return RandomPoolToProto(randomPool);
-            if (obj is ScatterGatherFirstCompletedPool scatterPool) return ScatterGatherFirstCompletedPoolToProto(scatterPool);
-            if (obj is TailChoppingPool tailChoppingPool) return TailChoppingPoolToProto(tailChoppingPool);
-            if (obj is ConsistentHashingPool hashingPool) return ConsistentHashingPoolToProto(hashingPool);
-            if (obj is RemoteRouterConfig remoteRouterConfig) return RemoteRouterConfigToProto(remoteRouterConfig);
-
-            throw new ArgumentException($"Cannot serialize object of type [{obj.GetType().TypeQualifiedName()}]");
+            switch (obj)
+            {
+                case Identify identify:
+                    return IdentifyToProto(identify);
+                case ActorIdentity actorIdentity:
+                    return ActorIdentityToProto(actorIdentity);
+                case IActorRef actorRef:
+                    return ActorRefToProto(actorRef);
+                case PoisonPill _:
+                case Kill _:
+                case RemoteWatcher.Heartbeat _:
+                    return EmptyBytes;
+                case RemoteWatcher.HeartbeatRsp heartbeatRsp:
+                    return HeartbeatRspToProto(heartbeatRsp);
+                case Status.Success success:
+                    return StatusSuccessToProto(success); 
+                case Status.Failure failure:
+                    return StatusFailureToProto(failure);
+                case LocalScope _:
+                    return EmptyBytes;
+                case RemoteScope remoteScope:
+                    return RemoteScopeToProto(remoteScope);
+                case Config config:
+                    return ConfigToProto(config);
+                case FromConfig fromConfig:
+                    return FromConfigToProto(fromConfig);
+                case DefaultResizer defaultResizer:
+                    return DefaultResizerToProto(defaultResizer);
+                case RoundRobinPool roundRobinPool:
+                    return RoundRobinPoolToProto(roundRobinPool);
+                case BroadcastPool broadcastPool:
+                    return BroadcastPoolToProto(broadcastPool);
+                case RandomPool randomPool:
+                    return RandomPoolToProto(randomPool);
+                case ScatterGatherFirstCompletedPool scatterPool:
+                    return ScatterGatherFirstCompletedPoolToProto(scatterPool);
+                case TailChoppingPool tailChoppingPool:
+                    return TailChoppingPoolToProto(tailChoppingPool);
+                case ConsistentHashingPool hashingPool:
+                    return ConsistentHashingPoolToProto(hashingPool);
+                case RemoteRouterConfig remoteRouterConfig:
+                    return RemoteRouterConfigToProto(remoteRouterConfig);
+                default:
+                    throw new ArgumentException($"Cannot serialize object of type [{obj.GetType().TypeQualifiedName()}]");
+            }
         }
 
         /// <inheritdoc />
         public override string Manifest(object obj)
         {
-            if (obj is Identify) return IdentifyManifest;
-            if (obj is ActorIdentity) return ActorIdentityManifest;
-            if (obj is IActorRef) return ActorRefManifest;
-            if (obj is PoisonPill) return PoisonPillManifest;
-            if (obj is Kill) return KillManifest;
-            if (obj is RemoteWatcher.Heartbeat) return RemoteWatcherHearthbeatManifest;
-            if (obj is RemoteWatcher.HeartbeatRsp) return RemoteWatcherHearthbeatRspManifest;
-            if (obj is LocalScope) return LocalScopeManifest;
-            if (obj is RemoteScope) return RemoteScopeManifest;
-            if (obj is Config) return ConfigManifest;
-            if (obj is FromConfig) return FromConfigManifest;
-            if (obj is DefaultResizer) return DefaultResizerManifest;
-            if (obj is RoundRobinPool) return RoundRobinPoolManifest;
-            if (obj is BroadcastPool) return BroadcastPoolManifest;
-            if (obj is RandomPool) return RandomPoolManifest;
-            if (obj is ScatterGatherFirstCompletedPool) return ScatterGatherPoolManifest;
-            if (obj is TailChoppingPool) return TailChoppingPoolManifest;
-            if (obj is ConsistentHashingPool) return ConsistentHashingPoolManifest;
-            if (obj is RemoteRouterConfig) return RemoteRouterConfigManifest;
-
-            throw new ArgumentException($"Cannot deserialize object of type [{obj.GetType().TypeQualifiedName()}]");
+            switch (obj)
+            {
+                case Identify _:
+                    return IdentifyManifest;
+                case ActorIdentity _:
+                    return ActorIdentityManifest;
+                case IActorRef _:
+                    return ActorRefManifest;
+                case PoisonPill _:
+                    return PoisonPillManifest;
+                case Kill _:
+                    return KillManifest;
+                case RemoteWatcher.Heartbeat _:
+                    return RemoteWatcherHearthbeatManifest;
+                case RemoteWatcher.HeartbeatRsp _:
+                    return RemoteWatcherHearthbeatRspManifest;
+                case Status.Success _:
+                    return StatusSuccessManifest;
+                case Status.Failure _:
+                    return StatusFailureManifest;
+                case LocalScope _:
+                    return LocalScopeManifest;
+                case RemoteScope _:
+                    return RemoteScopeManifest;
+                case Config _:
+                    return ConfigManifest;
+                case FromConfig _:
+                    return FromConfigManifest;
+                case DefaultResizer _:
+                    return DefaultResizerManifest;
+                case RoundRobinPool _:
+                    return RoundRobinPoolManifest;
+                case BroadcastPool _:
+                    return BroadcastPoolManifest;
+                case RandomPool _:
+                    return RandomPoolManifest;
+                case ScatterGatherFirstCompletedPool _:
+                    return ScatterGatherPoolManifest;
+                case TailChoppingPool _:
+                    return TailChoppingPoolManifest;
+                case ConsistentHashingPool _:
+                    return ConsistentHashingPoolManifest;
+                case RemoteRouterConfig _:
+                    return RemoteRouterConfigManifest;
+                default:
+                    throw new ArgumentException($"Cannot deserialize object of type [{obj.GetType().TypeQualifiedName()}]");
+            }
         }
 
         /// <inheritdoc />
         public override object FromBinary(byte[] bytes, string manifest)
         {
-            if (manifest == IdentifyManifest) return IdentifyFromProto(bytes);
-            if (manifest == ActorIdentityManifest) return ActorIdentityFromProto(bytes);
-            if (manifest == ActorRefManifest) return ActorRefFromProto(bytes);
-            if (manifest == PoisonPillManifest) return PoisonPill.Instance;
-            if (manifest == KillManifest) return Kill.Instance;
-            if (manifest == RemoteWatcherHearthbeatManifest) return RemoteWatcher.Heartbeat.Instance;
-            if (manifest == RemoteWatcherHearthbeatRspManifest) return HearthbeatRspFromProto(bytes);
-            if (manifest == LocalScopeManifest) return LocalScope.Instance;
-            if (manifest == RemoteScopeManifest) return RemoteScopeFromProto(bytes);
-            if (manifest == ConfigManifest) return ConfigFromProto(bytes);
-            if (manifest == FromConfigManifest) return FromConfigFromProto(bytes);
-            if (manifest == DefaultResizerManifest) return DefaultResizerFromProto(bytes);
-            if (manifest == RoundRobinPoolManifest) return RoundRobinPoolFromProto(bytes);
-            if (manifest == BroadcastPoolManifest) return BroadcastPoolFromProto(bytes);
-            if (manifest == RandomPoolManifest) return RandomPoolFromProto(bytes);
-            if (manifest == ScatterGatherPoolManifest) return ScatterGatherFirstCompletedPoolFromProto(bytes);
-            if (manifest == TailChoppingPoolManifest) return TailChoppingPoolFromProto(bytes);
-            if (manifest == ConsistentHashingPoolManifest) return ConsistentHashingPoolFromProto(bytes);
-            if (manifest == RemoteRouterConfigManifest) return RemoteRouterConfigFromProto(bytes);
- 
-            throw new SerializationException($"Unimplemented deserialization of message with manifest [{manifest}] in [{nameof(MiscMessageSerializer)}]");
+            switch (manifest)
+            {
+                case IdentifyManifest:
+                    return IdentifyFromProto(bytes);
+                case ActorIdentityManifest:
+                    return ActorIdentityFromProto(bytes);
+                case ActorRefManifest:
+                    return ActorRefFromProto(bytes);
+                case PoisonPillManifest:
+                    return PoisonPill.Instance;
+                case KillManifest:
+                    return Kill.Instance;
+                case RemoteWatcherHearthbeatManifest:
+                    return RemoteWatcher.Heartbeat.Instance;
+                case RemoteWatcherHearthbeatRspManifest:
+                    return HearthbeatRspFromProto(bytes);
+                case StatusSuccessManifest:
+                    return StatusSuccessFromProto(bytes);
+                case StatusFailureManifest:
+                    return StatusFailureFromProto(bytes);
+                case LocalScopeManifest:
+                    return LocalScope.Instance;
+                case RemoteScopeManifest:
+                    return RemoteScopeFromProto(bytes);
+                case ConfigManifest:
+                    return ConfigFromProto(bytes);
+                case FromConfigManifest:
+                    return FromConfigFromProto(bytes);
+                case DefaultResizerManifest:
+                    return DefaultResizerFromProto(bytes);
+                case RoundRobinPoolManifest:
+                    return RoundRobinPoolFromProto(bytes);
+                case BroadcastPoolManifest:
+                    return BroadcastPoolFromProto(bytes);
+                case RandomPoolManifest:
+                    return RandomPoolFromProto(bytes);
+                case ScatterGatherPoolManifest:
+                    return ScatterGatherFirstCompletedPoolFromProto(bytes);
+                case TailChoppingPoolManifest:
+                    return TailChoppingPoolFromProto(bytes);
+                case ConsistentHashingPoolManifest:
+                    return ConsistentHashingPoolFromProto(bytes);
+                case RemoteRouterConfigManifest:
+                    return RemoteRouterConfigFromProto(bytes);
+                default:
+                    throw new SerializationException($"Unimplemented deserialization of message with manifest [{manifest}] in [{nameof(MiscMessageSerializer)}]");
+            }
         }
 
         //
@@ -206,6 +286,42 @@ private RemoteWatcher.HeartbeatRsp HearthbeatRspFromProto(byte[] bytes)
             var message = Proto.Msg.RemoteWatcherHeartbeatResponse.Parser.ParseFrom(bytes);
             return new RemoteWatcher.HeartbeatRsp((int)message.Uid);
         }
+        
+        //
+        // Status.Success
+        //
+        
+        private byte[] StatusSuccessToProto(Status.Success success)
+        {
+            var message = new Proto.Msg.StatusSuccess();
+            message.Status = _payloadSupport.PayloadToProto(success.Status);
+            return message.ToByteArray();
+        }
+        
+        private Status.Success StatusSuccessFromProto(byte[] bytes)
+        {
+            var message = Proto.Msg.StatusSuccess.Parser.ParseFrom(bytes);
+            return new Status.Success(_payloadSupport.PayloadFrom(message.Status));
+        }
+        
+        //
+        // Status.Failure
+        //
+        
+        private byte[] StatusFailureToProto(Status.Failure failure)
+        {
+            var message = new Proto.Msg.StatusFailure();
+            message.Cause = _exceptionSupport.ExceptionToProto(failure.Cause);
+            message.State = _payloadSupport.PayloadToProto(failure.State);
+            return message.ToByteArray();
+        }
+        
+        private Status.Failure StatusFailureFromProto(byte[] bytes)
+        {
+            var message = Proto.Msg.StatusFailure.Parser.ParseFrom(bytes);
+            var payload = _payloadSupport.PayloadFrom(message.State);
+            return new Status.Failure(_exceptionSupport.ExceptionFromProto(message.Cause), payload);
+        }
 
         //
         // RemoteScope
diff --git a/src/core/Akka.Remote/Serialization/Proto/ContainerFormats.g.cs b/src/core/Akka.Remote/Serialization/Proto/ContainerFormats.g.cs
index 5ca35abaf08..b4d726a1fcf 100644
--- a/src/core/Akka.Remote/Serialization/Proto/ContainerFormats.g.cs
+++ b/src/core/Akka.Remote/Serialization/Proto/ContainerFormats.g.cs
@@ -49,7 +49,12 @@ static ContainerFormatsReflection() {
             "MkQuQWtrYS5SZW1vdGUuU2VyaWFsaXphdGlvbi5Qcm90by5Nc2cuRXhjZXB0",
             "aW9uRGF0YS5DdXN0b21GaWVsZHNFbnRyeRphChFDdXN0b21GaWVsZHNFbnRy",
             "eRILCgNrZXkYASABKAkSOwoFdmFsdWUYAiABKAsyLC5Ba2thLlJlbW90ZS5T",
-            "ZXJpYWxpemF0aW9uLlByb3RvLk1zZy5QYXlsb2FkOgI4AWIGcHJvdG8z"));
+            "ZXJpYWxpemF0aW9uLlByb3RvLk1zZy5QYXlsb2FkOgI4ASJNCg1TdGF0dXNT",
+            "dWNjZXNzEjwKBnN0YXR1cxgBIAEoCzIsLkFra2EuUmVtb3RlLlNlcmlhbGl6",
+            "YXRpb24uUHJvdG8uTXNnLlBheWxvYWQijwEKDVN0YXR1c0ZhaWx1cmUSQQoF",
+            "Y2F1c2UYASABKAsyMi5Ba2thLlJlbW90ZS5TZXJpYWxpemF0aW9uLlByb3Rv",
+            "Lk1zZy5FeGNlcHRpb25EYXRhEjsKBXN0YXRlGAIgASgLMiwuQWtrYS5SZW1v",
+            "dGUuU2VyaWFsaXphdGlvbi5Qcm90by5Nc2cuUGF5bG9hZGIGcHJvdG8z"));
       descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData,
           new pbr::FileDescriptor[] { },
           new pbr::GeneratedClrTypeInfo(null, null, new pbr::GeneratedClrTypeInfo[] {
@@ -61,7 +66,9 @@ static ContainerFormatsReflection() {
             new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.Identify), global::Akka.Remote.Serialization.Proto.Msg.Identify.Parser, new[]{ "MessageId" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.ActorIdentity), global::Akka.Remote.Serialization.Proto.Msg.ActorIdentity.Parser, new[]{ "CorrelationId", "Path" }, null, null, null, null),
             new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.RemoteWatcherHeartbeatResponse), global::Akka.Remote.Serialization.Proto.Msg.RemoteWatcherHeartbeatResponse.Parser, new[]{ "Uid" }, null, null, null, null),
-            new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.ExceptionData), global::Akka.Remote.Serialization.Proto.Msg.ExceptionData.Parser, new[]{ "TypeName", "Message", "StackTrace", "Source", "InnerException", "CustomFields" }, null, null, null, new pbr::GeneratedClrTypeInfo[] { null, })
+            new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.ExceptionData), global::Akka.Remote.Serialization.Proto.Msg.ExceptionData.Parser, new[]{ "TypeName", "Message", "StackTrace", "Source", "InnerException", "CustomFields" }, null, null, null, new pbr::GeneratedClrTypeInfo[] { null, }),
+            new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.StatusSuccess), global::Akka.Remote.Serialization.Proto.Msg.StatusSuccess.Parser, new[]{ "Status" }, null, null, null, null),
+            new pbr::GeneratedClrTypeInfo(typeof(global::Akka.Remote.Serialization.Proto.Msg.StatusFailure), global::Akka.Remote.Serialization.Proto.Msg.StatusFailure.Parser, new[]{ "Cause", "State" }, null, null, null, null)
           }));
     }
     #endregion
@@ -2288,6 +2295,448 @@ public void MergeFrom(pb::CodedInputStream input) {
 
   }
 
+  internal sealed partial class StatusSuccess : pb::IMessage<StatusSuccess>
+  #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+      , pb::IBufferMessage
+  #endif
+  {
+    private static readonly pb::MessageParser<StatusSuccess> _parser = new pb::MessageParser<StatusSuccess>(() => new StatusSuccess());
+    private pb::UnknownFieldSet _unknownFields;
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public static pb::MessageParser<StatusSuccess> Parser { get { return _parser; } }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public static pbr::MessageDescriptor Descriptor {
+      get { return global::Akka.Remote.Serialization.Proto.Msg.ContainerFormatsReflection.Descriptor.MessageTypes[9]; }
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    pbr::MessageDescriptor pb::IMessage.Descriptor {
+      get { return Descriptor; }
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public StatusSuccess() {
+      OnConstruction();
+    }
+
+    partial void OnConstruction();
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public StatusSuccess(StatusSuccess other) : this() {
+      status_ = other.status_ != null ? other.status_.Clone() : null;
+      _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public StatusSuccess Clone() {
+      return new StatusSuccess(this);
+    }
+
+    /// <summary>Field number for the "status" field.</summary>
+    public const int StatusFieldNumber = 1;
+    private global::Akka.Remote.Serialization.Proto.Msg.Payload status_;
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public global::Akka.Remote.Serialization.Proto.Msg.Payload Status {
+      get { return status_; }
+      set {
+        status_ = value;
+      }
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public override bool Equals(object other) {
+      return Equals(other as StatusSuccess);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public bool Equals(StatusSuccess other) {
+      if (ReferenceEquals(other, null)) {
+        return false;
+      }
+      if (ReferenceEquals(other, this)) {
+        return true;
+      }
+      if (!object.Equals(Status, other.Status)) return false;
+      return Equals(_unknownFields, other._unknownFields);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public override int GetHashCode() {
+      int hash = 1;
+      if (status_ != null) hash ^= Status.GetHashCode();
+      if (_unknownFields != null) {
+        hash ^= _unknownFields.GetHashCode();
+      }
+      return hash;
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public override string ToString() {
+      return pb::JsonFormatter.ToDiagnosticString(this);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public void WriteTo(pb::CodedOutputStream output) {
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+      output.WriteRawMessage(this);
+    #else
+      if (status_ != null) {
+        output.WriteRawTag(10);
+        output.WriteMessage(Status);
+      }
+      if (_unknownFields != null) {
+        _unknownFields.WriteTo(output);
+      }
+    #endif
+    }
+
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) {
+      if (status_ != null) {
+        output.WriteRawTag(10);
+        output.WriteMessage(Status);
+      }
+      if (_unknownFields != null) {
+        _unknownFields.WriteTo(ref output);
+      }
+    }
+    #endif
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public int CalculateSize() {
+      int size = 0;
+      if (status_ != null) {
+        size += 1 + pb::CodedOutputStream.ComputeMessageSize(Status);
+      }
+      if (_unknownFields != null) {
+        size += _unknownFields.CalculateSize();
+      }
+      return size;
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public void MergeFrom(StatusSuccess other) {
+      if (other == null) {
+        return;
+      }
+      if (other.status_ != null) {
+        if (status_ == null) {
+          Status = new global::Akka.Remote.Serialization.Proto.Msg.Payload();
+        }
+        Status.MergeFrom(other.Status);
+      }
+      _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public void MergeFrom(pb::CodedInputStream input) {
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+      input.ReadRawMessage(this);
+    #else
+      uint tag;
+      while ((tag = input.ReadTag()) != 0) {
+        switch(tag) {
+          default:
+            _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
+            break;
+          case 10: {
+            if (status_ == null) {
+              Status = new global::Akka.Remote.Serialization.Proto.Msg.Payload();
+            }
+            input.ReadMessage(Status);
+            break;
+          }
+        }
+      }
+    #endif
+    }
+
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
+      uint tag;
+      while ((tag = input.ReadTag()) != 0) {
+        switch(tag) {
+          default:
+            _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
+            break;
+          case 10: {
+            if (status_ == null) {
+              Status = new global::Akka.Remote.Serialization.Proto.Msg.Payload();
+            }
+            input.ReadMessage(Status);
+            break;
+          }
+        }
+      }
+    }
+    #endif
+
+  }
+
+  internal sealed partial class StatusFailure : pb::IMessage<StatusFailure>
+  #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+      , pb::IBufferMessage
+  #endif
+  {
+    private static readonly pb::MessageParser<StatusFailure> _parser = new pb::MessageParser<StatusFailure>(() => new StatusFailure());
+    private pb::UnknownFieldSet _unknownFields;
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public static pb::MessageParser<StatusFailure> Parser { get { return _parser; } }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public static pbr::MessageDescriptor Descriptor {
+      get { return global::Akka.Remote.Serialization.Proto.Msg.ContainerFormatsReflection.Descriptor.MessageTypes[10]; }
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    pbr::MessageDescriptor pb::IMessage.Descriptor {
+      get { return Descriptor; }
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public StatusFailure() {
+      OnConstruction();
+    }
+
+    partial void OnConstruction();
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public StatusFailure(StatusFailure other) : this() {
+      cause_ = other.cause_ != null ? other.cause_.Clone() : null;
+      state_ = other.state_ != null ? other.state_.Clone() : null;
+      _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public StatusFailure Clone() {
+      return new StatusFailure(this);
+    }
+
+    /// <summary>Field number for the "cause" field.</summary>
+    public const int CauseFieldNumber = 1;
+    private global::Akka.Remote.Serialization.Proto.Msg.ExceptionData cause_;
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public global::Akka.Remote.Serialization.Proto.Msg.ExceptionData Cause {
+      get { return cause_; }
+      set {
+        cause_ = value;
+      }
+    }
+
+    /// <summary>Field number for the "state" field.</summary>
+    public const int StateFieldNumber = 2;
+    private global::Akka.Remote.Serialization.Proto.Msg.Payload state_;
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public global::Akka.Remote.Serialization.Proto.Msg.Payload State {
+      get { return state_; }
+      set {
+        state_ = value;
+      }
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public override bool Equals(object other) {
+      return Equals(other as StatusFailure);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public bool Equals(StatusFailure other) {
+      if (ReferenceEquals(other, null)) {
+        return false;
+      }
+      if (ReferenceEquals(other, this)) {
+        return true;
+      }
+      if (!object.Equals(Cause, other.Cause)) return false;
+      if (!object.Equals(State, other.State)) return false;
+      return Equals(_unknownFields, other._unknownFields);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public override int GetHashCode() {
+      int hash = 1;
+      if (cause_ != null) hash ^= Cause.GetHashCode();
+      if (state_ != null) hash ^= State.GetHashCode();
+      if (_unknownFields != null) {
+        hash ^= _unknownFields.GetHashCode();
+      }
+      return hash;
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public override string ToString() {
+      return pb::JsonFormatter.ToDiagnosticString(this);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public void WriteTo(pb::CodedOutputStream output) {
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+      output.WriteRawMessage(this);
+    #else
+      if (cause_ != null) {
+        output.WriteRawTag(10);
+        output.WriteMessage(Cause);
+      }
+      if (state_ != null) {
+        output.WriteRawTag(18);
+        output.WriteMessage(State);
+      }
+      if (_unknownFields != null) {
+        _unknownFields.WriteTo(output);
+      }
+    #endif
+    }
+
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) {
+      if (cause_ != null) {
+        output.WriteRawTag(10);
+        output.WriteMessage(Cause);
+      }
+      if (state_ != null) {
+        output.WriteRawTag(18);
+        output.WriteMessage(State);
+      }
+      if (_unknownFields != null) {
+        _unknownFields.WriteTo(ref output);
+      }
+    }
+    #endif
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public int CalculateSize() {
+      int size = 0;
+      if (cause_ != null) {
+        size += 1 + pb::CodedOutputStream.ComputeMessageSize(Cause);
+      }
+      if (state_ != null) {
+        size += 1 + pb::CodedOutputStream.ComputeMessageSize(State);
+      }
+      if (_unknownFields != null) {
+        size += _unknownFields.CalculateSize();
+      }
+      return size;
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public void MergeFrom(StatusFailure other) {
+      if (other == null) {
+        return;
+      }
+      if (other.cause_ != null) {
+        if (cause_ == null) {
+          Cause = new global::Akka.Remote.Serialization.Proto.Msg.ExceptionData();
+        }
+        Cause.MergeFrom(other.Cause);
+      }
+      if (other.state_ != null) {
+        if (state_ == null) {
+          State = new global::Akka.Remote.Serialization.Proto.Msg.Payload();
+        }
+        State.MergeFrom(other.State);
+      }
+      _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields);
+    }
+
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    public void MergeFrom(pb::CodedInputStream input) {
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+      input.ReadRawMessage(this);
+    #else
+      uint tag;
+      while ((tag = input.ReadTag()) != 0) {
+        switch(tag) {
+          default:
+            _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input);
+            break;
+          case 10: {
+            if (cause_ == null) {
+              Cause = new global::Akka.Remote.Serialization.Proto.Msg.ExceptionData();
+            }
+            input.ReadMessage(Cause);
+            break;
+          }
+          case 18: {
+            if (state_ == null) {
+              State = new global::Akka.Remote.Serialization.Proto.Msg.Payload();
+            }
+            input.ReadMessage(State);
+            break;
+          }
+        }
+      }
+    #endif
+    }
+
+    #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE
+    [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+    [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
+    void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) {
+      uint tag;
+      while ((tag = input.ReadTag()) != 0) {
+        switch(tag) {
+          default:
+            _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input);
+            break;
+          case 10: {
+            if (cause_ == null) {
+              Cause = new global::Akka.Remote.Serialization.Proto.Msg.ExceptionData();
+            }
+            input.ReadMessage(Cause);
+            break;
+          }
+          case 18: {
+            if (state_ == null) {
+              State = new global::Akka.Remote.Serialization.Proto.Msg.Payload();
+            }
+            input.ReadMessage(State);
+            break;
+          }
+        }
+      }
+    }
+    #endif
+
+  }
+
   #endregion
 
 }
diff --git a/src/protobuf/ContainerFormats.proto b/src/protobuf/ContainerFormats.proto
index 0007ca45e4e..edef1ef90d5 100644
--- a/src/protobuf/ContainerFormats.proto
+++ b/src/protobuf/ContainerFormats.proto
@@ -70,3 +70,12 @@ message ExceptionData {
   ExceptionData innerException = 5;
   map<string, Payload> customFields = 6;
 }
+
+message StatusSuccess{
+  Payload status = 1;
+}
+
+message StatusFailure{
+  ExceptionData cause = 1;
+  Payload state = 2;
+}
\ No newline at end of file