[feat] integrate IP direct mode with desktop controls and history

2026-10-28 00:33:06 +08:00 · 2026-10-25 00:54:13 +08:00
parent 2445f79f92
commit ae6f81e83a
28 changed files with 1617 additions and 273 deletions
@@ -26,6 +26,7 @@ constexpr auto kPresenceRetryInterval = std::chrono::seconds(1);
 }  // namespace

 bool GuiRuntime::HasActiveSession() {
+  if (lan_ && !lan_->Quiescent()) return true;
  if (stream_window_inited_) return true;
  std::shared_lock lock(connection_status_mutex_);
  return std::any_of(connection_status_.begin(), connection_status_.end(),
@@ -188,7 +189,8 @@ void GuiRuntime::HandlePresenceProbeTimeout() {

 void GuiRuntime::HandleServerControllerDisconnected(
    const std::string& remote_id, const char* reason) {
-  transfers_.Cancel(peer_events_.get(), remote_id);
+  if (auto events = ControllerEvents(remote_id))
+    transfers_.Cancel(events.get(), remote_id);
  preview_subscriptions_.Remove(remote_id);

  bool has_connected_controller = false;
@@ -243,7 +245,15 @@ void GuiRuntime::HandleServerControllerDisconnected(
 #endif
  devices_.StopSpeakerCapturer();
  start_mouse_controller_ = false;
-  start_keyboard_capturer_ = false;
+  {
+    std::shared_lock lock(remote_sessions_mutex_);
+    start_keyboard_capturer_ = std::any_of(
+        remote_sessions_.begin(), remote_sessions_.end(), [](const auto& item) {
+          const auto& session = item.second;
+          return session && !session->closing_ &&
+                 session->connection_established_ && !session->wall_managed_;
+        });
+  }
  remote_client_id_.clear();
  audio_capture_ = false;
 }
@@ -294,10 +304,15 @@ int GuiRuntime::ForceSignalReconnect() {
 }

 void GuiRuntime::CloseRemoteSession(std::shared_ptr<RemoteSession> props) {
-  if (!props || props->closing_.exchange(true)) return;
+  if (!props) return;
+  if (props->lan_session_ && !props->lan_session_->closing) {
+    if (lan_) lan_->Close(props->lan_session_);
+    return;
+  }
+  if (props->closing_.exchange(true)) return;
  MuteRemoteAudio(props);
  if (props->peer_events_) {
-    props->peer_events_->Deactivate();
+    props->peer_events_->StopAccepting();
    transfers_.Cancel(props->peer_events_.get());
  }
  props->connection_established_ = false;
@@ -350,7 +365,8 @@ void GuiRuntime::CloseRemoteSession(std::shared_ptr<RemoteSession> props) {
  session_cleanup_tasks_[props->remote_id_] = session_cleanup_queue_.PostTask(
      [props, peer, frame_snapshot, video_width, video_height, platform,
       thumbnail = thumbnail_, queued_at]() mutable {
-        if (peer) {
+        if (props->peer_events_) props->peer_events_->Deactivate();
+        if (peer && !props->lan_session_) {
          LOG_INFO("[{}] Background leave connection [{}]", props->local_id_,
                   props->remote_id_);
          LeaveConnection(peer, props->remote_id_.c_str());
@@ -359,11 +375,20 @@ void GuiRuntime::CloseRemoteSession(std::shared_ptr<RemoteSession> props) {
        if (thumbnail && props->was_connected_) {
          const char* nv12 = frame_snapshot && !frame_snapshot->empty()
              ? reinterpret_cast<const char*>(frame_snapshot->data()) : nullptr;
+          const auto direct = props->lan_session_;
+          const Thumbnail::LanTarget target =
+              direct ? Thumbnail::LanTarget{direct->target.address,
+                                            direct->target.device_id,
+                                            direct->target.port}
+                     : Thumbnail::LanTarget{};
+          const auto record_id = direct
+                                     ? Thumbnail::LanRecordId(target.device_id)
+                                     : props->remote_id_;
          if (thumbnail->SaveToThumbnail(
-                  nv12, video_width, video_height, props->remote_id_,
+                  nv12, video_width, video_height, record_id,
                  props->remote_host_name_,
                  props->remember_password_ ? props->remote_password_ : "",
-                  platform) != 0) {
+                  platform, direct ? &target : nullptr) != 0) {
            LOG_ERROR("[{}] Failed to save recent connection", props->remote_id_);
          }
        }
@@ -375,6 +400,17 @@ void GuiRuntime::CloseRemoteSession(std::shared_ptr<RemoteSession> props) {
 }

 void GuiRuntime::CloseServerController(const std::string& remote_id) {
+  if (remote_id.rfind("lan/", 0) == 0) {
+    std::shared_ptr<lan::Session> session;
+    {
+      std::shared_lock lock(remote_sessions_mutex_);
+      if (auto it = lan_controllers_.find(remote_id);
+          it != lan_controllers_.end())
+        session = it->second.session;
+    }
+    if (session && lan_) lan_->Close(session);
+    return;
+  }
  if (!peer_ || remote_id.empty() || controller_cleanup_tasks_.count(remote_id))
    return;
  controller_cleanup_tasks_[remote_id] = session_cleanup_queue_.PostTask(
@@ -392,20 +428,17 @@ void GuiRuntime::CloseConnectionPeer() {
    announcements_.Reset();
    return;
  }
-  peer_events_->Deactivate();
+  peer_events_->StopAccepting();
  transfers_.Cancel(peer_events_.get());
  announcements_.Reset();
  std::vector<std::string> controllers;
  {
    std::shared_lock lock(connection_status_mutex_);
-    for (const auto& [id, status] : connection_status_) controllers.push_back(id);
+    for (const auto& [id, status] : connection_status_)
+      if (id.rfind("lan/", 0) != 0) controllers.push_back(id);
  }
  for (const auto& id : controllers)
    HandleServerControllerDisconnected(id, "peer_closed");
-  devices_.StopMouseController();
-  devices_.StopScreenCapturer();
-  devices_.StopSpeakerCapturer();
-  devices_.StopKeyboardCapturer();
  signal_connected_ = false;
  device_presence_cache_.SetSignalConnected(false);
  signal_status_ = SignalStatus::SignalClosed;
@@ -418,6 +451,7 @@ void GuiRuntime::CloseConnectionPeer() {
  // these finish before the shared peer is destroyed.
  connection_peer_cleanup_ = session_cleanup_queue_.PostTask(
      [peer, handler = peer_events_, id = std::string(client_id_)]() mutable {
+        handler->Deactivate();
        LOG_INFO("[{}] Background connection peer cleanup", id);
        LeaveConnection(peer, id.c_str());
        DestroyPeer(&peer);
@@ -551,6 +585,12 @@ std::shared_ptr<GuiRuntime::RemoteSession> GuiRuntime::FindRemoteSession(
  std::shared_lock lock(remote_sessions_mutex_);
  auto it = remote_sessions_.find(remote_id);
  if (it == remote_sessions_.end()) {
+    if (remote_id.rfind("lan-", 0) == 0)
+      for (const auto& [_, props] : remote_sessions_)
+        if (props->lan_session_ && !props->closing_ &&
+            Thumbnail::LanRecordId(props->lan_session_->target.device_id) ==
+                remote_id)
+          return props;
    return nullptr;
  }
  return it->second;
@@ -22,6 +22,7 @@ inline nlohmann::json BuildDevicePresenceRequest(
  const auto append = [&](const std::string& id) {
    // Recent connection keys contain Y/N plus display settings after the ID.
    const std::string pure_id = id.substr(0, id.find_first_of("YN"));
+    if (pure_id.rfind("lan-", 0) == 0 || pure_id.rfind("lan/", 0) == 0) return;
    if (!pure_id.empty() &&
        std::find(devices.begin(), devices.end(), pure_id) == devices.end()) {
      devices.push_back(pure_id);
@@ -70,6 +70,7 @@ GuiRuntime::~GuiRuntime() {
  windows_rdp_handoff_connected_->store(false, std::memory_order_relaxed);
  if (windows_service_worker_) windows_service_worker_->Reset();
 #endif
+  lan_.reset();
  transfers_.Shutdown();
  WaitForSessionCleanup();
 }
@@ -292,7 +293,7 @@ int GuiRuntime::CreateConnectionPeer() {
    LOG_INFO("Create peer [{}] instance failed", client_id_);
  }

-  if (0 == devices_.InitializeScreenCapturer()) {
+  if (0 == devices_.EnsureCaptureReady()) {
    const auto &displays = devices_.display_info_list();
    for (size_t index = 0; index < displays.size(); ++index) {
      const std::string stream_id = MakeDisplayStreamId(index);
@@ -334,10 +335,18 @@ void GuiRuntime::HandleRecentConnections() {
      // Record a successful connection before a preview or host info arrives.
      // The close path still updates its final thumbnail. Remember password
      // controls only credential storage, not whether this device is listed.
+      const auto direct = props->lan_session_;
+      const Thumbnail::LanTarget target =
+          direct ? Thumbnail::LanTarget{direct->target.address,
+                                        direct->target.device_id,
+                                        direct->target.port}
+                 : Thumbnail::LanTarget{};
+      const auto record_id =
+          direct ? Thumbnail::LanRecordId(target.device_id) : props->remote_id_;
      if (thumbnail_->SaveToThumbnail(
-              nullptr, 0, 0, props->remote_id_, host_name,
+              nullptr, 0, 0, record_id, host_name,
              props->remember_password_ ? props->remote_password_ : "",
-              platform) == 0)
+              platform, direct ? &target : nullptr) == 0)
        reload_recent_connections_ = true;
      else
        LOG_ERROR("[{}] Failed to save recent connection", props->remote_id_);
@@ -356,7 +365,8 @@ void GuiRuntime::HandleRecentConnections() {

      recent_connection_ids_.clear();
      for (const auto &conn : recent_connections_) {
-        recent_connection_ids_.push_back(conn.first);
+        if (conn.second.lan.device_id.empty())
+          recent_connection_ids_.push_back(conn.first);
      }
      need_to_send_recent_connections_ = true;
    }
@@ -19,6 +19,7 @@
 #include "features/input/keyboard_controller.h"
 #include "features/input/remote_input_dispatcher.h"
 #include "features/settings/settings_manager.h"
+#include "network/lan/lan_session_manager.h"
 #include "privacy_controller.h"
 #include "runtime/gui_state.h"
 #include "runtime/announcement_inbox.h"
@@ -50,6 +51,35 @@ class GuiRuntime : protected gui_detail::GuiState {
  static void SdlCaptureAudioIn(void* userdata, uint8_t* stream, int len);
  static void SdlCaptureAudioOut(void* userdata, uint8_t* stream, int len);

+  void InitializeLan();
+  void ApplyLanSettings();
+  bool LanAvailable() const;
+  void TickLan();
+  void BeginShutdown();
+  bool ShutdownDrained();
+  void RegisterStreams(PeerPtr* peer, size_t display_count);
+  void ConnectLan(const std::string& address, unsigned port,
+                  std::string password, bool remember_password = false);
+  int WithController(
+      const std::string& id,
+      const std::function<int(PeerPtr*, const std::string&)>& operation);
+  int SendControllerData(const std::string& id, const char* data, size_t size,
+                         const char* label);
+  std::shared_ptr<PeerEventHandler> ControllerEvents(const std::string& id);
+  std::vector<std::string> ConnectedControllers();
+  std::unique_ptr<lan::SessionManager> lan_;
+  std::string lan_error_;
+  std::shared_ptr<lan::Session> lan_connecting_;
+  bool lan_stopping_ = false;
+  bool stopping_ = false;
+  std::chrono::steady_clock::time_point shutdown_report_at_{};
+  struct LanController {
+    std::shared_ptr<lan::Session> session;
+    std::shared_ptr<PeerEventHandler> events;
+  };
+  // Protected by remote_sessions_mutex_; snapshots acquire callback leases
+  // before releasing that lock, so removal cannot race a raw Peer use.
+  std::map<std::string, LanController> lan_controllers_;
  int CreateConnectionPeer();
  int ConnectTo(const std::string& remote_id, const char* password,
                bool remember_password, bool bypass_presence_check = false,
@@ -0,0 +1,236 @@
+#include <display_stream_id.h>
+
+#include "rd_log.h"
+#include "runtime/gui_runtime.h"
+#if defined(__linux__) && !defined(__OHOS__)
+#include "platform/linux/headless/headless_console.h"
+#endif
+
+namespace crossdesk {
+void GuiRuntime::RegisterStreams(PeerPtr* peer, size_t displays) {
+  for (size_t i = 0; i < displays; ++i) {
+    const auto id = MakeDisplayStreamId(i);
+    AddVideoStream(peer, id.c_str());
+  }
+  AddAudioStream(peer, audio_label_.c_str());
+  AddDataStream(peer, data_label_.c_str(), false);
+  for (const auto& label :
+       {mouse_label_, keyboard_label_, control_data_label_, file_label_,
+        file_feedback_label_, clipboard_label_})
+    AddDataStream(peer, label.c_str(), true);
+}
+void GuiRuntime::InitializeLan() {
+  if (unattended_host_ || stopping_ || lan_ ||
+      !config_center_->IsIpDirectEnabled()) return;
+#if defined(__linux__) && !defined(__OHOS__)
+  if (HeadlessConsole::Instance().active()) return;
+#endif
+  lan_error_.clear();
+  lan::SessionManager::Hooks hooks;
+  hooks.error = [this](const std::string& error) {
+    if (!lan_stopping_ && !stopping_) lan_error_ = error;
+  };
+  hooks.prepare = [this](const std::shared_ptr<lan::Session>& session) {
+    if (stopping_ || lan_stopping_ || devices_.EnsureCaptureReady() != 0)
+      throw std::runtime_error("Capture initialization failed");
+    const size_t displays = devices_.display_info_list().size();
+    auto events = std::make_shared<PeerEventHandler>(*this);
+    events->BindDirect(session);
+    // Direct initialization must not depend on cloud login/configuration having
+    // successfully populated the managed peer's Params snapshot.
+    Params params{};
+    std::snprintf(params.log_path, sizeof(params.log_path), "%s",
+                  dll_log_path_.c_str());
+    params.hardware_acceleration = true;
+    params.native_video_output = true;
+    params.av1_encoding = config_center_->GetVideoEncodeFormat() ==
+                          ConfigCenter::VIDEO_ENCODE_FORMAT::AV1;
+    params.enable_srtp = true;
+    params.turn_mode = TurnMode::TurnDisabled;
+    params.video_content_type = VideoContentType::ScreenContent;
+    params.video_quality = VideoQuality::QualityHigh;
+    params.video_frame_rate = 60;
+    params.video_degradation_preference =
+        VideoDegradationPreference::MaintainResolution;
+    params.on_receive_video_frame = PeerEventHandler::OnReceiveVideoBuffer;
+    params.on_receive_audio_buffer = PeerEventHandler::OnReceiveAudioBuffer;
+    params.on_receive_data_buffer = PeerEventHandler::OnReceiveDataBuffer;
+    params.on_connection_status = PeerEventHandler::OnConnectionStatus;
+    params.on_net_status_report = PeerEventHandler::OnNetStatusReport;
+    params.user_id = session->local_id.c_str();
+    params.user_data = events.get();
+    if (!session->incoming) {
+      auto props = FindRemoteSession(session->key);
+      if (!props) throw std::runtime_error("LAN session cancelled");
+      props->peer_events_ = events;
+      props->local_id_ = session->local_id;
+    }
+    return [this, session, events, params, displays] {
+      lan::SessionManager::Prepared result;
+      result.callback_owner = events;
+      result.drain_callbacks = [events] { events->Deactivate(); };
+      result.peer = CreatePeer(&params);
+      events->SetDirectPeer(result.peer);
+      if (result.peer) RegisterStreams(result.peer, displays);
+      return result;
+    };
+  };
+  hooks.attached = [this](const std::shared_ptr<lan::Session>& session) {
+    auto events =
+        std::static_pointer_cast<PeerEventHandler>(session->callback_owner);
+    std::unique_lock lock(remote_sessions_mutex_);
+    if (session->incoming)
+      lan_controllers_[session->key] = {session, events};
+    else if (auto it = remote_sessions_.find(session->key);
+             it != remote_sessions_.end())
+      it->second->peer_ = session->peer;
+  };
+  hooks.connected = [this](const std::shared_ptr<lan::Session>& session) {
+    auto events =
+        std::static_pointer_cast<PeerEventHandler>(session->callback_owner);
+    events->ApplyConnectionStatus(ConnectionStatus::Connected, session->key);
+    events->FlushEarlyData();
+  };
+  hooks.closed = [this](const std::shared_ptr<lan::Session>& session) {
+    auto events =
+        std::static_pointer_cast<PeerEventHandler>(session->callback_owner);
+    bool attached = !session->incoming;
+    if (session->incoming) {
+      std::shared_lock lock(remote_sessions_mutex_);
+      attached = lan_controllers_.count(session->key) != 0;
+    }
+    if (events) {
+      if (attached)
+        events->ApplyConnectionStatus(ConnectionStatus::Closed, session->key);
+      events->StopAccepting();
+      transfers_.Cancel(events.get());
+    }
+    if (session->incoming) {
+      std::unique_lock lock(remote_sessions_mutex_);
+      lan_controllers_.erase(session->key);
+    } else if (auto props = FindRemoteSession(session->key))
+      CloseRemoteSession(props);
+  };
+  try {
+    lan_ = std::make_unique<lan::SessionManager>(
+        path_manager_->GetConfigPath() / "lan", std::move(hooks));
+  } catch (const std::exception& error) {
+    lan_error_ = error.what();
+    LOG_ERROR("LAN initialization failed: {}", error.what());
+  }
+}
+void GuiRuntime::ApplyLanSettings() {
+  if (config_center_->IsIpDirectEnabled()) {
+    InitializeLan();
+  } else {
+    lan_connecting_.reset();
+    lan_error_.clear();
+    if (lan_ && !lan_stopping_) {
+      lan_stopping_ = true;
+      lan_->Stop();
+    }
+  }
+}
+bool GuiRuntime::LanAvailable() const {
+  return lan_ && !lan_stopping_ && !stopping_;
+}
+void GuiRuntime::TickLan() {
+  if (lan_) lan_->Tick();
+  if (lan_ && lan_stopping_ && lan_->Quiescent()) {
+    // Drain async authentication, media preparation and callbacks before
+    // releasing the manager. A quick off/on waits here before rebinding.
+    lan_.reset();
+    lan_stopping_ = false;
+    InitializeLan();
+  }
+}
+void GuiRuntime::ConnectLan(const std::string& address, unsigned port,
+                            std::string password, bool remember_password) {
+  if (!LanAvailable() || (lan_connecting_ && !lan_connecting_->closing))
+    return;
+  try {
+    auto props = std::make_shared<RemoteSession>();
+    props->remember_password_ = remember_password;
+    if (remember_password && lan::ValidPassword(password))
+      std::memcpy(props->remote_password_, password.data(), password.size());
+    auto session = lan_->Connect(address, port, std::move(password));
+    lan_connecting_ = session;
+    props->lan_session_ = session;
+    props->remote_id_ = session->key;
+    props->connection_status_ = ConnectionStatus::Connecting;
+    props->connection_auth_.Complete();
+    {
+      std::unique_lock lock(remote_sessions_mutex_);
+      remote_sessions_[session->key] = props;
+    }
+    focused_remote_id_ = session->key;
+    show_connection_status_window_ = false;
+    lan_error_.clear();
+  } catch (const std::exception& error) {
+    lan_error_ = error.what();
+  }
+}
+std::shared_ptr<PeerEventHandler> GuiRuntime::ControllerEvents(
+    const std::string& id) {
+  std::shared_lock lock(remote_sessions_mutex_);
+  auto it = lan_controllers_.find(id);
+  return it == lan_controllers_.end()
+             ? (id.rfind("lan/", 0) == 0 ? nullptr : peer_events_)
+             : it->second.events;
+}
+int GuiRuntime::WithController(
+    const std::string& id,
+    const std::function<int(PeerPtr*, const std::string&)>& operation) {
+  std::shared_lock lock(remote_sessions_mutex_);
+  auto it = lan_controllers_.find(id);
+  auto events = it == lan_controllers_.end() ? peer_events_ : it->second.events;
+  if (it == lan_controllers_.end() && id.rfind("lan/", 0) == 0) return -1;
+  PeerPtr* peer =
+      it == lan_controllers_.end() ? peer_ : it->second.session->peer;
+  if (!peer || !events || !events->DataAllowed()) return -1;
+  auto callback = events->EnterCallback();
+  if (!callback) return -1;
+  const auto wire_id = events->WireId(id);
+  lock.unlock();
+  return operation(peer, wire_id);
+}
+int GuiRuntime::SendControllerData(const std::string& id, const char* data,
+                                   size_t size, const char* label) {
+  return WithController(id, [&](PeerPtr* peer, const std::string& wire) {
+    return SendReliableDataFrameToPeer(peer, data, size, label, wire.data(),
+                                       wire.size());
+  });
+}
+std::vector<std::string> GuiRuntime::ConnectedControllers() {
+  std::vector<std::string> ids;
+  std::shared_lock lock(connection_status_mutex_);
+  ids.reserve(connection_status_.size());
+  for (const auto& [id, status] : connection_status_)
+    if (status == ConnectionStatus::Connected) ids.push_back(id);
+  return ids;
+}
+void GuiRuntime::BeginShutdown() {
+  if (stopping_) return;
+  stopping_ = true;
+  shutdown_report_at_ =
+      std::chrono::steady_clock::now() + std::chrono::seconds(5);
+  if (lan_) lan_->Stop();
+  CloseAllRemoteSessions();
+}
+bool GuiRuntime::ShutdownDrained() {
+  TickLan();
+  HandleSessionCleanup();
+  const bool drained =
+      (!lan_ || lan_->Quiescent()) && !connection_peer_cleanup_.valid() &&
+      session_cleanup_tasks_.empty() && controller_cleanup_tasks_.empty();
+  if (!drained && std::chrono::steady_clock::now() >= shutdown_report_at_) {
+    LOG_WARN(
+        "Shutdown waiting: LAN [{}], cloud peer={}, viewers={}, controllers={}",
+        lan_ ? lan_->PendingWork() : "none", connection_peer_cleanup_.valid(),
+        session_cleanup_tasks_.size(), controller_cleanup_tasks_.size());
+    shutdown_report_at_ =
+        std::chrono::steady_clock::now() + std::chrono::seconds(5);
+  }
+  return drained;
+}
+}  // namespace crossdesk
@@ -84,17 +84,52 @@ void LogSecureDesktopInputBlocked(uint32_t *last_tick, const char *side,

 } // namespace

+bool PeerEventHandler::QueueEarlyData(const char* data, size_t size,
+                                      const char* label, size_t label_size) {
+  if (!direct_bound_) return false;
+  auto session = direct_.lock();
+  if (!session || session->closing) return true;
+  std::lock_guard lock(early_data_mutex_);
+  if (session->authorized) return false;
+  // Keep bounded reliable metadata until the manager commits Connected. Mouse,
+  // keyboard, clipboard and files never enter this pre-authorization queue.
+  if (std::string_view(label, label_size) != owner_.control_data_label_)
+    return true;
+  if (size > 8192 || early_data_.size() >= 16 ||
+      early_data_bytes_ + size > 65536) {
+    session->failed = true;
+    return true;
+  }
+  early_data_bytes_ += size;
+  early_data_.emplace_back(data, size);
+  return true;
+}
+void PeerEventHandler::FlushEarlyData() {
+  std::vector<std::string> queued;
+  {
+    std::lock_guard lock(early_data_mutex_);
+    queued.swap(early_data_);
+    early_data_bytes_ = 0;
+  }
+  const auto session = direct_.lock();
+  if (!session || !session->authorized || session->closing) return;
+  const auto& label = owner_.control_data_label_;
+  for (const auto& data : queued)
+    OnReceiveDataBuffer(data.data(), data.size(), session->remote_id.data(),
+                        session->remote_id.size(), label.data(), label.size(),
+                        this);
+}
+
 void PeerEventHandler::OnReceiveDataBuffer(
    const char *data, size_t size, const char *user_id, size_t user_id_size,
    const char *src_id, size_t src_id_size, void *user_data) {
  auto *handler = static_cast<PeerEventHandler *>(user_data);
  if (!handler) return;
  auto callback = handler->EnterCallback();
-  if (!callback) return;
-  GuiRuntime *runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime) {
+  if (!callback || handler->QueueEarlyData(data, size, src_id, src_id_size) ||
+      !handler->DataAllowed())
    return;
-  }
+  GuiRuntime* runtime = &handler->owner_;

  std::string source_id = std::string(src_id, src_id_size);
  // The service-owned host runs as SYSTEM. Do not expose filesystem or
@@ -105,11 +140,13 @@ void PeerEventHandler::OnReceiveDataBuffer(
       source_id == runtime->clipboard_label_))
    return;
  if (source_id == runtime->file_label_) {
-    runtime->transfers_.Receive(data, size, handler, std::string(user_id, user_id_size));
+    runtime->transfers_.Receive(data, size, handler,
+                                handler->ConnectionKey(user_id, user_id_size));
    return;
  } else if (source_id == runtime->clipboard_label_) {
    if (size > 0) {
-      std::string remote_user_id(user_id, user_id_size);
+      std::string remote_user_id =
+          handler->ConnectionKey(user_id, user_id_size);
      auto props =
          runtime->FindRemoteSession(remote_user_id);
      if (props && !props->clipboard_allowed_.load()) {
@@ -121,11 +158,15 @@ void PeerEventHandler::OnReceiveDataBuffer(
    }
    return;
  } else if (source_id == runtime->file_feedback_label_) {
-    runtime->transfers_.HandleAck(data, size, handler, std::string(user_id, user_id_size));
+    runtime->transfers_.HandleAck(
+        data, size, handler, handler->ConnectionKey(user_id, user_id_size));
    return;
  }

-  const std::string remote_id(user_id, user_id_size);
+  const std::string remote_id = handler->ConnectionKey(user_id, user_id_size);
+  const std::string wire_id = handler->WireId(remote_id);
+  PeerPtr* bound_peer =
+      handler->IsDirect() ? handler->DirectPeer() : runtime->peer_;
  std::string json_str(data, size);
  const auto preview_json =
      json_str.find("\"deskhub_preview") != std::string::npos
@@ -145,29 +186,31 @@ void PeerEventHandler::OnReceiveDataBuffer(
    if (!previous || subscription->IsNewerThan(*previous)) {
      if (subscription->fps > 0)
        accepted = UpdateVideoSettings(
-                     runtime->peer_, remote_id.data(), remote_id.size(),
-                     static_cast<VideoQuality>(subscription->quality),
-                     subscription->frame_rate,
-                     static_cast<VideoDegradationPreference>(
-                         subscription->EncoderPreference())) == 0;
+                       bound_peer, wire_id.data(), wire_id.size(),
+                       static_cast<VideoQuality>(subscription->quality),
+                       subscription->frame_rate,
+                       static_cast<VideoDegradationPreference>(
+                           subscription->EncoderPreference())) == 0;
      if (accepted && (!previous || previous->max_bitrate_bps.value_or(0) !=
                                       subscription->max_bitrate_bps.value_or(0)))
        accepted = UpdateVideoBitrateLimit(
-                       runtime->peer_, remote_id.data(), remote_id.size(),
+                       bound_peer, wire_id.data(), wire_id.size(),
                       subscription->max_bitrate_bps.value_or(0)) == 0;
      // Rebalancing bandwidth alone must not force every device to send a
      // keyframe. The pause API requests its own keyframes on resume.
      if (accepted && (!previous || previous->width != subscription->width ||
                       previous->height != subscription->height))
-        accepted = UpdateVideoResolution(
-                       runtime->peer_, remote_id.data(), remote_id.size(),
-                       subscription->width, subscription->height) == 0;
+        accepted = UpdateVideoResolution(bound_peer, wire_id.data(),
+                                         wire_id.size(), subscription->width,
+                                         subscription->height) == 0;
      if (accepted)
-        accepted = SetVideoSendPaused(runtime->peer_, remote_id.data(),
-                                      remote_id.size(), subscription->fps == 0) == 0;
+        accepted =
+            SetVideoSendPaused(bound_peer, wire_id.data(), wire_id.size(),
+                               subscription->fps == 0) == 0;
      if (accepted)
-        accepted = SetAudioSendPaused(runtime->peer_, remote_id.data(),
-                                      remote_id.size(), !subscription->audio) == 0;
+        accepted =
+            SetAudioSendPaused(bound_peer, wire_id.data(), wire_id.size(),
+                               !subscription->audio) == 0;
      if (accepted && subscription->audio)
        runtime->devices_.StartSpeakerCapturer();
    }
@@ -178,9 +221,9 @@ void PeerEventHandler::OnReceiveDataBuffer(
        {"request_id", subscription->request_id},
        {"accepted",
         accepted}}.dump();
-    SendReliableDataFrameToPeer(runtime->peer_, reply.data(), reply.size(),
+    SendReliableDataFrameToPeer(bound_peer, reply.data(), reply.size(),
                                runtime->control_data_label_.c_str(),
-                                remote_id.data(), remote_id.size());
+                                wire_id.data(), wire_id.size());
    return;
  }
  if (preview_json.is_object() &&
@@ -253,7 +296,7 @@ void PeerEventHandler::OnReceiveDataBuffer(
    const auto& settings = remote_action.vs;
    remote_action.vs.accepted =
        UpdateVideoSettings(
-            runtime->peer_, remote_id.data(), remote_id.size(),
+            bound_peer, wire_id.data(), wire_id.size(),
            static_cast<VideoQuality>(settings.quality), settings.frame_rate,
            static_cast<VideoDegradationPreference>(settings.preference)) == 0;
    LOG_INFO("Video settings: peer={} source={} request={} quality={} fps={} "
@@ -262,9 +305,9 @@ void PeerEventHandler::OnReceiveDataBuffer(
             settings.frame_rate, settings.preference, remote_action.vs.accepted);
    remote_action.type = ControlType::video_settings_status;
    const auto message = remote_action.to_json();
-    SendReliableDataFrameToPeer(runtime->peer_, message.data(), message.size(),
+    SendReliableDataFrameToPeer(bound_peer, message.data(), message.size(),
                                runtime->control_data_label_.c_str(),
-                                remote_id.data(), remote_id.size());
+                                wire_id.data(), wire_id.size());
    return;
  }
  if (remote_action.type == ControlType::privacy_status) {
@@ -407,8 +450,10 @@ void PeerEventHandler::OnReceiveDataBuffer(
          props->video_settings_request_tick_ = SDL_GetTicks();
          settings.vs = props->video_settings_;
          const auto message = settings.to_json();
-          SendReliableDataFrame(props->peer_, message.data(), message.size(),
-                                props->control_data_label_.c_str());
+          SendReliableDataFrame(
+              handler->IsDirect() ? handler->DirectPeer() : props->peer_,
+              message.data(), message.size(),
+              props->control_data_label_.c_str());
        }
      }
      // client mode
@@ -456,8 +501,8 @@ void PeerEventHandler::OnReceiveDataBuffer(
      // command arrives after it (e.g. when unmuting an inactive window).
      const auto subscription = runtime->preview_subscriptions_.Get(remote_id);
      const bool paused = subscription ? !subscription->audio : !remote_action.a;
-      const int ret = SetAudioSendPaused(runtime->peer_, remote_id.data(),
-                                         remote_id.size(), paused);
+      const int ret = SetAudioSendPaused(bound_peer, wire_id.data(),
+                                         wire_id.size(), paused);
      if (remote_action.a)
        runtime->devices_.StartSpeakerCapturer();
      else if (ret != 0)
@@ -49,6 +49,22 @@ constexpr HostPlatform kClientPlatform = HostPlatform::Unknown;

 }  // namespace

+std::string PeerEventHandler::ConnectionKey(const char* id, size_t size) const {
+  if (auto session = direct_.lock()) return session->key;
+  return direct_bound_ ? std::string{} : std::string(id, size);
+}
+
+std::string PeerEventHandler::WireId(const std::string& key) const {
+  if (auto session = direct_.lock()) return session->remote_id;
+  return key;
+}
+
+bool PeerEventHandler::DataAllowed() const {
+  if (!direct_bound_) return true;
+  auto session = direct_.lock();
+  return session && session->authorized && !session->closing;
+}
+
 void PeerEventHandler::SendClientInfo(PeerPtr* peer,
                                      const std::string& client_id) {
  if (!peer) {
@@ -74,8 +90,8 @@ void PeerEventHandler::OnSignalMessage(const char* message, size_t size,
  if (!handler) return;
  auto callback = handler->EnterCallback();
  if (!callback) return;
-  GuiRuntime* runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime || !message || size == 0) {
+  GuiRuntime* runtime = &handler->owner_;
+  if (!message || size == 0) {
    return;
  }
  std::string s(message, size);
@@ -179,10 +195,7 @@ void PeerEventHandler::OnSignalStatus(SignalStatus status, const char* user_id,
  if (!handler) return;
  auto callback = handler->EnterCallback();
  if (!callback) return;
-  GuiRuntime* runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime) {
-    return;
-  }
+  GuiRuntime* runtime = &handler->owner_;

  std::string client_id(user_id, user_id_size);
  if (client_id == runtime->client_id_) {
@@ -251,17 +264,30 @@ void PeerEventHandler::OnSignalStatus(SignalStatus status, const char* user_id,
 }

 void PeerEventHandler::OnConnectionStatus(ConnectionStatus status,
-                                          const char* user_id,
-                                          const size_t user_id_size,
-                                          void* user_data) {
+                                          const char* user_id, size_t size,
+                                          void* context) {
+  auto* handler = static_cast<PeerEventHandler*>(context);
+  if (!handler) return;
+  if (handler->direct_bound_) {
+    auto callback = handler->EnterCallback();
+    if (callback)
+      if (auto session = handler->direct_.lock()) session->Status(status);
+    return;
+  }
+  DispatchConnectionStatus(status, user_id, size, context);
+}
+
+void PeerEventHandler::DispatchConnectionStatus(ConnectionStatus status,
+                                                const char* user_id,
+                                                const size_t user_id_size,
+                                                void* user_data) {
  auto* handler = static_cast<PeerEventHandler*>(user_data);
  if (!handler) return;
  auto callback = handler->EnterCallback();
  if (!callback) return;
-  GuiRuntime* runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime) return;
+  GuiRuntime* runtime = &handler->owner_;

-  std::string remote_id(user_id, user_id_size);
+  std::string remote_id = handler->ConnectionKey(user_id, user_id_size);
  if (status == ConnectionStatus::Connected) runtime->transfers_.Connected(handler, remote_id);
  else if (status == ConnectionStatus::Disconnected || status == ConnectionStatus::Failed ||
           status == ConnectionStatus::Closed) runtime->transfers_.Cancel(handler, remote_id);
@@ -474,8 +500,8 @@ void PeerEventHandler::OnConnectionStatus(ConnectionStatus status,
          }
          std::string msg = remote_action.to_json();
          const int ret =
-              SendReliableDataFrame(runtime->peer_, msg.data(), msg.size(),
-                                    runtime->control_data_label_.c_str());
+              runtime->SendControllerData(remote_id, msg.data(), msg.size(),
+                                          runtime->control_data_label_.c_str());
          if (ret != 0) {
            LOG_WARN("[{}] host information send failed, ret={}", remote_id,
                     ret);
@@ -516,21 +542,12 @@ void PeerEventHandler::OnNetStatusReport(
  if (!handler) return;
  auto callback = handler->EnterCallback();
  if (!callback) return;
-  GuiRuntime* runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime) {
-    return;
-  }
+  GuiRuntime* runtime = &handler->owner_;

-  if (strchr(client_id, '@') != nullptr && strchr(user_id, '-') == nullptr) {
-    std::string id, password;
-    const char* at_pos = strchr(client_id, '@');
-    if (at_pos == nullptr) {
-      id = client_id;
-      password.clear();
-    } else {
-      id.assign(client_id, at_pos - client_id);
-      password = at_pos + 1;
-    }
+  if (const char* at_pos = handler->IsDirect() ? nullptr : strchr(client_id, '@');
+      at_pos && strchr(user_id, '-') == nullptr) {
+    const std::string id(client_id, at_pos - client_id);
+    const std::string password(at_pos + 1);

    bool is_self_hosted = runtime->config_center_->IsSelfHosted();

@@ -578,7 +595,7 @@ void PeerEventHandler::OnNetStatusReport(
 #endif
  }

-  std::string remote_id(user_id, user_id_size);
+  std::string remote_id = handler->ConnectionKey(user_id, user_id_size);
  auto props = runtime->FindRemoteSession(remote_id);
  if (!props) {
    bool report_codecs = false;
@@ -597,9 +614,8 @@ void PeerEventHandler::OnNetStatusReport(
                    net_traffic_stats->video_encoder);
      const auto message = action.to_json();
      // Report only to this controller, after releasing the status lock.
-      SendReliableDataFrameToPeer(runtime->peer_, message.data(), message.size(),
-                                  runtime->control_data_label_.c_str(),
-                                  remote_id.data(), remote_id.size());
+      runtime->SendControllerData(remote_id, message.data(), message.size(),
+                                  runtime->control_data_label_.c_str());
    }
    return;
  }
@@ -8,7 +8,10 @@
 #define _PEER_EVENT_HANDLER_H_

 #include <cstddef>
+#include <memory>
+#include <mutex>
 #include <string>
+#include <vector>

 #include "minirtc.h"
 #include "runtime/session_lifecycle.h"
@@ -16,12 +19,30 @@
 namespace crossdesk {

 class GuiRuntime;
+namespace lan {
+struct Session;
+}

 // Adapts MiniRTC C callbacks to the GUI runtime.
 class PeerEventHandler {
 public:
  explicit PeerEventHandler(GuiRuntime &owner);

+  void BindDirect(const std::shared_ptr<lan::Session>& session) {
+    direct_ = session;
+    direct_bound_ = true;
+  }
+  std::string ConnectionKey(const char* id, size_t size) const;
+  std::string WireId(const std::string& key) const;
+  bool DataAllowed() const;
+  bool IsDirect() const { return direct_bound_; }
+  void SetDirectPeer(PeerPtr* peer) { direct_peer_ = peer; }
+  PeerPtr* DirectPeer() const { return direct_peer_; }
+  void FlushEarlyData();
+  void ApplyConnectionStatus(ConnectionStatus status, const std::string& id) {
+    DispatchConnectionStatus(status, id.data(), id.size(), this);
+  }
+  void StopAccepting() { callbacks_.StopAccepting(); }
  bool IsActive() const { return callbacks_.IsActive(); }
  auto EnterCallback() { return callbacks_.Enter(); }
  void Deactivate() { callbacks_.Close(); }
@@ -51,12 +72,22 @@ public:
                                void *user_data);

 private:
-  static void SendClientInfo(PeerPtr *peer, const std::string &client_id);
+  static void DispatchConnectionStatus(ConnectionStatus status, const char* id,
+                                       size_t size, void* context);
+  static void SendClientInfo(PeerPtr* peer, const std::string& client_id);
+  bool QueueEarlyData(const char* data, size_t size, const char* label,
+                      size_t label_size);
+  std::weak_ptr<lan::Session> direct_;
+  bool direct_bound_ = false;
+  PeerPtr* direct_peer_ = nullptr;
+  std::mutex early_data_mutex_;
+  std::vector<std::string> early_data_;
+  size_t early_data_bytes_ = 0;

-  GuiRuntime &owner_;
+  GuiRuntime& owner_;
  gui_detail::SessionCallbackGate callbacks_;
 };

 } // namespace crossdesk

-#endif
+#endif
@@ -38,20 +38,21 @@ void PeerEventHandler::OnReceiveVideoBuffer(
  auto* handler = static_cast<PeerEventHandler*>(user_data);
  if (!handler) return;
  auto callback = handler->EnterCallback();
-  if (!callback) return;
-  GuiRuntime* runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime || !video_frame) {
+  if (!callback || !handler->DataAllowed()) return;
+  GuiRuntime* runtime = &handler->owner_;
+  if (!video_frame) {
    return;
  }

-  std::string remote_id(user_id, user_id_size);
+  std::string remote_id = handler->ConnectionKey(user_id, user_id_size);
  auto props = runtime->FindRemoteSession(remote_id);
  if (!props) return;

  if (props->connection_established_) {
    const auto timing_now = VideoLatencyFrame::Clock::now();
    // Pair the clocks before preview conversion so its cost is included.
-    const auto local_now_us = GetSystemTimeMicros(props->peer_);
+    const auto local_now_us = GetSystemTimeMicros(
+        handler->IsDirect() ? handler->DirectPeer() : props->peer_);
    const auto* received_native = GetNativeVideoFrame(video_frame);
    props->received_video_stats_.Record(
        received_native ? received_native->width : video_frame->width,
@@ -274,15 +275,15 @@ void PeerEventHandler::OnReceiveAudioBuffer(
  auto* handler = static_cast<PeerEventHandler*>(user_data);
  if (!handler) return;
  auto callback = handler->EnterCallback();
-  if (!callback) return;
-  GuiRuntime* runtime = handler ? &handler->owner_ : nullptr;
-  if (!runtime || runtime->unattended_host_) {
+  if (!callback || !handler->DataAllowed()) return;
+  GuiRuntime* runtime = &handler->owner_;
+  if (runtime->unattended_host_) {
    return;
  }

  std::lock_guard delivery_lock(runtime->audio_delivery_mutex_);
  const auto session =
-      runtime->FindRemoteSession(std::string(user_id, user_id_size));
+      runtime->FindRemoteSession(handler->ConnectionKey(user_id, user_id_size));
  if (!session || session->closing_ || !session->audio_allowed_.load()) return;
  runtime->audio_buffer_fresh_ = true;

@@ -15,6 +15,10 @@
 namespace crossdesk {

 bool GuiRuntime::IsAuthorizedController(const std::string& remote_id) {
+  if (remote_id.rfind("lan/", 0) == 0) {
+    const auto events = ControllerEvents(remote_id);
+    if (!events || !events->DataAllowed()) return false;
+  }
  // Connections appear here only after the existing password/signalling/RTC
  // admission path. Outgoing/viewer sessions live in remote_sessions_ instead.
  std::shared_lock lock(connection_status_mutex_);
@@ -109,9 +113,9 @@ void GuiRuntime::HandlePrivacy() {
                  "A connected controller does not support privacy screen");
    }
  }
-  if (!peer_ || (last_privacy_status_tick_ != 0 &&
-                 status.revision == last_privacy_revision_ &&
-                 now - last_privacy_status_tick_ < 1000))
+  if (last_privacy_status_tick_ != 0 &&
+       status.revision == last_privacy_revision_ &&
+       now - last_privacy_status_tick_ < 1000)
    return;
  RemoteAction action{};
  action.type = ControlType::privacy_status;
@@ -119,9 +123,8 @@ void GuiRuntime::HandlePrivacy() {
  const auto message = action.to_json();
  bool sent = true;
  for (const auto& id : controllers) {
-    const int result = SendReliableDataFrameToPeer(
-        peer_, message.data(), message.size(), control_data_label_.c_str(),
-        id.data(), id.size());
+    const int result = SendControllerData(id, message.data(), message.size(),
+                                          control_data_label_.c_str());
    if (result != 0) {
      sent = false;
      LOG_WARN("Privacy status send failed, peer={}, ret={}", id, result);
@@ -31,7 +31,12 @@
 #include "runtime/received_video_stats.h"
 #include "runtime/connection_auth_state.h"

-namespace crossdesk { class PeerEventHandler; }
+namespace crossdesk {
+class PeerEventHandler;
+namespace lan {
+struct Session;
+}
+}  // namespace crossdesk

 namespace crossdesk::gui_detail {

@@ -105,6 +110,7 @@ struct RemoteSession {
  // Decoder-thread scratch memory, never shared with the UI.
  std::mutex gallery_conversion_mutex_;
  std::vector<unsigned char> gallery_nv12_, gallery_scaled_, gallery_scratch_;
+  std::shared_ptr<lan::Session> lan_session_;
  Params params_;
  PeerPtr* peer_ = nullptr;
  std::shared_ptr<PeerEventHandler> peer_events_;
@@ -37,6 +37,11 @@ void SessionCallbackGate::EndCallback() {
  }
 }

+void SessionCallbackGate::StopAccepting() {
+  std::lock_guard lock(mutex_);
+  accepting_callbacks_ = false;
+}
+
 void SessionCallbackGate::Close() {
  std::unique_lock lock(mutex_);
  accepting_callbacks_ = false;
@@ -39,6 +39,7 @@ class SessionCallbackGate {
  CallbackGuard Enter();

  // Called by the session owner, never from within this session's callbacks.
+  void StopAccepting();
  void Close();

 private: