#include "Control_Service.hpp"
#include "Control_Service.ipp"
#include "Control_Requests.hpp"
#include "Protocol_Type.hpp"
#include "runtime/Gallery_Plots.hpp"
#include "runtime/Datoviz_Observation_Json.hpp"
#include "runtime/Input_Event.hpp"
#include "runtime/Taskflow_Trace_Json.hpp"
#include <Frame_Policy/Fixed_Rate_Frame_Policy.hpp>
#include <Frame_Policy/Manual_Frame_Policy.hpp>
#include <Frame_Policy/Maximum_Rate_Frame_Policy.hpp>
#include <magic_enum/magic_enum.hpp>
#include <render_2D/plottable/Plottables.hpp>
#include <render_3D/Render_3D.hpp>
#include <render_3D/detail/Gpu_Completion_Service.hpp>
#include <render_common.hpp>
#include <array>
#include <chrono>
#include <cmath>
#include <optional>
#include <ranges>
#include <stdexcept>
#include <unordered_set>
#include <utility>

namespace aethera::mcp {
namespace {

using Invoke = Tool_Call_Output (*)(Control_Service&, const nlohmann::json&);
using Schema = nlohmann::json (*)();

struct Operation {
    std::string_view name;                                  /* MCP 稳定 tool name。 */
    std::string_view description;                           /* 模型选择工具时使用的业务说明。 */
    Schema schema;                                          /* PFR 自动生成的输入结构。 */
    Invoke invoke;                                          /* 协议无关的业务调用入口。 */
};

template <typename Request>
[[nodiscard]] nlohmann::json request_schema() {
    return describe_protocol_type<Request>();
}

template <typename Request, typename Callback>
[[nodiscard]] Tool_Call_Output decode_and_call(
    const nlohmann::json& arguments, Callback&& callback) {
    try {
        Request request{};
        decode_protocol_value(request, arguments);
        return std::forward<Callback>(callback)(request);
    } catch (const nlohmann::json::exception& failure) {
        return {Tool_Call_Result::invalid_arguments, {}, failure.what()};
    } catch (const std::invalid_argument& failure) {
        return {Tool_Call_Result::invalid_arguments, {}, failure.what()};
    }
}

[[nodiscard]] std::string_view pacing_mode_name(Frame_Policy_Type type) {
    const auto name = magic_enum::enum_name(type);
    if (name.empty()) throw std::logic_error("unknown frame policy type");
    return name;
}

[[nodiscard]] nlohmann::json frame_policy_schema(
    const Frame_Policy& policy) {
    const auto* fixed = dynamic_cast<const Fixed_Rate_Frame_Policy*>(&policy);
    return {
        {"id", "frame-analysis"}, {"label", "渲染与媒体流水线"},
        {"kind", "analysis"},
        {"fields", nlohmann::json::array({
            {{"key", "render_enabled"}, {"editor", "boolean"},
             {"editable", true},
             {"value", policy.get<&Frame_Policy::Prop::render_enabled>()}},
            {{"key", "pixel_delivery_enabled"}, {"editor", "boolean"},
             {"editable", true},
             {"value", policy.get<&Frame_Policy::Prop::pixel_delivery_enabled>()}},
            {{"key", "pacing_mode"}, {"editor", "select"},
             {"editable", true}, {"value", pacing_mode_name(policy.type())},
             {"options", nlohmann::json::array({
                 {{"value", "manual"}, {"label", "手动渲染"}},
                 {{"value", "fixed_rate"}, {"label", "固定帧率"}},
                 {{"value", "maximum_rate"}, {"label", "最大吞吐"}}})}},
            {{"key", "fixed_rate_fps"}, {"editor", "number"},
             {"editable", true}, {"minimum", 0.1}, {"maximum", 100.0},
             {"value", fixed ? fixed->frame_rate() : 100.0}}
        })}
    };
}

[[nodiscard]] nlohmann::json frame_policy_state_json(
    const Frame_Policy_State& state, const Frame_Policy& policy) {
    const auto milliseconds = [](std::uint64_t value) {
        return static_cast<double>(value) / 1'000'000.0;
    };
    const auto observed_ns = state.observed_until_ns > state.observation_started_ns
        ? state.observed_until_ns - state.observation_started_ns : 0U;
    const auto observed_seconds = static_cast<double>(observed_ns) / 1e9;
    const auto rate = [observed_seconds](std::uint64_t count) {
        return observed_seconds == 0.0 ? 0.0 :
            static_cast<double>(count) / observed_seconds;
    };
    const auto* fixed = dynamic_cast<const Fixed_Rate_Frame_Policy*>(&policy);
    const double target = fixed ? fixed->frame_rate() : 0.0;
    const auto completion_span =
        state.last_completion_ns > state.first_completion_ns
            ? state.last_completion_ns - state.first_completion_ns : 0U;
    const double completed_fps = completion_span != 0 &&
        state.completed_frame_count > 1
            ? static_cast<double>(state.completed_frame_count - 1) * 1e9 /
                static_cast<double>(completion_span)
            : 0.0;
    return {
        {"generation", state.generation},
        {"configuration", {
            {"mode", pacing_mode_name(policy.type())},
            {"render_enabled", policy.get<&Frame_Policy::Prop::render_enabled>()},
            {"pixel_delivery_enabled", policy.get<&Frame_Policy::Prop::pixel_delivery_enabled>()},
            {"fixed_rate_fps", target}}},
        {"observation", {
            {"duration_ms", milliseconds(observed_ns)},
            {"request_count", state.request_count},
            {"submitted_frame_count", state.submitted_frame_count},
            {"completed_frame_count", state.completed_frame_count},
            {"active_frame_count", state.active_frame_count}}},
        {"lifecycle", {
            {"state", magic_enum::enum_name(
                policy.read_state<Frame_Policy::Base_Tag>().lifecycle)},
            {"active_frame_count", state.active_frame_count},
            {"published_frame_count", state.published_frame_count},
            {"publication_failed_count", state.publication_failed_count}}},
        {"throughput", {
            {"request_rate_fps", rate(state.request_count)},
            {"submission_rate_fps", rate(state.submitted_frame_count)},
            {"completion_rate_fps", completed_fps},
            {"target_achievement_ratio", target == 0.0 ? 0.0 : completed_fps / target},
            {"latest_frame_interval_ms", milliseconds(state.latest_completion_interval_ns)}}},
        {"latency", {
            {"latest_tick_queue_ms", milliseconds(state.latest_tick_queue_ns)},
            {"average_tick_queue_ms", milliseconds(
                state.submitted_frame_count == 0 ? 0U :
                state.tick_queue_total_ns / state.submitted_frame_count)},
            {"maximum_tick_queue_ms", milliseconds(state.maximum_tick_queue_ns)},
            {"latest_completion_ms", milliseconds(state.latest_completion_latency_ns)},
            {"average_completion_ms", milliseconds(
                state.completed_frame_count == 0 ? 0U :
                state.completion_latency_total_ns /
                    state.completed_frame_count)},
            {"maximum_completion_ms", milliseconds(state.maximum_completion_latency_ns)}}},
        {"requests", {
            {"periodic", state.periodic_request_count},
            {"immediate", state.immediate_request_count},
            {"maximum_rate", state.maximum_rate_request_count},
            {"accepted", state.accepted_request_count},
            {"policy_rejected", state.policy_rejection_count},
            {"frame_slot_backpressure", state.frame_slot_backpressure_count},
            {"scene_rejected", state.scene_rejection_count}}},
        {"last_frame", {
            {"sequence", state.last_frame_sequence},
            {"request_source", magic_enum::enum_name(state.last_request_source)}}}
    };
}

void append_frame_statistics_json(
    nlohmann::json& output, const Frame_Statistics_State& state) {
    for (const auto statistic : magic_enum::enum_values<Frame_Statistic>()) {
        if (statistic == Frame_Statistic::count) continue;
        const auto& value = state.values[static_cast<std::size_t>(statistic)];
        if (value.count == 0) continue;
        output[magic_enum::enum_name(statistic)] = {
            {"count", value.count}, {"latest", value.latest},
            {"minimum", value.minimum}, {"maximum", value.maximum},
            {"average", value.average},
            {"trimmed_average", value.trimmed_average},
            {"variability", value.variability}, {"p50", value.p50},
            {"p95", value.p95}, {"p99", value.p99}};
    }
}

void append_event_statistics_json(
    nlohmann::json& output, const Event_Statistics_State& state) {
    for (const auto type : magic_enum::enum_values<Event_Type>()) {
        auto& event = output[magic_enum::enum_name(type)];
        const auto& values = state.values[static_cast<std::size_t>(type)];
        for (const auto statistic : magic_enum::enum_values<Event_Statistic>()) {
            if (statistic == Event_Statistic::count) continue;
            const auto& value = values[static_cast<std::size_t>(statistic)];
            if (value.count == 0) continue;
            event[magic_enum::enum_name(statistic)] = {
                {"count", value.count}, {"latest", value.latest},
                {"minimum", value.minimum}, {"maximum", value.maximum},
                {"average", value.average},
                {"trimmed_average", value.trimmed_average},
                {"variability", value.variability}, {"p50", value.p50},
                {"p95", value.p95}, {"p99", value.p99}};
        }
        if (event.empty())
            output.erase(std::string{magic_enum::enum_name(type)});
    }
}

[[nodiscard]] Tool_Call_Output list_plots(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Empty_Request>(arguments, [&service](const auto&) {
        return Tool_Call_Output{Tool_Call_Result::ok, service.plot_catalog(), {}};
    });
}

[[nodiscard]] Tool_Call_Output plot_schema(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        return Tool_Call_Output{Tool_Call_Result::ok,
                                service.plot_schema(request.plot), {}};
    });
}

