#include "remote_display_topology.h"

#include <algorithm>
#include <set>
#include <unordered_set>

#include <sstream>

// The composer is built on its own by a component test, which links no logging backend. So the
// lines are compiled out there and go to the project loggers in the product: this file stays
// dependency-free enough to be tested, and still says what it is doing on a real host.
#ifdef SUNSHINE_TESTS
  namespace {
    // Swallows the stream without naming a logger, so a component test can build this file
    // without a logging backend.
    struct discarded_log_t {
      template <typename T>
      discarded_log_t &operator<<(const T &) { return *this; }
    };
  }  // namespace
  #define REMOTE_MONITOR_LOG(level) discarded_log_t {}
#else
  // Only in the product build. The component test links no logging backend and carries only
  // Boost's core headers, so this include is a compile error there: boost/log/trivial.hpp lives
  // in Boost's log module, which the test does not have. It sat above the #ifdef and broke the
  // Linux build the first time the suite reached this file (2026-11-04 17:40:54).
  #include <boost/log/trivial.hpp>
  #include "logging.h"
  #define REMOTE_MONITOR_LOG(level) BOOST_LOG(level)
#endif

namespace remote_display_topology {

  std::string_view remote_monitor_suffix() { return ":remote-monitor"; }

  std::string remote_monitor_stable_id(const std::string &client_uuid) {
    if (client_uuid.empty()) {
      return {};
    }
    // A role suffix, not a second uuid: the game role keeps hashing the bare client uuid, so
    // every existing game display and its recovery journal entry stays valid, and only the
    // monitor moves to an identity of its own.
    return client_uuid + std::string {remote_monitor_suffix()};
  }
  namespace {
    constexpr int max_gap_px = 100000;

    bool contains(const std::vector<std::string> &values, const std::string &value) {
      return std::find(values.begin(), values.end(), value) != values.end();
    }

    const char *lifecycle_name(lifecycle_e lifecycle) {
      switch (lifecycle) {
        case lifecycle_e::desired: return "desired";
        case lifecycle_e::leased: return "leased";
        case lifecycle_e::applying: return "applying";
        case lifecycle_e::ready: return "ready";
        case lifecycle_e::retryable: return "retryable";
        case lifecycle_e::released: return "released";
      }
      return "retryable";
    }
  }  // namespace

  namespace {
    nlohmann::json empty_layout() { return {{"version", layout_version}, {"placements", nlohmann::json::object()}}; }
  }

  bool layout_schema_valid(const nlohmann::json &layout, std::string &error) {
    if (!layout.is_object() || !layout.contains("version") || !layout["version"].is_number_integer() || layout["version"].get<int>() != static_cast<int>(layout_version) || !layout.contains("placements") || !layout["placements"].is_object()) {
      error = "Layout must be version 1 with an object of placements.";
      return false;
    }
    const std::set<std::string> edges {"left", "right", "above", "below"};
    const std::set<std::string> alignments {"start", "center", "end"};
    for (const auto &[client, placement] : layout["placements"].items()) {
      if (client.empty() || !placement.is_object() || !placement.contains("anchor_kind") || !placement["anchor_kind"].is_string() || !placement.contains("anchor_id") || !placement["anchor_id"].is_string() || !placement.contains("edge") || !placement["edge"].is_string() || !placement.contains("alignment") || !placement["alignment"].is_string() || !placement.contains("gap_px") || !placement["gap_px"].is_number_integer()) {
        error = "Every layout placement must use the expected field types.";
        return false;
      }
      if (placement.contains("primary") && !placement["primary"].is_boolean()) {
        error = "Placement primary must be a boolean.";
        return false;
      }
      const auto gap = placement["gap_px"].get<int>();
      if ((placement["anchor_kind"].get<std::string>() != "physical" && placement["anchor_kind"].get<std::string>() != "client") || placement["anchor_id"].get<std::string>().empty() || !edges.contains(placement["edge"].get<std::string>()) || !alignments.contains(placement["alignment"].get<std::string>()) || gap < 0 || gap > max_gap_px) {
        error = "Placement has an unsupported anchor, edge, alignment, or gap.";
        return false;
      }
    }
    return true;
  }

  nlohmann::json normalize_layout(const nlohmann::json &layout) {
    std::string error;
    return layout_schema_valid(layout, error) ? layout : empty_layout();
  }

  bool validate_layout(const nlohmann::json &layout, const std::vector<std::string> &known_clients, const std::vector<std::string> &known_physical_ids, std::string &error) {
    if (!layout_schema_valid(layout, error)) return false;
    unsigned int primary_count = 0;
    std::unordered_map<std::string, std::string> client_anchors;
    const std::set<std::string> edges {"left", "right", "above", "below"};
    const std::set<std::string> alignments {"start", "center", "end"};
    for (const auto &[client, placement] : layout["placements"].items()) {
      if (!contains(known_clients, client) || !placement.is_object()) {
        error = "Every placement must name a paired client.";
        return false;
      }
      const auto anchor_kind = placement["anchor_kind"].get<std::string>();
      const auto anchor_id = placement["anchor_id"].get<std::string>();
      if (placement.value("primary", false) && ++primary_count > 1) {
        error = "Only one client display may be primary.";
        return false;
      }
      if (anchor_kind == "client") {
        if (!contains(known_clients, anchor_id) || anchor_id == client) {
          error = "A client placement cannot anchor to itself or an unknown client.";
          return false;
        }
        client_anchors.emplace(client, anchor_id);
      } else if (!contains(known_physical_ids, anchor_id)) {
        error = "A physical placement must anchor to a currently known monitor device path.";
        return false;
      }
    }
    for (const auto &[start, ignored] : client_anchors) {
      std::unordered_set<std::string> visited;
      auto current = start;
      for (auto it = client_anchors.find(current); it != client_anchors.end(); it = client_anchors.find(current)) {
        if (!visited.insert(current).second) {
          error = "Client placements must not contain an anchor cycle.";
          return false;
        }
        current = it->second;
      }
    }
    return true;
  }

