#include "Plot.hpp"
#include "Renderable_Adapter.hpp"
#include "Taskflow_Trace_Json.hpp"
#include <Frame_Policy/Frame_Policy.hpp>
#include <Frame_Policy/Frame_Policy_3D.hpp>
#include <Frame_Policy/Frame_Scheduler.hpp>
#include <concurrentqueue-1.0.5/concurrentqueue.h>
#include <nlohmann/json.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 <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cmath>
#include <concepts>
#include <exception>
#include <initializer_list>
#include <limits>
#include <memory>
#include <optional>
#include <span>
#include <stdexcept>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <variant>
#include <vector>
namespace aethera::web {
namespace {
using namespace render_2d;
using namespace render_3d;
using Scene_2D = Render_Scene_2D;
using Scene_3D = Render_Scene_3D;
constexpr std::uint16_t plot_stream_protocol_version{9};
constexpr std::size_t diagnostic_window_capacity{600};
std::string exception_description(const std::exception_ptr& failure) {
    try {
        if (failure) std::rethrow_exception(failure);
    }
    catch (const std::exception& error) {
        return error.what();
    }
    catch (...) {
        return "non-standard Plot failure";
    }
    return "empty Plot failure";
}
std::string_view pacing_mode_name(Frame_Pacing_Mode mode) {
    const auto name = magic_enum::enum_name(mode);
    if (name.empty()) throw std::logic_error("unknown frame pacing mode");
    return name;
}
std::optional<Frame_Pacing_Mode> parse_pacing_mode(std::string_view value) {
    return magic_enum::enum_cast<Frame_Pacing_Mode>(value);
}
std::string_view pixel_format_name(render_2d::Pixel_Format format) {
    const auto name = magic_enum::enum_name(format);
    if (name.empty()) throw std::logic_error("unknown 2D pixel format");
    return name;
}
std::string_view pixel_format_name(render_3d::Pixel_Format format) {
    switch (format) {
    case render_3d::Pixel_Format::rgba8_unorm: return "rgba8";
    }
    throw std::logic_error("unknown 3D pixel format");
}
const Frame_Policy::State& frame_policy_common_state(
    const Frame_Policy& policy) noexcept {
    return policy.read_state<Frame_Policy::Base_Tag>();
}
const Frame_Policy::State& frame_policy_common_state(
    const Frame_Policy_3D& policy) noexcept {
    return policy.read_state<Frame_Policy_3D::Base_Tag>().common;
}
template <typename Policy>
nlohmann::json frame_policy_schema(const Policy& policy) {
    const auto& current = frame_policy_common_state(policy);
    return {
        {"id", "frame-analysis"}, {"label", "渲染与媒体流水线"}, {"kind", "analysis"},
        {
            "fields", nlohmann::json::array({
                {
                    {"key", "render_enabled"}, {"label", "持续渲染与采样"}, {"editor", "boolean"},
                    {"editable", true}, {"description", "控制当前 Scene 的周期刷新；画面隐藏不会修改此项。"},
                    {"technical_description", "Authoritative per-scene periodic render switch."},
                    {"value", current.render_enabled}
                },
                {
                    {"key", "pixel_delivery_enabled"}, {"label", "图集像素传输"}, {"editor", "boolean"},
                    {"editable", true}, {"description", "控制完成帧是否进入页面级采样器；2D BGRA 与 3D RGBA 均保持原生格式。"},
                    {"technical_description", "Authoritative tile publication switch for the shared gallery pixel stream."},
                    {"value", current.pixel_delivery_enabled}
                },
                {
                    {"key", "pacing_mode"}, {"label", "服务端帧策略"}, {"editor", "select"},
                    {"editable", true}, {"description", "只控制 Scene::render(Frame*) 的调用节奏；Scene 的 Frame 所有权与接口保持不变。"},
                    {"technical_description", "Per-scene frame pacing policy backed by the Kernel scheduler."},
                    {"value", pacing_mode_name(current.mode)},
                    {
                        "options", nlohmann::json::array({
                            {{"value", "manual"}, {"label", "手动渲染"}},
                            {{"value", "fixed_rate"}, {"label", "固定频率"}},
                            {{"value", "maximum_rate"}, {"label", "最大吞吐"}}
                        })
                    }
                },
                {
                    {"key", "fixed_rate_fps"}, {"label", "目标帧率"}, {"editor", "number"},
                    {"editable", true}, {"minimum", 0.1}, {"maximum", 100.0}, {"step", 0.1},
                    {"description", "仅 fixed_rate 使用；maximum_rate 在完成准入释放后异步自驱下一帧。"},
                    {"technical_description", "Independent per-scene target frame rate."},
                    {"value", current.fixed_rate_fps}
                }
            })
        }
    };
}
template <typename Policy>
nlohmann::json write_frame_policy_prop(Policy& pacing,
                                       std::string_view key,
                                       const nlohmann::json& value) {
    if (key == "render_enabled" || key == "pixel_delivery_enabled") {
        if (!value.is_boolean()) return {{"success", false}, {"error", "frame policy switch requires a boolean"}};
        const bool target = value.get<bool>();
        if (key == "render_enabled") pacing.set_render_enabled(target);
        else pacing.set_pixel_delivery_enabled(target);
        return {
            {"success", true}, {"component", "frame-analysis"}, {"key", key},
            {"value", target}
        };
    }
    if (key == "pacing_mode") {
        if (!value.is_string()) return {{"success", false}, {"error", "pacing_mode requires a string"}};
        const auto parsed = parse_pacing_mode(value.get_ref<const std::string&>());
        if (!parsed) return {{"success", false}, {"error", "unknown frame pacing mode"}};
        pacing.set_mode(*parsed);
        return {
            {"success", true}, {"component", "frame-analysis"}, {"key", key},
            {"value", pacing_mode_name(*parsed)}
        };
    }
    if (key == "fixed_rate_fps") {
        if (!value.is_number()) return {{"success", false}, {"error", "fixed_rate_fps requires a number"}};
        const double next = value.get<double>();
        if (!std::isfinite(next) || next < 0.1 || next > 100.0) return {{"success", false}, {"error", "fixed_rate_fps must be between 0.1 and 100"}};
        pacing.set_fixed_rate(next);
        return {
            {"success", true}, {"component", "frame-analysis"}, {"key", key},
            {"value", next}
        };
    }
    return {{"success", false}, {"error", "unknown frame runtime property"}};
}
nlohmann::json frame_policy_state_json(const Frame_Policy::State& state) {
    const auto milliseconds = [](std::uint64_t nanoseconds) {
        return static_cast<double>(nanoseconds) / 1'000'000.0;
    };
    const auto ratio = [](std::uint64_t numerator, std::uint64_t denominator) {
        return denominator == 0 ? 0.0 : static_cast<double>(numerator) / static_cast<double>(denominator);
    };
    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) / 1'000'000'000.0;
    const auto rate = [observed_seconds](std::uint64_t count) {
        return observed_seconds > 0.0
                   ? static_cast<double>(count) / observed_seconds
                   : 0.0;
    };
    const auto completion_span = state.last_completion_ns > state.first_completion_ns
                                     ? state.last_completion_ns - state.first_completion_ns
                                     : 0U;
    const auto effective_fps = completion_span != 0 && state.completed_frame_count > 1
                                   ? static_cast<double>(state.completed_frame_count - 1U) *
                                   1'000'000'000.0 / static_cast<double>(completion_span)
                                   : 0.0;
    const auto interval_mean = state.completion_interval_count == 0
                                   ? 0.0
                                   : static_cast<double>(state.completion_interval_total_ns) /
                                   static_cast<double>(state.completion_interval_count);
    const auto interval_variance = state.completion_interval_count == 0
                                       ? 0.0
                                       : std::max(0.0,
                                                  state.completion_interval_squared_total_ns2 /
                                                  static_cast<double>(state.completion_interval_count) -
                                                  interval_mean * interval_mean);
    const auto target_achievement =
    state.mode == Frame_Pacing_Mode::fixed_rate && state.fixed_rate_fps > 0.0
        ? effective_fps / state.fixed_rate_fps
        : 0.0;
    return {
        {"generation", state.generation},
        {
            "configuration", {
                {"mode", pacing_mode_name(state.mode)},
                {"render_enabled", state.render_enabled},
                {"pixel_delivery_enabled", state.pixel_delivery_enabled},
                {"fixed_rate_fps", state.fixed_rate_fps}
            }
        },
        {
            "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}
            }
        },
        {
            "throughput", {
                {"request_rate_fps", rate(state.request_count)},
                {"submission_rate_fps", rate(state.submitted_frame_count)},
                {"completion_rate_fps", effective_fps},
                {"target_achievement_ratio", target_achievement},
                {
                    "latest_frame_interval_ms",
                    milliseconds(state.latest_completion_interval_ns)
                },
                {
                    "average_frame_interval_ms", milliseconds(
                        state.completion_interval_count == 0
                            ? 0U
                            : state.completion_interval_total_ns /
                            state.completion_interval_count)
                },
                {
                    "frame_interval_jitter_ms",
                    std::sqrt(interval_variance) / 1'000'000.0
                }
            }
        },
        {
            "requests", {
                {"periodic", state.periodic_request_count},
                {"immediate", state.immediate_request_count},
                {"maximum_rate", state.maximum_rate_request_count},
                {"accepted", state.accepted_request_count},
                {"coalesced", state.coalesced_request_count},
                {"policy_rejected", state.policy_rejection_count},
                {"frame_slot_backpressure", state.frame_slot_backpressure_count},
                {"scene_rejected", state.scene_rejection_count},
                {
                    "acceptance_ratio", ratio(
                        state.accepted_request_count, state.request_count)
                },
                {
                    "coalescing_ratio", ratio(
                        state.coalesced_request_count, state.request_count)
                },
                {
                    "backpressure_ratio", ratio(
                        state.frame_slot_backpressure_count,
                        state.accepted_request_count)
                }
            }
        },
        {
            "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)
                }
            }
        },
        {
            "last_frame", {
                {"sequence", state.last_frame_sequence},
                {
                    "request_source", magic_enum::enum_name(
                        state.last_request_source)
                }
            }
        }
    };
}
void append_statistic_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)});
    }
}
nlohmann::json datoviz_observation_json(
    const Datoviz_Frame_Observation& value) {
    const auto milliseconds = [](std::uint64_t nanoseconds) {
        return static_cast<double>(nanoseconds) / 1'000'000.0;
    };
    nlohmann::json result{
        {"render_sequence", value.render_sequence},
        {"path", magic_enum::enum_name(value.path)},
        {"gpu_timing_requested", value.gpu_timing_requested},
        {"readback_requested", value.readback_requested},
        {"controller_input_applied", value.controller_input_applied},
        {
            "prepare_released_after_submission",
            value.prepare_released_after_submission
        },
        {
            "timings_ms", {
                {
                    "queue_submit_wait",
                    milliseconds(value.queue_submit_wait_ns)
                },
                {"target_acquire", milliseconds(value.target_acquire_ns)},
                {"structure_check", milliseconds(value.structure_check_ns)},
                {"apply", milliseconds(value.apply_ns)},
                {"query", milliseconds(value.query_ns)},
                {"runtime_plan", milliseconds(value.runtime_plan_ns)},
                {"runtime_execute", milliseconds(value.runtime_execute_ns)},
                {"mvp_update", milliseconds(value.mvp_update_ns)},
                {"frame_begin", milliseconds(value.frame_begin_ns)},
                {"frame_plan", milliseconds(value.frame_plan_ns)},
                {"external_register", milliseconds(value.external_register_ns)},
                {"frame_attach", milliseconds(value.frame_attach_ns)},
                {"frame_execute", milliseconds(value.frame_execute_ns)},
                {"frame_finish", milliseconds(value.frame_finish_ns)},
                {"drp_validation", milliseconds(value.drp_validation_ns)},
                {"drp_state", milliseconds(value.drp_state_ns)},
                {"drp_buffer_create", milliseconds(value.drp_buffer_create_ns)},
                {"drp_texture_create", milliseconds(value.drp_texture_create_ns)},
                {"drp_shader_create", milliseconds(value.drp_shader_create_ns)},
                {"drp_shader_compile", milliseconds(value.drp_shader_compile_ns)},
                {
                    "drp_shader_module_create",
                    milliseconds(value.drp_shader_module_create_ns)
                },
                {"drp_pipeline_create", milliseconds(value.drp_pipeline_create_ns)},
                {"drp_binding_create", milliseconds(value.drp_binding_create_ns)},
                {"drp_upload", milliseconds(value.drp_upload_ns)},
                {"drp_transfer", milliseconds(value.drp_transfer_ns)},
                {"drp_record", milliseconds(value.drp_record_ns)},
                {"submit", milliseconds(value.submit_ns)},
                {
                    "gpu_completion_observation",
                    milliseconds(value.gpu_completion_observation_ns)
                },
                {
                    "completion_task_queue",
                    milliseconds(value.completion_task_queue_ns)
                },
                {"readback", milliseconds(value.readback_ns)}
            }
        },
        {
            "traffic", {
                {"uploaded_bytes", value.uploaded_bytes},
                {"readback_bytes", value.readback_bytes},
                {"pipeline_create_count", value.drp_pipeline_create_count}
            }
        },
        {
            "frame_plan", {
                {"resource_version", value.artifact_resource_version},
                {"frame_index", value.artifact_frame_index},
                {"status", value.artifact_status}
            }
        },
        {
            "validation", {
                {"performed", value.validation_performed},
                {"ok", value.validation_ok},
                {"code", value.validation_code},
                {"command_index", value.validation_command_index}
            }
        }
    };
    if (value.gpu) {
        result["gpu_ms"] = {
            {"render", milliseconds(value.gpu->render_ns)},
            {"transition", milliseconds(value.gpu->transition_ns)},
            {"copy", milliseconds(value.gpu->copy_ns)},
            {"total", milliseconds(value.gpu->total_ns)}
        };
    }
    if (!value.artifact_json.empty()) result["frame_plan"]["artifact_json"] = value.artifact_json;
    return result;
}
}
nlohmann::json taskflow_trace_json(
    const Taskflow_Frame_Trace& trace,
    const nlohmann::json& captured_components,
    const nlohmann::json& captured_backend) {
    nlohmann::json markers = nlohmann::json::object();
    for (const auto& marker : trace.markers)
        markers[magic_enum::enum_name(marker.marker)] =
        static_cast<double>(marker.elapsed_ns) / 1'000'000.0;
    nlohmann::json measurements = nlohmann::json::object();
    for (const auto& measurement : trace.measurements)
        measurements[magic_enum::enum_name(measurement.measurement)] =
        static_cast<double>(measurement.value_ns) / 1'000'000.0;
    nlohmann::json graphs = nlohmann::json::array();
    std::unordered_map<std::uint64_t, std::string> node_ids;
    for (const auto& graph : trace.graphs) {
        nlohmann::json nodes = nlohmann::json::array();
        for (const auto& node : graph.nodes) {
            node_ids.emplace(node.native_id, node.node_id);
            nlohmann::json predecessors = nlohmann::json::array();
            for (const auto native_id : node.predecessors) predecessors.push_back(std::to_string(native_id));
            nlohmann::json successors = nlohmann::json::array();
            for (const auto native_id : node.successors) successors.push_back(std::to_string(native_id));
            nlohmann::json attributes = nlohmann::json::object();
            for (const auto& [key, value] : node.attributes) attributes[key] = value;
            nlohmann::json encoded{
                {"native_id", std::to_string(node.native_id)}, {"id", node.node_id},
                {"parent_id", node.parent_node_id}, {"name", node.name},
                {"type", node.type}, {"predecessors", std::move(predecessors)},
                {"successors", std::move(successors)},
                {"attributes", std::move(attributes)}
            };
            const auto owner = encoded["attributes"].value(
                "owner_component", std::string{});
            if (!owner.empty() && captured_components.contains(owner)) {
                const auto& captured = captured_components.at(owner);
                encoded["owner"] = {
                    {"component", owner},
                    {"label", captured.value("label", owner)},
                    {"kind", captured.value("kind", std::string{})}
                };
                encoded["prop"] = captured.value("prop", nlohmann::json::object());
                encoded["state"] = captured.value("state", nlohmann::json::object());
            }
            nodes.push_back(std::move(encoded));
        }
        graphs.push_back({
            {"stage", graph.stage}, {"name", graph.taskflow_name},
            {"submitted_ms", graph.submitted_ms},
            {"finished_ms", graph.finished_ms},
            {"completed", graph.completed}, {"nodes", std::move(nodes)}
        });
    }
    nlohmann::json executions = nlohmann::json::array();
    for (const auto& task : trace.tasks) {
        const auto found = node_ids.find(task.native_id);
        nlohmann::json cooperative_waits = nlohmann::json::array();
        for (const auto& wait : task.cooperative_waits) {
            cooperative_waits.push_back({
                {"started_ms", wait.started_ms},
                {"finished_ms", wait.finished_ms}});
        }
        executions.push_back({
            {"native_id", std::to_string(task.native_id)},
            {"node_id", found == node_ids.end() ? std::string{} : found->second},
            {"worker_id", task.worker_id},
            {"worker_queue_size", task.worker_queue_size},
            {"worker_queue_capacity", task.worker_queue_capacity},
            {"ready_ms", task.ready_ms}, {"entered_ms", task.entered_ms},
            {"started_ms", task.started_ms}, {"finished_ms", task.finished_ms},
            {"completed_ms", task.completed_ms},
            {"duration_ms", task.duration_ms},
            {"cooperative_wait_ms", task.cooperative_wait_ms},
            {"cooperative_waits", std::move(cooperative_waits)},
            {"observer_entry_ms", task.observer_entry_ms},
            {"observer_exit_ms", task.observer_exit_ms},
            {"queue_wait_ms", task.queue_wait_ms}
        });
    }
    nlohmann::json result{
        {"sequence", trace.identity.sequence},
        {"correlation_id", trace.identity.correlation_id},
        {"request_source", magic_enum::enum_name(trace.request_source)},
        {"frame_policy", frame_policy_state_json(trace.frame_policy)},
        {"created_time_unix_ns", trace.created_time_unix_ns},
        {"worker_count", trace.worker_count},
        {"markers", std::move(markers)},
        {"measurements", std::move(measurements)},
        {"graphs", std::move(graphs)},
        {"executions", std::move(executions)}
    };
    if (!captured_backend.empty()) result["datoviz"] = captured_backend;
    return result;
}
namespace {
[[nodiscard]] Event_Timeline_Time input_timeline_time(
    double time_milliseconds) {
    constexpr long double nanoseconds_per_millisecond{1'000'000.0L};
    constexpr long double maximum_milliseconds =
    static_cast<long double>(std::numeric_limits<std::uint64_t>::max()) /
    nanoseconds_per_millisecond;
    if (!std::isfinite(time_milliseconds) || time_milliseconds < 0.0 ||
        static_cast<long double>(time_milliseconds) > maximum_milliseconds)
        throw std::invalid_argument(
            "input time_milliseconds must be finite, non-negative and representable");
    return Event_Timeline_Time{
        static_cast<std::uint64_t>(
            static_cast<long double>(time_milliseconds) *
            nanoseconds_per_millisecond)
    };
}
template <typename Scene_Object>
void dispatch_plot_input(Scene_Object& scene, const Plot_Input_Event& input) {
    const auto occurred_at = input_timeline_time(input.time_milliseconds);
    const auto dispatch = [&](auto event) {
        scene.template submit_stream<aethera::Scene_Event_Stream_Tag>(std::move(event));
    };
    const auto apply_pointer = [&](auto& event) {
        event.position = input.position;
        event.global_position = input.global_position;
        event.button = input.button;
        event.buttons = input.buttons;
        event.modifiers = input.modifiers;
    };
    switch (input.type) {
    case Event_Type::pointer_move:
    case Event_Type::pointer_press:
    case Event_Type::pointer_release: {
        auto event = scene.template make_event<Basic_Pointer_Event<Point_F>>(
            input.type, occurred_at);
        apply_pointer(*event);
        dispatch(std::move(event));
        break;
    }
    case Event_Type::wheel: {
        auto event = scene.template make_event<Basic_Wheel_Event<Point_F>>(
            occurred_at);
        apply_pointer(*event);
        event->pixel_delta_x = input.pixel_delta_x;
        event->pixel_delta_y = input.pixel_delta_y;
        event->angle_delta_x = input.angle_delta_x;
        event->angle_delta_y = input.angle_delta_y;
        dispatch(std::move(event));
        break;
    }
    case Event_Type::key_press:
    case Event_Type::key_release: {
        auto event = scene.template make_event<Key_Event>(
            input.type, occurred_at);
        event->key = input.key;
        event->native_key = input.native_key;
        event->modifiers = input.modifiers;
        event->auto_repeat = input.auto_repeat;
        dispatch(std::move(event));
        break;
    }
    default: dispatch(scene.template make_event<Event>(input.type, occurred_at));
        break;
    }
}
}
struct Plot::Private {
    using Scene = std::variant<std::unique_ptr<Scene_2D>, std::unique_ptr<Scene_3D>>;
    using Frame = std::variant<std::unique_ptr<Frame_2D>, std::unique_ptr<Frame_3D>>;
    enum struct Frame_State : std::uint8_t {
        available,
        rendering, /* Scene::advance -> Plot pixel publish，不可重入。 */
        consuming /* 外接 Taskflow 正在消费已发布帧；允许下一帧渲染。 */
    };
    enum struct Render_Admission_State : std::uint8_t {
        ready,
        rendering,
        frame_slots_exhausted
    };
    struct Managed_Frame {
        std::chrono::microseconds presentation_time{}; /* 共享页面时钟产生的媒体时间戳。 */
        std::chrono::steady_clock::time_point tick_issued_at{}; /* 本逻辑帧请求进入 Plot 的时刻。 */
        std::chrono::steady_clock::time_point submitted_at{}; /* Scene 接受本逻辑帧的时刻。 */
        Frame frame{}; /* 三缓冲物理槽拥有且反复承载逻辑帧。 */
        std::atomic<Frame_State> state{Frame_State::available}; /* 本槽唯一生命周期状态。 */
        std::atomic_size_t diagnostic_readers{}; /* 无锁诊断读取认领；非零时该槽不可复用。 */
        std::uint64_t statistics_generation{}; /* 与本槽中完成帧统计共同发布。 */
        Frame_Statistics_State statistics{}; /* 统计结果归属当前完成帧，不在 Plot 复制。 */
        std::atomic<Managed_Frame*> retired_next{}; /* 无锁退役队列的槽内侵入链接。 */
    };
    struct Consumer {
        Stream_Handler handler;
        std::uint32_t width{};
        std::uint32_t height{};
    };
    using Consumer_Map = std::unordered_map<Stream_Id, Consumer>;
    struct Stream_Snapshot {
        std::shared_ptr<const Consumer_Map> consumers;
        std::uint32_t width{};
        std::uint32_t height{};
    };
    std::unique_ptr<Scene_View> view;
    std::once_flag start_once;
    std::weak_ptr<Plot> lifetime{}; /* 仅用于 completion 后重新投递 Taskflow，避免在 Scene callback 内重入 render。 */
    std::atomic<std::shared_ptr<Plot>> frame_policy_lifetime{}; /* 任一物理帧被借用期间由策略保活整个 Plot；最后一槽归还后释放。 */
    std::atomic<std::shared_ptr<const Consumer_Map>> consumers{
        std::make_shared<const Consumer_Map>()
    }; /* 低频订阅修改发布不可变版本。 */
    std::atomic_uint64_t next_stream_id{1};
    std::atomic<std::shared_ptr<const std::string>> terminal_failure{}; /* 首次 Plot Unknown Failure 的唯一终止状态。 */
    std::uint64_t next_frame_sequence{1};
    std::unique_ptr<Frame_Policy> frame_policy_2d{};
    std::unique_ptr<Frame_Policy_3D> frame_policy_3d{};
    Frame_Scheduler::Timer frame_timer{}; /* 每 Plot/Scene 只有轻量时间轮节点，不持有线程。 */
    static constexpr std::size_t scene_frame_capacity{3};
    std::array<Managed_Frame, scene_frame_capacity> frame_slots{}; /* 帧策略拥有并反复调度的稳定三缓冲；Scene 只借用。 */
    std::atomic<Managed_Frame*> latest_statistics_frame{}; /* 只定位权威帧槽，不保存统计副本。 */
    std::atomic<Managed_Frame*> retired_frames{}; /* 完成回调返回、唯一帧策略写者消费的物理帧。 */
    Scene scene; /* 析构顺序保证 Scene 先停止，再释放物理帧。 */
    moodycamel::ConcurrentQueue<Plot_Render_Tick> tick_requests{}; /* 多生产者提交、唯一短任务消费的帧请求流。 */
    std::optional<Plot_Render_Tick> deferred_tick{}; /* 仅 tick consumer 任务访问的 latest 延后请求。 */
    std::atomic_uint64_t consumer_work_generation{}; /* tick 或退休帧入队后推进，关闭 consumer 尾部唤醒竞争窗口。 */
    std::atomic_bool tick_task_scheduled{}; /* 唯一短任务准入；不占用 Worker 等待。 */
    std::atomic<Render_Admission_State> render_admission{Render_Admission_State::ready}; /* Plot 渲染准入及物理槽背压的唯一状态源。 */
    std::chrono::steady_clock::time_point clock_origin{std::chrono::steady_clock::now()};
    std::atomic_size_t taskflow_trace_remaining{}; /* 尚待标记的实际渲染帧数。 */
    static constexpr std::size_t maximum_taskflow_trace_frames{120};
    /* 高 32 位 requested，低 32 位 captured。每槽只发布一次不可变 Trace，
     * GET 直接读取已发布槽位，不复制或重排整个历史容器。 */
    std::atomic_uint64_t taskflow_trace_control{};
    std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
               maximum_taskflow_trace_frames> taskflow_trace_slots{};
    Frame_Statistics_Accumulator completed_frame_statistics{diagnostic_window_capacity};
    std::uint64_t applied_statistics_generation{}; /* 仅完成帧退役任务读写。 */
    std::atomic_uint64_t statistics_generation{}; /* reset 只推进代次，不触碰单写者累加器。 */
    template <typename Scene_Object>
    Private(std::unique_ptr<Scene_Object> value_scene,
            std::unique_ptr<Scene_View> value_view) : view(std::move(value_view)), scene(std::move(value_scene)) {
        if constexpr (std::same_as<Scene_Object, Scene_2D>) {
            auto built_policy = Frame_Policy::Builder<Frame_Policy>{}.build();
            if (!built_policy) throw std::logic_error("2D Frame Policy dependency graph is invalid");
            frame_policy_2d = std::move(*built_policy);
        }
        else {
            auto built_policy =
            Frame_Policy_3D::Builder<Frame_Policy_3D>{}.build();
            if (!built_policy) throw std::logic_error("3D Frame Policy dependency graph is invalid");
            frame_policy_3d = std::move(*built_policy);
            frame_policy_3d->set_mode(Frame_Pacing_Mode::maximum_rate);
            static_cast<void>(frame_policy_3d->consume_events());
        }
        for (auto& slot : frame_slots) {
            if constexpr (std::same_as<Scene_Object, Scene_2D>) slot.frame = std::make_unique<Frame_2D>(Frame_Identity{});
            else slot.frame = std::make_unique<Frame_3D>(Frame_Identity{});
        }
    }
    template <typename Callback>
    decltype(auto) with_frame_policy(Callback&& callback) {
        if (frame_policy_2d) return std::forward<Callback>(callback)(*frame_policy_2d);
        return std::forward<Callback>(callback)(*frame_policy_3d);
    }
    template <typename Callback>
    decltype(auto) with_frame_policy(Callback&& callback) const {
        if (frame_policy_2d) return std::forward<Callback>(callback)(*frame_policy_2d);
        return std::forward<Callback>(callback)(*frame_policy_3d);
    }
    [[nodiscard]] const Frame_Policy::State& pacing_state() const noexcept {
        if (frame_policy_2d) return frame_policy_2d->read_state<Frame_Policy::Base_Tag>();
        return frame_policy_3d->read_state<Frame_Policy_3D::Base_Tag>().common;
    }
    [[nodiscard]] bool consume_frame_policy_events() {
        return with_frame_policy(
            [](auto& policy) {
                return policy.consume_events();
            });
    }
    [[nodiscard]] nlohmann::json schema() const;
    [[nodiscard]] Stream_Snapshot stream_snapshot() const;
    void publish(std::shared_ptr<const Plot_Stream_Frame> frame) noexcept;
    void submit_tick_request(Plot_Render_Tick tick);
    void keep_latest_tick(const Plot_Render_Tick& tick);
    void arm_tick_consumer(std::weak_ptr<Plot> lifetime);
    void release_render_admission(std::weak_ptr<Plot> lifetime);
    void retain_frame_policy_lifetime();
    void release_frame_policy_lifetime_if_idle();
    void consume_tick(std::weak_ptr<Plot> lifetime);
    void refresh_schedule();
    void clock_tick(const Plot_Render_Tick& tick);
    void render_frame(Plot_Render_Tick tick);
    void publish_completed_frame(not_null<Render_Frame*> completed);
    void consume_completed_frame(not_null<Render_Frame*> frame);
    void retire_completed_frame(not_null<Render_Frame*> frame);
    void consume_retired_frames();
    void finalize_retired_frame(not_null<Render_Frame*> frame);
    [[nodiscard]] bool mark_taskflow_trace(Render_Frame& frame);
    void store_trace(std::atomic_uint64_t& control,
                     std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
                                maximum_taskflow_trace_frames>& slots,
                     const Taskflow_Frame_Trace& trace,
                     const nlohmann::json& captured_components = {},
                     const nlohmann::json& captured_backend = {});
    [[nodiscard]] nlohmann::json trace_response(
        const std::atomic_uint64_t& control,
        const std::atomic_size_t& remaining,
        const std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
                         maximum_taskflow_trace_frames>& slots) const;
    void fail(std::exception_ptr failure) noexcept;
};
void Plot::Private::fail(std::exception_ptr failure) noexcept {
    try {
        auto description = std::make_shared<const std::string>(
            exception_description(failure));
        std::shared_ptr<const std::string> empty;
        if (!terminal_failure.compare_exchange_strong(
            empty, description, std::memory_order_acq_rel,
            std::memory_order_acquire))
            return;
        const auto output = std::make_shared<const Plot_Stream_Frame>(
            Plot_Stream_Frame{
                nlohmann::json{
                    {"kind", "plot_error"},
                    {"protocol", "aethera.plot.stream"},
                    {"version", plot_stream_protocol_version},
                    {"message", *description}
                }.dump(),
                {}
            });
        publish(std::move(output));
    }
    catch (...) {}
}
nlohmann::json Plot::Private::schema() const {
    auto result = view->schema();
    auto analysis = with_frame_policy(
        [](const auto& policy) {
            return frame_policy_schema(policy);
        });
    if (frame_policy_3d) {
        analysis["fields"].push_back({
            {"key", "pipeline_capacity"}, {"label", "3D 帧槽容量"},
            {"editor", "integer"}, {"editable", false}, {"value", 3},
            {"description", "3D 独立策略最多允许三帧处于 Prepare、GPU 和退休阶段。"},
            {"technical_description", "3D-only submitted-driven pipeline capacity; 2D policy is unchanged."}
        });
    }
    const auto generator = view->data_generator_schema();
    if (!generator.is_null()) analysis["data_generator"] = generator;
    result["frame_analysis"] = std::move(analysis);
    return result;
}
Plot::Private::Stream_Snapshot Plot::Private::stream_snapshot() const {
    Stream_Snapshot result;
    result.consumers = consumers.load(std::memory_order_acquire);
    for (const auto& [id, consumer] : *result.consumers) {
        static_cast<void>(id);
        if (consumer.width == 0 || consumer.height == 0) continue;
        result.width = std::max(result.width, consumer.width);
        result.height = std::max(result.height, consumer.height);
    }
    result.width = std::clamp(result.width == 0 ? 320U : result.width, 160U, 1920U) & ~1U;
    result.height = std::clamp(result.height == 0 ? 192U : result.height, 120U, 1080U) & ~1U;
    return result;
}
void Plot::Private::publish(
    std::shared_ptr<const Plot_Stream_Frame> frame) noexcept {
    if (!frame) return;
    try {
        const auto snapshot = stream_snapshot();
        std::vector<Stream_Id> failed_consumers;
        for (const auto& [id, consumer] : *snapshot.consumers) {
            if (!consumer.handler) continue;
            try {
                consumer.handler(frame);
            }
            catch (...) {
                failed_consumers.push_back(id);
            }
        }
        if (failed_consumers.empty()) return;
        auto current = consumers.load(std::memory_order_acquire);
        for (;;) {
            auto next = std::make_shared<Consumer_Map>(*current);
            for (const auto id : failed_consumers) next->erase(id);
            std::shared_ptr<const Consumer_Map> desired = next;
            if (consumers.compare_exchange_weak(
                current, desired, std::memory_order_release,
                std::memory_order_acquire))
                break;
        }
    }
    catch (...) {}
}
void Plot::Private::refresh_schedule() {
    if (!frame_timer.valid()) return;
    const auto current_consumers = consumers.load(std::memory_order_acquire);
    const auto& pacing = pacing_state();
    if (!pacing.render_enabled || current_consumers->empty()) {
        frame_timer.cancel();
        return;
    }
    if (pacing.mode == Frame_Pacing_Mode::fixed_rate) {
        frame_timer.start_periodic(pacing.fixed_rate_fps);
        return;
    }
    frame_timer.cancel();
    if (pacing.mode != Frame_Pacing_Mode::maximum_rate) return;
    const auto now = std::chrono::steady_clock::now();
    submit_tick_request(Plot_Render_Tick{
        .issued_at = now,
        .time_milliseconds = std::chrono::duration<double, std::milli>(
            now - clock_origin).count(),
        .source = Frame_Request_Source::maximum_rate
    });
    arm_tick_consumer(lifetime);
}
void Plot::Private::submit_tick_request(Plot_Render_Tick tick) {
    const auto source = tick.source;
    const auto issued_at = tick.issued_at;
    if (!tick_requests.enqueue(std::move(tick))) throw std::bad_alloc{};
    with_frame_policy([&](auto& policy) {
        policy.record_request(source, issued_at);
    });
    consumer_work_generation.fetch_add(1, std::memory_order_release);
}
void Plot::Private::keep_latest_tick(const Plot_Render_Tick& tick) {
    const auto priority = [](Frame_Request_Source source) {
        switch (source) {
        case Frame_Request_Source::unspecified: return 0;
        case Frame_Request_Source::periodic: return 0;
        case Frame_Request_Source::maximum_rate: return 1;
        case Frame_Request_Source::immediate: return 2;
        }
        return 0;
    };
    if (deferred_tick) {
        const auto current_priority = priority(deferred_tick->source);
        const auto next_priority = priority(tick.source);
        with_frame_policy([](auto& policy) {
            policy.record_request_coalesced();
        });
        if (current_priority > next_priority ||
            (current_priority == next_priority &&
                deferred_tick->issued_at >= tick.issued_at))
            return;
    }
    deferred_tick = tick;
}
void Plot::Private::arm_tick_consumer(std::weak_ptr<Plot> lifetime) {
    if (terminal_failure.load(std::memory_order_acquire)) return;
    if (tick_task_scheduled.exchange(true, std::memory_order_acq_rel)) return;
    aethera::schedule_task("web.plot.tick.consume", [lifetime] {
        const auto plot = lifetime.lock();
        if (!plot) return;
        try {
            plot->d->consume_tick(lifetime);
        }
        catch (...) {
            plot->d->fail(std::current_exception());
        }
    });
}
void Plot::Private::release_render_admission(std::weak_ptr<Plot> lifetime) {
    auto expected = Render_Admission_State::rendering;
    if (!render_admission.compare_exchange_strong(
        expected, Render_Admission_State::ready,
        std::memory_order_acq_rel, std::memory_order_acquire))
        return;
    const auto& pacing = pacing_state();
    if (pacing.render_enabled && pacing.mode == Frame_Pacing_Mode::maximum_rate) {
        const auto current_consumers = consumers.load(std::memory_order_acquire);
        if (!current_consumers->empty()) {
            const auto now = std::chrono::steady_clock::now();
            submit_tick_request(Plot_Render_Tick{
                .issued_at = now,
                .time_milliseconds = std::chrono::duration<double, std::milli>(
                    now - clock_origin).count(),
                .source = Frame_Request_Source::maximum_rate
            });
        }
    }
    arm_tick_consumer(std::move(lifetime));
}
void Plot::Private::retain_frame_policy_lifetime() {
    if (frame_policy_lifetime.load(std::memory_order_acquire)) return;
    const auto owner = lifetime.lock();
    if (!owner)
        throw std::logic_error(
            "Plot frame policy cannot retain an expired Plot");
    std::shared_ptr<Plot> empty;
    static_cast<void>(frame_policy_lifetime.compare_exchange_strong(
        empty, owner, std::memory_order_release, std::memory_order_acquire));
}
void Plot::Private::release_frame_policy_lifetime_if_idle() {
    if (std::ranges::any_of(frame_slots, [](const Managed_Frame& slot) {
        return slot.state.load(std::memory_order_acquire) !=
        Frame_State::available;
    }))
        return;
    frame_policy_lifetime.store({}, std::memory_order_release);
}
void Plot::Private::consume_tick(std::weak_ptr<Plot> lifetime) {
    const auto observed_generation =
    consumer_work_generation.load(std::memory_order_acquire);
    consume_retired_frames();
    if (consume_frame_policy_events()) refresh_schedule();
    Plot_Render_Tick requested;
    while (tick_requests.try_dequeue(requested)) keep_latest_tick(requested);
    if (render_admission.load(std::memory_order_acquire) ==
        Render_Admission_State::ready) {
        auto tick = std::exchange(deferred_tick, {});
        if (tick) clock_tick(*tick);
    }
    consume_retired_frames();
    if (consume_frame_policy_events()) refresh_schedule();
    tick_task_scheduled.store(false, std::memory_order_release);
    if (consumer_work_generation.load(std::memory_order_acquire) !=
        observed_generation ||
        retired_frames.load(std::memory_order_acquire) ||
        (render_admission.load(std::memory_order_acquire) ==
            Render_Admission_State::ready && deferred_tick))
        arm_tick_consumer(std::move(lifetime));
}
void Plot::Private::clock_tick(const Plot_Render_Tick& tick) {
    if (terminal_failure.load(std::memory_order_acquire)) return;
    const auto& pacing = pacing_state();
    const bool accepted = pacing.render_enabled &&
    (tick.source == Frame_Request_Source::immediate ||
        (tick.source == Frame_Request_Source::periodic &&
            pacing.mode == Frame_Pacing_Mode::fixed_rate) ||
        (tick.source == Frame_Request_Source::maximum_rate &&
            pacing.mode == Frame_Pacing_Mode::maximum_rate));
    if (!accepted) {
        with_frame_policy([](auto& policy) {
            policy.record_policy_rejection();
        });
        return;
    }
    with_frame_policy([&](auto& policy) {
        policy.record_request_accepted(tick.source);
    });
    render_frame(tick);
}
bool Plot::Private::mark_taskflow_trace(Render_Frame& frame) {
    auto remaining = taskflow_trace_remaining.load(std::memory_order_acquire);
    while (remaining != 0) {
        if (taskflow_trace_remaining.compare_exchange_weak(
            remaining, remaining - 1, std::memory_order_acq_rel,
            std::memory_order_acquire)) {
            frame.request_taskflow_trace();
            return true;
        }
    }
    return false;
}
void Plot::Private::store_trace(
    std::atomic_uint64_t& control,
    std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
               maximum_taskflow_trace_frames>& slots,
    const Taskflow_Frame_Trace& value,
    const nlohmann::json& captured_components,
    const nlohmann::json& captured_backend) {
    auto trace = std::make_shared<const nlohmann::json>(
        taskflow_trace_json(value, captured_components, captured_backend));
    auto state = 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) return;
        slots[captured].store(trace, std::memory_order_release);
        const auto next = (static_cast<std::uint64_t>(requested) << 32U) |
        static_cast<std::uint64_t>(captured + 1U);
        if (control.compare_exchange_weak(
            state, next, std::memory_order_release,
            std::memory_order_acquire))
            return;
    }
}
nlohmann::json Plot::Private::trace_response(
    const std::atomic_uint64_t& control,
    const std::atomic_size_t& remaining,
    const std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
                     maximum_taskflow_trace_frames>& slots) const {
    nlohmann::json frames = nlohmann::json::array();
    const auto state = 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 trace = slots[index].load(std::memory_order_acquire)) frames.push_back(*trace);
    const auto left = remaining.load(std::memory_order_acquire);
    return {
        {"protocol", "aethera.taskflow.frames"}, {"version", 1},
        {"requested", requested}, {"remaining", left},
        {"captured", frames.size()},
        {"complete", requested != 0 && frames.size() == requested},
        {"frames", std::move(frames)}
    };
}
void Plot::Private::render_frame(Plot_Render_Tick tick) {
    if (terminal_failure.load(std::memory_order_acquire)) return;
    const auto streams = stream_snapshot();
    const auto& pacing = pacing_state();
    const bool direct_diagnostics_frame =
    streams.consumers->empty() &&
    tick.source == Frame_Request_Source::immediate;
    if (!pacing.render_enabled ||
        (streams.consumers->empty() && !direct_diagnostics_frame))
        return;
    auto admission_expected = Render_Admission_State::ready;
    if (!render_admission.compare_exchange_strong(
        admission_expected, Render_Admission_State::rendering,
        std::memory_order_acq_rel,
        std::memory_order_acquire)) {
        keep_latest_tick(tick);
        return;
    }
    std::size_t slot_index{};
    Managed_Frame* managed{};
    /*
     * 只有 rendering 槽受 Scene 不可重入门约束；consuming 槽表示上一帧
     * 已经完成 Plot 像素发布，外接 Drogon 图库流仍可继续持有该物理帧的
     * 诊断生命周期。只要还有 available 槽，下一帧即可进入。
     */
    for (std::size_t index = 0; index < frame_slots.size(); ++index) {
        auto* candidate = &frame_slots[index];
        auto* published = candidate;
        const bool was_latest = latest_statistics_frame.compare_exchange_strong(
            published, nullptr, std::memory_order_acq_rel,
            std::memory_order_acquire);
        if (candidate->diagnostic_readers.load(std::memory_order_acquire) != 0) {
            if (was_latest) {
                Managed_Frame* empty{};
                static_cast<void>(latest_statistics_frame.compare_exchange_strong(
                    empty, candidate, std::memory_order_release,
                    std::memory_order_relaxed));
            }
            continue;
        }
        auto expected = Frame_State::available;
        if (!candidate->state.compare_exchange_strong(
            expected, Frame_State::rendering,
            std::memory_order_acq_rel, std::memory_order_acquire)) {
            if (was_latest) {
                Managed_Frame* empty{};
                static_cast<void>(latest_statistics_frame.compare_exchange_strong(
                    empty, candidate, std::memory_order_release,
                    std::memory_order_relaxed));
            }
            continue;
        }
        slot_index = index;
        managed = &frame_slots[index];
        break;
    }
    if (!managed) {
        with_frame_policy([](auto& policy) {
            policy.record_frame_slot_backpressure();
        });
        keep_latest_tick(tick);
        /*
         * 三个槽都仍被外接消费者持有时，只保留 latest pending。这里绝不能
         * 立即 arm tick consumer，否则会在没有任何槽可用期间形成
         * consume -> no slot -> consume 的 Taskflow 任务风暴。真正的唤醒点
         * 是 retire_completed_frame：某个 consuming 槽变回 available 后只唤醒一次。
         */
        render_admission.store(
            Render_Admission_State::frame_slots_exhausted,
            std::memory_order_release);
        return;
    }
    retain_frame_policy_lifetime();
    managed->presentation_time =
    std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::duration<double, std::milli>(tick.time_milliseconds));
    managed->tick_issued_at = tick.issued_at;
    managed->submitted_at = {};
    const auto rollback_unsubmitted = [this, slot_index] {
        auto& slot = frame_slots[slot_index];
        auto expected = Frame_State::rendering;
        static_cast<void>(slot.state.compare_exchange_strong(
            expected, Frame_State::available, std::memory_order_acq_rel,
            std::memory_order_acquire));
        release_frame_policy_lifetime_if_idle();
    };
    bool taskflow_trace_claimed{};
    const auto restore_taskflow_trace_claim = [this, &taskflow_trace_claimed] {
        if (!std::exchange(taskflow_trace_claimed, false)) return;
        taskflow_trace_remaining.fetch_add(1, std::memory_order_release);
    };
    try {
        if (streams.consumers->empty()) {
            tick.width = std::clamp(tick.width, 160U, 1920U) & ~1U;
            tick.height = std::clamp(tick.height, 120U, 1080U) & ~1U;
        }
        else {
            tick.width = streams.width;
            tick.height = streams.height;
        }
        const std::uint64_t sequence = next_frame_sequence++;
        const Frame_Identity identity{
            sequence, tick.sequence == 0 ? sequence : tick.sequence
        };
        const auto request_source = tick.source;
        const auto& policy_state = pacing_state();
        Render_Frame* logical_frame{};
        if (auto* frame_2d = std::get_if<std::unique_ptr<Frame_2D>>(&managed->frame)) {
            (*frame_2d)->begin(identity, Frame_2D::native_pixel_format,
                               request_source, policy_state);
            logical_frame = frame_2d->get();
        }
        else {
            auto& frame_3d = std::get<std::unique_ptr<Frame_3D>>(managed->frame);
            frame_3d->begin(identity,
                            pacing.pixel_delivery_enabled
                                ? Frame_3D_Output::pixels
                                : Frame_3D_Output::diagnostics,
                            Frame_3D::native_pixel_format, request_source,
                            policy_state);
            logical_frame = frame_3d.get();
        }
        taskflow_trace_claimed = mark_taskflow_trace(*logical_frame);
        /*
         * 各图的采样、网格构造和属性读取都在 Plot 自己的准备域完成。
         * 进入 Scene::render 后只剩已经准备好的 Visual 批次与轻量提交；
         * 共享 Render Domain 不承担业务数据生成。
         */
        logical_frame->mark(Frame_Trace_Marker::plot_update_started);
        const auto update_started = std::chrono::steady_clock::now();
        view->update(tick);
        const auto update_elapsed = std::chrono::steady_clock::now() - update_started;
        logical_frame->mark(Frame_Trace_Marker::plot_update_finished);
        const auto tick_queue_elapsed = tick.issued_at.time_since_epoch().count() == 0
                                            ? std::chrono::steady_clock::duration::zero()
                                            : update_started - tick.issued_at;
        const auto record_plot_measurements = [&](Render_Frame& frame) {
            const auto nanoseconds = [](std::chrono::steady_clock::duration duration) {
                return static_cast<std::uint64_t>(std::max<std::int64_t>(0,
                                                                         std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count()));
            };
            frame.record(Frame_Trace_Measurement::plot_tick_queue_ns,
                         nanoseconds(tick_queue_elapsed));
            frame.record(Frame_Trace_Measurement::plot_update_ns,
                         nanoseconds(update_elapsed));
        };
        if (auto* scene_2d = std::get_if<std::unique_ptr<Scene_2D>>(&scene)) {
            auto& output = *std::get<std::unique_ptr<Frame_2D>>(managed->frame);
            record_plot_measurements(output);
            (*scene_2d)->set<&Render_Scene_2D::Prop::viewport>(
                Size{static_cast<int>(tick.width), static_cast<int>(tick.height)});
            const auto result = (*scene_2d)->render(
                &output,
                [weak = lifetime](not_null<Frame_2D*> frame) {
                    if (auto owner = weak.lock()) {
                        try {
                            owner->d->publish_completed_frame(frame);
                            owner->d->consume_completed_frame(frame);
                            owner->d->retire_completed_frame(frame);
                        }
                        catch (...) {
                            owner->d->fail(std::current_exception());
                        }
                    }
                });
            if (!result) {
                with_frame_policy([](auto& policy) {
                    policy.record_scene_rejection();
                });
                rollback_unsubmitted();
                restore_taskflow_trace_claim();
                release_render_admission(lifetime);
            }
            else {
                taskflow_trace_claimed = false;
                managed->submitted_at = std::chrono::steady_clock::now();
                with_frame_policy([&](auto& policy) {
                    policy.record_frame_submitted(
                        sequence, request_source,
                        managed->submitted_at - managed->tick_issued_at);
                });
            }
            return;
        }
        auto& output = *std::get<std::unique_ptr<Frame_3D>>(managed->frame);
        record_plot_measurements(output);
        auto& scene_3d = std::get<std::unique_ptr<Scene_3D>>(scene);
        scene_3d->set<&Render_Scene_3D::Prop::viewport>(Extent{tick.width, tick.height});
        const auto weak = lifetime;
        const auto result = scene_3d->render(
            &output,
            Render_Scene_3D::Frame_Callbacks{
                .submitted = [weak](not_null<Frame_3D*>, bool) {
                    if (auto owner = weak.lock()) {
                        try {
                            owner->d->release_render_admission(weak);
                        }
                        catch (...) {
                            owner->d->fail(std::current_exception());
                        }
                    }
                },
                .completed = [weak](not_null<Frame_3D*> frame) {
                    if (auto owner = weak.lock()) {
                        try {
                            owner->d->publish_completed_frame(frame);
                            owner->d->consume_completed_frame(frame);
                            owner->d->retire_completed_frame(frame);
                        }
                        catch (...) {
                            owner->d->fail(std::current_exception());
                        }
                    }
                }
            });
        if (result == Render_Scene_3D::Render_Result::submitted) {
            taskflow_trace_claimed = false;
            managed->submitted_at = std::chrono::steady_clock::now();
            with_frame_policy([&](auto& policy) {
                policy.record_frame_submitted(
                    sequence, request_source,
                    managed->submitted_at - managed->tick_issued_at);
            });
            return;
        }
        with_frame_policy([](auto& policy) {
            policy.record_scene_rejection();
        });
        rollback_unsubmitted();
        restore_taskflow_trace_claim();
        release_render_admission(lifetime);
        if (result == Render_Scene_3D::Render_Result::backend_unavailable) throw std::runtime_error("3D render backend became unavailable before submission");
    }
    catch (...) {
        rollback_unsubmitted();
        restore_taskflow_trace_claim();
        release_render_admission(lifetime);
        throw;
    }
}
void Plot::Private::publish_completed_frame(
    not_null<Render_Frame*> completed) {
    Managed_Frame* managed{};
    for (auto& slot : frame_slots) {
        const auto frame = std::visit(
            [](const auto& value) -> not_null<Render_Frame*> {
                return not_null{value.get()};
            },
            slot.frame);
        if (frame.get() != completed.get()) continue;
        managed = &slot;
        break;
    }
    if (!managed) throw std::logic_error("completed frame has no owning Plot policy slot");
    if (managed->state.load(std::memory_order_acquire) !=
        Frame_State::rendering)
        throw std::logic_error("completed Plot frame is not rendering");
    const auto frame = completed;
    const auto& pacing = pacing_state();
    const auto identity = frame->identity();
    Frame_Identity rendered_identity = identity;
    std::shared_ptr<const std::vector<std::byte>> pixel_storage;
    Plot_Pixel_Layout pixel_layout{Plot_Pixel_Layout::rgba8};
    std::uint32_t width{};
    std::uint32_t height{};
    if (auto* frame_2d =
        std::get_if<std::unique_ptr<Frame_2D>>(&managed->frame)) {
        pixel_layout = Plot_Pixel_Layout::bgra8;
        const auto image = (*frame_2d)->image();
        width = static_cast<std::uint32_t>(image.width);
        height = static_cast<std::uint32_t>(image.height);
        if (pacing.pixel_delivery_enabled) {
            auto output = (*frame_2d)->output_pixels();
            pixel_storage = std::make_shared<const std::vector<std::byte>>(
                std::move(output.bytes));
            width = static_cast<std::uint32_t>(output.width);
            height = static_cast<std::uint32_t>(output.height);
        }
    }
    else {
        auto& frame_3d = std::get<std::unique_ptr<Frame_3D>>(managed->frame);
        rendered_identity = frame_3d->rendered_identity();
        const auto extent = frame_3d->extent();
        width = extent.width;
        height = extent.height;
        if (pacing.pixel_delivery_enabled && frame_3d->output() == Frame_3D_Output::pixels) pixel_storage = frame_3d->share_pixels();
    }
    auto pixels = std::make_shared<const Plot_Pixel_Frame>(Plot_Pixel_Frame{
        std::move(pixel_storage), pixel_layout, managed->presentation_time,
        identity.sequence, identity.correlation_id, rendered_identity.sequence,
        rendered_identity.correlation_id, width, height
    });
    const auto published = std::make_shared<const Plot_Stream_Frame>(
        Plot_Stream_Frame{{}, std::move(pixels)});
    const auto publish_started = std::chrono::steady_clock::now();
    publish(std::move(published));
    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>(
                          std::chrono::steady_clock::now() - publish_started)
                      .count())));
    auto expected = Frame_State::rendering;
    if (!managed->state.compare_exchange_strong(
        expected, Frame_State::consuming, std::memory_order_acq_rel,
        std::memory_order_acquire))
        throw std::logic_error("Plot frame left rendering before pixel publish");
}
void Plot::Private::consume_completed_frame(not_null<Render_Frame*> frame) {
    Managed_Frame* managed{};
    for (auto& slot : frame_slots) {
        const auto address = std::visit(
            [](const auto& value) -> not_null<Render_Frame*> {
                return not_null{value.get()};
            },
            slot.frame);
        if (address.get() != frame.get()) continue;
        if (slot.state.load(std::memory_order_acquire) != Frame_State::consuming) throw std::logic_error("completed Plot frame was not published");
        managed = &slot;
        break;
    }
    if (!managed) throw std::logic_error("frame callback has no owned Plot frame");
    /* Scene 完成即释放 Plot admission。媒体采样属于 Gallery 自己的独立时钟和
     * DAG，不再借 Plot 保存一套 post-publish 管线状态。 */
    if (std::holds_alternative<std::unique_ptr<Frame_2D>>(managed->frame))
        release_render_admission(lifetime);
}
void Plot::Private::consume_retired_frames() {
    for (;;) {
        auto* list = retired_frames.exchange(nullptr, std::memory_order_acq_rel);
        if (!list) break;
        std::vector<Managed_Frame*> frames;
        while (list) {
            auto* next = list->retired_next.exchange(
                nullptr, std::memory_order_relaxed);
            frames.push_back(list);
            list = next;
        }
        std::ranges::sort(frames, {}, [](const Managed_Frame* managed) {
            return std::visit(
                [](const auto& value) {
                    return value->identity().sequence;
                },
                managed->frame);
        });
        for (auto* managed : frames) {
            auto* frame = std::visit(
                [](const auto& value) -> Render_Frame* {
                    return value.get();
                },
                managed->frame);
            finalize_retired_frame(frame);
        }
    }
}
void Plot::Private::retire_completed_frame(not_null<Render_Frame*> frame) {
    Managed_Frame* managed{};
    for (auto& slot : frame_slots) {
        const auto address = std::visit(
            [](const auto& value) -> not_null<Render_Frame*> {
                return not_null{value.get()};
            },
            slot.frame);
        if (address.get() != frame.get()) continue;
        managed = &slot;
        break;
    }
    if (!managed) throw std::logic_error("retired frame has no owned Plot slot");
    auto* head = retired_frames.load(std::memory_order_relaxed);
    do {
        managed->retired_next.store(head, std::memory_order_relaxed);
    }
    while (!retired_frames.compare_exchange_weak(
        head, managed, std::memory_order_release,
        std::memory_order_relaxed));
    consumer_work_generation.fetch_add(1, std::memory_order_release);
    arm_tick_consumer(lifetime);
}
void Plot::Private::finalize_retired_frame(not_null<Render_Frame*> frame) {
    Managed_Frame* managed{};
    for (auto& slot : frame_slots) {
        const auto address = std::visit(
            [](const auto& value) -> not_null<Render_Frame*> {
                return not_null{value.get()};
            },
            slot.frame);
        if (address.get() == frame.get()) {
            managed = &slot;
            break;
        }
    }
    if (!managed) throw std::logic_error("retired frame has no owned Plot slot");
    const auto completion_latency =
    managed->submitted_at.time_since_epoch().count() == 0
        ? std::chrono::steady_clock::duration::zero()
        : std::chrono::steady_clock::now() - managed->submitted_at;
    std::optional<Datoviz_Frame_Observation> datoviz_observation;
    if (auto* frame_3d = dynamic_cast<Frame_3D*>(frame.get())) {
        datoviz_observation = frame_3d->take_datoviz_observation();
        if (!datoviz_observation)
            throw std::logic_error(
                "completed 3D frame has no Datoviz observation");
        frame_policy_3d->record_frame_completed(
            frame->identity().sequence, completion_latency,
            datoviz_observation->prepare_released_after_submission);
    }
    else {
        frame_policy_2d->record_frame_completed(
            frame->identity().sequence, completion_latency);
    }
    const auto statistics_generation_value =
    statistics_generation.load(std::memory_order_acquire);
    if (applied_statistics_generation != statistics_generation_value) {
        completed_frame_statistics.reset();
        applied_statistics_generation = statistics_generation_value;
    }
    managed->statistics = completed_frame_statistics.submit(*frame);
    managed->statistics_generation = statistics_generation_value;
    std::optional<Taskflow_Frame_Trace> captured_trace;
    nlohmann::json captured_components;
    nlohmann::json captured_backend;
    if (frame->taskflow_trace_requested()) {
        captured_trace.emplace(frame->take_taskflow_trace());
        std::unordered_set<std::uint64_t> executed_nodes;
        for (const auto& execution : captured_trace->tasks) executed_nodes.insert(execution.native_id);
        std::vector<std::string> executed_components;
        for (const auto& graph : captured_trace->graphs) {
            for (const auto& node : graph.nodes) {
                if (!executed_nodes.contains(node.native_id)) continue;
                const auto owner = std::ranges::find(
                    node.attributes, "owner_component",
                    &std::pair<std::string, std::string>::first);
                if (owner == node.attributes.end() || owner->second.empty() ||
                    std::ranges::find(executed_components, owner->second) !=
                    executed_components.end())
                    continue;
                executed_components.push_back(owner->second);
            }
        }
        captured_components = view->capture_components(executed_components);
        if (datoviz_observation) captured_backend = datoviz_observation_json(*datoviz_observation);
    }
    auto expected = Frame_State::consuming;
    if (!managed->state.compare_exchange_strong(
        expected, Frame_State::available, std::memory_order_acq_rel,
        std::memory_order_acquire))
        throw std::logic_error("retired Plot frame is not consuming");
    latest_statistics_frame.store(managed, std::memory_order_release);
    /* 物理槽是唯一背压原因；归还任意槽后只解除一次耗尽状态。 */
    auto admission = Render_Admission_State::frame_slots_exhausted;
    static_cast<void>(render_admission.compare_exchange_strong(
        admission, Render_Admission_State::ready,
        std::memory_order_acq_rel, std::memory_order_acquire));
    arm_tick_consumer(lifetime);
    release_frame_policy_lifetime_if_idle();
    if (captured_trace) {
        auto owner = lifetime;
        schedule_task("plot.taskflow.serialize", [owner,
                          trace = std::move(*captured_trace),
                          components = std::move(captured_components),
                          backend = std::move(captured_backend)]() mutable {
                          if (const auto plot = owner.lock())
                              plot->d->store_trace(
                                  plot->d->taskflow_trace_control,
                                  plot->d->taskflow_trace_slots, trace, components,
                                  backend);
                      });
    }
}
Plot::Plot(std::unique_ptr<Scene_2D> scene,
           std::unique_ptr<Scene_View> view) : d(std::make_unique<Private>(std::move(scene), std::move(view))) {}