[[nodiscard]] Tool_Call_Output plot_diagnostics(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        return Tool_Call_Output{Tool_Call_Result::ok,
                                service.plot_diagnostics(request.plot), {}};
    });
}

[[nodiscard]] Tool_Call_Output reset_plot_diagnostics(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        service.reset_plot_diagnostics(request.plot);
        return Tool_Call_Output{Tool_Call_Result::ok, {{"accepted", true}}, {}};
    });
}

[[nodiscard]] Tool_Call_Output component_state(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Component_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        auto result = service.plot_component_state(
            request.plot, request.component);
        if (!result.value("success", true))
            return Tool_Call_Output{Tool_Call_Result::rejected, std::move(result), "unknown component"};
        return Tool_Call_Output{Tool_Call_Result::ok, std::move(result), {}};
    });
}

[[nodiscard]] Tool_Call_Output write_property(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Write_Property_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        auto result = service.write_plot_property(
            request.plot, request.component, request.property, request.value);
        const bool success = result.value("success", false);
        return Tool_Call_Output{success ? Tool_Call_Result::ok : Tool_Call_Result::rejected,
                                std::move(result), success ? "" : "property write rejected"};
    });
}

[[nodiscard]] Tool_Call_Output generate_data(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Generate_Data_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        return Tool_Call_Output{Tool_Call_Result::ok,
            service.generate_plot_data(request.plot, request.input), {}};
    });
}

[[nodiscard]] Tool_Call_Output submit_plot_input(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Input_Request>(
        arguments, [&service](auto request) {
            if (!service.contains_plot(request.plot))
                return Tool_Call_Output{
                    Tool_Call_Result::unknown_plot, {}, "unknown plot"};
            const auto time_milliseconds =
                request.event.at("time_milliseconds").get<double>();
            service.submit_input(
                request.plot, web::make_input_event(request.event));
            return Tool_Call_Output{
                Tool_Call_Result::ok,
                {{"accepted", true},
                 {"plot", request.plot},
                 {"time_milliseconds", time_milliseconds}}, {}};
        });
}

[[nodiscard]] Tool_Call_Output render_plot_at(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Render_Request>(
        arguments, [&service](const auto& request) {
            if (!service.contains_plot(request.plot))
                return Tool_Call_Output{
                    Tool_Call_Result::unknown_plot, {}, "unknown plot"};
            service.request_frame(request.plot, Frame_Request{
                .issued_at = std::chrono::steady_clock::now(),
                .time_milliseconds = request.time_milliseconds,
                .width = request.width,
                .height = request.height,
                .source = Frame_Request_Source::immediate});
            return Tool_Call_Output{
                Tool_Call_Result::ok,
                {{"accepted", true},
                 {"plot", request.plot},
                 {"time_milliseconds", request.time_milliseconds},
                 {"width", request.width}, {"height", request.height}}, {}};
        });
}

[[nodiscard]] Tool_Call_Output begin_frame_trace(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Frame_Trace_Request>(
        arguments, [&service](const auto& request) {
            if (!service.contains_plot(request.plot))
                return Tool_Call_Output{
                    Tool_Call_Result::unknown_plot, {}, "unknown plot"};
            service.request_plot_taskflow_trace(
                request.plot, request.frame_count);
            return Tool_Call_Output{
                Tool_Call_Result::ok,
                {{"accepted", true}, {"plot", request.plot},
                 {"frame_count", request.frame_count},
                 {"status_tool", "aethera_frame_trace_read"}}, {}};
        });
}

[[nodiscard]] Tool_Call_Output read_frame_trace(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Request>(
        arguments, [&service](const auto& request) {
            if (!service.contains_plot(request.plot))
                return Tool_Call_Output{
                    Tool_Call_Result::unknown_plot, {}, "unknown plot"};
            return Tool_Call_Output{
                Tool_Call_Result::ok,
                service.plot_taskflow_trace(request.plot), {}};
        });
}

[[nodiscard]] Tool_Call_Output begin_benchmark(
    Control_Service& service, const nlohmann::json& arguments) {
    return decode_and_call<Plot_Request>(arguments, [&service](const auto& request) {
        if (!service.contains_plot(request.plot))
            return Tool_Call_Output{Tool_Call_Result::unknown_plot, {}, "unknown plot"};
        service.reset_plot_diagnostics(request.plot);
        service.request_frame(request.plot, Frame_Request{
            .issued_at = std::chrono::steady_clock::now(),
            .source = Frame_Request_Source::immediate});
        return Tool_Call_Output{Tool_Call_Result::ok,
            {{"accepted", true}, {"plot", request.plot},
             {"status_tool", "aethera_benchmark_read"}}, {}};
    });
}

[[nodiscard]] nlohmann::json task_runtime_json() {
    const auto state = task_runtime_state();
    nlohmann::json workers = nlohmann::json::array();
    for (const auto& worker : state.workers)
        workers.push_back({
            {"id", worker.id}, {"task_count", worker.task_count},
            {"entry_queue_size", worker.current_queue_size},
            {"entry_queue_capacity", worker.current_queue_capacity},
            {"peak_queue_size", worker.peak_observed_queue_size},
            {"max_queue_capacity", worker.max_observed_queue_capacity},
            {"active_task", {{"native_id", std::to_string(worker.active_task_hash)},
                              {"type", worker.active_task_type},
                              {"time_ns", worker.active_task_time_ns}}},
            {"task_time_ns", worker.task_time_ns},
            {"busy_time_ns", worker.busy_time_ns},
            {"cpu_time_ns", worker.cpu_time_ns},
            {"cooperative_wait_count", worker.cooperative_wait_count},
            {"cooperative_wait_time_ns", worker.cooperative_wait_time_ns},
            {"idle_time_ns", worker.idle_time_ns},
            {"min_task_time_ns", worker.min_task_time_ns},
            {"max_task_time_ns", worker.max_task_time_ns},
            {"utilization", worker.utilization},
            {"cpu_utilization", worker.cpu_utilization}});
    nlohmann::json task_types = nlohmann::json::array();
    for (const auto& type : state.task_types)
        task_types.push_back({
            {"name", type.name}, {"count", type.count},
            {"total_time_ns", type.total_time_ns},
            {"min_time_ns", type.min_time_ns},
            {"max_time_ns", type.max_time_ns}});
    return {
        {"protocol", "aethera.taskflow.runtime"}, {"version", 2},
        {"worker_count", state.worker_count},
        {"active_topologies", state.active_topology_count},
        {"active_taskflows", state.active_taskflow_count},
        {"peak_active_taskflows", state.peak_active_taskflow_count},
        {"completed_taskflows", state.completed_taskflow_count},
        {"failed_taskflows", state.failed_taskflow_count},
        {"active_tasks", state.active_task_count},
        {"peak_active_tasks", state.peak_active_task_count},
        {"active_workers", state.active_worker_count},
        {"peak_active_workers", state.peak_active_worker_count},
        {"observed_tasks", state.observed_task_count},
        {"named_tasks", state.named_task_count},
        {"peak_worker_queue_size", state.peak_observed_worker_queue_size},
        {"max_worker_queue_capacity", state.max_observed_worker_queue_capacity},
        {"longest_task", {{"native_id", std::to_string(state.longest_task_hash)},
                           {"name", state.longest_task_name},
                           {"type", state.longest_task_type},
                           {"time_ns", state.longest_task_time_ns}}},
        {"total_task_time_ns", state.total_task_time_ns},
        {"worker_busy_time_ns", state.worker_busy_time_ns},
        {"worker_cpu_time_ns", state.worker_cpu_time_ns},
        {"cooperative_wait_count", state.cooperative_wait_count},
        {"cooperative_wait_time_ns", state.cooperative_wait_time_ns},
        {"observed_wall_time_ns", state.observed_wall_time_ns},
        {"worker_utilization", state.worker_utilization},
        {"worker_cpu_utilization", state.worker_cpu_utilization},
        {"task_types", std::move(task_types)}, {"workers", std::move(workers)}};
}

[[nodiscard]] Tool_Call_Output task_runtime(
    Control_Service&, const nlohmann::json& arguments) {
    return decode_and_call<Empty_Request>(arguments, [](const auto&) {
        return Tool_Call_Output{Tool_Call_Result::ok, task_runtime_json(), {}};
    });
}

constexpr std::array operations{
    Operation{"aethera_plot_list", "List every 2D and 3D gallery plot.",
              &request_schema<Empty_Request>, &list_plots},
    Operation{"aethera_plot_schema", "Describe a plot and its editable render components.",
              &request_schema<Plot_Request>, &plot_schema},
    Operation{"aethera_plot_diagnostics", "Read current frame, scene, GPU and Datoviz diagnostics.",
              &request_schema<Plot_Request>, &plot_diagnostics},
    Operation{"aethera_plot_diagnostics_reset", "Reset the plot's authoritative diagnostic counters.",
              &request_schema<Plot_Request>, &reset_plot_diagnostics},
    Operation{"aethera_component_state", "Read one render component's published state.",
              &request_schema<Component_Request>, &component_state},
    Operation{"aethera_component_write", "Write one editable render component property.",
              &request_schema<Write_Property_Request>, &write_property},
    Operation{"aethera_data_generate", "Generate input data for a gallery plot.",
              &request_schema<Generate_Data_Request>, &generate_data},
    Operation{"aethera_plot_input", "Submit one timestamped input event without waiting for rendering.",
              &request_schema<Plot_Input_Request>, &submit_plot_input},
    Operation{"aethera_plot_render_at", "Request one diagnostic frame at a caller-owned timeline time.",
              &request_schema<Plot_Render_Request>, &render_plot_at},
    Operation{"aethera_frame_trace_begin", "Capture existing per-frame Taskflow and Datoviz observations.",
              &request_schema<Frame_Trace_Request>, &begin_frame_trace},
    Operation{"aethera_frame_trace_read", "Read the current physical-frame trace capture.",
              &request_schema<Plot_Request>, &read_frame_trace},
    Operation{"aethera_benchmark_begin", "Reset diagnostics and asynchronously start rendering a plot.",
              &request_schema<Plot_Request>, &begin_benchmark},
    Operation{"aethera_benchmark_read", "Read benchmark results from the plot's current diagnostics.",
              &request_schema<Plot_Request>, &plot_diagnostics},
    Operation{"aethera_task_runtime", "Read Taskflow executor utilization and queue diagnostics.",
              &request_schema<Empty_Request>, &task_runtime},
};

}

Control_Service::Private::Entry::Entry(
    std::string value_id, web::Gallery_Build build,
    Gallery_Output value_output)
    : components(std::move(build.first)), id(std::move(value_id)),
      output(std::move(value_output)) {
    runtime.first = std::move(build.second);
}

Control_Service::Private::Entry::~Entry() = default;

std::shared_ptr<Frame_Policy>
Control_Service::Private::Entry::policy() const noexcept {
    return runtime.second.load(std::memory_order_acquire);
}

void Control_Service::Private::Entry::fail(
    std::exception_ptr failure) noexcept {
    try {
        std::string description{"unknown Gallery runtime failure"};
        try {
            if (failure) std::rethrow_exception(failure);
        }
        catch (const std::exception& error) { description = error.what(); }
        catch (...) { description = "non-standard Gallery runtime failure"; }
        terminal_failure.store(
            std::make_shared<const std::string>(std::move(description)),
            std::memory_order_release);
    }
    catch (...) {}
}

std::shared_ptr<Frame_Policy>
Control_Service::Private::Entry::create_policy(
    Frame_Policy_Type type, double fixed_rate_fps) {
    auto* scene = std::visit(
        [](auto& value) -> Render_Frame_Scene* { return value.get(); },
        runtime.first);
    const bool is_3d = std::holds_alternative<
        std::unique_ptr<render_3d::Render_Scene_3D>>(runtime.first);
    const auto weak = weak_from_this();
    Frame_Policy::Dependencies dependencies{
        .scene = not_null{scene},
        .create_frame = [is_3d]() -> std::unique_ptr<Render_Frame> {
            if (is_3d)
                return std::make_unique<render_3d::Frame_3D>(Frame_Identity{});
            return std::make_unique<render_2d::Frame_2D>(Frame_Identity{});
        },
        .prepare_frame = [weak](not_null<Render_Frame*> frame,
                                const Frame_Production& production) {
            if (const auto owner = weak.lock())
                owner->prepare_frame(frame, production);
        },
        .publish_frame = [weak](not_null<Render_Frame*> frame,
                                const Frame_Production& production) {
            if (const auto owner = weak.lock())
                return owner->publish_frame(frame, production);
            return Frame_Publication_Dispatch{};
        },
        .retain_owner = [weak]() -> std::shared_ptr<void> {
            return weak.lock();
        },
        .report_failure = [weak](std::exception_ptr failure) {
            if (const auto owner = weak.lock())
                owner->fail(std::move(failure));
        },
        .frame_capacity = 3};
    switch (type) {
    case Frame_Policy_Type::manual:
        return std::make_shared<Manual_Frame_Policy>(std::move(dependencies));
    case Frame_Policy_Type::fixed_rate:
        return std::make_shared<Fixed_Rate_Frame_Policy>(
            std::move(dependencies), fixed_rate_fps);
    case Frame_Policy_Type::maximum_rate:
        return std::make_shared<Maximum_Rate_Frame_Policy>(
            std::move(dependencies));
    }
    throw std::logic_error("unknown Frame Policy type");
}

void Control_Service::Private::Entry::start() {
    auto next = create_policy(Frame_Policy_Type::fixed_rate, 100.0);
    runtime.second.store(next, std::memory_order_release);
    next->start();
}

void Control_Service::Private::Entry::stop() noexcept {
    auto current = runtime.second.exchange({}, std::memory_order_acq_rel);
    if (!current) return;
    try {
        current->stop([owner = shared_from_this(), current]() mutable {
            current.reset();
            owner.reset();
        });
    }
    catch (...) { fail(std::current_exception()); }
}

void Control_Service::Private::Entry::replace_policy(
    Frame_Policy_Type type) {
    auto current = runtime.second.exchange({}, std::memory_order_acq_rel);
    if (!current) throw std::logic_error("Frame Policy switch is active");
    const auto* fixed = dynamic_cast<const Fixed_Rate_Frame_Policy*>(
        current.get());
    const auto fixed_rate_fps = fixed ? fixed->frame_rate() : 100.0;
    current->stop([
        owner = shared_from_this(), current, type, fixed_rate_fps]() mutable {
        try {
            auto next = owner->create_policy(type, fixed_rate_fps);
            owner->runtime.second.store(next, std::memory_order_release);
            next->start();
        }
        catch (...) { owner->fail(std::current_exception()); }
        current.reset();
    });
}

void Control_Service::Private::Entry::prepare_frame(
    not_null<Render_Frame*> frame, const Frame_Production& production) {
    auto request = production.request;
    request.width = output.width;
    request.height = output.height;
    const Frame_Identity identity{
        production.sequence,
        request.sequence == 0 ? production.sequence : request.sequence};
    if (auto* frame_2d = dynamic_cast<render_2d::Frame_2D*>(frame.get())) {
        frame_2d->begin(identity, render_2d::Frame_2D::native_pixel_format,
                        request.source);
        std::get<std::unique_ptr<render_2d::Render_Scene_2D>>(runtime.first)
            ->set<&render_2d::Render_Scene_2D::Prop::viewport>(
                render_2d::Size{static_cast<int>(request.width),
                                static_cast<int>(request.height)});
    }
    else if (auto* frame_3d = dynamic_cast<render_3d::Frame_3D*>(frame.get())) {
        frame_3d->begin(identity, render_3d::Frame_3D_Output::pixels,
                        render_3d::Frame_3D::native_pixel_format,
                        request.source);
        std::get<std::unique_ptr<render_3d::Render_Scene_3D>>(runtime.first)
            ->set<&render_3d::Render_Scene_3D::Prop::viewport>(
                render_3d::Extent{request.width, request.height});
    }
    else {
        throw std::logic_error("Gallery policy created an unsupported Frame");
    }
    auto remaining = taskflow_trace_remaining.load(std::memory_order_acquire);
    while (remaining != 0 &&
           !taskflow_trace_remaining.compare_exchange_weak(
               remaining, remaining - 1, std::memory_order_acq_rel,
               std::memory_order_acquire)) {}
    if (remaining != 0) frame->request_taskflow_trace();
    frame->mark(Frame_Trace_Marker::plot_update_started);
    const auto started = std::chrono::steady_clock::now();
    components->update(request);
    frame->mark(Frame_Trace_Marker::plot_update_finished);
    const auto elapsed = std::chrono::steady_clock::now() - started;
    frame->record(Frame_Trace_Measurement::plot_update_ns,
        static_cast<std::uint64_t>(std::max<std::int64_t>(0,
            std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed)
                .count())));
}

Frame_Publication_Dispatch
Control_Service::Private::Entry::publish_frame(
    not_null<Render_Frame*> frame, const Frame_Production& production) {
    const auto current = policy();
    const bool deliver_pixels = current && current->get<
        &Frame_Policy::Prop::pixel_delivery_enabled>();
    const auto identity = frame->identity();
    auto rendered = identity;
    std::shared_ptr<const std::vector<std::byte>> storage;
    std::optional<render_3d::Datoviz_Frame_Observation> datoviz_observation;
    web::Gallery_Pixel_Layout layout{web::Gallery_Pixel_Layout::rgba8};
    std::uint32_t width{}, height{};
    if (auto* frame_2d = dynamic_cast<render_2d::Frame_2D*>(frame.get())) {
        layout = web::Gallery_Pixel_Layout::bgra8;
        const auto image = frame_2d->image();
        width = static_cast<std::uint32_t>(std::max(0, image.width));
        height = static_cast<std::uint32_t>(std::max(0, image.height));
        if (deliver_pixels) {
            auto pixels = frame_2d->output_pixels();
            width = static_cast<std::uint32_t>(pixels.width);
            height = static_cast<std::uint32_t>(pixels.height);
            storage = std::make_shared<const std::vector<std::byte>>(
                std::move(pixels.bytes));
        }
    }
    else if (auto* frame_3d = dynamic_cast<render_3d::Frame_3D*>(frame.get())) {
        rendered = frame_3d->rendered_identity();
        datoviz_observation = frame_3d->take_datoviz_observation();
        const auto extent = frame_3d->extent();
        width = extent.width;
        height = extent.height;
        if (deliver_pixels) storage = frame_3d->share_pixels();
    }
    const auto timestamp = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::duration<double, std::milli>(
            production.request.time_milliseconds));
    auto pixels = std::make_shared<const web::Gallery_Pixel_Frame>(
        web::Gallery_Pixel_Frame{
            std::move(storage), layout, timestamp, identity.sequence,
            identity.correlation_id, rendered.sequence,
            rendered.correlation_id, width, height});
    auto publication = std::make_shared<const web::Gallery_Frame>(
        web::Gallery_Frame{std::move(pixels)});
    const auto started = std::chrono::steady_clock::now();
    auto result = web::Gallery_Frame_Publication::ignored;
    if (output.publish) result = output.publish(id, std::move(publication));
    const auto completed = std::chrono::steady_clock::now();
    frame->record(Frame_Trace_Measurement::plot_publish_ns,
        static_cast<std::uint64_t>(std::max<std::int64_t>(0,
            std::chrono::duration_cast<std::chrono::nanoseconds>(
                completed - started).count())));
    if (frame->taskflow_trace_requested()) {
        const auto trace_value = frame->take_taskflow_trace();
        std::unordered_set<std::uint64_t> executed;
        for (const auto& task : trace_value.tasks)
            executed.insert(task.native_id);
        std::vector<std::string> component_ids;
        for (const auto& graph : trace_value.graphs)
            for (const auto& node : graph.nodes) {
                if (!executed.contains(node.native_id)) continue;
                const auto found = std::ranges::find(
                    node.attributes, "owner_component",
                    &std::pair<std::string, std::string>::first);
                if (found != node.attributes.end() && !found->second.empty() &&
                    std::ranges::find(component_ids, found->second) ==
                        component_ids.end())
                    component_ids.push_back(found->second);
            }
        nlohmann::json captured_backend;
        if (datoviz_observation)
            captured_backend = datoviz_observation_json(*datoviz_observation);
        const auto encoded = std::make_shared<const nlohmann::json>(
            web::taskflow_trace_json(trace_value,
                components->capture_components(component_ids),
                captured_backend));
        auto state = taskflow_trace_control.load(std::memory_order_acquire);
        for (;;) {
            const auto requested = static_cast<std::uint32_t>(state >> 32U);
            const auto captured = static_cast<std::uint32_t>(state);
            if (captured >= requested) break;
            taskflow_trace_slots[captured].store(encoded,
                                                  std::memory_order_release);
            const auto next = (static_cast<std::uint64_t>(requested) << 32U) |
                              static_cast<std::uint64_t>(captured + 1U);
            if (taskflow_trace_control.compare_exchange_weak(
                    state, next, std::memory_order_release,
                    std::memory_order_acquire))
                break;
        }
    }
    return {
        .started_at = started,
        .completed_at = completed,
        .asynchronous_feedback_count =
            result == web::Gallery_Frame_Publication::asynchronous ? 1U : 0U,
        .succeeded = result == web::Gallery_Frame_Publication::completed,
        .pixel_width = width,
        .pixel_height = height};
}