  coordinator_t &instance() { static coordinator_t coordinator; return coordinator; }

  void coordinator_t::set_runtime_callbacks(runtime_callbacks_t callbacks) { std::lock_guard lock(mutex_); callbacks_ = std::move(callbacks); }
  void coordinator_t::set_layout(nlohmann::json layout) { std::lock_guard lock(mutex_); layout_ = normalize_layout(layout); }
  void coordinator_t::set_physical_baseline(std::vector<node_t> nodes) { std::lock_guard lock(mutex_); physical_baseline_ = std::move(nodes); }
  std::vector<std::string> coordinator_t::physical_node_ids() const {
    std::lock_guard lock(mutex_);
    std::vector<std::string> ids;
    for (const auto &node : physical_baseline_) {
      if (node.physical) ids.push_back(node.id);
    }
    return ids;
  }

  bool coordinator_t::normal_game_release_pending() const {
    std::lock_guard lock(mutex_);
    return std::any_of(clients_.begin(), clients_.end(), [](const auto &entry) {
      return entry.second.normal_release_pending;
    });
  }

  std::size_t coordinator_t::managed_client_identity_count() const {
    std::lock_guard lock(mutex_);
    return static_cast<std::size_t>(std::count_if(clients_.begin(), clients_.end(), [](const auto &entry) {
      return entry.second.normal_game || entry.second.remote_monitor;
    }));
  }
  std::vector<std::string> coordinator_t::managed_client_identity_ids() const {
    std::lock_guard lock(mutex_);
    std::vector<std::string> ids;
    for (const auto &[uuid, state] : clients_) {
      if (state.normal_game || state.remote_monitor) ids.push_back(uuid);
    }
    std::sort(ids.begin(), ids.end());
    return ids;
  }
  std::vector<std::string> coordinator_t::protected_remote_monitor_client_ids() const {
    std::lock_guard lock(mutex_);
    std::vector<std::string> ids;
    for (const auto &[uuid, state] : clients_) {
      if (state.remote_monitor) ids.push_back(uuid);
    }
    std::sort(ids.begin(), ids.end());
    return ids;
  }
  bool coordinator_t::generic_virtual_display_cleanup_allowed() const {
    std::lock_guard lock(mutex_);
    return std::none_of(clients_.begin(), clients_.end(), [](const auto &entry) {
      return entry.second.normal_game || entry.second.remote_monitor;
    });
  }
  void coordinator_t::set_plaintext_rtsp_warning_provider(std::function<std::string(const std::string &)> provider) { std::lock_guard lock(mutex_); plaintext_rtsp_warning_provider_ = std::move(provider); }

  normal_game_reservation_t coordinator_t::reserve_normal_game_identity(const std::string &client_uuid, const std::string &label, mode_t mode) {
    std::lock_guard lock(mutex_);
    if (client_uuid.empty() || (!clients_.contains(client_uuid) && clients_.size() >= max_client_identities)) return {};
    auto [state_it, inserted] = clients_.try_emplace(client_uuid);
    auto &state = state_it->second;
    if (inserted) state.placement_order = ++next_placement_order_;
    state.label = label;
    state.normal_requested_mode = mode;
    if (!state.remote_monitor) state.effective_mode = mode;
    if (state.normal_game && !state.normal_release_pending) return {true, false, state.normal_game_token};
    state.normal_game = true;
    state.normal_release_pending = false;
    state.normal_game_token = ++next_normal_game_token_;
    return {true, true, state.normal_game_token};
  }

  bool coordinator_t::reapply_composed_topology() {
    std::lock_guard lock(mutex_);
    if (!callbacks_.apply_composed_topology) {
      return false;
    }
    if (callbacks_.resolve_mode) {
      for (auto &[uuid, state] : clients_) {
        if (state.normal_game || state.remote_monitor) {
          resolve_effective_mode_locked(uuid, state);
        }
      }
    } else {
      for (auto &[_, state] : clients_) {
        if (state.normal_game || state.remote_monitor) state.effective_mode = desired_mode(state);
      }
    }
    if (!refresh_unverified_shared_geometry_locked()) {
      // A geometry change is still landing. Replacing the topology now would move live peers
      // onto coordinates that do not exist yet, and the compositor would refuse the lot.
      return false;
    }
    std::vector<std::string> ignored;
    return callbacks_.apply_composed_topology(compose_locked(ignored));
  }