Plot::Plot(std::unique_ptr<Scene_3D> scene,
           std::unique_ptr<Scene_View> view) : d(std::make_unique<Private>(std::move(scene), std::move(view))) {}
Plot::~Plot() = default;
void Plot::ensure_started() {
    std::call_once(d->start_once, [this] {
        const auto weak = weak_from_this();
        d->lifetime = weak;
        d->frame_timer = Frame_Scheduler::instance().make_timer(
            [weak](Frame_Scheduler::Tick tick) {
                if (const auto owner = weak.lock()) {
                    owner->schedule_render(Plot_Render_Tick{
                        .issued_at = tick.issued_at,
                        .sequence = tick.sequence,
                        .time_milliseconds = tick.time_milliseconds,
                        .source = Frame_Request_Source::periodic
                    });
                }
            });
        d->refresh_schedule();
    });
}
Plot::Stream_Id Plot::subscribe(Stream_Handler handler) {
    if (!handler) throw std::invalid_argument("Plot subscription requires a handler");
    ensure_started();
    const auto id = d->next_stream_id.fetch_add(1, std::memory_order_relaxed);
    const auto notification = handler;
    auto current = d->consumers.load(std::memory_order_acquire);
    for (;;) {
        auto next = std::make_shared<Private::Consumer_Map>(*current);
        next->emplace(id, Private::Consumer{handler});
        std::shared_ptr<const Private::Consumer_Map> desired = next;
        if (d->consumers.compare_exchange_weak(
            current, desired, std::memory_order_release,
            std::memory_order_acquire))
            break;
    }
    d->refresh_schedule();
    if (d->terminal_failure.load(std::memory_order_acquire)) {
        const auto failure = d->terminal_failure.load(std::memory_order_acquire);
        try {
            notification(std::make_shared<const Plot_Stream_Frame>(
                Plot_Stream_Frame{
                    nlohmann::json{
                        {"kind", "plot_error"},
                        {"protocol", "aethera.plot.stream"},
                        {"version", plot_stream_protocol_version},
                        {"message", failure ? *failure : "Plot unavailable"}
                    }.dump(),
                    {}
                }));
        }
        catch (...) {
            unsubscribe(id);
        }
    }
    return id;
}
void Plot::unsubscribe(Stream_Id stream) {
    auto current = d->consumers.load(std::memory_order_acquire);
    while (current->contains(stream)) {
        auto next = std::make_shared<Private::Consumer_Map>(*current);
        next->erase(stream);
        std::shared_ptr<const Private::Consumer_Map> desired = next;
        if (d->consumers.compare_exchange_weak(
            current, desired, std::memory_order_release,
            std::memory_order_acquire))
            break;
    }
    d->refresh_schedule();
}
void Plot::configure_stream(Stream_Id stream, std::uint32_t width,
                            std::uint32_t height) {
    auto current = d->consumers.load(std::memory_order_acquire);
    for (;;) {
        const auto found = current->find(stream);
        if (found == current->end()) return;
        auto next = std::make_shared<Private::Consumer_Map>(*current);
        auto& consumer = next->at(stream);
        consumer.width = width;
        consumer.height = height;
        std::shared_ptr<const Private::Consumer_Map> desired = next;
        if (d->consumers.compare_exchange_weak(
            current, desired, std::memory_order_release,
            std::memory_order_acquire))
            return;
    }
}
void Plot::schedule_render(Plot_Render_Tick tick) {
    if (!std::isfinite(tick.time_milliseconds) ||
        tick.time_milliseconds < 0.0)
        throw std::invalid_argument(
            "render time_milliseconds must be finite and non-negative");
    if (tick.width == 0 || tick.height == 0) throw std::invalid_argument("render viewport must be non-zero");
    ensure_started();
    if (d->terminal_failure.load(std::memory_order_acquire)) return;
    d->submit_tick_request(std::move(tick));
    d->arm_tick_consumer(weak_from_this());
}
void Plot::render_once() {
    ensure_started();
    const auto now = std::chrono::steady_clock::now();
    const auto elapsed = now - d->clock_origin;
    schedule_render(Plot_Render_Tick{
        .issued_at = now,
        .time_milliseconds =
        std::chrono::duration<double, std::milli>(elapsed).count(),
        .source = Frame_Request_Source::immediate
    });
}
void Plot::submit_input(Plot_Input_Event event) {
    static_cast<void>(input_timeline_time(event.time_milliseconds));
    ensure_started();
    if (d->terminal_failure.load(std::memory_order_acquire)) return;
    /*
     * WebSocket 线程只向 Scene 的当前事件缓冲追加一个由 Scene
     * memory_resource 分配的基类指针。Prepare 边界交换完整批次，
     * Scene 在 Renderable 完成消费时按 Event_Type 增量统计，并随自身
     * State 双缓冲发布；Web 层只在低频 diagnostics 请求中读取结果。
     */
    try {
        if (auto* scene_2d = std::get_if<std::unique_ptr<Scene_2D>>(&d->scene)) dispatch_plot_input(**scene_2d, event);
        else dispatch_plot_input(*std::get<std::unique_ptr<Scene_3D>>(d->scene), event);
    }
    catch (...) {
        d->fail(std::current_exception());
    }
}
nlohmann::json Plot::schema() {
    ensure_started();
    return d->schema();
}
nlohmann::json Plot::write_prop(std::string_view component,
                                std::string_view key,
                                const nlohmann::json& value) {
    ensure_started();
    if (component != "frame-analysis") return d->view->write_prop(component, key, value);
    auto result = d->with_frame_policy([&](auto& policy) {
        return write_frame_policy_prop(policy, key, value);
    });
    if (result.value("success", false)) d->arm_tick_consumer(weak_from_this());
    return result;
}
nlohmann::json Plot::component_state(std::string_view component) const {
    return d->view->component_state(component);
}
nlohmann::json Plot::generate_data(const nlohmann::json& input) {
    ensure_started();
    return d->view->generate_data(input);
}
nlohmann::json Plot::diagnostics() const {
    nlohmann::json frame_statistics = nlohmann::json::object();
    nlohmann::json input_statistics = nlohmann::json::object();
    Frame_Identity identity{};
    std::uint64_t created_time_unix_ns{};
    std::uint64_t dropped_sequences{};
    std::uint32_t completed_width{};
    std::uint32_t completed_height{};
    double frame_rate{};
    bool is_3d{};
    const auto read_scene_statistics = [&](const auto& state) {
        append_event_statistics_json(input_statistics, state.event_statistics);
    };
    std::visit([&](const auto& scene) {
        using Scene_Pointer = std::remove_cvref_t<decltype(scene)>;
        if constexpr (std::same_as<Scene_Pointer, std::unique_ptr<Scene_2D>>) {
            scene->template access_state<Render_Scene_2D::Base_Tag>(
                read_scene_statistics);
        }
        else {
            is_3d = true;
            scene->template access_state<Render_Scene_3D::Base_Tag>(
                read_scene_statistics);
        }
    }, d->scene);
    {
        const auto generation =
        d->statistics_generation.load(std::memory_order_acquire);
        auto* completed_frame =
        d->latest_statistics_frame.load(std::memory_order_acquire);
        Frame_Statistics_State statistics{};
        if (completed_frame) {
            completed_frame->diagnostic_readers.fetch_add(
                1, std::memory_order_acq_rel);
            if (d->latest_statistics_frame.load(std::memory_order_acquire) ==
                completed_frame &&
                completed_frame->state.load(std::memory_order_acquire) ==
                Private::Frame_State::available &&
                completed_frame->statistics_generation == generation) {
                statistics = completed_frame->statistics;
                std::visit([&](const auto& frame) {
                    using Frame_Pointer =
                    std::remove_cvref_t<decltype(frame)>;
                    if constexpr (std::same_as<
                        Frame_Pointer,
                        std::unique_ptr<Frame_2D>>) {
                        const auto image = frame->image();
                        completed_width = static_cast<std::uint32_t>(
                            std::max(0, image.width));
                        completed_height = static_cast<std::uint32_t>(
                            std::max(0, image.height));
                    }
                    else {
                        const auto extent = frame->extent();
                        completed_width = extent.width;
                        completed_height = extent.height;
                    }
                }, completed_frame->frame);
            }
            completed_frame->diagnostic_readers.fetch_sub(
                1, std::memory_order_release);
        }
        append_statistic_json(frame_statistics, statistics);
        identity = statistics.identity;
        created_time_unix_ns = statistics.created_time_unix_ns;
        dropped_sequences = statistics.dropped_sequences;
        const auto& interval = statistics.values[
            static_cast<std::size_t>(Frame_Statistic::frame_interval_ms)];
        frame_rate = interval.trimmed_average > 0.0
                         ? 1'000.0 / interval.trimmed_average
                         : 0.0;
    }
    const auto& pacing = d->pacing_state();
    const auto stream = d->stream_snapshot();
    const auto admission = d->render_admission.load(std::memory_order_acquire);
    const auto admission_name = [&] {
        switch (admission) {
        case Private::Render_Admission_State::ready: return "ready";
        case Private::Render_Admission_State::rendering: return "rendering";
        case Private::Render_Admission_State::frame_slots_exhausted: return "frame_slots_exhausted";
        }
        return "unknown";
    }();
    nlohmann::json supported_formats = nlohmann::json::array();
    if (is_3d) {
        for (const auto format : Frame_3D::supported_pixel_formats) supported_formats.push_back(pixel_format_name(format));
    }
    else {
        for (const auto format : Frame_2D::supported_pixel_formats) supported_formats.push_back(pixel_format_name(format));
    }
    const auto format = is_3d
                            ? pixel_format_name(Frame_3D::native_pixel_format)
                            : pixel_format_name(Frame_2D::native_pixel_format);
    const auto native_format = is_3d
                                   ? pixel_format_name(Frame_3D::native_pixel_format)
                                   : pixel_format_name(Frame_2D::native_pixel_format);
    const auto pixel_width = completed_width == 0
                                 ? stream.width
                                 : completed_width;
    const auto pixel_height = completed_height == 0
                                  ? stream.height
                                  : completed_height;
    const std::size_t byte_length = pacing.pixel_delivery_enabled
                                        ? static_cast<std::size_t>(pixel_width) * pixel_height * 4U
                                        : 0U;
    nlohmann::json output{
        {"protocol", "aethera.plot.diagnostics"}, {"version", 4},
        {"dimension", is_3d ? "3D" : "2D"},
        {"sequence", identity.sequence},
        {"correlation_id", identity.correlation_id},
        {"rendered_sequence", identity.sequence},
        {"rendered_correlation_id", identity.correlation_id},
        {
            "generated_time_unix_ms",
            static_cast<double>(created_time_unix_ns) / 1'000'000.0
        },
        {"delivery", pacing.pixel_delivery_enabled ? "gallery-pixels" : "diagnostics"},
        {"frame_rate_fps", frame_rate},
        {"dropped_sequence_count", dropped_sequences},
        {"window_capacity", diagnostic_window_capacity},
        {
            "pixel", {
                {"width", pixel_width}, {"height", pixel_height},
                {"format", format}, {"native_format", native_format},
                {"supported_formats", std::move(supported_formats)},
                {"byte_length", byte_length}
            }
        },
        {"frame_policy", frame_policy_state_json(pacing)},
        {"render_admission", admission_name},
        {"frame_statistics", std::move(frame_statistics)},
        {"input_statistics", std::move(input_statistics)}
    };
    if (is_3d) {
        const auto& pipeline = d->frame_policy_3d->read_state<
            Frame_Policy_3D::Base_Tag>();
        output["frame_policy"]["three_dimensional_pipeline"] = {
            {"capacity", Frame_Policy_3D::pipeline_capacity},
            {"overlapped_release_count", pipeline.overlapped_release_count},
            {
                "completion_gated_release_count",
                pipeline.completion_gated_release_count
            },
            {"gpu_completion_count", pipeline.gpu_completion_count},
            {"last_completed_sequence", pipeline.last_completed_sequence}
        };
        const auto& gpu = render_3d::detail::Gpu_Completion_Service::instance().
        read_state<render_3d::detail::Gpu_Completion_Service::Base_Tag>();
        const auto milliseconds = [](std::uint64_t nanoseconds) {
            return static_cast<double>(nanoseconds) / 1'000'000.0;
        };
        output["gpu_completion_domain"] = {
            {"capacity", gpu.capacity}, {"in_flight", gpu.in_flight},
            {"peak_in_flight", gpu.peak_in_flight}, {"watched", gpu.watched},
            {"peak_watched", gpu.peak_watched},
            {"active_fences", gpu.active_fences},
            {"pending_fences", gpu.pending_fences},
            {"reservation_count", gpu.reservation_count},
            {"completion_count", gpu.completion_count},
            {"cancellation_count", gpu.cancellation_count},
            {"fence_probe_count", gpu.fence_probe_count},
            {"fence_wait_count", gpu.fence_wait_count},
            {"fence_wait_timeout_count", gpu.fence_wait_timeout_count},
            {"fence_wait_total_ms", milliseconds(gpu.fence_wait_total_ns)},
            {"fence_wait_max_ms", milliseconds(gpu.fence_wait_max_ns)},
            {"callback_total_ms", milliseconds(gpu.callback_total_ns)},
            {"callback_max_ms", milliseconds(gpu.callback_max_ns)},
            {"callback_failure_count", gpu.callback_failure_count},
            {"backpressure_count", gpu.backpressure_count},
            {"fault_count", gpu.fault_count},
            {"abandoned_count", gpu.abandoned_count}
        };
    }
    if (const auto failure = d->terminal_failure.load(std::memory_order_acquire)) output["terminal_failure"] = *failure;
    return output;
}
void Plot::request_taskflow_trace(std::size_t frame_count) {
    if (frame_count == 0 ||
        frame_count > Private::maximum_taskflow_trace_frames)
        throw std::invalid_argument("Taskflow trace frame_count must be between 1 and 120");
    ensure_started();
    auto control = d->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 frame trace request is already active");
        const auto next = static_cast<std::uint64_t>(frame_count) << 32U;
        if (d->taskflow_trace_control.compare_exchange_weak(
            control, next, std::memory_order_release,
            std::memory_order_acquire))
            break;
    }
    for (auto& slot : d->taskflow_trace_slots) slot.store({}, std::memory_order_release);
    d->taskflow_trace_remaining.store(frame_count, std::memory_order_release);
}
nlohmann::json Plot::taskflow_trace() const {
    return d->trace_response(d->taskflow_trace_control,
                             d->taskflow_trace_remaining,
                             d->taskflow_trace_slots);
}
void Plot::reset_diagnostics() {
    std::visit([](auto& scene) {
        scene->template update_state<&Scene::State::event_statistics>(
            Event_Statistics_State{});
    }, d->scene);
    d->statistics_generation.fetch_add(1, std::memory_order_acq_rel);
    d->with_frame_policy([](auto& policy) {
        policy.reset_statistics();
    });
    d->arm_tick_consumer(weak_from_this());
}
}