nlohmann::json Control_Service::Private::Entry::schema() const {
    auto result = components->schema();
    if (const auto current = policy()) {
        auto analysis = frame_policy_schema(*current);
        const auto generator = components->data_generator_schema();
        if (!generator.is_null()) analysis["data_generator"] = generator;
        result["frame_analysis"] = std::move(analysis);
    }
    return result;
}

nlohmann::json Control_Service::Private::Entry::diagnostics() const {
    const auto current = policy();
    if (!current)
        return {{"protocol", "aethera.plot.diagnostics"}, {"version", 5},
                {"available", false}};
    Frame_Policy_State observation{};
    current->access_state<Frame_Policy::Base_Tag>(
        [&](const Frame_Policy::State& state) {
            observation = state.observation;
        });
    nlohmann::json input_statistics = nlohmann::json::object();
    nlohmann::json frame_statistics = nlohmann::json::object();
    append_frame_statistics_json(
        frame_statistics, observation.frame_statistics);
    std::visit([&](const auto& scene) {
        scene->template access_state<Scene::Base_Tag>(
            [&](const Scene::State& state) {
                append_event_statistics_json(
                    input_statistics, state.event_statistics);
            });
    }, runtime.first);
    nlohmann::json result{
        {"protocol", "aethera.plot.diagnostics"}, {"version", 5},
        {"available", true}, {"plot", id},
        {"dimension", std::holds_alternative<
            std::unique_ptr<render_3d::Render_Scene_3D>>(runtime.first)
                ? "3D" : "2D"},
        {"sequence", observation.last_frame_sequence},
        {"pixel", {{"width", observation.pixel_width},
                    {"height", observation.pixel_height}}},
        {"frame_policy", frame_policy_state_json(observation, *current)},
        {"frame_lifecycle", magic_enum::enum_name(
            current->read_state<Frame_Policy::Base_Tag>().lifecycle)},
        {"frame_statistics", std::move(frame_statistics)},
        {"input_statistics", std::move(input_statistics)}};
    if (const auto failure = terminal_failure.load(std::memory_order_acquire))
        result["terminal_failure"] = *failure;
    return result;
}

void Control_Service::Private::Entry::request_trace(
    std::size_t frame_count) {
    if (frame_count == 0 || frame_count > maximum_taskflow_trace_frames)
        throw std::invalid_argument(
            "Taskflow trace frame_count must be between 1 and 120");
    auto control = taskflow_trace_control.load(std::memory_order_acquire);
    for (;;) {
        const auto requested = static_cast<std::uint32_t>(control >> 32U);
        const auto captured = static_cast<std::uint32_t>(control);
        if (requested != captured)
            throw std::logic_error("A Taskflow trace request is active");
        const auto next = static_cast<std::uint64_t>(frame_count) << 32U;
        if (taskflow_trace_control.compare_exchange_weak(
                control, next, std::memory_order_release,
                std::memory_order_acquire))
            break;
    }
    for (auto& slot : taskflow_trace_slots)
        slot.store({}, std::memory_order_release);
    taskflow_trace_remaining.store(frame_count, std::memory_order_release);
}

nlohmann::json Control_Service::Private::Entry::trace() const {
    nlohmann::json frames = nlohmann::json::array();
    const auto state = taskflow_trace_control.load(std::memory_order_acquire);
    const auto requested = static_cast<std::uint32_t>(state >> 32U);
    const auto captured = static_cast<std::uint32_t>(state);
    for (std::uint32_t index = 0; index < captured; ++index)
        if (const auto value =
                taskflow_trace_slots[index].load(std::memory_order_acquire))
            frames.push_back(*value);
    return {{"protocol", "aethera.taskflow.frames"}, {"version", 1},
            {"requested", requested},
            {"remaining", taskflow_trace_remaining.load(
                std::memory_order_acquire)},
            {"captured", frames.size()},
            {"complete", requested != 0 && frames.size() == requested},
            {"frames", std::move(frames)}};
}

Control_Service::Control_Service(Gallery_Output output)
    : d(std::make_unique<Private>()) {
    d->output = std::move(output);
}

Control_Service::~Control_Service() {
    for (const auto& [id, entry] : d->plots) {
        static_cast<void>(id);
        entry->stop();
    }
}

std::shared_ptr<Control_Service> Control_Service::create(
    Gallery_Output output) {
    auto service = std::shared_ptr<Control_Service>(
        new Control_Service(std::move(output)));
    service->d->plots.reserve(web::gallery_plot_definitions().size());
    for (const auto& definition : web::gallery_plot_definitions()) {
        auto entry = std::make_shared<Private::Entry>(
            std::string{definition.id}, definition.create(),
            service->d->output);
        service->d->plots.emplace(entry->id, entry);
        entry->start();
    }
    return service;
}