  bool coordinator_t::rollback_normal_game_identity(const std::string &client_uuid, const std::uint64_t token) {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end() || !it->second.normal_game || it->second.normal_game_token != token) {
      return false;
    }
    if (!it->second.normal_capture_references.empty()) {
      it->second.normal_release_pending = true;
      return false;
    }
    it->second.normal_game = false;
    it->second.normal_requested_mode.reset();
    it->second.normal_game_token = 0;
    it->second.normal_release_pending = false;
    if (it->second.remote_monitor) {
      return false;
    }
    clients_.erase(it);
    return true;
  }

  void coordinator_t::release_normal_game_identity(const std::string &client_uuid, const std::uint64_t token) {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end() || !it->second.normal_game || token == 0 || it->second.normal_game_token != token) return;
    auto &state = it->second;
    state.normal_release_pending = true;
    if (!state.normal_capture_references.empty()) {
      return;
    }
    release_normal_game_identity_locked(client_uuid, state);
  }

  struct coordinator_t::capture_reference_t {
    coordinator_t &owner;
    std::string client_uuid;
    std::uint64_t token;
    bool held = false;

    capture_reference_t(coordinator_t &owner, const std::string &client_uuid, const std::uint64_t token):
        owner(owner),
        client_uuid(client_uuid),
        token(token) {}

    ~capture_reference_t() {
      if (!held) {
        return;
      }
      std::lock_guard lock(owner.mutex_);
      const auto client = owner.clients_.find(client_uuid);
      if (client == owner.clients_.end()) {
        return;
      }
      auto &references = client->second.normal_capture_references;
      const auto reference = references.find(token);
      if (reference != references.end() && --reference->second == 0) {
        references.erase(reference);
      }
    }
  };

  std::shared_ptr<void> coordinator_t::retain_normal_game_capture(const std::string &client_uuid, const std::uint64_t token) {
    auto reference = std::make_shared<capture_reference_t>(*this, client_uuid, token);
    std::lock_guard lock(mutex_);
    const auto client = clients_.find(client_uuid);
    if (client == clients_.end() || !client->second.normal_game || token == 0 || client->second.normal_game_token != token || client->second.normal_release_pending) {
      return {};
    }
    ++client->second.normal_capture_references[token];
    reference->held = true;
    return reference;
  }

  void coordinator_t::release_drained_normal_game_identities() {
    std::lock_guard lock(mutex_);
    for (auto it = clients_.begin(); it != clients_.end();) {
      auto current = it++;
      if (current->second.normal_release_pending && current->second.normal_capture_references.empty()) {
        release_normal_game_identity_locked(current->first, current->second);
      }
    }
  }

  void coordinator_t::release_normal_game_identity_locked(const std::string &client_uuid, client_state_t &state) {
    const auto requested_mode = state.normal_requested_mode;
    const auto token = state.normal_game_token;
    state.normal_game = false;
    state.normal_requested_mode.reset();
    state.normal_game_token = 0;
    if (state.remote_monitor) {
      state.normal_release_pending = false;
      return;
    }

    // Retire the output from KWin only after the remaining/saved topology is
    // active.  Disconnecting the connector first can transiently leave the
    // compositor with zero outputs and force a physical-link retrain.
    if (callbacks_.apply_composed_topology) {
      std::vector<std::string> ignored;
      if (!apply_composed_or_accept_empty_locked(ignored)) {
        state.normal_game = true;
        state.normal_requested_mode = requested_mode;
        state.normal_game_token = token;
        return;
      }
    } else {
      state.normal_game = true;
      state.normal_requested_mode = requested_mode;
      state.normal_game_token = token;
      return;
    }
    if (callbacks_.remove_owned_display && !callbacks_.remove_owned_display(client_uuid)) {
      state.normal_game = true;
      state.normal_requested_mode = requested_mode;
      state.normal_game_token = token;
      std::vector<std::string> ignored;
      (void) callbacks_.apply_composed_topology(compose_locked(ignored));
      return;
    }
    clients_.erase(client_uuid);
  }

  void coordinator_t::release_all_normal_game_identities() {
    std::vector<std::pair<std::string, std::uint64_t>> owners;
    {
      std::lock_guard lock(mutex_);
      for (const auto &[uuid, state] : clients_) {
        if (state.normal_game) owners.emplace_back(uuid, state.normal_game_token);
      }
    }
    for (const auto &[uuid, token] : owners) {
      release_normal_game_identity(uuid, token);
    }
  }

  activation_result_t coordinator_t::activate_remote_monitor(const std::string &client_uuid, const std::string &label, mode_t mode) {
    const auto result = activate_or_resume(client_uuid, label, mode, 0);
    return {result.accepted, result.ready, result.error};
  }

  activation_result_t coordinator_t::resume_remote_monitor(const std::string &client_uuid) {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end() || !it->second.remote_monitor) {
      REMOTE_MONITOR_LOG(warning) << "Remote Monitor resume refused: client '" << client_uuid << "' does not hold the monitor role.";
      return {false, false, "Remote Monitor is not owned by this paired client."};
    }
    return activate_locked(client_uuid, it->second);
  }

  monitor_runtime_state_t coordinator_t::activate_or_resume(const std::string &client_uuid, const std::string &label, mode_t mode, uint64_t generation) {
    std::lock_guard lock(mutex_);
    if (!clients_.contains(client_uuid) && clients_.size() >= max_client_identities) {
      REMOTE_MONITOR_LOG(warning) << "Remote Monitor refused: capacity reached (" << clients_.size() << "/"
                         << max_client_identities << " client identities).";
      return {false, false, true, {}, "Remote display capacity is four paired-client identities."};
    }
    auto [state_it, inserted] = clients_.try_emplace(client_uuid);
    auto &state = state_it->second;
    if (inserted) state.placement_order = ++next_placement_order_;
    if (generation < state.generation) {
      return {true, state.lifecycle == lifecycle_e::ready, state.lifecycle == lifecycle_e::retryable, state.exact_output, state.warning, state.effective_mode.hdr};
    }
    state.generation = generation;
    state.label = label;
    state.monitor_requested_mode = mode;
    state.effective_mode = mode;
    state.remote_monitor = true;
    const auto result = activate_locked(client_uuid, state);
    return {result.accepted, result.capture_ready, state.lifecycle == lifecycle_e::retryable, state.exact_output, result.warning, state.effective_mode.hdr};
  }

  monitor_runtime_state_t coordinator_t::snapshot(const std::string &client_uuid, uint64_t generation) const {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end() || generation != it->second.generation) return {};
    const auto &state = it->second;
    return {true, state.lifecycle == lifecycle_e::ready, state.lifecycle == lifecycle_e::retryable, state.exact_output, state.warning, state.effective_mode.hdr};
  }

  bool coordinator_t::is_ready(const std::string &client_uuid, uint64_t generation) const { return snapshot(client_uuid, generation).ready; }

  void coordinator_t::explicit_release(const std::string &client_uuid, uint64_t generation, const std::string &reason) {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end() || generation != it->second.generation) return;
    release_locked(client_uuid, it->second, reason);
    if (!it->second.normal_game && !it->second.remote_monitor) {
      clients_.erase(it);
    }
  }

  void coordinator_t::transport_lost(const std::string &client_uuid, uint64_t generation) {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end() || generation != it->second.generation) return;
    it->second.lifecycle = lifecycle_e::retryable;
    it->second.warning = "Transport was lost; Remote Monitor ownership and desired settings were retained for Resume.";
  }

  activation_result_t coordinator_t::activate_locked(const std::string &client_uuid, client_state_t &state) {
    if (!callbacks_.create_or_reclaim || !callbacks_.apply_composed_topology || !callbacks_.exact_target_has_current_mode_and_dxgi) {
      state.lifecycle = lifecycle_e::retryable;
      state.warning = "Remote Monitor runtime wiring is unavailable; ownership was retained for Resume.";
      return {true, false, state.warning};
    }
    // Another role's shared output may still be landing a mode it was asked for. Every node is
    // placed from these numbers, so composing a requested one that has not taken effect moves
    // live peers onto a rectangle the desktop does not have - and the compositor refuses the
    // whole topology and keeps the old one, which is exactly what the refusal looks like.
    if (!refresh_unverified_shared_geometry_locked(client_uuid)) {
      state.lifecycle = lifecycle_e::retryable;
      state.warning = "A shared display mode is awaiting verification; topology changes are deferred.";
      return {true, false, state.warning};
    }
    // A lease taken on this call is a connector nothing else is capturing: its mode is the
    // pool's default rather than a promise, and the apply below is what sets it. Verifying it
    // beforehand can only ever disagree, and returning a 503 to a client that asks once per
    // launch turns that disagreement into a Remote Monitor that never opens. A lease that was
    // already held is the opposite case - a live peer we must not move onto unverified numbers.
    bool took_lease_now = false;
    if (!state.lease_held) {
      state.lifecycle = lifecycle_e::leased;
      if (!callbacks_.create_or_reclaim(client_uuid, state.label, desired_mode(state))) {
        const auto want = desired_mode(state);
        REMOTE_MONITOR_LOG(warning) << "Remote monitor activation: create_or_reclaim failed for client '" << client_uuid
                           << "' (label '" << state.label << "') mode=" << want.width << "x" << want.height << "x"
                           << want.refresh_hz << "; ownership retained for Resume.";
        state.lifecycle = lifecycle_e::retryable;
        state.warning = "The owned virtual display could not be created or reclaimed; Resume will retry the same identity.";
        return {true, false, state.warning};
      }
      state.lease_held = true;
      took_lease_now = true;
    }
    resolve_effective_mode_locked(client_uuid, state);
    // A retained lease is a live peer: its geometry must be observed before anything is placed
    // against it. A lease taken on this call is the only case that skips the check, because its
    // mode is the pool default until the apply below sets it.
    if (callbacks_.observe_shared_mode && !took_lease_now) {
      // A successful create/reclaim does not prove the geometry took effect, and a role that
      // owns a live output can be asked for more than its panel or driver granted.
      const auto observed = callbacks_.observe_shared_mode(client_uuid, state.remote_monitor);
      const bool disagrees = observed && (observed->width != state.effective_mode.width ||
                                          observed->height != state.effective_mode.height ||
                                          observed->refresh_hz != state.effective_mode.refresh_hz);
      if (disagrees && state.shared_mode_deferrals < max_shared_geometry_deferrals) {
        ++state.shared_mode_deferrals;
        REMOTE_MONITOR_LOG(warning) << "Remote monitor activation: shared geometry for client '" << client_uuid
                           << "' is " << observed->width << "x" << observed->height << "x" << observed->refresh_hz
                           << " but " << state.effective_mode.width << "x" << state.effective_mode.height << "x"
                           << state.effective_mode.refresh_hz << " was requested; deferring the topology.";
        state.effective_mode.width = observed->width;
        state.effective_mode.height = observed->height;
        state.effective_mode.refresh_hz = observed->refresh_hz;
        state.shared_mode_unverified = true;
        state.lifecycle = lifecycle_e::retryable;
        state.warning = "The shared display does not yet match the requested geometry; Resume will retry.";
        return {true, false, state.warning};
      }
      if (!disagrees) {
        // Unknown geometry is not a disagreement: the platform simply could not read it, and
        // refusing to stream because a query failed would be worse than the old behaviour.
        state.shared_mode_unverified = false;
        state.shared_mode_deferrals = 0;
      }
    }
    state.lifecycle = lifecycle_e::applying;
    std::vector<std::string> warnings;
    const auto composed = compose_locked(warnings);
    if (!callbacks_.apply_composed_topology(composed)) {
      REMOTE_MONITOR_LOG(warning) << "Remote monitor activation: composed topology did not apply for client '" << client_uuid
                         << "' (" << composed.size() << " node(s)); ownership retained for Resume.";
      state.shared_mode_unverified = true;
      state.lifecycle = lifecycle_e::retryable;
      state.warning = "The composed display topology did not apply; existing owners and their displays were retained.";
      return {true, false, state.warning};
    }
    if (!warnings.empty()) state.warning = warnings.front();
    const auto exact_output = callbacks_.exact_target_has_current_mode_and_dxgi(client_uuid, state.effective_mode);
    if (!exact_output || exact_output->empty()) {
      REMOTE_MONITOR_LOG(warning) << "Remote monitor activation: no exact capture output for client '" << client_uuid
                         << "' mode=" << state.effective_mode.width << "x" << state.effective_mode.height << "x"
                         << state.effective_mode.refresh_hz << "; ownership retained for Resume.";
      // The output is not capture-ready yet, which is the same obligation as an unverified
      // geometry: do not let the next composition assume the mode landed.
      state.shared_mode_unverified = true;
      state.lifecycle = lifecycle_e::retryable;
      state.warning = "The requested client display is not yet capture-ready. Existing streamed physical or virtual displays remain topology anchors, but capture will wait rather than mirror another screen.";
      return {true, false, state.warning};
    }
    state.exact_output = *exact_output;
    state.shared_mode_unverified = false;
    state.shared_mode_deferrals = 0;
    state.lifecycle = lifecycle_e::ready;
    state.warning.clear();
    return {true, true, {}};
  }

  void coordinator_t::note_lease_lost(const std::string &client_uuid) {
    std::lock_guard lock(mutex_);
    if (auto it = clients_.find(client_uuid); it != clients_.end()) {
      it->second.lease_held = false;
      it->second.lifecycle = lifecycle_e::retryable;
      it->second.warning = "Remote Monitor lease was lost; desired settings and identity are retained for Resume.";
    }
  }

  void coordinator_t::release_abandoned_remote_monitors(const std::string &reason) {
    std::lock_guard lock(mutex_);
    for (auto it = clients_.begin(); it != clients_.end();) {
      // ready means a live transport is still capturing this role: a monitor someone is
      // actually watching is never abandoned. Everything else - retryable after its transport
      // went away, leased or applying with no session left - is a display kept only for a
      // resume that cannot happen now that the app has ended.
      if (!it->second.remote_monitor || it->second.lifecycle == lifecycle_e::ready) {
        ++it;
        continue;
      }
      const auto released_uuid = it->first;
      REMOTE_MONITOR_LOG(info) << "Releasing Remote Monitor role for client '" << released_uuid << "' (" << reason << ").";
      release_locked(released_uuid, it->second, reason);
      // The composition no longer names this client and an empty compose is accepted without
      // asking Windows for anything, so nothing else will retire this output. Whoever releases a
      // role on the app's last breath removes the display it owned, right here.
      if (callbacks_.remove_owned_display) {
        callbacks_.remove_owned_display(released_uuid);
      }
      if (!it->second.normal_game && !it->second.remote_monitor) {
        it = clients_.erase(it);
      } else {
        ++it;
      }
    }
  }

  void coordinator_t::disconnect_monitor(const std::string &client_uuid) {
    std::lock_guard lock(mutex_);
    const auto it = clients_.find(client_uuid);
    if (it == clients_.end()) return;
    release_locked(client_uuid, it->second, "Remote Monitor was disconnected.");
    if (!it->second.normal_game && !it->second.remote_monitor) {
      clients_.erase(it);
    }
  }

  void coordinator_t::unpair_client(const std::string &client_uuid) {
    // Unpair releases Remote Monitor ownership, but it cannot discard a normal
    // game's still-live ownership record. The process teardown releases that
    // role with its reservation token.
    disconnect_monitor(client_uuid);
  }

  void coordinator_t::shutdown(const bool preserve_owned_displays) {
    std::lock_guard lock(mutex_);
    if (preserve_owned_displays) {
      // A supervised host hands the still-connected machine display to its
      // successor.  Process teardown must not issue KScreen or hotplug work.
      clients_.clear();
      return;
    }
    std::vector<std::string> managed_ids;
    for (const auto &[uuid, state] : clients_) {
      if (state.normal_game || state.remote_monitor) managed_ids.push_back(uuid);
    }
    std::sort(managed_ids.begin(), managed_ids.end());
    clients_.clear();
    if (callbacks_.apply_composed_topology) {
      std::vector<std::string> ignored;
      (void) callbacks_.apply_composed_topology(compose_locked(ignored));
    }
    for (const auto &uuid : managed_ids) {
      if (callbacks_.remove_owned_display) {
        (void) callbacks_.remove_owned_display(uuid);
      }
    }
  }

  void coordinator_t::release_locked(const std::string &client_uuid, client_state_t &state, const std::string &reason) {
    if (!state.remote_monitor) return;
    const auto monitor_mode = state.monitor_requested_mode;
    const auto previous_lease = state.lease_held;
    const auto previous_lifecycle = state.lifecycle;
    const auto previous_warning = state.warning;
    /*
     * Each role owns its own VDD identity - the monitor's display is created under
     * "<uuid>:remote-monitor", the game's under the bare uuid - so ending the monitor role must
     * retire the monitor's display whether or not the same client's game role is still streaming.
     * The old rule ("ending either role cannot remove the other's display") was true while the two
     * roles shared one connector. Since the split it retires nothing and leaves the monitor's
     * display behind, which is the ghost that makes every later setDisplayConfig call fail.
     */
    const bool remove_display = callbacks_.remove_owned_display != nullptr;
    bool display_removed = false;
    state.remote_monitor = false;
    state.monitor_requested_mode.reset();
    if (state.normal_game) {
      resolve_effective_mode_locked(client_uuid, state);
    }
    // The normal-game role still owns the shared connector. Keep the lease
    // truthfully held so a later monitor resume cannot appear to reacquire or
    // disconnect it out from under the game.
    state.lease_held = state.normal_game;
    state.lifecycle = lifecycle_e::released;
    state.warning = reason;

    // Recompose only the remaining explicit owners.  This intentionally does
    // not restore a saved/global topology or remove any peer identity.
    if (callbacks_.apply_composed_topology) {
      std::vector<std::string> ignored;
      bool applied = apply_composed_or_accept_empty_locked(ignored);
      if (!applied && remove_display) {
        /*
         * A departing display whose own session has ended can still sit in Windows' database as an
         * active path with no target mode. setDisplayConfig then rejects the request during
         * validation with ERROR_INVALID_PARAMETER before it looks at it at all (2026-11-04
         * 15:20:34: one requested node, game display on DISPLAY15, monitor display left active on
         * DISPLAY16 with target_mode absent), so shrinking the topology first can never succeed
         * while that ghost exists. Retire the display this role owns - the other role's display has
         * a different identity and is not touched - and ask once more: with the ghost gone the
         * request is valid and the handoff completes.
         */
        display_removed = callbacks_.remove_owned_display(client_uuid);
        applied = display_removed && apply_composed_or_accept_empty_locked(ignored);
      }
      if (!applied) {
        state.remote_monitor = true;
        state.monitor_requested_mode = monitor_mode;
        state.lease_held = previous_lease;
        state.lifecycle = previous_lifecycle;
        state.warning = previous_warning;
        resolve_effective_mode_locked(client_uuid, state);
        return;
      }
    } else if (remove_display) {
      state.remote_monitor = true;
      state.monitor_requested_mode = monitor_mode;
      state.lease_held = previous_lease;
      state.lifecycle = previous_lifecycle;
      state.warning = previous_warning;
      resolve_effective_mode_locked(client_uuid, state);
      return;
    }
    if (remove_display && !display_removed && !callbacks_.remove_owned_display(client_uuid)) {
      state.remote_monitor = true;
      state.monitor_requested_mode = monitor_mode;
      state.lease_held = previous_lease;
      state.lifecycle = previous_lifecycle;
      state.warning = previous_warning;
      resolve_effective_mode_locked(client_uuid, state);
      std::vector<std::string> ignored;
      (void) callbacks_.apply_composed_topology(compose_locked(ignored));
    }
  }

  mode_t coordinator_t::effective_mode(const node_t &node) { return node.current_mode.value_or(node.last_requested_mode.value_or(node.configured_mode)); }

  int coordinator_t::layout_width(const node_t &node) {
    return node.layout_width.value_or(effective_mode(node).width);
  }

  int coordinator_t::layout_height(const node_t &node) {
    return node.layout_height.value_or(effective_mode(node).height);
  }

  mode_t coordinator_t::desired_mode(const client_state_t &state) {
    if (state.remote_monitor && state.monitor_requested_mode) return *state.monitor_requested_mode;
    if (state.normal_game && state.normal_requested_mode) return *state.normal_requested_mode;
    return {};
  }

  void coordinator_t::resolve_effective_mode_locked(const std::string &client_uuid, client_state_t &state) {
    state.effective_mode = desired_mode(state);
    if (callbacks_.resolve_mode) callbacks_.resolve_mode(client_uuid, state.effective_mode);
  }

  bool coordinator_t::refresh_unverified_shared_geometry_locked(const std::string &except) {
    if (!callbacks_.observe_shared_mode) {
      return true;
    }
    for (auto &[uuid, state] : clients_) {
      if (uuid == except || (!state.normal_game && !state.remote_monitor)) continue;
      if (!state.shared_mode_unverified) continue;
      const auto observed = callbacks_.observe_shared_mode(uuid, state.remote_monitor);
      if (!observed || observed->width <= 0 || observed->height <= 0 || observed->refresh_hz <= 0) {
        // The output exists but its geometry is not readable yet. Composing the requested
        // numbers here is what made the compositor refuse and retain the old topology.
        return false;
      }
      state.effective_mode.width = observed->width;
      state.effective_mode.height = observed->height;
      state.effective_mode.refresh_hz = observed->refresh_hz;
    }
    return true;
  }

  bool coordinator_t::apply_composed_or_accept_empty_locked(std::vector<std::string> &warnings) {
    const auto composed = compose_locked(warnings);
    const bool nothing_to_place = composed.empty();
    const bool applied = callbacks_.apply_composed_topology ? callbacks_.apply_composed_topology(composed) : false;
    if (!applied && nothing_to_place) {
      // No role wants a display: there is no topology to apply, and asking for one would be a
      // request to place nothing rather than a request that failed. Reported as a refusal, the
      // last release rolled its identity back (release_locked and release_normal_game_identity_
      // locked both undo on a false return), which kept normal_game set for the life of the
      // process: generic_virtual_display_cleanup_allowed() then answered false forever, every
      // teardown deferred, and the physical outputs were never restored - the black KVM after a
      // session ended (2026-11-04 16:24:36: three deferrals in thirty-one milliseconds, with the
      // roster already reading game=false monitor=false).
      REMOTE_MONITOR_LOG(info) << "Remote monitor composition: no role wants a display; accepting the refusal.";
      return true;
    }
    return applied;
  }

  std::vector<node_t> coordinator_t::compose_locked(std::vector<std::string> &warnings) const {
    // Name the roster every time a topology is built. A compose that silently carries one client
    // fewer than the caller expected is indistinguishable from one that was correct, and the log
    // of a refusal then reads like a clean run - which is exactly how a missing node stayed
    // invisible for a night.
    for (const auto &[uuid, state] : clients_) {
      REMOTE_MONITOR_LOG(info) << "Remote monitor compose roster: client '" << uuid << "' game=" << state.normal_game
                      << " monitor=" << state.remote_monitor << " lifecycle=" << static_cast<int>(state.lifecycle)
                      << " mode=" << state.effective_mode.width << "x" << state.effective_mode.height << "x"
                      << state.effective_mode.refresh_hz;
    }
    auto nodes = physical_baseline_;
    int rightmost = 0;
    for (const auto &node : nodes) rightmost = std::max(rightmost, node.x + layout_width(node));

    std::vector<std::string> active_ids;
    for (const auto &[uuid, state] : clients_) {
      if (state.remote_monitor || state.normal_game) active_ids.push_back(uuid);
    }
    std::sort(active_ids.begin(), active_ids.end(), [&](const std::string &lhs, const std::string &rhs) {
      const auto lhs_order = clients_.at(lhs).placement_order;
      const auto rhs_order = clients_.at(rhs).placement_order;
      return lhs_order == rhs_order ? lhs < rhs : lhs_order < rhs_order;
    });

    std::unordered_set<std::string> emitted;
    std::unordered_set<std::string> visiting;
    std::function<void(const std::string &)> emit = [&](const std::string &uuid) {
      if (emitted.contains(uuid)) return;
      const auto state_it = clients_.find(uuid);
      if (state_it == clients_.end() || (!state_it->second.remote_monitor && !state_it->second.normal_game)) return;

      /*
       * Per role, not per client: effective_mode is a single field and the monitor's activation
       * overwrites it, so the game node was wearing the monitor's mode. Each role's own request is
       * what belongs on its node.
       */
      const auto role_mode = state_it->second.remote_monitor && state_it->second.normal_game
                                 ? state_it->second.normal_requested_mode.value_or(state_it->second.effective_mode)
                                 : state_it->second.effective_mode;
      /*
       * The monitor role's display is created under "<uuid>:remote-monitor", never the bare uuid,
       * so a client holding only the monitor role must carry that identity on its node or the
       * compose step looks the display up under an identity nothing was ever created with. Before
       * the per-role identity existed the monitor borrowed the client's display, which is why a
       * second device used to work and stopped once the role got its own display.
       */
      const bool monitor_only = state_it->second.remote_monitor && !state_it->second.normal_game;
      node_t node {
        .id = monitor_only ? remote_monitor_stable_id(uuid) : uuid,
        .label = state_it->second.label,
        .active = true,
        .configured_mode = role_mode,
        .last_requested_mode = role_mode,
      };
      const auto append_right = [&](const std::string &warning) {
        node.x = rightmost;
        rightmost += layout_width(node);
        if (!warning.empty()) warnings.push_back(warning);
        nodes.push_back(std::move(node));
        emitted.insert(uuid);
      };

      if (!visiting.insert(uuid).second) {
        append_right("Saved client anchors contain a cycle; '" + uuid + "' was appended to the right for this activation.");
        return;
      }

      const auto placement_it = layout_["placements"].find(uuid);
      if (placement_it == layout_["placements"].end()) {
        append_right({});
        visiting.erase(uuid);
        return;
      }

      const auto &placement = *placement_it;
      node.primary = placement.value("primary", false);
      const auto anchor_id = placement.value("anchor_id", "");
      if (placement.value("anchor_kind", "") == "client") {
        emit(anchor_id);
        if (emitted.contains(uuid)) {
          visiting.erase(uuid);
          return;
        }
      }

      // Saved anchors are client uuids, but a monitor-only node carries the role suffix, so match
      // the anchor on either form or an existing placement stops resolving.
      const auto anchor = std::find_if(nodes.begin(), nodes.end(), [&](const node_t &candidate) {
        return candidate.id == anchor_id || candidate.id == remote_monitor_stable_id(anchor_id);
      });
      if (anchor == nodes.end()) {
        append_right("Saved anchor '" + anchor_id + "' is unavailable; the client display was appended to the right for this activation.");
        visiting.erase(uuid);
        return;
      }

      const auto anchor_width = layout_width(*anchor);
      const auto anchor_height = layout_height(*anchor);
      const auto width = layout_width(node);
      const auto height = layout_height(node);
      const auto gap = placement.value("gap_px", 0);
      const auto edge = placement.value("edge", "right");
      const auto alignment = placement.value("alignment", "center");
      if (edge == "left") node.x = anchor->x - width - gap;
      if (edge == "right") node.x = anchor->x + anchor_width + gap;
      if (edge == "above") node.y = anchor->y - height - gap;
      if (edge == "below") node.y = anchor->y + anchor_height + gap;
      if (edge == "left" || edge == "right") node.y = alignment == "start" ? anchor->y : alignment == "end" ? anchor->y + anchor_height - height : anchor->y + (anchor_height - height) / 2;
      if (edge == "above" || edge == "below") node.x = alignment == "start" ? anchor->x : alignment == "end" ? anchor->x + anchor_width - width : anchor->x + (anchor_width - width) / 2;
      rightmost = std::max(rightmost, node.x + width);
      nodes.push_back(std::move(node));
      emitted.insert(uuid);
      visiting.erase(uuid);
    };

    for (const auto &uuid : active_ids) emit(uuid);


    // A paired client can hold a game role and a Remote Monitor role at once, and each role owns
    // a separate display. The list above carries one node per client, so the monitor's display
    // had no place in the model at all - the composer kept the game's node and the capture was
    // refused ("capture will wait rather than mirror another screen") instead of being aimed at
    // the wrong screen. Give the monitor its own node, on its own stable identity, beside its
    // owner's. Two-device setups are untouched by construction: each client there holds a single
    // role, so this loop never runs and the node list is byte-identical to before.
    for (const auto &uuid : active_ids) {
      const auto state_it = clients_.find(uuid);
      if (state_it == clients_.end()) continue;
      const auto &state = state_it->second;
      if (!state.remote_monitor || !state.normal_game) continue;

      const auto monitor_mode = state.monitor_requested_mode.value_or(state.effective_mode);
      node_t monitor {
        .id = remote_monitor_stable_id(uuid),
        .label = state.label,
        .active = true,
        .configured_mode = monitor_mode,
        .last_requested_mode = monitor_mode,
      };
      monitor.x = rightmost;
      rightmost += layout_width(monitor);
      nodes.push_back(std::move(monitor));
      warnings.push_back("Client '" + uuid + "' owns both a game display and a Remote Monitor display; the monitor was placed to the right for this activation.");
    }

    // One place that says what the topology will actually ask Windows for. A refusal is otherwise
    // read against the roster and the node list guessed at, and a node that never got emitted looks
    // exactly like a node that was emitted and rejected.
    REMOTE_MONITOR_LOG(info) << "Remote monitor compose result: " << nodes.size() << " node(s)";
    for (const auto &node : nodes) {
      REMOTE_MONITOR_LOG(info) << "Remote monitor compose result: node id='" << node.id << "' label='" << node.label
                      << "' device_id='" << node.device_id << "' physical=" << node.physical
                      << " preexisting=" << node.preexisting << " mode=" << node.configured_mode.width << "x"
                      << node.configured_mode.height << "x" << node.configured_mode.refresh_hz << " at " << node.x
                      << "," << node.y << (node.primary ? " primary" : "");
    }

    return nodes;
  }

  nlohmann::json coordinator_t::snapshot(const std::vector<nlohmann::json> &paired_clients) const {
    std::lock_guard lock(mutex_);
    std::vector<std::string> warnings;
    const auto nodes = compose_locked(warnings);
    nlohmann::json result {{"status", true}, {"version", layout_version}, {"layout", layout_}, {"capacity", {{"max", max_client_identities}, {"used", clients_.size()}}}, {"warnings", warnings}, {"nodes", nlohmann::json::array()}, {"clients", paired_clients}};
    for (const auto &node : nodes) {
      const auto mode = effective_mode(node);
      result["nodes"].push_back({{"id", node.id}, {"label", node.label}, {"kind", node.physical ? "physical" : "client"}, {"active", node.active}, {"primary", node.primary}, {"desired_position", {{"x", node.x}, {"y", node.y}}}, {"current_position", {{"x", node.x}, {"y", node.y}}}, {"mode", {{"width", mode.width}, {"height", mode.height}, {"refresh_hz", mode.refresh_hz}, {"hdr", mode.hdr}}}});
    }
    for (const auto &[uuid, state] : clients_) result["runtime"][uuid] = {{"lifecycle", lifecycle_name(state.lifecycle)}, {"ready", state.lifecycle == lifecycle_e::ready}, {"retryable", state.lifecycle == lifecycle_e::retryable}, {"lease_held", state.lease_held}, {"warning", state.warning}, {"plaintext_rtsp_warning", plaintext_rtsp_warning_provider_ ? plaintext_rtsp_warning_provider_(uuid) : ""}};
    return result;
  }
}  // namespace remote_display_topology