diff --git a/.github/workflows/ci-cppcheck.yml b/.github/workflows/ci-cppcheck.yml
index 625aa164..0cc7e06f 100644
--- a/.github/workflows/ci-cppcheck.yml
+++ b/.github/workflows/ci-cppcheck.yml
@@ -27,23 +27,23 @@ jobs:
         uses: actions/cache@v4
         with:
           path: _cppcheck
-          key: ${{ runner.os }}-cppcheck-2.19.0
-          restore-keys: ${{ runner.os }}-cppcheck-2.19.0
+          key: ${{ runner.os }}-cppcheck-2.21.0
+          restore-keys: ${{ runner.os }}-cppcheck-2.21.0
 
       - if: ${{ steps.cache-cppcheck.outputs.cache-hit != 'true' }}
         name: download and build cppcheck
         run: |
           mkdir -p _cppcheck
-          wget -O _cppcheck/cppcheck.tar.gz https://github.com/danmar/cppcheck/archive/refs/tags/2.19.0.tar.gz
+          wget -O _cppcheck/cppcheck.tar.gz https://github.com/danmar/cppcheck/archive/refs/tags/2.21.0.tar.gz
           cd _cppcheck
           tar -xzf cppcheck.tar.gz
-          cd cppcheck-2.19.0
+          cd cppcheck-2.21.0
           make -j
           pwd
           ls ./cppcheck
           ./cppcheck --version
           cd ../..
-          _cppcheck/cppcheck-2.19.0/cppcheck --version
+          _cppcheck/cppcheck-2.21.0/cppcheck --version
 
       # we need configure cmake to get the compile_commands.json file
       - name: cmake configure
@@ -71,7 +71,7 @@ jobs:
       - name: cppcheck
         run: >
           cd build &&
-          ../_cppcheck/cppcheck-2.19.0/cppcheck --project=compile_commands.json
+          ../_cppcheck/cppcheck-2.21.0/cppcheck --project=compile_commands.json
           -D__GNUC__=4
           --safety
           --error-exitcode=-1
diff --git a/AGENTS.md b/AGENTS.md
index f0949518..6ee89322 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -93,9 +93,13 @@
  To run tests, go into `<build-dir>` and run `ctest . --progress -j -E "((sodium)|(hydro)|(bcrypt)).*" --output-on-failure`.
  Or run specific test by name from `<build-dir>/tests/run/<test-name>`.
 
- To run smoke test, run `<build-dir>/aether-client-cpp-cloud`. 
+### Smoke test
+
+ The first smoke test is a `<build-dir>/aether-client-cpp-cloud`.
+ To run it simply `cd <build-dir>` and ran the `./aether-client-cpp-cloud` binary. 
  Notice! Run `aether-client-cpp-cloud` generates `state` dir there object state is saved. 
  Remove this `state` dir before run to make clean run. Keep it to run with previous state.
 
 ## Operational Rules
+
 - Do not analyze logs until everything is working fine.
diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt
index 9d579e17..2348b6f5 100644
--- a/aether/CMakeLists.txt
+++ b/aether/CMakeLists.txt
@@ -213,7 +213,8 @@ list(APPEND aether_srcs
 list(APPEND aether_srcs
             "client_messages/p2p_message_stream.cpp"
             "client_messages/p2p_message_stream_manager.cpp"
-            "client_messages/p2p_safe_message_stream.cpp")
+            "client_messages/p2p_safe_message_stream.cpp"
+            "client_messages/p2p_port_handle.cpp")
 
 list(APPEND aether_srcs
             "domain_storage/domain_storage_factory.cpp"
diff --git a/aether/aether_c/aether_capi.cpp b/aether/aether_c/aether_capi.cpp
index 7bcf1b27..fd3fbbe0 100644
--- a/aether/aether_c/aether_capi.cpp
+++ b/aether/aether_c/aether_capi.cpp
@@ -17,6 +17,7 @@
 #include "aether/aether_c/aether_capi.h"
 
 #include <cassert>
+#include <memory>
 
 #include <string_view>
 
@@ -52,9 +53,9 @@
 static ae::RcPtr<ae::AetherApp> aether_app;
 
 struct AetherClient {
-  ClientConfig config;
+  ClientConfig config{};
   ae::Client::ptr client;
-  std::map<ae::Uid, ae::RcPtr<ae::P2pStream>> streams;
+  std::map<ae::Uid, std::shared_ptr<ae::ByteIStream>> message_streams;
   std::unique_ptr<ae::ActionsQueue> actions_queue_;
 };
 
@@ -63,8 +64,9 @@ std::unique_ptr<AetherClient, void (*)(AetherClient*)> default_client{
 
 void Listen(AetherClient* client, void* user_data);
 
-void ListenStream(AetherClient* client, void* user_data, CUid const& c_dest,
-                  ae::RcPtr<ae::P2pStream> const& stream);
+void ListenMessageStream(AetherClient* client, void* user_data,
+                         CUid const& c_dest,
+                         std::shared_ptr<ae::ByteIStream> const& stream);
 
 ae::SelectClientAction& SelectClientImpl(AetherClient* client,
                                          ClientConfig const* config) {
@@ -110,16 +112,20 @@ ae::WriteAction& WriteMessageImpl(AetherClient* client, ae::Uid destination,
                                   ae::DataBuffer&& data,
                                   ActionStatusCb status_cb, void* user_data) {
   assert(client->client);
-  auto it = client->streams.find(destination);
-  if (it == std::end(client->streams)) {
+  auto it = client->message_streams.find(destination);
+  if (it == std::end(client->message_streams)) {
     // create new stream
-    auto stream =
-        client->client->message_stream_manager().CreateStream(destination);
+    auto handle =
+        client->client->message_stream_manager().CreatePort(destination);
+    auto stream = std::make_shared<ae::P2pStream>(
+        ae::AeContext{*aether_app}, client->client.Load(), destination,
+        std::move(handle));
     auto c_dest =
         CUidFromBytes(destination.value.data(), destination.value.size());
-    ListenStream(client, user_data, c_dest, stream);
+    ListenMessageStream(client, user_data, c_dest, stream);
 
-    std::tie(it, std::ignore) = client->streams.emplace(destination, stream);
+    std::tie(it, std::ignore) =
+        client->message_streams.emplace(destination, stream);
   }
 
   auto& action = it->second->Write(std::move(data));
@@ -169,21 +175,22 @@ void Listen(AetherClient* client, void* user_data) {
     return;
   }
 
-  client->client->message_stream_manager().new_stream_event().Subscribe(
-      [client, user_data](ae::RcPtr<ae::P2pStream> stream) {
-        auto dest = stream->destination();
+  client->client->message_stream_manager().new_port_event().Subscribe(
+      [client, user_data](ae::P2pPortHandle handle) {
+        auto dest = handle.destination();
+        auto stream = std::make_shared<ae::P2pStream>(
+            ae::AeContext{*aether_app}, client->client.Load(), dest,
+            std::move(handle));
         // store the stream for future use
-        client->streams.emplace(dest, stream);
-
-        // create cuid for callback
+        client->message_streams.emplace(dest, stream);
         auto c_dest = CUidFromBytes(dest.value.data(), dest.value.size());
-        // listen for incoming messages
-        ListenStream(client, user_data, c_dest, stream);
+        ListenMessageStream(client, user_data, c_dest, stream);
       });
 }
 
-void ListenStream(AetherClient* client, void* user_data, CUid const& c_dest,
-                  ae::RcPtr<ae::P2pStream> const& stream) {
+void ListenMessageStream(AetherClient* client, void* user_data,
+                         CUid const& c_dest,
+                         std::shared_ptr<ae::ByteIStream> const& stream) {
   // special context saved in heap
   struct ListenContext {
     void* user_data;
diff --git a/aether/client_messages/p2p_message_stream.cpp b/aether/client_messages/p2p_message_stream.cpp
index 4e941085..650a9470 100644
--- a/aether/client_messages/p2p_message_stream.cpp
+++ b/aether/client_messages/p2p_message_stream.cpp
@@ -144,45 +144,16 @@ class MessageSendStream final : public IStream<AeMessage, AeMessage> {
   Subscription servers_update_sub_;
   MultiSubscription streams_update_sub_;
 };
-
-class ReadMessageGate {
- public:
-  using OutDataEvent = Event<void(DataBuffer const& data)>;
-
-  ReadMessageGate(Uid uid, CloudServerConnections& cloud_connection,
-                  RequestPolicy::Variant request_policy)
-      : uid_{uid},
-        message_event_sub_{
-            ApiEventSubscriber{[this](ClientApiSafe& client_api, auto*) {
-              return client_api.send_message_event().Subscribe(
-                  [this](auto const& message) {
-                    if (message.uid == uid_) {
-                      out_data_event_.Emit(message.data);
-                    }
-                  });
-            }},
-            cloud_connection,
-            request_policy,
-        } {}
-
-  OutDataEvent::Subscriber out_data_event() { return out_data_event_; }
-
- private:
-  Uid uid_;
-  CloudEventListener message_event_sub_;
-  OutDataEvent out_data_event_;
-};
-
 }  // namespace p2p_stream_internal
 
 P2pStream::P2pStream(AeContext const& ae_context, Ptr<Client> const& client,
-                     Uid destination)
+                     Uid destination, P2pPortHandle handle)
     : ae_context_{ae_context},
       client_{client},
-      destination_{destination},
+      destination_{std::move(destination)},
+      handle_{std::move(handle)},
       buffer_write_{ae_context_, MethodPtr<&P2pStream::OnWrite>{this}} {
   AE_TELE_DEBUG(kP2pMessageStreamNew, "P2pStream created for {}", destination_);
-  // destination uid must not be empty
   assert(!destination_.empty());
 
   ConnectReceive();
@@ -203,7 +174,6 @@ P2pStream::StreamUpdateEvent::Subscriber P2pStream::stream_update_event() {
 }
 
 StreamInfo P2pStream::stream_info() const {
-  // TODO: Combine info for receive and send streams
   if (message_send_stream_) {
     return message_send_stream_->stream_info();
   }
@@ -232,14 +202,7 @@ void P2pStream::WriteOut(DataBuffer const& data) {
 Uid const& P2pStream::destination() const { return destination_; }
 
 void P2pStream::ConnectReceive() {
-  auto client_ptr = client_.Lock();
-  assert(client_ptr);
-  // TODO: config request policy
-  read_message_gate_ = std::make_unique<p2p_stream_internal::ReadMessageGate>(
-      destination_, client_ptr->cloud_connection(),
-      RequestPolicy::Replica{client_ptr->cloud_connection().max_connections()});
-  // write out received data
-  read_message_gate_->out_data_event().Subscribe(
+  out_data_sub_ = handle_.out_data_event().Subscribe(
       MethodPtr<&P2pStream::WriteOut>{this});
 }
 
@@ -256,7 +219,6 @@ void P2pStream::ConnectSend() {
           dest_cloud_conn_ = MakeDestinationCloudConn(
               cloud.Load(), client_ptr->server_connection_manager()
                                 .GetServerConnectionFactory());
-          // TODO: add config for request policy
           message_send_stream_ =
               std::make_unique<p2p_stream_internal::MessageSendStream>(
                   *dest_cloud_conn_, RequestPolicy::MainServer{});
diff --git a/aether/client_messages/p2p_message_stream.h b/aether/client_messages/p2p_message_stream.h
index 449ec815..9bb6c371 100644
--- a/aether/client_messages/p2p_message_stream.h
+++ b/aether/client_messages/p2p_message_stream.h
@@ -25,38 +25,33 @@
 
 #include "aether/write_action/buffer_write.h"
 
-#include "aether/cloud_connections/cloud_server_connections.h"
+#include "aether/client_messages/p2p_port_handle.h"
 #include "aether/connection_manager/client_cloud_manager.h"
+#include "aether/cloud_connections/cloud_server_connections.h"
 
 namespace ae {
 class Client;
 class Cloud;
 namespace p2p_stream_internal {
 class MessageSendStream;
-class ReadMessageGate;
 }  // namespace p2p_stream_internal
 
 class P2pStream final : public ByteIStream {
  public:
   P2pStream(AeContext const& ae_context, Ptr<Client> const& client,
-            Uid destination);
+            Uid destination, P2pPortHandle handle);
 
   ~P2pStream() override;
 
   AE_CLASS_NO_COPY_MOVE(P2pStream);
 
   WriteAction& Write(DataBuffer&& data) override;
-
   StreamUpdateEvent::Subscriber stream_update_event() override;
-
   StreamInfo stream_info() const override;
-
   OutDataEvent::Subscriber out_data_event() override;
-
   void Restream() override;
 
   void WriteOut(DataBuffer const& data);
-
   Uid const& destination() const;
 
  private:
@@ -72,16 +67,18 @@ class P2pStream final : public ByteIStream {
   PtrView<Client> client_;
   Uid destination_{};
 
+  P2pPortHandle handle_;
+
   // connection to destination cloud
   std::unique_ptr<CloudServerConnections> dest_cloud_conn_;
   BufferWrite<AeMessage, 100> buffer_write_;
   std::unique_ptr<p2p_stream_internal::MessageSendStream> message_send_stream_;
-  std::unique_ptr<p2p_stream_internal::ReadMessageGate> read_message_gate_;
 
   OutDataEvent out_data_event_;
   StreamUpdateEvent stream_update_event_;
 
   Subscription get_client_cloud_sub_;
+  Subscription out_data_sub_;
 };
 
 }  // namespace ae
diff --git a/aether/client_messages/p2p_message_stream_manager.cpp b/aether/client_messages/p2p_message_stream_manager.cpp
index ef55e898..a45e2552 100644
--- a/aether/client_messages/p2p_message_stream_manager.cpp
+++ b/aether/client_messages/p2p_message_stream_manager.cpp
@@ -16,12 +16,15 @@
 
 #include "aether/client_messages/p2p_message_stream_manager.h"
 
+#include <cassert>
+#include <utility>
+
 #include "aether/client.h"
 #include "aether/work_cloud_api/client_api/client_api_safe.h"
-
 #include "aether/client_messages/client_messages_tele.h"
 
 namespace ae {
+
 P2pMessageStreamManager::P2pMessageStreamManager(AeContext const& ae_context,
                                                  Ptr<Client> const& client)
     : ae_context_{ae_context},
@@ -35,70 +38,46 @@ P2pMessageStreamManager::P2pMessageStreamManager(AeContext const& ae_context,
           *cloud_connection_,
           RequestPolicy::Replica{cloud_connection_->count_connections()}}} {}
 
-RcPtr<P2pStream> P2pMessageStreamManager::CreateStream(Uid destination) {
-  CleanUpStreams();
-  auto it = streams_.find(destination);
-  if (it != std::end(streams_)) {
-    // after cleanup only used streams should stay
-    assert(it->second);
-    return it->second.lock();
-  }
-  // create new stream
-  auto stream = MakeStream(destination);
-  streams_.emplace(destination, stream);
-  message_stream_update_subs_ += stream->stream_update_event().Subscribe(
-      [this, dest{destination}]() { OnStreamUpdated(dest); });
-  return stream;
+P2pPortHandle P2pMessageStreamManager::CreatePort(Uid const& destination) {
+  auto [port, is_new] = GetOrCreatePort(destination);
+  return P2pPortHandle{std::move(port)};
 }
 
-P2pMessageStreamManager::NewStreamEvent::Subscriber
-P2pMessageStreamManager::new_stream_event() {
-  return EventSubscriber{new_stream_event_};
+P2pMessageStreamManager::NewPortEvent::Subscriber
+P2pMessageStreamManager::new_port_event() {
+  return EventSubscriber{new_port_event_};
 }
 
-void P2pMessageStreamManager::NewMessageReceived(AeMessage const& message) {
-  AE_TELED_DEBUG("New message received {}", message.uid);
-  auto it = streams_.find(message.uid);
-  // if there is no stream or stream was closed
-  if ((it == std::end(streams_)) || !it->second) {
-    auto stream = CreateStream(message.uid);
-    new_stream_event_.Emit(stream);
-    // write out first data
-    stream->WriteOut(message.data);
+std::pair<std::shared_ptr<p2p_stream_internal::P2pReceivePort>, bool>
+P2pMessageStreamManager::GetOrCreatePort(Uid const& destination) {
+  CleanUpPorts();
+  auto it = ports_.find(destination);
+  if (it != std::end(ports_)) {
+    if (auto existing = it->second.lock()) {
+      return {std::move(existing), false};
+    }
   }
+  auto port =
+      std::make_shared<p2p_stream_internal::P2pReceivePort>(destination);
+  ports_[destination] = port;
+  return {port, true};
 }
 
-void P2pMessageStreamManager::CleanUpStreams() {
-  // remove unused streams
-  for (auto it = streams_.begin(); it != streams_.end();) {
-    if (!it->second) {
-      it = streams_.erase(it);
-    } else {
-      ++it;
-    }
+void P2pMessageStreamManager::NewMessageReceived(AeMessage const& message) {
+  AE_TELED_DEBUG("New message received {}", message.uid);
+
+  auto [port, is_new] = GetOrCreatePort(message.uid);
+  assert(port != nullptr);
+
+  if (is_new) {
+    new_port_event_.Emit(P2pPortHandle{port});
   }
-}
 
-RcPtr<P2pStream> P2pMessageStreamManager::MakeStream(Uid destination) {
-  auto client_ptr = client_.Lock();
-  assert(client_ptr);
-  return MakeRcPtr<P2pStream>(ae_context_, client_ptr, destination);
+  port->Deliver(message.data);
 }
 
-void P2pMessageStreamManager::OnStreamUpdated(Uid destination) {
-  auto it = streams_.find(destination);
-  if (it == std::end(streams_)) {
-    return;
-  }
-  auto stream = it->second.lock();
-  if (!stream) {
-    return;
-  }
-  auto info = stream->stream_info();
-  // remove failed streams
-  if (info.link_state == LinkState::kLinkError) {
-    streams_.erase(it);
-  }
+void P2pMessageStreamManager::CleanUpPorts() {
+  std::erase_if(ports_, [](auto const& p) { return p.second.expired(); });
 }
 
 }  // namespace ae
diff --git a/aether/client_messages/p2p_message_stream_manager.h b/aether/client_messages/p2p_message_stream_manager.h
index 46e97d41..ee2c0415 100644
--- a/aether/client_messages/p2p_message_stream_manager.h
+++ b/aether/client_messages/p2p_message_stream_manager.h
@@ -18,44 +18,51 @@
 #define AETHER_CLIENT_MESSAGES_P2P_MESSAGE_STREAM_MANAGER_H_
 
 #include <map>
+#include <memory>
 
 #include "aether/ptr/ptr.h"
+#include "aether/ptr/ptr_view.h"
 #include "aether/types/uid.h"
 #include "aether/ae_context.h"
-#include "aether/ptr/rc_ptr.h"
-#include "aether/ptr/ptr_view.h"
 #include "aether/events/events.h"
-#include "aether/client_messages/p2p_message_stream.h"
-#include "aether/cloud_connections/cloud_server_connections.h"
 #include "aether/cloud_connections/cloud_subscription.h"
+#include "aether/client_messages/p2p_port_handle.h"
 
 namespace ae {
+
 class Client;
+
+namespace p2p_stream_internal {
+class P2pReceivePort;
+}  // namespace p2p_stream_internal
+
 class P2pMessageStreamManager {
  public:
-  using NewStreamEvent = Event<void(RcPtr<P2pStream>)>;
+  using NewPortEvent = Event<void(P2pPortHandle)>;
 
   P2pMessageStreamManager(AeContext const& ae_context,
                           Ptr<Client> const& client);
 
-  RcPtr<P2pStream> CreateStream(Uid destination);
-  NewStreamEvent::Subscriber new_stream_event();
+  P2pPortHandle CreatePort(Uid const& destination);
+  NewPortEvent::Subscriber new_port_event();
 
  private:
   void NewMessageReceived(AeMessage const& message);
-  void CleanUpStreams();
-  RcPtr<P2pStream> MakeStream(Uid destination);
+  void CleanUpPorts();
 
-  void OnStreamUpdated(Uid destination);
+  using PortsMap =
+      std::map<Uid, std::weak_ptr<p2p_stream_internal::P2pReceivePort>>;
+  std::pair<std::shared_ptr<p2p_stream_internal::P2pReceivePort>, bool>
+  GetOrCreatePort(Uid const& destination);
 
   AeContext ae_context_;
   PtrView<Client> client_;
   CloudServerConnections* cloud_connection_;
-  std::map<Uid, RcPtrView<P2pStream>> streams_;
-  NewStreamEvent new_stream_event_;
+  PortsMap ports_;
+  NewPortEvent new_port_event_;
   CloudEventListener on_message_received_sub_;
-  MultiSubscription message_stream_update_subs_;
 };
+
 }  // namespace ae
 
 #endif  // AETHER_CLIENT_MESSAGES_P2P_MESSAGE_STREAM_MANAGER_H_
diff --git a/aether/client_messages/p2p_port_handle.cpp b/aether/client_messages/p2p_port_handle.cpp
new file mode 100644
index 00000000..c3be51df
--- /dev/null
+++ b/aether/client_messages/p2p_port_handle.cpp
@@ -0,0 +1,22 @@
+#include "aether/client_messages/p2p_port_handle.h"
+
+#include <utility>
+
+namespace ae {
+namespace p2p_stream_internal {
+
+P2pReceivePort::P2pReceivePort(Uid destination)
+    : destination_{std::move(destination)} {}
+
+Uid const& P2pReceivePort::destination() const { return destination_; }
+
+P2pReceivePort::OutDataEvent::Subscriber P2pReceivePort::out_data_event() {
+  return EventSubscriber{out_data_event_};
+}
+
+void P2pReceivePort::Deliver(DataBuffer const& data) {
+  out_data_event_.Emit(data);
+}
+
+}  // namespace p2p_stream_internal
+}  // namespace ae
diff --git a/aether/client_messages/p2p_port_handle.h b/aether/client_messages/p2p_port_handle.h
new file mode 100644
index 00000000..2e88cf81
--- /dev/null
+++ b/aether/client_messages/p2p_port_handle.h
@@ -0,0 +1,66 @@
+#ifndef AETHER_CLIENT_MESSAGES_P2P_PORT_HANDLE_H_
+#define AETHER_CLIENT_MESSAGES_P2P_PORT_HANDLE_H_
+
+#include <cassert>
+#include <memory>
+
+#include "aether/common.h"
+#include "aether/types/uid.h"
+#include "aether/types/data_buffer.h"
+#include "aether/events/events.h"
+
+namespace ae {
+
+namespace p2p_stream_internal {
+
+class P2pReceivePort {
+ public:
+  using OutDataEvent = Event<void(DataBuffer const&)>;
+
+  explicit P2pReceivePort(Uid destination);
+  ~P2pReceivePort() = default;
+
+  AE_CLASS_NO_COPY_MOVE(P2pReceivePort)
+
+  Uid const& destination() const;
+  OutDataEvent::Subscriber out_data_event();
+  void Deliver(DataBuffer const& data);
+
+ private:
+  Uid destination_;
+  OutDataEvent out_data_event_;
+};
+
+}  // namespace p2p_stream_internal
+
+class P2pPortHandle {
+ public:
+  using OutDataEvent = Event<void(DataBuffer const&)>;
+
+  P2pPortHandle() = default;
+  explicit P2pPortHandle(
+      std::shared_ptr<p2p_stream_internal::P2pReceivePort> port)
+      : port_{std::move(port)} {}
+
+  AE_CLASS_MOVE_ONLY(P2pPortHandle);
+
+  void Reset() { port_.reset(); }
+  explicit operator bool() const { return port_ != nullptr; }
+
+  Uid const& destination() const {
+    assert(port_ != nullptr && "P2pPortHandle is empty — was it moved from?");
+    return port_->destination();
+  }
+
+  OutDataEvent::Subscriber out_data_event() {
+    assert(port_ != nullptr && "P2pPortHandle is empty — was it moved from?");
+    return port_->out_data_event();
+  }
+
+ private:
+  std::shared_ptr<p2p_stream_internal::P2pReceivePort> port_;
+};
+
+}  // namespace ae
+
+#endif  // AETHER_CLIENT_MESSAGES_P2P_PORT_HANDLE_H_
diff --git a/aether/client_messages/p2p_safe_message_stream.cpp b/aether/client_messages/p2p_safe_message_stream.cpp
index b5f43000..ad74164a 100644
--- a/aether/client_messages/p2p_safe_message_stream.cpp
+++ b/aether/client_messages/p2p_safe_message_stream.cpp
@@ -24,7 +24,7 @@
 namespace ae {
 P2pSafeStream::P2pSafeStream(AeContext const& ae_context,
                              SafeStreamConfig const& config,
-                             RcPtr<P2pStream> p2p_stream)
+                             std::shared_ptr<ByteIStream> p2p_stream)
     : sized_packet_gate_{},
       safe_stream_{std::make_unique<SafeStreamImpl>(ae_context, config)},
       p2p_stream_{std::move(p2p_stream)},
diff --git a/aether/client_messages/p2p_safe_message_stream.h b/aether/client_messages/p2p_safe_message_stream.h
index 75d5dc28..bad726d2 100644
--- a/aether/client_messages/p2p_safe_message_stream.h
+++ b/aether/client_messages/p2p_safe_message_stream.h
@@ -18,6 +18,7 @@
 #define AETHER_CLIENT_MESSAGES_P2P_SAFE_MESSAGE_STREAM_H_
 
 #include "aether/common.h"
+#include "aether/ae_context.h"
 #include "aether/memory.h"
 #include "aether/config.h"
 #include "aether/actions/action_context.h"
@@ -26,8 +27,6 @@
 #include "aether/stream_api/sized_packet_gate.h"
 #include "aether/safe_stream/safe_stream_config.h"
 
-#include "aether/client_messages/p2p_message_stream.h"
-
 namespace ae {
 template <std::size_t Capacity>
 class SafeStream;
@@ -37,7 +36,7 @@ class P2pSafeStream final : public ByteIStream {
   using SafeStreamImpl = SafeStream<AE_SAFE_STREAM_CAPACITY>;
 
   P2pSafeStream(AeContext const& ae_context, SafeStreamConfig const& config,
-                RcPtr<P2pStream> p2p_stream);
+                std::shared_ptr<ByteIStream> p2p_stream);
   ~P2pSafeStream() override;
 
   AE_CLASS_NO_COPY_MOVE(P2pSafeStream)
@@ -52,7 +51,7 @@ class P2pSafeStream final : public ByteIStream {
   SizedPacketGate sized_packet_gate_;
   // TODO: add config
   std::unique_ptr<SafeStreamImpl> safe_stream_;
-  RcPtr<P2pStream> p2p_stream_;
+  std::shared_ptr<ByteIStream> p2p_stream_;
   OutDataEvent out_data_event_;
   std::array<Subscription, 2> out_data_sub_;
 };
diff --git a/cppcheck_suppressions.txt b/cppcheck_suppressions.txt
index 62fcffb4..3c0f1a69 100644
--- a/cppcheck_suppressions.txt
+++ b/cppcheck_suppressions.txt
@@ -14,6 +14,7 @@ accessMoved:tests/*        // sometimes it's specially checked if var is moved
 cstyleCast:tests/*         // c-style cast is heavily used in tests
 dangerousTypeCast:tests/*  // c-style cast is heavily used in tests
 unusedStructMember:tests/test-reflect/* // there is a lot of members that looks unused
+uninitMemberVarNoCtor:tests/*
 
 constParameterCallback:capi/*
 uninitMemberVar:aligned_storage.h
@@ -23,3 +24,5 @@ uninitMemberVar:ptr_view.h
 
 noExplicitConstructor:crc.h
 noExplicitConstructor:small_function.h
+
+uninitMemberVarNoCtor:*.h
diff --git a/examples/benches/send_message_delays/receiver.cpp b/examples/benches/send_message_delays/receiver.cpp
index 826538a4..c7c915b5 100644
--- a/examples/benches/send_message_delays/receiver.cpp
+++ b/examples/benches/send_message_delays/receiver.cpp
@@ -20,6 +20,7 @@
 #include <utility>
 
 #include "aether-miscpp/meta/arg_at.h"
+#include "aether/client_messages/p2p_message_stream.h"
 #include "aether/client_messages/p2p_safe_message_stream.h"
 
 #include "send_message_delays/api/bench_delays_api.h"
@@ -38,10 +39,12 @@ void Receiver::ConnectP2pStream() {
   AE_TELED_DEBUG("Receiver::ConnectP2pStream()");
 
   message_stream_subscription_ =
-      client_->message_stream_manager().new_stream_event().Subscribe(
-          [this](RcPtr<P2pStream> message_stream) {
+      client_->message_stream_manager().new_port_event().Subscribe(
+          [this](ae::P2pPortHandle handle) {
             AE_TELED_DEBUG("Receive new connection");
-            receive_message_stream_ = std::move(message_stream);
+            auto dest = handle.destination();
+            receive_message_stream_ = std::make_shared<P2pStream>(
+                ae_context_, client_.Load(), dest, std::move(handle));
             recv_data_sub_ =
                 receive_message_stream_->out_data_event().Subscribe(
                     MethodPtr<&Receiver::OnRecvData>{this});
@@ -60,11 +63,14 @@ void Receiver::ConnectP2pSafeStream() {
         MethodPtr<&Receiver::OnRecvData>{this});
   } else {
     message_stream_subscription_ =
-        client_->message_stream_manager().new_stream_event().Subscribe(
-            [this](RcPtr<P2pStream> message_stream) {
+        client_->message_stream_manager().new_port_event().Subscribe(
+            [this](ae::P2pPortHandle handle) {
               AE_TELED_DEBUG("Receive new safe stream connection");
+              auto dest = handle.destination();
+              auto p2p_stream = std::make_shared<P2pStream>(
+                  ae_context_, client_.Load(), dest, std::move(handle));
               receive_message_safe_stream_ = make_unique<P2pSafeStream>(
-                  ae_context_, safe_stream_config_, std::move(message_stream));
+                  ae_context_, safe_stream_config_, std::move(p2p_stream));
               recv_data_sub_ =
                   receive_message_safe_stream_->out_data_event().Subscribe(
                       MethodPtr<&Receiver::OnRecvData>{this});
diff --git a/examples/benches/send_message_delays/receiver.h b/examples/benches/send_message_delays/receiver.h
index 854bcbf0..0b8aa49e 100644
--- a/examples/benches/send_message_delays/receiver.h
+++ b/examples/benches/send_message_delays/receiver.h
@@ -17,6 +17,8 @@
 #ifndef EXAMPLES_BENCHES_RECEIVER_H_
 #define EXAMPLES_BENCHES_RECEIVER_H_
 
+#include <memory>
+
 #include "aether/memory.h"
 #include "aether/client.h"
 #include "aether/ae_context.h"
@@ -56,7 +58,7 @@ class Receiver {
   ProtocolContext protocol_context_;
   BenchDelaysApi bench_delays_api_;
 
-  RcPtr<P2pStream> receive_message_stream_;
+  std::shared_ptr<ae::ByteIStream> receive_message_stream_;
   std::unique_ptr<P2pSafeStream> receive_message_safe_stream_;
   std::unique_ptr<TimedReceiver> receiver_action_;
 
diff --git a/examples/benches/send_message_delays/sender.cpp b/examples/benches/send_message_delays/sender.cpp
index 9f77c2af..9b1bcbef 100644
--- a/examples/benches/send_message_delays/sender.cpp
+++ b/examples/benches/send_message_delays/sender.cpp
@@ -20,6 +20,7 @@
 
 #include "aether/api_protocol/api_context.h"
 #include "aether/stream_api/api_call_adapter.h"
+#include "aether/client_messages/p2p_message_stream.h"
 #include "aether/client_messages/p2p_safe_message_stream.h"
 
 #include "aether/tele/tele.h"
@@ -39,22 +40,27 @@ Sender::Sender(AeContext const& ae_context, Client::ptr client,
 void Sender::ConnectP2pStream() {
   AE_TELED_DEBUG("Sender::ConnectP2pStream()");
 
-  send_message_stream_ =
-      client_->message_stream_manager().CreateStream(destination_uid_);
+  auto handle =
+      client_->message_stream_manager().CreatePort(destination_uid_);
+  send_message_stream_ = std::make_shared<P2pStream>(
+      ae_context_, client_.Load(), destination_uid_, std::move(handle));
   connected_stream_ = send_message_stream_.get();
 }
 
 void Sender::ConnectP2pSafeStream() {
   AE_TELED_DEBUG("Sender::ConnectP2pSafeStream()");
+  auto handle =
+      client_->message_stream_manager().CreatePort(destination_uid_);
+  auto p2p_stream = std::make_shared<P2pStream>(
+      ae_context_, client_.Load(), destination_uid_, std::move(handle));
   send_message_safe_stream_ = make_unique<P2pSafeStream>(
-      ae_context_, safe_stream_config_,
-      client_->message_stream_manager().CreateStream(destination_uid_));
+      ae_context_, safe_stream_config_, std::move(p2p_stream));
   connected_stream_ = send_message_safe_stream_.get();
 }
 
 void Sender::Disconnect() {
   AE_TELED_DEBUG("Sender::Disconnect()");
-  send_message_stream_.Reset();
+  send_message_stream_.reset();
   send_message_safe_stream_.reset();
 }
 
diff --git a/examples/benches/send_message_delays/sender.h b/examples/benches/send_message_delays/sender.h
index 926e33b8..c5e5b528 100644
--- a/examples/benches/send_message_delays/sender.h
+++ b/examples/benches/send_message_delays/sender.h
@@ -17,12 +17,13 @@
 #ifndef EXAMPLES_BENCHES_SEND_MESSAGE_DELAYS_SENDER_H_
 #define EXAMPLES_BENCHES_SEND_MESSAGE_DELAYS_SENDER_H_
 
+#include <memory>
+
 #include "aether/memory.h"
 #include "aether/client.h"
 #include "aether/types/uid.h"
 #include "aether/ae_context.h"
 #include "aether/events/multi_subscription.h"
-#include "aether/client_messages/p2p_message_stream.h"
 #include "aether/client_messages/p2p_safe_message_stream.h"
 
 #include "send_message_delays/timed_sender.h"
@@ -53,7 +54,7 @@ class Sender {
   Uid destination_uid_;
   SafeStreamConfig safe_stream_config_;
 
-  RcPtr<P2pStream> send_message_stream_;
+  std::shared_ptr<ae::ByteIStream> send_message_stream_;
   std::unique_ptr<P2pSafeStream> send_message_safe_stream_;
   ByteIStream* connected_stream_;
   ProtocolContext protocol_context_;
diff --git a/examples/benches/send_messages_bandwidth/receiver/receiver.cpp b/examples/benches/send_messages_bandwidth/receiver/receiver.cpp
index 4a8a1fdc..d4cdaac4 100644
--- a/examples/benches/send_messages_bandwidth/receiver/receiver.cpp
+++ b/examples/benches/send_messages_bandwidth/receiver/receiver.cpp
@@ -30,17 +30,19 @@ Receiver::Receiver(AeContext const& ae_context, Client::ptr client)
 EventSubscriber<void()> Receiver::error_event() { return error_event_; }
 
 void Receiver::Connect() {
-  client_->message_stream_manager().new_stream_event().Subscribe(
-      [this](RcPtr<P2pStream> stream) {
-        AE_TELED_DEBUG("Received message stream from {}",
-                       stream->destination());
-        message_stream_ = std::move(stream);
-        message_stream_->out_data_event().Subscribe(
-            MethodPtr<&Receiver::OnRecvData>{this});
-      });
+  message_stream_subscription_ =
+      client_->message_stream_manager().new_port_event().Subscribe(
+          [this](ae::P2pPortHandle handle) {
+            auto dest = handle.destination();
+            AE_TELED_DEBUG("Received message stream from {}", dest);
+            message_stream_ = std::make_shared<P2pStream>(
+                ae_context_, client_.Load(), dest, std::move(handle));
+            message_stream_->out_data_event().Subscribe(
+                MethodPtr<&Receiver::OnRecvData>{this});
+          });
 }
 
-void Receiver::Disconnect() { message_stream_.Reset(); }
+void Receiver::Disconnect() { message_stream_.reset(); }
 
 EventSubscriber<void()> Receiver::Handshake() {
   bandwidth_api_.handshake_event().Subscribe([this](RequestId req_id) {
diff --git a/examples/benches/send_messages_bandwidth/receiver/receiver.h b/examples/benches/send_messages_bandwidth/receiver/receiver.h
index 32f855eb..b83f2158 100644
--- a/examples/benches/send_messages_bandwidth/receiver/receiver.h
+++ b/examples/benches/send_messages_bandwidth/receiver/receiver.h
@@ -17,6 +17,8 @@
 #ifndef EXAMPLES_BENCHES_SEND_MESSAGES_BANDWIDTH_RECEIVER_RECEIVER_H_
 #define EXAMPLES_BENCHES_SEND_MESSAGES_BANDWIDTH_RECEIVER_RECEIVER_H_
 
+#include <memory>
+
 #include "aether/client.h"
 #include "aether/memory.h"
 #include "aether/ae_context.h"
@@ -51,7 +53,7 @@ class Receiver {
   ProtocolContext protocol_context_;
   BandwidthApi bandwidth_api_;
 
-  RcPtr<P2pStream> message_stream_;
+  std::shared_ptr<ae::ByteIStream> message_stream_;
 
   std::unique_ptr<MessageReceiver> message_receiver_;
 
@@ -65,6 +67,7 @@ class Receiver {
   Subscription test_start_sub_;
   Subscription test_stop_sub_;
   Subscription message_recv_sub_;
+  Subscription message_stream_subscription_;
 };
 }  // namespace ae::bench
 
diff --git a/examples/benches/send_messages_bandwidth/sender/sender.cpp b/examples/benches/send_messages_bandwidth/sender/sender.cpp
index 3d2dc3ff..9a2923c8 100644
--- a/examples/benches/send_messages_bandwidth/sender/sender.cpp
+++ b/examples/benches/send_messages_bandwidth/sender/sender.cpp
@@ -32,14 +32,16 @@ Sender::Sender(AeContext const& ae_context, Client::ptr client, Uid destination)
 EventSubscriber<void()> Sender::error_event() { return error_event_; }
 
 void Sender::Connect() {
-  message_stream_ =
-      client_->message_stream_manager().CreateStream(destination_);
+  auto handle =
+      client_->message_stream_manager().CreatePort(destination_);
+  message_stream_ = std::make_shared<P2pStream>(
+      ae_context_, client_.Load(), destination_, std::move(handle));
 
   on_recv_data_sub_ = message_stream_->out_data_event().Subscribe(
       MethodPtr<&Sender::OnRecvData>{this});
 }
 
-void Sender::Disconnect() { message_stream_.Reset(); }
+void Sender::Disconnect() { message_stream_.reset(); }
 
 EventSubscriber<void()> Sender::Handshake() {
   auto api = ApiCallAdapter{ApiContext{bandwidth_api_}, *message_stream_};
diff --git a/examples/benches/send_messages_bandwidth/sender/sender.h b/examples/benches/send_messages_bandwidth/sender/sender.h
index b7ce4416..7ac853a9 100644
--- a/examples/benches/send_messages_bandwidth/sender/sender.h
+++ b/examples/benches/send_messages_bandwidth/sender/sender.h
@@ -17,6 +17,7 @@
 #ifndef EXAMPLES_BENCHES_SEND_MESSAGES_BANDWIDTH_COMMON_SENDER_H_
 #define EXAMPLES_BENCHES_SEND_MESSAGES_BANDWIDTH_COMMON_SENDER_H_
 
+#include <memory>
 #include <optional>
 
 #include "aether/client.h"
@@ -64,7 +65,7 @@ class Sender {
   Event<void()> test_stopped_event_;
   Event<void()> error_event_;
 
-  RcPtr<P2pStream> message_stream_;
+  std::shared_ptr<ae::ByteIStream> message_stream_;
 
   std::optional<RepeatableTask<AeContext>> start_test_action_;
   std::optional<RepeatableTask<AeContext>> stop_test_action_;
diff --git a/examples/cloud/cloud_test.cpp b/examples/cloud/cloud_test.cpp
index d3d0f3ae..bf2b3ee9 100644
--- a/examples/cloud/cloud_test.cpp
+++ b/examples/cloud/cloud_test.cpp
@@ -18,6 +18,7 @@
 #include <cstdint>
 
 #include "aether/all.h"
+#include "aether/client_messages/p2p_message_stream.h"
 
 #define CLOUD_TEST_LORA_MODULE 0
 #define CLOUD_TEST_MODEM 0
@@ -101,8 +102,11 @@ int AetherCloudExample() {
    * Send confirmation to received message.
    */
   std::unique_ptr<ae::ByteIStream> receiver_stream;
-  client_a->message_stream_manager().new_stream_event().Subscribe(
-      [&](auto p2p_stream) {
+  client_a->message_stream_manager().new_port_event().Subscribe(
+      [&](ae::P2pPortHandle handle) {
+        auto dest = handle.destination();
+        auto p2p_stream = std::make_shared<ae::P2pStream>(
+            *aether_app, client_a.Load(), dest, std::move(handle));
         receiver_stream = ae::make_unique<ae::P2pSafeStream>(
             *aether_app, ae::cloud_test::kSafeStreamConfig,
             std::move(p2p_stream));
@@ -129,9 +133,12 @@ int AetherCloudExample() {
    * Create a sender to receiver stream.
    * Subscribe to receiving message event for confirmations.
    */
+  auto handle = client_b->message_stream_manager().CreatePort(client_a->uid());
+  auto p2p_stream = std::make_shared<ae::P2pStream>(
+      *aether_app, client_b.Load(), client_a->uid(), std::move(handle));
   auto sender_stream = ae::make_unique<ae::P2pSafeStream>(
       *aether_app, ae::cloud_test::kSafeStreamConfig,
-      client_b->message_stream_manager().CreateStream(client_a->uid()));
+      std::move(p2p_stream));
 
   sender_stream->out_data_event().Subscribe([&](auto const& data) {
     auto str_response =
diff --git a/opencode.json b/opencode.json
index 2e813a3a..2959c5e5 100644
--- a/opencode.json
+++ b/opencode.json
@@ -14,22 +14,23 @@
         }
       },
       "temperature": 0.1,
-      "steps": 10
+      "steps": 5
     },
     "tester": {
       "mode": "subagent",
       "description": "Run tests and analyze results",
-      "prompt": "You are highly qualified quality assurance specialist. Your task is to run unit tests and smoke tests. If unit test fails, analyze the log and point the cases are failed. If smoke test fails, analyze the log and make a report. Do not try to fix the issues yourself or provide solutions. There are other agents to fix them.",
+      "prompt": "You are highly qualified quality assurance specialist. Your task is to run unit tests and smoke tests. If any test failed immediately make a report what test case is failed and what error in the logs.",
       "permission": {
         "edit": "deny",
         "bash": {
           "*": "deny",
+          "*aether-client-cpp-cloud*": "allow",
           "ninja test": "allow",
           "ctest *": "allow"
         }
       },
       "temperature": 0.1,
-      "steps": 10
+      "steps": 5
     },
     "coder": {
       "mode": "subagent",
@@ -38,7 +39,10 @@
       "permission": {
         "edit": "allow",
         "grep": "allow",
-        "bash": "deny",
+        "bash": {
+          "*": "deny",
+          "rm*": "allow"
+        },
         "external_directory": "deny",
         "repo_clone": "deny"
       }
@@ -73,7 +77,7 @@
     "team-lead": {
       "mode": "primary",
       "description": "The main agent to rule the others on the way to work on code.",
-      "prompt": "You are team-lead. You don't write code, you don't build, you don't test. You manage team of highly qualified agents and thats all. You have @architect - to create a solution based on requirements and existent code; @coder - to write actual code by your detailed instructions; @builder_cpp to validate builds, analyze compiler errors; @tester to run tests and analyze test logs; @code-reviewer to work in pair with @coder and check if everything made as it's intended. Print plan first and wait for user approve.",
+      "prompt": "You are team-manager. You don't write code, you don't build, you don't test. You manage team of highly qualified agents and thats all. You have @architect - to create a solution based on requirements and existent code; @coder - to write actual code by your detailed instructions; @builder_cpp to validate builds, analyze compiler errors; @tester to run tests and analyze test logs; @code-reviewer to work in pair with @coder and check if everything made as it's intended. Then you got task keep one strategy: analyze request, ask @architect to build the solution, let user review it, if user approved ask @coder to implement it, ask @builder_cpp to validate it's building, ask @tester to run unit tests and then smoke tests, ask @code-reviewer to proov everything is implemented without an issues.",
       "permission": {
         "edit": "deny",
         "bash": "deny"
diff --git a/third_party/libsodium_cmake.patch b/third_party/libsodium_cmake.patch
index c1f4e3f9..7f2e0c4e 100644
--- a/third_party/libsodium_cmake.patch
+++ b/third_party/libsodium_cmake.patch
@@ -1,9 +1,9 @@
 diff --git a/CMakeLists.txt b/CMakeLists.txt
 new file mode 100644
-index 0000000..bf05fcd
+index 0000000..3d1ce59
 --- /dev/null
 +++ b/CMakeLists.txt
-@@ -0,0 +1,63 @@
+@@ -0,0 +1,62 @@
 +cmake_minimum_required(VERSION 3.16.0)
 +
 +include(CMakePackageConfigHelpers)
@@ -43,8 +43,7 @@ index 0000000..bf05fcd
 +target_compile_definitions(${PROJECT_NAME} PUBLIC SODIUM_STATIC)
 +
 +target_compile_options(${PROJECT_NAME} PRIVATE
-+  $<$<C_COMPILER_ID:Clang>:-Wno-error>
-+  $<$<C_COMPILER_ID:GNU>:-Wno-error>
++  $<$<C_COMPILER_ID:GNU,Clang>: -Wno-error -Wno-error=array-parameter  -Wno-error=unknown-pragmas>
 +)
 +
 +option(ENABLE_INSTALL "Enable installation" ON)