bool Control_Service::contains_plot(std::string_view id) const noexcept {
    return d->plots.contains(std::string{id});
}

void Control_Service::submit_input(
    std::string_view id, Scene::Event_Pointer event) {
    if (!event) throw std::invalid_argument("Gallery input event is null");
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    std::visit([event = std::move(event)](auto& scene) mutable {
        scene->submit_event(std::move(event));
    }, found->second->runtime.first);
}

void Control_Service::request_frame(
    std::string_view id, Frame_Request request) {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    const auto current = found->second->policy();
    if (!current) throw std::logic_error("Frame Policy switch is active");
    current->request_frame(std::move(request));
}

void Control_Service::submit_publication_feedback(
    std::string_view id, Frame_Publication_Feedback feedback) noexcept {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) return;
    if (const auto current = found->second->policy())
        current->submit_publication_feedback(std::move(feedback));
}

nlohmann::json Control_Service::plot_schema(std::string_view id) const {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    return found->second->schema();
}

nlohmann::json Control_Service::write_plot_property(
    std::string_view id, std::string_view component, std::string_view key,
    const nlohmann::json& value) {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    auto& entry = *found->second;
    if (component != "frame-analysis")
        return entry.components->write_prop(component, key, value);
    if (key == "pacing_mode") {
        if (!value.is_string())
            return {{"success", false},
                    {"error", "pacing_mode requires a string"}};
        const auto parsed = magic_enum::enum_cast<Frame_Policy_Type>(
            value.get_ref<const std::string&>());
        if (!parsed)
            return {{"success", false},
                    {"error", "unknown frame pacing mode"}};
        entry.replace_policy(*parsed);
        return {{"success", true}, {"component", component}, {"key", key},
                {"value", pacing_mode_name(*parsed)}};
    }
    const auto current = entry.policy();
    if (!current)
        return {{"success", false}, {"error", "Frame Policy switch is active"}};
    if (key == "render_enabled" || key == "pixel_delivery_enabled") {
        if (!value.is_boolean())
            return {{"success", false}, {"error", "boolean required"}};
        const auto enabled = value.get<bool>();
        if (key == "render_enabled")
            current->set<&Frame_Policy::Prop::render_enabled>(enabled);
        else
            current->set<&Frame_Policy::Prop::pixel_delivery_enabled>(enabled);
        return {{"success", true}, {"component", component}, {"key", key},
                {"value", enabled}};
    }
    if (key == "fixed_rate_fps") {
        if (!value.is_number())
            return {{"success", false}, {"error", "number required"}};
        const auto rate = value.get<double>();
        auto* fixed = dynamic_cast<Fixed_Rate_Frame_Policy*>(current.get());
        if (!fixed || !std::isfinite(rate) || rate < 0.1 || rate > 100.0)
            return {{"success", false}, {"error", "invalid fixed rate"}};
        fixed->set_frame_rate(rate);
        return {{"success", true}, {"component", component}, {"key", key},
                {"value", rate}};
    }
    return {{"success", false}, {"error", "unknown frame property"}};
}

nlohmann::json Control_Service::plot_component_state(
    std::string_view id, std::string_view component) const {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    return found->second->components->component_state(component);
}

nlohmann::json Control_Service::generate_plot_data(
    std::string_view id, const nlohmann::json& input) {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    return found->second->components->generate_data(input);
}

nlohmann::json Control_Service::plot_diagnostics(std::string_view id) const {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    return found->second->diagnostics();
}

void Control_Service::reset_plot_diagnostics(std::string_view id) {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    std::visit([](auto& scene) {
        scene->template update_state<&Scene::State::event_statistics>(
            Event_Statistics_State{});
    }, found->second->runtime.first);
    if (const auto current = found->second->policy())
        current->reset_statistics();
}

void Control_Service::request_plot_taskflow_trace(
    std::string_view id, std::size_t frame_count) {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    found->second->request_trace(frame_count);
}

nlohmann::json Control_Service::plot_taskflow_trace(
    std::string_view id) const {
    const auto found = d->plots.find(std::string{id});
    if (found == d->plots.end()) throw std::invalid_argument("unknown plot");
    return found->second->trace();
}

nlohmann::json Control_Service::plot_catalog() const {
    nlohmann::json result = nlohmann::json::array();
    for (const auto& definition : web::gallery_plot_definitions())
        result.push_back({
            {"id", definition.id}, {"title", definition.title},
            {"category", definition.category},
            {"description", definition.description},
            {"dimension", web::plot_dimension_name(definition.dimension)}});
    return result;
}

nlohmann::json Control_Service::tool_catalog() const {
    nlohmann::json result = nlohmann::json::array();
    for (const auto& operation : operations)
        result.push_back({{"name", operation.name},
                          {"description", operation.description},
                          {"inputSchema", operation.schema()}});
    return result;
}

Tool_Call_Output Control_Service::call_tool(
    std::string_view name, const nlohmann::json& arguments) {
    for (const auto& operation : operations)
        if (operation.name == name) return operation.invoke(*this, arguments);
    return {Tool_Call_Result::unknown_tool, {}, "unknown tool"};
}

}