原始像素太卡 走ffmpeg

Project_detail_specification.md

@@ -29,9 +29,10 @@
   层只向帧写入固定语义的单调时间点与原始耗时，不保存平均值、分位数或波动等衍生统计。
 * Plot 使用服务端帧时钟调用 `Scene::render(frame*)`；`fixed_rate` 只由 Frame_Scheduler 周期时钟驱动，完成回调不得改变其
   deadline。`maximum_rate` 是唯一允许完成回调在释放 Plot 准入后异步投递下一帧请求的特殊模式；回调不得直接重入 `render()`
-  、不得同步等待，并且物理帧槽耗尽时只能由帧退役机制解除背压。Scene 回调发布 2D 原生 BGRA 或 3D 原生 RGBA；Gallery
-  像素流按固定时钟采样各 Plot 的 latest，只把本周期真实变化的 tile 通过 Drogon WebSocket 原始像素批协议发送；服务端不组装图集，浏览器由各 Plot 的 WebGL Canvas 直接上传自身 tile。每个连接只允许一批在途，
-  浏览器实际提交 Canvas 后 ACK，服务端只发送当时最新帧；Plot 输入和控制也统一使用 Drogon WebSocket。
+  、不得同步等待，并且物理帧槽耗尽时只能由帧退役机制解除背压。Scene 回调发布 2D 原生 BGRA 或 3D 原生 RGBA；每个 Plot
+  只服从自身 Frame_Policy。Gallery 仅按 Plot 完成帧的到达顺序更新对应图集槽位，不拥有采样时钟、帧准入、合并或丢帧策略；
+  每次槽位更新均在唯一串行 Taskflow 媒体 DAG 中组成图集并由 FFmpeg/libx264 编码 H.264 Annex-B。Drogon WebSocket 只传输压缩
+  access unit，不得用在途窗口反向跳过已通过 Plot 策略的帧；浏览器由 WebCodecs 解码后按槽位提交各 Plot Canvas。Plot 输入和控制也统一使用 Drogon WebSocket。
 * 帧策略及其时钟实现统一归属 `kernel/src/kernel/Frame_Policy`。配置变更和运行事实通过 MPMC 事件流进入唯一 consumer，由
   `double_buffer/model.hpp` 发布 `Frame_Policy::State`；查询只读取已发布 State。手动帧捕获必须把帧创建边界的策略 State
   绑定到该帧并沿现有帧捕获协议返回，禁止另建策略历史、轮询累加器或重复诊断字段。

mcp/main.cmake

@@ -111,6 +111,30 @@ if (Aethera_BUILD_TESTS)
             --benchmark_report_aggregates_only=true
             DEPENDS Aethera_MCP_Benchmarks
             USES_TERMINAL)
+    add_executable(Aethera_MCP_Event_Latency_Benchmarks
+            "${CMAKE_CURRENT_LIST_DIR}/tests/Event_Latency_Benchmarks.cpp")
+    target_link_libraries(Aethera_MCP_Event_Latency_Benchmarks PRIVATE
+            Aethera_MCP_Core
+            benchmark::benchmark
+            TBB::tbbmalloc_proxy)
+    renderive_stage_render_3D_runtime(Aethera_MCP_Event_Latency_Benchmarks)
+    if (MSVC)
+        target_compile_options(Aethera_MCP_Event_Latency_Benchmarks PRIVATE
+                /utf-8 /bigobj)
+        target_link_options(Aethera_MCP_Event_Latency_Benchmarks PRIVATE
+                "/INCLUDE:__TBB_malloc_proxy")
+        add_custom_command(TARGET Aethera_MCP_Event_Latency_Benchmarks POST_BUILD
+                COMMAND "${CMAKE_COMMAND}" -E copy_if_different
+                "$<TARGET_FILE:TBB::tbbmalloc>"
+                "$<TARGET_FILE:TBB::tbbmalloc_proxy>"
+                "$<TARGET_FILE_DIR:Aethera_MCP_Event_Latency_Benchmarks>"
+                VERBATIM)
+    endif ()
+    add_custom_target(Aethera_MCP_event_latency_benchmark
+            COMMAND Aethera_MCP_Event_Latency_Benchmarks
+            --aethera_event_samples=12
+            DEPENDS Aethera_MCP_Event_Latency_Benchmarks
+            USES_TERMINAL)
     add_custom_target(Aethera_MCP_check
             COMMAND "${CMAKE_CTEST_COMMAND}"
             --test-dir "${CMAKE_BINARY_DIR}"

mcp/tests/Control_Path_Benchmarks.cpp

@@ -98,8 +98,8 @@ void append_dimension(web::Plot_Dimension dimension) {
 [[nodiscard]] Parse_Benchmark_Arguments_Result parse_arguments(
     int& argc, char** argv) {
     configuration = {};
-    if (!select_plots("3d")) {
-        configuration_error = "the Gallery has no 3D plots";
+    if (!select_plots("all")) {
+        configuration_error = "the Gallery has no plots";
         return Parse_Benchmark_Arguments_Result::invalid_argument;
     }
     int retained_count{1};

mcp/tests/Event_Latency_Benchmarks.cpp

@@ -0,0 +1,554 @@
+#include <mcp/core/Control_Requests.hpp>
+#include <mcp/core/Control_Service.hpp>
+#include <mcp/core/Protocol_Type.hpp>
+#include <mcp/core/runtime/Gallery_Plots.hpp>
+#include <benchmark/benchmark.h>
+#include <magic_enum/magic_enum.hpp>
+#include <render_common.hpp>
+#include <algorithm>
+#include <array>
+#include <atomic>
+#include <charconv>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <iomanip>
+#include <iostream>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <vector>
+
+namespace aethera::mcp::event_latency_benchmarks {
+namespace {
+
+using Clock = std::chrono::steady_clock;
+inline constexpr auto event_types = magic_enum::enum_values<Event_Type>();
+inline constexpr std::size_t event_count = event_types.size();
+
+struct Configuration {
+    std::uint32_t width{720};
+    std::uint32_t height{420};
+    std::uint32_t samples{12};
+};
+
+struct Summary {
+    double p50{};
+    double p95{};
+    double p99{};
+    double maximum{};
+};
+
+struct Distribution {
+    std::vector<double> samples{};
+
+    void add(double value) { samples.push_back(value); }
+
+    [[nodiscard]] Summary summarize() const {
+        if (samples.empty()) return {};
+        auto ordered = samples;
+        std::ranges::sort(ordered);
+        const auto percentile = [&](double probability) {
+            const auto rank = probability *
+                static_cast<double>(ordered.size() - 1U);
+            const auto lower = static_cast<std::size_t>(rank);
+            const auto upper = std::min(lower + 1U, ordered.size() - 1U);
+            const auto fraction = rank - static_cast<double>(lower);
+            return ordered[lower] +
+                (ordered[upper] - ordered[lower]) * fraction;
+        };
+        return {
+            percentile(0.50), percentile(0.95), percentile(0.99),
+            ordered.back()
+        };
+    }
+};
+
+struct Event_Result {
+    Distribution mcp_accept_ms{};
+    Distribution scene_queue_ms{};
+    Distribution dispatch_ms{};
+    Distribution scene_total_ms{};
+    Distribution accepted_to_frame_ms{};
+    Distribution end_to_end_ms{};
+};
+
+struct Plot_Result {
+    std::string id{};
+    web::Plot_Dimension dimension{web::Plot_Dimension::two_d};
+    std::array<Event_Result, event_count> events{};
+};
+
+struct Plot_Probe {
+    std::atomic_uint64_t expected_correlation{};
+    std::atomic_uint64_t completed_steady_ns{};
+};
+
+struct Active_Plot {
+    std::shared_ptr<web::Plot> plot{};
+    web::Plot::Stream_Id stream{};
+    std::shared_ptr<Plot_Probe> probe{};
+};
+
+struct Batch_Timing {
+    std::array<std::uint64_t, event_count> submitted_ns{};
+    std::array<std::uint64_t, event_count> accepted_ns{};
+};
+
+Configuration configuration{};
+std::vector<Plot_Result> published_results{};
+
+[[nodiscard]] std::uint64_t steady_time_ns() noexcept {
+    return static_cast<std::uint64_t>(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(
+            Clock::now().time_since_epoch()).count());
+}
+
+[[nodiscard]] double elapsed_ms(std::uint64_t begin,
+                                std::uint64_t end) noexcept {
+    return end >= begin
+        ? static_cast<double>(end - begin) / 1'000'000.0 : 0.0;
+}
+
+[[nodiscard]] web::Plot_Input_Event make_event(
+    Event_Type type, std::size_t plot_index, std::uint64_t batch) {
+    web::Plot_Input_Event event{};
+    event.type = type;
+    event.time_milliseconds = static_cast<double>(steady_time_ns()) /
+        1'000'000.0;
+    event.position = {
+        120.0 + static_cast<double>((batch * 13U + plot_index * 7U) % 400U),
+        80.0 + static_cast<double>((batch * 11U + plot_index * 5U) % 240U)};
+    event.global_position = event.position;
+    event.button = Mouse_Button::left;
+    event.buttons = type == Event_Type::pointer_release ? 0U : 1U;
+    event.modifiers = Keyboard_Modifier::control;
+    event.pixel_delta_x = 3.0;
+    event.pixel_delta_y = (batch & 1U) == 0U ? 120.0 : -120.0;
+    event.angle_delta_x = event.pixel_delta_x;
+    event.angle_delta_y = event.pixel_delta_y;
+    event.key = Key::space;
+    event.native_key = 32;
+    event.auto_repeat = (batch & 1U) != 0U;
+    return event;
+}
+
+[[nodiscard]] const nlohmann::json& event_statistic(
+    const nlohmann::json& diagnostics, Event_Type type,
+    std::string_view statistic) {
+    const auto type_name = std::string{magic_enum::enum_name(type)};
+    const auto& input = diagnostics.at("input_statistics");
+    const auto found_type = input.find(type_name);
+    if (found_type == input.end())
+        throw std::runtime_error(
+            "completed frame did not publish input statistics for " +
+            type_name);
+    const auto found_statistic = found_type->find(std::string{statistic});
+    if (found_statistic == found_type->end())
+        throw std::runtime_error(
+            "completed frame did not publish " + std::string{statistic} +
+            " for " + type_name);
+    return *found_statistic;
+}
+
+void configure_manual_backend_plot(const std::shared_ptr<web::Plot>& plot) {
+    const auto manual = plot->write_prop(
+        "frame-analysis", "pacing_mode", "manual");
+    if (!manual.value("success", false))
+        throw std::runtime_error("failed to select manual frame pacing");
+    const auto diagnostics_only = plot->write_prop(
+        "frame-analysis", "pixel_delivery_enabled", false);
+    if (!diagnostics_only.value("success", false))
+        throw std::runtime_error("failed to disable backend pixel readback");
+}
+
+[[nodiscard]] std::uint64_t correlation_id(
+    std::uint64_t batch, std::size_t plot_index) noexcept {
+    return (batch + 1U) * 1'000U +
+        static_cast<std::uint64_t>(plot_index + 1U);
+}
+
+void add_distribution(Distribution& destination,
+                      const Distribution& source) {
+    destination.samples.insert(destination.samples.end(),
+                               source.samples.begin(), source.samples.end());
+}
+
+class Event_Latency_Reporter final : public benchmark::ConsoleReporter {
+public:
+    bool ReportContext(const Context& context) override {
+        return ConsoleReporter::ReportContext(context);
+    }
+
+    void ReportRuns(const std::vector<Run>& reports) override {
+        auto& output = GetOutputStream();
+        for (const auto& report : reports) {
+            output << report.benchmark_name();
+            if (report.skipped != benchmark::internal::NotSkipped)
+                output << " ERROR: " << report.skip_message << '\n';
+            else
+                output << "  elapsed=" << report.real_accumulated_time
+                       << " s\n";
+        }
+        if (published_results.empty()) return;
+        const auto flags = output.flags();
+        const auto precision = output.precision();
+        output << std::fixed << std::setprecision(3)
+               << "\nPure backend event timeline (ms)\n"
+               << "MCP = emit -> aethera_plot_input accepted; "
+                  "Queue/Dispatch = authoritative Scene Event timing; "
+                  "Frame = accepted -> correlated backend frame complete.\n\n";
+
+        output << std::left << std::setw(24) << "Plot"
+               << std::setw(6) << "Dim"
+               << std::setw(18) << "Worst event"
+               << std::right << std::setw(11) << "MCP P95"
+               << std::setw(11) << "Queue P95"
+               << std::setw(11) << "Disp P95"
+               << std::setw(11) << "Frame P95"
+               << std::setw(11) << "E2E P95" << '\n';
+        for (const auto& plot : published_results) {
+            std::size_t worst_index{};
+            double worst_e2e{};
+            for (std::size_t index = 0; index < event_count; ++index) {
+                const auto p95 = plot.events[index].end_to_end_ms.summarize().p95;
+                if (p95 <= worst_e2e) continue;
+                worst_e2e = p95;
+                worst_index = index;
+            }
+            const auto& event = plot.events[worst_index];
+            output << std::left << std::setw(24) << plot.id
+                   << std::setw(6) << web::plot_dimension_name(plot.dimension)
+                   << std::setw(18)
+                   << magic_enum::enum_name(event_types[worst_index])
+                   << std::right
+                   << std::setw(11) << event.mcp_accept_ms.summarize().p95
+                   << std::setw(11) << event.scene_queue_ms.summarize().p95
+                   << std::setw(11) << event.dispatch_ms.summarize().p95
+                   << std::setw(11)
+                   << event.accepted_to_frame_ms.summarize().p95
+                   << std::setw(11) << worst_e2e << '\n';
+        }
+
+        output << "\nAll plots by event type\n"
+               << std::left << std::setw(18) << "Event"
+               << std::right << std::setw(11) << "MCP P50"
+               << std::setw(11) << "MCP P95"
+               << std::setw(11) << "Queue P95"
+               << std::setw(11) << "Disp P95"
+               << std::setw(11) << "Scene P95"
+               << std::setw(11) << "Frame P95"
+               << std::setw(11) << "E2E P50"
+               << std::setw(11) << "E2E P95"
+               << std::setw(11) << "E2E P99" << '\n';
+        for (std::size_t event_index = 0; event_index < event_count;
+             ++event_index) {
+            Event_Result aggregate{};
+            for (const auto& plot : published_results) {
+                const auto& event = plot.events[event_index];
+                add_distribution(aggregate.mcp_accept_ms, event.mcp_accept_ms);
+                add_distribution(aggregate.scene_queue_ms, event.scene_queue_ms);
+                add_distribution(aggregate.dispatch_ms, event.dispatch_ms);
+                add_distribution(aggregate.scene_total_ms, event.scene_total_ms);
+                add_distribution(
+                    aggregate.accepted_to_frame_ms,
+                    event.accepted_to_frame_ms);
+                add_distribution(aggregate.end_to_end_ms, event.end_to_end_ms);
+            }
+            const auto mcp = aggregate.mcp_accept_ms.summarize();
+            const auto queue = aggregate.scene_queue_ms.summarize();
+            const auto dispatch = aggregate.dispatch_ms.summarize();
+            const auto scene = aggregate.scene_total_ms.summarize();
+            const auto frame = aggregate.accepted_to_frame_ms.summarize();
+            const auto end_to_end = aggregate.end_to_end_ms.summarize();
+            output << std::left << std::setw(18)
+                   << magic_enum::enum_name(event_types[event_index])
+                   << std::right << std::setw(11) << mcp.p50
+                   << std::setw(11) << mcp.p95
+                   << std::setw(11) << queue.p95
+                   << std::setw(11) << dispatch.p95
+                   << std::setw(11) << scene.p95
+                   << std::setw(11) << frame.p95
+                   << std::setw(11) << end_to_end.p50
+                   << std::setw(11) << end_to_end.p95
+                   << std::setw(11) << end_to_end.p99 << '\n';
+        }
+        output.flags(flags);
+        output.precision(precision);
+    }
+};
+
+void run_event_latency(benchmark::State& state) {
+    published_results.clear();
+    auto service = Control_Service::create();
+    const auto definitions = web::gallery_plot_definitions();
+    std::vector<Active_Plot> active;
+    active.reserve(definitions.size());
+    published_results.reserve(definitions.size());
+
+    try {
+        for (const auto& definition : definitions) {
+            auto plot = service->find_plot(definition.id);
+            if (!plot)
+                throw std::runtime_error(
+                    "Gallery plot is unavailable: " +
+                    std::string{definition.id});
+            configure_manual_backend_plot(plot);
+            auto probe = std::make_shared<Plot_Probe>();
+            active.push_back({std::move(plot), 0, std::move(probe)});
+            published_results.push_back({
+                std::string{definition.id}, definition.dimension, {}});
+        }
+
+        std::string configuration_failure;
+        Task_Graph configuration_barrier{"mcp.event-latency.configure"};
+        configuration_barrier.add("await.manual-policy", [&] {
+            const auto deadline = Clock::now() + std::chrono::seconds{10};
+            Task_Graph::corun_until([&] {
+                const bool ready = std::ranges::all_of(
+                    active, [](const Active_Plot& item) {
+                        const auto diagnostics = item.plot->diagnostics();
+                        const auto& policy = diagnostics.at("frame_policy").
+                            at("configuration");
+                        return policy.at("mode") == "manual" &&
+                            !policy.at("pixel_delivery_enabled").get<bool>();
+                    });
+                if (ready) return true;
+                if (Clock::now() < deadline) return false;
+                configuration_failure =
+                    "timed out applying manual backend frame policy";
+                return true;
+            });
+        });
+        aethera::detail::run_taskflow(configuration_barrier);
+        if (!configuration_failure.empty())
+            throw std::runtime_error(configuration_failure);
+
+        for (auto& item : active) {
+            const auto probe = item.probe;
+            item.stream = item.plot->subscribe(
+                [probe](std::shared_ptr<const web::Plot_Stream_Frame> frame) {
+                    if (!frame || !frame->pixels) return;
+                    const auto expected = probe->expected_correlation.load(
+                        std::memory_order_acquire);
+                    if (expected == 0 ||
+                        frame->pixels->correlation_id != expected)
+                        return;
+                    std::uint64_t incomplete{};
+                    static_cast<void>(
+                        probe->completed_steady_ns.compare_exchange_strong(
+                            incomplete, steady_time_ns(),
+                            std::memory_order_release,
+                            std::memory_order_relaxed));
+                });
+            item.plot->configure_stream(
+                item.stream, configuration.width, configuration.height);
+        }
+
+        for (auto& item : active) item.plot->reset_diagnostics();
+        std::vector<Batch_Timing> batch_timing(active.size());
+        std::string failure;
+        Task_Graph coordinator{"mcp.event-latency"};
+        coordinator.add("measure.all-plots", [&] {
+            for (std::uint64_t batch = 0; batch < configuration.samples;
+                 ++batch) {
+                const auto deadline = Clock::now() + std::chrono::seconds{30};
+                for (std::size_t plot_index = 0;
+                     plot_index < active.size(); ++plot_index) {
+                    auto& item = active[plot_index];
+                    auto& timing = batch_timing[plot_index];
+                    const auto correlation = correlation_id(batch, plot_index);
+                    item.probe->completed_steady_ns.store(
+                        0, std::memory_order_relaxed);
+                    item.probe->expected_correlation.store(
+                        correlation, std::memory_order_release);
+                    for (std::size_t event_index = 0;
+                         event_index < event_count; ++event_index) {
+                        auto request = Plot_Input_Request{
+                            .plot = published_results[plot_index].id,
+                            .event = make_event(
+                                event_types[event_index], plot_index, batch)
+                        };
+                        timing.submitted_ns[event_index] = steady_time_ns();
+                        const auto result = service->call_tool(
+                            "aethera_plot_input",
+                            encode_protocol_value(request));
+                        timing.accepted_ns[event_index] = steady_time_ns();
+                        if (result.result != Tool_Call_Result::ok) {
+                            failure = "aethera_plot_input rejected " +
+                                published_results[plot_index].id + "/" +
+                                std::string{magic_enum::enum_name(
+                                    event_types[event_index])} + ": " +
+                                result.message;
+                            return;
+                        }
+                    }
+                    item.plot->schedule_render(web::Plot_Render_Tick{
+                        .issued_at = Clock::now(),
+                        .sequence = correlation,
+                        .time_milliseconds =
+                            static_cast<double>(steady_time_ns()) / 1'000'000.0,
+                        .width = configuration.width,
+                        .height = configuration.height,
+                        .source = Frame_Request_Source::immediate
+                    });
+                }
+
+                Task_Graph::corun_until([&] {
+                    const bool complete = std::ranges::all_of(
+                        active, [](const Active_Plot& item) {
+                            return item.probe->completed_steady_ns.load(
+                                std::memory_order_acquire) != 0;
+                        });
+                    if (complete) return true;
+                    if (Clock::now() < deadline) return false;
+                    failure = "timed out waiting for a correlated backend frame";
+                    return true;
+                });
+                if (!failure.empty()) return;
+
+                for (std::size_t plot_index = 0;
+                     plot_index < active.size(); ++plot_index) {
+                    const auto completed = active[plot_index].probe->
+                        completed_steady_ns.load(std::memory_order_acquire);
+                    const auto diagnostics = active[plot_index].plot->diagnostics();
+                    for (std::size_t event_index = 0;
+                         event_index < event_count; ++event_index) {
+                        auto& result = published_results[plot_index].
+                            events[event_index];
+                        const auto& timing = batch_timing[plot_index];
+                        result.mcp_accept_ms.add(elapsed_ms(
+                            timing.submitted_ns[event_index],
+                            timing.accepted_ns[event_index]));
+                        result.accepted_to_frame_ms.add(elapsed_ms(
+                            timing.accepted_ns[event_index], completed));
+                        result.end_to_end_ms.add(elapsed_ms(
+                            timing.submitted_ns[event_index], completed));
+                        const auto type = event_types[event_index];
+                        result.scene_queue_ms.add(event_statistic(
+                            diagnostics, type, "queue_wait_ms").at("latest").
+                            get<double>());
+                        result.dispatch_ms.add(event_statistic(
+                            diagnostics, type, "dispatch_ms").at("latest").
+                            get<double>());
+                        result.scene_total_ms.add(event_statistic(
+                            diagnostics, type, "total_ms").at("latest").
+                            get<double>());
+                    }
+                    active[plot_index].probe->expected_correlation.store(
+                        0, std::memory_order_release);
+                }
+            }
+        });
+
+        for ([[maybe_unused]] auto iteration : state)
+            aethera::detail::run_taskflow(coordinator);
+        if (!failure.empty()) state.SkipWithError(failure);
+
+        Distribution all_end_to_end;
+        for (const auto& plot : published_results) {
+            Distribution plot_end_to_end;
+            for (std::size_t event_index = 0;
+                 event_index < event_count; ++event_index) {
+                const auto event_name = std::string{
+                    magic_enum::enum_name(event_types[event_index])};
+                const auto& event = plot.events[event_index];
+                add_distribution(plot_end_to_end, event.end_to_end_ms);
+                add_distribution(all_end_to_end, event.end_to_end_ms);
+                const auto prefix = plot.id + "/" + event_name + "/";
+                state.counters[prefix + "queue_p95_ms"] =
+                    event.scene_queue_ms.summarize().p95;
+                state.counters[prefix + "dispatch_p95_ms"] =
+                    event.dispatch_ms.summarize().p95;
+                state.counters[prefix + "e2e_p95_ms"] =
+                    event.end_to_end_ms.summarize().p95;
+            }
+            state.counters[plot.id + "/e2e_p95_ms"] =
+                plot_end_to_end.summarize().p95;
+        }
+        const auto end_to_end = all_end_to_end.summarize();
+        state.counters["all/e2e_p50_ms"] = end_to_end.p50;
+        state.counters["all/e2e_p95_ms"] = end_to_end.p95;
+        state.counters["all/e2e_p99_ms"] = end_to_end.p99;
+        state.SetItemsProcessed(static_cast<std::int64_t>(
+            configuration.samples * active.size() * event_count));
+    }
+    catch (const std::exception& error) {
+        state.SkipWithError(error.what());
+    }
+
+    for (auto& item : active) item.plot->unsubscribe(item.stream);
+}
+
+BENCHMARK(run_event_latency)->Iterations(1)->UseRealTime();
+
+[[nodiscard]] bool parse_unsigned(std::string_view value,
+                                  std::uint32_t& output) {
+    const auto* begin = value.data();
+    const auto* end = begin + value.size();
+    const auto parsed = std::from_chars(begin, end, output);
+    return parsed.ec == std::errc{} && parsed.ptr == end;
+}
+
+[[nodiscard]] bool configure(int& argc, char** argv) {
+    int retained{1};
+    for (int index = 1; index < argc; ++index) {
+        const std::string_view argument{argv[index]};
+        constexpr std::string_view samples_prefix{
+            "--aethera_event_samples="};
+        constexpr std::string_view size_prefix{"--aethera_size="};
+        if (argument.starts_with(samples_prefix)) {
+            if (!parse_unsigned(argument.substr(samples_prefix.size()),
+                                configuration.samples) ||
+                configuration.samples == 0 || configuration.samples > 600) {
+                std::cerr << "--aethera_event_samples must be in [1, 600]\n";
+                return false;
+            }
+            continue;
+        }
+        if (argument.starts_with(size_prefix)) {
+            const auto value = argument.substr(size_prefix.size());
+            const auto separator = value.find('x');
+            if (separator == std::string_view::npos ||
+                !parse_unsigned(value.substr(0, separator),
+                                configuration.width) ||
+                !parse_unsigned(value.substr(separator + 1U),
+                                configuration.height) ||
+                configuration.width == 0 || configuration.height == 0) {
+                std::cerr << "--aethera_size requires WIDTHxHEIGHT\n";
+                return false;
+            }
+            continue;
+        }
+        argv[retained++] = argv[index];
+    }
+    argc = retained;
+    return true;
+}
+
+} // namespace
+} // namespace aethera::mcp::event_latency_benchmarks
+
+int main(int argc, char** argv) {
+    if (!aethera::mcp::event_latency_benchmarks::configure(argc, argv))
+        return 2;
+    aethera::initialize_runtime({});
+    benchmark::Initialize(&argc, argv);
+    if (benchmark::ReportUnrecognizedArguments(argc, argv)) return 2;
+    benchmark::AddCustomContext("aethera_plots", "all");
+    benchmark::AddCustomContext(
+        "aethera_event_samples",
+        std::to_string(
+            aethera::mcp::event_latency_benchmarks::configuration.samples));
+    benchmark::AddCustomContext(
+        "aethera_size",
+        std::to_string(
+            aethera::mcp::event_latency_benchmarks::configuration.width) +
+        "x" + std::to_string(
+            aethera::mcp::event_latency_benchmarks::configuration.height));
+    aethera::mcp::event_latency_benchmarks::Event_Latency_Reporter reporter;
+    benchmark::RunSpecifiedBenchmarks(&reporter);
+    benchmark::Shutdown();
+    return 0;
+}

render_3D/render_3D/scene/Render_Scene_3D.ipp

@@ -670,10 +670,12 @@ void Render_Scene_3D::Private::render_datoviz(
                 scene, *visuals, sequence, observe, readback);
         }
         if (!prepared) {
+            frame->mark(Frame_Trace_Marker::event_dispatch_started);
             for (const auto& event : context->events) {
                 if (event) event->mark_dispatch_started(sequence);
                 dispatch_datoviz(event, scene.viewport);
             }
+            frame->mark(Frame_Trace_Marker::event_dispatch_finished);
             context->events.clear();
             frame->mark(Frame_Trace_Marker::backend_prepare_started);
             prepared = prepare(
@@ -1034,9 +1036,7 @@ void Render_Scene_3D::Private::ensure_frame_taskflow(Object* object) {
         if (!active_context || !active_context->frame) throw std::logic_error("3D frame DAG lost its active frame");
         const auto frame = active_context->frame;
         frame->mark(Frame_Trace_Marker::scene_render_started);
-        frame->mark(Frame_Trace_Marker::event_dispatch_started);
         active_context->events = take_events(object);
-        frame->mark(Frame_Trace_Marker::event_dispatch_finished);
         camera_component->advance_object();
         axes_component->advance_object();
         auto context = std::static_pointer_cast<
@@ -1070,11 +1070,11 @@ void Render_Scene_3D::Private::ensure_frame_taskflow(Object* object) {
         if (!detail::prepared_visual_batch_complete(*context->visuals))
             throw std::logic_error(
                 "render scene published an incomplete Visual batch");
+        render_datoviz(object, execution, context->visuals,
+                       context->parameters);
         auto& state = static_cast<State&>(
             double_buffer::detail::Internal_Access::pending_state(object));
         state.event_statistics = event_statistics.state();
-        render_datoviz(object, execution, context->visuals,
-                       context->parameters);
     });
     submit.describe("dimension", "3D")
           .describe("backend", "Datoviz")

web_server/src/Gallery_Video_Stream.ipp

@@ -1,84 +0,0 @@
-#pragma once
-#include "frame_sampling/Frame_Sampler.hpp"
-#include <Frame_Policy/Frame_Scheduler.hpp>
-#include <frame_statistics.hpp>
-#include <ownership.hpp>
-#include <array>
-#include <atomic>
-#include <chrono>
-namespace aethera::web {
-struct Gallery_Video_Stream::Private : Prev_Private {
-    using Object = Gallery_Video_Stream;
-    struct Source {
-        Plot_Entry entry; /* 布局槽位关联的实际 Plot。 */
-        Plot::Stream_Id stream{}; /* Plot 完成帧的唯一订阅标识。 */
-    };
-    struct Consumer {
-        Stream_Id id{}; /* 当前网页媒体连接的订阅身份。 */
-        std::string connection{}; /* 同一浏览器页重连时保持稳定的协议身份。 */
-        std::shared_ptr<const Stream_Handler> handler{}; /* 发布期间保持回调存活。 */
-        std::shared_ptr<const Transport_Readiness> readiness{}; /* 采样前查询对应传输窗口。 */
-        std::shared_ptr<const Transport_Diagnostics> diagnostics{}; /* 连接自己的传输诊断来源。 */
-    };
-    static constexpr std::size_t maximum_consumers{32};
-    Object* object{}; /* Def 机制所属最终对象；bind_private_crtp 后有效。 */
-    Sliding_Statistics compose_ms{600}; /* 图集采样与变更槽复制统计计算器。 */
-    Sliding_Statistics sample_delay_ms{600}; /* 固定 deadline 到实际采样的调度延迟。 */
-    Sliding_Statistics encode_ms{600}; /* H.264 编码统计计算器。 */
-    Sliding_Statistics publish_ms{600}; /* H.264 WebSocket 发布统计计算器。 */
-    std::vector<Source> sources{}; /* 已按业务标识排序的稳定图集来源。 */
-    std::unique_ptr<frame_sampling::Frame_Sampler> sampler{}; /* latest 图像与 30 FPS deadline 的唯一状态源。 */
-    std::shared_ptr<Task_Graph> media_graph{}; /* 独立媒体时钟运行的持久 DAG；异步完成回调延长其生命周期。 */
-    Frame_Scheduler::Timer sample_timer{}; /* 与任意单张 Plot 完成速率解耦的 30 FPS 控制时钟。 */
-    std::atomic_bool media_busy{}; /* 单个媒体 DAG 的 latest-only 准入。 */
-    frame_sampling::ffmpeg::FFmpeg_Frame_Transport ffmpeg_transport; /* FFmpeg 子模型的唯一编码上下文。 */
-    std::optional<frame_sampling::Sampled_Gallery_Frame> active_sample{}; /* 当前 deadline 取得的原生图集视图。 */
-    std::shared_ptr<const Encoded_Video_Frame> active_video{}; /* 当前编码结果不可变共享所有权。 */
-    std::atomic_uint64_t next_consumer_id{1}; /* 订阅身份生成器。 */
-    std::array<std::atomic<std::shared_ptr<const Consumer>>,
-               maximum_consumers> consumers{}; /* 每个槽位是订阅关系的唯一权威入口；热路径原子查询，不复制集合。 */
-    std::atomic_bool stopping{}; /* 关闭开始后拒绝新工作。 */
-    std::atomic_uint64_t skipped_sample_ticks{}; /* 未到 deadline 或无可写消费者的采样触发数。 */
-    std::atomic_uint64_t sampling_clock_ticks{}; /* 有消费者期间触发的独立媒体时钟 tick。 */
-    std::atomic_bool failed{}; /* 首次 Unknown Failure 后停止热路径。 */
-    std::optional<std::string> terminal_failure{}; /* 首次终止失败文本。 */
-    std::uint64_t encoded_frame_count{}; /* 累计编码成功帧数。 */
-    std::chrono::steady_clock::time_point metric_started{}; /* 当前指标窗口起点。 */
-    std::uint64_t metric_encoded_start{}; /* 窗口起点累计编码帧数。 */
-    std::uint64_t metric_sample_start{}; /* 窗口起点累计采样帧数。 */
-    std::uint64_t metric_clock_start{}; /* 窗口起点累计时钟数。 */
-    std::vector<std::uint64_t> metric_completion_starts{}; /* 各 Plot 窗口起点逻辑完成数。 */
-    std::vector<std::uint64_t> metric_rendered_starts{}; /* 各 Plot 窗口起点真实画面数。 */
-    static constexpr std::size_t maximum_taskflow_trace_frames{120};
-    std::atomic_size_t taskflow_trace_remaining{}; /* 尚待捕获的真实媒体 DAG 次数。 */
-    std::atomic_uint64_t taskflow_trace_control{}; /* 高 32 位 requested，低 32 位 captured。 */
-    std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
-               maximum_taskflow_trace_frames> taskflow_trace_slots{};
-    Private();
-    ~Private();
-    template <Attached Attached_Object>
-    void bind_private_crtp(Attached_Object* attached) {
-        Prev_Private::bind_private_crtp(attached);
-        object = not_null{static_cast<Object*>(attached)};
-    }
-    void initialize(std::vector<Plot_Entry> plots);
-    [[nodiscard]] bool consumer_accepts() const noexcept;
-    [[nodiscard]] bool remove_consumer(Stream_Id stream) noexcept;
-    void publish(std::shared_ptr<const Encoded_Video_Frame> video,
-                 std::string notification) noexcept;
-    void fail(std::exception_ptr failure) noexcept;
-    void accept_frame(std::size_t slot,
-                      std::shared_ptr<const Plot_Stream_Frame> frame);
-    void update_metrics(const frame_sampling::Sampled_Gallery_Frame& sample,
-                        std::size_t encoded_bytes,
-                        std::optional<Video_Encoder_Backend> encoder_backend);
-    void sample_media_frame();
-    void encode_media_frame();
-    void publish_video_frame();
-    void complete_media_frame();
-    [[nodiscard]] bool mark_taskflow_trace();
-    void restore_taskflow_trace() noexcept;
-    void store_taskflow_trace(const Taskflow_Frame_Trace& trace);
-    [[nodiscard]] nlohmann::json taskflow_trace_response() const;
-};
-}

web_server/src/H264_Encoder.hpp

@@ -1,43 +0,0 @@
-#pragma once
-#include <chrono>
-#include <cstddef>
-#include <cstdint>
-#include <memory>
-#include <optional>
-#include <span>
-#include <string_view>
-#include <vector>
-namespace aethera::web {
-enum struct Video_Pixel_Layout : std::uint8_t {
-    bgra,
-    rgba
-};
-enum struct Video_Encoder_Backend : std::uint8_t {
-    libx264
-};
-[[nodiscard]] std::string_view video_encoder_backend_name(
-    Video_Encoder_Backend backend) noexcept;
-[[nodiscard]] std::string_view h264_profile_level_id() noexcept;
-struct Encoded_Video_Frame {
-    std::vector<std::byte> annex_b{}; /* 带起始码的完整 H.264 access unit，可直接交给 RTP packetizer。 */
-    std::chrono::microseconds presentation_time{}; /* 从 Plot 帧时钟起点累计的媒体时间戳。 */
-    std::uint64_t sequence{}; /* 对应 Render_Frame 的单调序号。 */
-    Video_Encoder_Backend backend{Video_Encoder_Backend::libx264}; /* 实际产生本 access unit 的 FFmpeg 后端。 */
-    bool key_frame{}; /* 本 access unit 是否可独立解码。 */
-};
-struct H264_Encoder final {
-public:
-    explicit H264_Encoder(double frame_rate);
-    ~H264_Encoder();
-    H264_Encoder(const H264_Encoder&) = delete;
-    H264_Encoder& operator=(const H264_Encoder&) = delete;
-    void request_key_frame();
-    [[nodiscard]] std::optional<Encoded_Video_Frame> encode(
-        std::span<const std::byte> pixels, std::uint32_t width,
-        std::uint32_t height, Video_Pixel_Layout layout,
-        std::uint64_t sequence, std::chrono::microseconds presentation_time);
-private:
-    struct Private;
-    std::unique_ptr<Private> d;
-};
-}

web_server/src/Web_Server.cpp

@@ -1,6 +1,6 @@
 #include "Web_Server.hpp"
-#include "Gallery_Video_Stream.hpp"
-#include "Gallery_WebSocket.hpp"
+#include "media/Gallery_Video_Stream.hpp"
+#include "media/Gallery_Video_WebSocket.hpp"
 #include "Graph_WebSocket.hpp"
 #include <mcp/core/runtime/Gallery_Plots.hpp>
 #include <mcp/core/Control_Service.hpp>
@@ -20,7 +20,8 @@
 namespace aethera::web {
 namespace {
 using Gallery_Stream_Map =
-    std::unordered_map<std::string, std::shared_ptr<Gallery_Video_Stream>>;
+    std::unordered_map<std::string,
+                       std::shared_ptr<media::Gallery_Video_Stream>>;
 
 drogon::HttpResponsePtr json_response(nlohmann::json value) {
     auto response = drogon::HttpResponse::newHttpResponse();
@@ -108,8 +109,8 @@ nlohmann::json merge_gallery_media_trace(nlohmann::json plot_trace,
 int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root) {
     auto control = mcp::Control_Service::create();
 
-    std::vector<Gallery_Video_Stream::Plot_Entry> gallery_2d;
-    std::vector<Gallery_Video_Stream::Plot_Entry> gallery_3d;
+    std::vector<media::Gallery_Video_Stream::Plot_Entry> gallery_2d;
+    std::vector<media::Gallery_Video_Stream::Plot_Entry> gallery_3d;
     gallery_2d.reserve(8);
     gallery_3d.reserve(16);
     for (const auto& definition : gallery_plot_definitions()) {
@@ -131,7 +132,7 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root)
         std::unordered_map<std::string, std::string>>();
     const auto add_media_group = [&gallery_streams, &plot_media, &plot_media_group](
         std::string id,
-        std::vector<Gallery_Video_Stream::Plot_Entry> entries) {
+        std::vector<media::Gallery_Video_Stream::Plot_Entry> entries) {
         if (entries.empty()) return;
         const auto path = "/ws/gallery?group=" + id;
         for (const auto& entry : entries) {
@@ -139,18 +140,19 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root)
             plot_media_group->emplace(entry.id, id);
         }
         gallery_streams->emplace(
-            std::move(id), Gallery_Video_Stream::create(std::move(entries)));
+            std::move(id),
+            media::Gallery_Video_Stream::create(std::move(entries)));
     };
     constexpr std::size_t plots_per_media_group{4};
     const auto add_media_groups = [&add_media_group](
         std::string_view prefix,
-        std::vector<Gallery_Video_Stream::Plot_Entry> entries) {
+        std::vector<media::Gallery_Video_Stream::Plot_Entry> entries) {
         std::size_t group_index{};
         for (std::size_t offset = 0; offset < entries.size();
              offset += plots_per_media_group) {
             const auto end = std::min(
                 entries.size(), offset + plots_per_media_group);
-            std::vector<Gallery_Video_Stream::Plot_Entry> group(
+            std::vector<media::Gallery_Video_Stream::Plot_Entry> group(
                 std::make_move_iterator(entries.begin() +
                     static_cast<std::ptrdiff_t>(offset)),
                 std::make_move_iterator(entries.begin() +
@@ -165,7 +167,8 @@ int run_web_server(std::uint16_t port, const std::filesystem::path& asset_root)
         return control->find_plot(id);
     };
     auto websocket = std::make_shared<Graph_WebSocket_Controller>(resolve_plot);
-    auto gallery_websocket = std::make_shared<Gallery_WebSocket_Controller>(
+    auto gallery_websocket =
+        std::make_shared<media::Gallery_Video_WebSocket_Controller>(
         [gallery_streams](std::string_view id) {
             const auto found = gallery_streams->find(std::string{id});
             return found == gallery_streams->end() ? nullptr : found->second;

web_server/src/frame_sampling/Frame_Sampler.cpp

@@ -1,144 +0,0 @@
-#include "Frame_Sampler.hpp"
-#include <algorithm>
-#include <atomic>
-#include <cmath>
-#include <limits>
-#include <stdexcept>
-
-namespace aethera::web::frame_sampling {
-namespace {
-using Clock = std::chrono::steady_clock;
-
-std::int64_t clock_nanoseconds(Clock::time_point value) noexcept {
-    return std::chrono::duration_cast<std::chrono::nanoseconds>(
-        value.time_since_epoch()).count();
-}
-}
-
-struct Frame_Sampler::Private {
-    detail::Gallery_Frame_Atlas atlas;                      /* 各 Plot 最近原生帧与持久图集的唯一状态源。 */
-    std::chrono::nanoseconds period{};                      /* 固定采样周期；构造后不变。 */
-    double frame_rate_fps{};                               /* 对外诊断使用的精确目标频率。 */
-    std::size_t sampling_source_slot{};                    /* 为样本提供页面媒体时间的固定业务来源。 */
-    std::atomic_int64_t next_deadline_ns{};                 /* 下一次允许生成样本的单调 deadline。 */
-    std::atomic_uint64_t accepted_source_frames{};          /* 成功发布到 latest 的源帧计数。 */
-    std::atomic_uint64_t rejected_source_frames{};          /* 无效或过期源帧计数。 */
-    std::atomic_uint64_t sampled_frames{};                  /* 实际采样计数。 */
-    std::atomic_uint64_t early_ticks{};                     /* deadline 前被合并的触发计数。 */
-    std::atomic_uint64_t missed_periods{};                  /* 因迟到直接跳过的完整周期数。 */
-
-    Private(double target_frame_rate_fps,
-            std::uint32_t tile_width,
-            std::uint32_t tile_height,
-            std::uint32_t columns,
-            std::vector<std::string> source_ids,
-            std::size_t source_slot)
-        : atlas(tile_width, tile_height, columns, std::move(source_ids)),
-          period(static_cast<std::int64_t>(std::llround(
-              1'000'000'000.0 / target_frame_rate_fps))),
-          frame_rate_fps(target_frame_rate_fps),
-          sampling_source_slot(source_slot) {}
-};
-
-Frame_Sampler::Frame_Sampler(
-    double frame_rate_fps, std::uint32_t tile_width,
-    std::uint32_t tile_height, std::uint32_t columns,
-    std::vector<std::string> source_ids,
-    std::size_t sampling_source_slot) {
-    if (!std::isfinite(frame_rate_fps) || frame_rate_fps <= 0.0)
-        throw std::invalid_argument(
-            "frame sampler rate must be finite and positive");
-    if (sampling_source_slot >= source_ids.size())
-        throw std::out_of_range("frame sampler source slot is invalid");
-    d = std::make_unique<Private>(
-        frame_rate_fps, tile_width, tile_height, columns,
-        std::move(source_ids), sampling_source_slot);
-    if (d->period <= std::chrono::nanoseconds::zero())
-        throw std::invalid_argument("frame sampler period is too small");
-}
-
-Frame_Sampler::~Frame_Sampler() = default;
-
-Frame_Sampler::Accept_Frame_Result Frame_Sampler::accept_frame(
-    std::size_t slot, std::shared_ptr<const Plot_Pixel_Frame> frame) {
-    const auto result = d->atlas.accept_frame(slot, std::move(frame));
-    if (result == detail::Gallery_Frame_Atlas::Accept_Frame_Result::accepted) {
-        d->accepted_source_frames.fetch_add(1, std::memory_order_release);
-        return Accept_Frame_Result::accepted;
-    }
-    d->rejected_source_frames.fetch_add(1, std::memory_order_relaxed);
-    return result == detail::Gallery_Frame_Atlas::Accept_Frame_Result::stale_frame
-        ? Accept_Frame_Result::stale_frame
-        : Accept_Frame_Result::invalid_frame;
-}
-
-std::optional<Sampled_Gallery_Frame> Frame_Sampler::sample() {
-    const auto now = Clock::now();
-    const auto now_ns = clock_nanoseconds(now);
-    const auto compose_sample = [this, now](double deadline_delay_ms,
-                                            std::uint64_t sample_sequence) {
-        auto composition = d->atlas.compose();
-        const auto& source = composition.sources[d->sampling_source_slot];
-        Plot_Render_Tick tick{
-            now,
-            source.completion_correlation_id,
-            static_cast<double>(source.completion_presentation_time.count()) /
-                1'000.0,
-            composition.width,
-            composition.height};
-        return Sampled_Gallery_Frame{
-            std::move(composition), tick, sample_sequence,
-            deadline_delay_ms};
-    };
-    /* 首帧以前没有可编码内容；首帧以后每个 deadline 都编码当前图集。
-     * 这不是 changed-only：静态画面仍持续产生 H.264 帧并维持解码参考链。 */
-    if (d->accepted_source_frames.load(std::memory_order_acquire) == 0)
-        return std::nullopt;
-    auto deadline_ns = d->next_deadline_ns.load(std::memory_order_acquire);
-    for (;;) {
-        if (deadline_ns == 0) {
-            const auto next = now_ns + d->period.count();
-            if (!d->next_deadline_ns.compare_exchange_weak(
-                    deadline_ns, next, std::memory_order_acq_rel,
-                    std::memory_order_acquire))
-                continue;
-            const auto sequence = d->sampled_frames.fetch_add(
-                1, std::memory_order_relaxed) + 1U;
-            return compose_sample(0.0, sequence);
-        }
-        if (now_ns < deadline_ns) {
-            d->early_ticks.fetch_add(1, std::memory_order_relaxed);
-            return std::nullopt;
-        }
-        const auto late_ns = now_ns - deadline_ns;
-        const auto missed = static_cast<std::uint64_t>(
-            late_ns / d->period.count());
-        const auto next = deadline_ns +
-            static_cast<std::int64_t>(missed + 1U) * d->period.count();
-        if (!d->next_deadline_ns.compare_exchange_weak(
-                deadline_ns, next, std::memory_order_acq_rel,
-                std::memory_order_acquire))
-            continue;
-        const auto sequence = d->sampled_frames.fetch_add(
-            1, std::memory_order_relaxed) + 1U;
-        d->missed_periods.fetch_add(missed, std::memory_order_relaxed);
-        return compose_sample(static_cast<double>(late_ns) / 1'000'000.0,
-                              sequence);
-    }
-}
-
-detail::Gallery_Atlas_Description Frame_Sampler::describe() const {
-    return d->atlas.describe();
-}
-
-Frame_Sampler_State Frame_Sampler::state() const noexcept {
-    return {
-        d->frame_rate_fps,
-        d->accepted_source_frames.load(std::memory_order_relaxed),
-        d->rejected_source_frames.load(std::memory_order_relaxed),
-        d->sampled_frames.load(std::memory_order_relaxed),
-        d->early_ticks.load(std::memory_order_relaxed),
-        d->missed_periods.load(std::memory_order_relaxed)};
-}
-
-}

web_server/src/frame_sampling/Frame_Sampler.hpp

@@ -1,63 +0,0 @@
-#pragma once
-#include <mcp/core/runtime/Plot.hpp>
-#include "../detail/Gallery_Frame_Atlas.hpp"
-#include <chrono>
-#include <cstddef>
-#include <cstdint>
-#include <memory>
-#include <optional>
-#include <string>
-#include <vector>
-
-namespace aethera::web::frame_sampling {
-
-struct Sampled_Gallery_Frame {
-    detail::Gallery_Atlas_Composition composition{};       /* 本次采样得到的原生像素图集；下一次成功采样前有效。 */
-    Plot_Render_Tick tick{};                               /* 触发采样的服务端帧时钟。 */
-    std::uint64_t sample_sequence{};                       /* 媒体流自身严格递增的帧序号。 */
-    double deadline_delay_ms{};                            /* 实际采样相对固定周期 deadline 的延迟。 */
-};
-
-struct Frame_Sampler_State {
-    double target_frame_rate_fps{};                        /* 采样器固定目标频率。 */
-    std::uint64_t accepted_source_frames{};                /* 已进入权威 latest 的 Plot 完成帧数。 */
-    std::uint64_t rejected_source_frames{};                /* 因尺寸、布局或顺序不合法而拒绝的完成帧数。 */
-    std::uint64_t sampled_frames{};                        /* 到达 deadline 并实际生成的图集样本数。 */
-    std::uint64_t early_ticks{};                           /* deadline 前到达、未触发采样的外接图次数。 */
-    std::uint64_t missed_periods{};                        /* 调度延迟跨过的完整采样周期数。 */
-};
-
-/*
- * Gallery 像素采样的唯一权威入口。Plot 可以按任意帧策略发布，采样器只在
- * 固定 deadline 到达时读取每个槽位的 latest，不排队、不补历史帧。
- */
-struct Frame_Sampler final {
-public:
-    enum struct Accept_Frame_Result : std::uint8_t {
-        accepted,
-        invalid_frame,
-        stale_frame
-    };
-
-    Frame_Sampler(double frame_rate_fps,
-                  std::uint32_t tile_width,
-                  std::uint32_t tile_height,
-                  std::uint32_t columns,
-                  std::vector<std::string> source_ids,
-                  std::size_t sampling_source_slot);
-    ~Frame_Sampler();
-    Frame_Sampler(const Frame_Sampler&) = delete;
-    Frame_Sampler& operator=(const Frame_Sampler&) = delete;
-
-    [[nodiscard]] Accept_Frame_Result accept_frame(
-        std::size_t slot, std::shared_ptr<const Plot_Pixel_Frame> frame);
-    [[nodiscard]] std::optional<Sampled_Gallery_Frame> sample();
-    [[nodiscard]] detail::Gallery_Atlas_Description describe() const;
-    [[nodiscard]] Frame_Sampler_State state() const noexcept;
-
-private:
-    struct Private;
-    std::unique_ptr<Private> d;
-};
-
-}

web_server/src/frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp

@@ -1,26 +0,0 @@
-#pragma once
-#include "../../H264_Encoder.hpp"
-#include "../Frame_Sampler.hpp"
-#include <memory>
-#include <optional>
-
-namespace aethera::web::frame_sampling::ffmpeg {
-
-/* 固定采样帧到 FFmpeg H.264 access unit 的单一编码位置。 */
-struct FFmpeg_Frame_Transport final {
-public:
-    explicit FFmpeg_Frame_Transport(double frame_rate_fps);
-    ~FFmpeg_Frame_Transport();
-    FFmpeg_Frame_Transport(const FFmpeg_Frame_Transport&) = delete;
-    FFmpeg_Frame_Transport& operator=(const FFmpeg_Frame_Transport&) = delete;
-
-    void request_key_frame() noexcept;
-    [[nodiscard]] std::optional<Encoded_Video_Frame> encode(
-        const Sampled_Gallery_Frame& frame);
-
-private:
-    struct Private;
-    std::unique_ptr<Private> d;
-};
-
-}

web_server/src/media/Encoded_Video_Frame.hpp

@@ -0,0 +1,32 @@
+#pragma once
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <string_view>
+#include <vector>
+
+namespace aethera::web::media {
+
+enum struct Video_Pixel_Layout : std::uint8_t {
+    bgra,
+    rgba
+};
+
+enum struct Video_Encoder_Backend : std::uint8_t {
+    libx264
+};
+
+[[nodiscard]] std::string_view video_encoder_backend_name(
+    Video_Encoder_Backend backend) noexcept;
+[[nodiscard]] std::string_view h264_profile_level_id() noexcept;
+
+struct Encoded_Video_Frame {
+    std::vector<std::byte> annex_b{};                              /* 带起始码的完整 H.264 access unit，可直接交给 WebCodecs。 */
+    std::chrono::microseconds presentation_time{};                 /* 从媒体流起点累计的显示时间戳。 */
+    std::chrono::nanoseconds source_time_unix{};                   /* 对应 Plot 像素进入媒体流水线的 Unix 时间。 */
+    std::uint64_t sequence{};                                      /* 图集媒体流的严格单调帧序号。 */
+    Video_Encoder_Backend backend{Video_Encoder_Backend::libx264}; /* 实际产生本 access unit 的编码后端。 */
+    bool key_frame{};                                              /* 本 access unit 是否可独立解码。 */
+};
+
+}

web_server/src/media/Gallery_Video_Stream.cpp

@@ -2,7 +2,6 @@
 #include <mcp/core/runtime/Taskflow_Trace_Json.hpp>
 #include <frame_statistics.hpp>
 #include <render_common.hpp>
-#include "frame_sampling/Frame_Sampler.hpp"
 #include <nlohmann/json.hpp>
 #include <algorithm>
 #include <atomic>
@@ -12,13 +11,13 @@
 #include <stdexcept>
 #include <string>
 #include <utility>
-namespace aethera::web {
+namespace aethera::web::media {
 namespace {
 // Scene 的业务视口是 720x420；每组四图形成一行 2880x420 H.264 图集。
 constexpr std::uint32_t tile_width{720};
 constexpr std::uint32_t tile_height{420};
 constexpr std::uint32_t atlas_columns{4};
-constexpr double gallery_frame_rate{30.0};
+constexpr double encoder_nominal_frame_rate{30.0};
 constexpr auto metric_interval{std::chrono::seconds(1)};
 nlohmann::json statistic_json(const Statistic_State& value) {
     return {
@@ -44,7 +43,7 @@ std::string exception_description(const std::exception_ptr& failure) {
 }
 }
 Gallery_Video_Stream::Private::Private()
-    : ffmpeg_transport(gallery_frame_rate) {}
+    : ffmpeg_transport(encoder_nominal_frame_rate) {}
 Gallery_Video_Stream::Private::~Private() = default;
 void Gallery_Video_Stream::Private::initialize(
     std::vector<Plot_Entry> plots) {
@@ -57,9 +56,8 @@ void Gallery_Video_Stream::Private::initialize(
         source_ids.push_back(plot.id);
         sources.push_back(Source{std::move(plot)});
     }
-    sampler = std::make_unique<frame_sampling::Frame_Sampler>(
-        gallery_frame_rate, tile_width, tile_height, atlas_columns,
-        std::move(source_ids), sources.size() - 1U);
+    atlas = std::make_unique<detail::Gallery_Frame_Atlas>(
+        tile_width, tile_height, atlas_columns, std::move(source_ids));
     metric_completion_starts.resize(sources.size());
     metric_rendered_starts.resize(sources.size());
 }
@@ -167,7 +165,7 @@ void Gallery_Video_Stream::Private::fail(
         const auto notification = nlohmann::json{
                     {"kind", "gallery_error"},
                     {"protocol", "aethera.gallery.video"},
-                    {"version", 4},
+                    {"version", 5},
                     {"message", *terminal_failure}
                 }.dump();
         publish({}, notification);
@@ -177,37 +175,48 @@ void Gallery_Video_Stream::Private::fail(
 void Gallery_Video_Stream::Private::accept_frame(
     std::size_t slot, std::shared_ptr<const Plot_Stream_Frame> frame) {
     if (stopping.load(std::memory_order_acquire) ||
-        failed.load(std::memory_order_acquire) || !frame)
+        failed.load(std::memory_order_acquire) || !frame || !frame->pixels ||
+        !frame->pixels->pixels || !consumer_accepts())
         return;
-    static_cast<void>(sampler->accept_frame(slot, frame->pixels));
+    Pending_Plot_Frame pending{
+        slot, std::move(frame->pixels), std::chrono::steady_clock::now(),
+        std::chrono::duration_cast<std::chrono::nanoseconds>(
+            std::chrono::system_clock::now().time_since_epoch())};
+    if (!pending_frames.enqueue(std::move(pending))) throw std::bad_alloc{};
+    received_frame_count.fetch_add(1, std::memory_order_relaxed);
+    media_work_generation.fetch_add(1, std::memory_order_release);
+    arm_media_task(object->weak_from_this());
 }
 void Gallery_Video_Stream::Private::update_metrics(
-    const frame_sampling::Sampled_Gallery_Frame& sample,
+    const Active_Media_Frame& frame,
     std::size_t encoded_bytes,
     std::optional<Video_Encoder_Backend> encoder_backend) {
-    const auto& composition = sample.composition;
-    const auto sampler_state = sampler->state();
+    const auto& composition = frame.composition;
     const auto now = std::chrono::steady_clock::now();
+    const auto received_total = received_frame_count.load(
+        std::memory_order_relaxed);
+    const auto processed_total = processed_frame_count.load(
+        std::memory_order_relaxed);
+    const auto encoded_total = encoded_frame_count.load(
+        std::memory_order_relaxed);
     if (metric_started.time_since_epoch().count() == 0) {
         metric_started = now;
-        metric_encoded_start = encoded_frame_count;
-        metric_sample_start = sampler_state.sampled_frames;
-        metric_clock_start = sampling_clock_ticks.load(std::memory_order_relaxed);
+        metric_received_start = received_total;
+        metric_processed_start = processed_total;
+        metric_encoded_start = encoded_total;
         for (std::size_t slot = 0; slot < composition.sources.size(); ++slot) {
             metric_completion_starts[slot] =
-            composition.sources[slot].completion_count;
+                composition.sources[slot].completion_count;
             metric_rendered_starts[slot] =
-            composition.sources[slot].rendered_frame_count;
+                composition.sources[slot].rendered_frame_count;
         }
         return;
     }
     const auto elapsed = now - metric_started;
     if (elapsed < metric_interval) return;
     const double seconds = std::chrono::duration<double>(elapsed).count();
-    const auto clock_total =
-    sampling_clock_ticks.load(std::memory_order_relaxed);
     nlohmann::json source_metrics = nlohmann::json::object();
-    const auto description = sampler->describe();
+    const auto description = atlas->describe();
     for (std::size_t slot = 0; slot < composition.sources.size(); ++slot) {
         const auto& progress = composition.sources[slot];
         const auto completed = progress.completion_count -
@@ -244,38 +253,39 @@ void Gallery_Video_Stream::Private::update_metrics(
         metric_rendered_starts[slot] = progress.rendered_frame_count;
     }
     const auto& compose = compose_ms.state();
-    const auto& sample_delay = sample_delay_ms.state();
+    const auto& queue_delay = queue_delay_ms.state();
     const auto& encode = encode_ms.state();
     const auto& publish_time = publish_ms.state();
     auto output = nlohmann::json{
         {"kind", "gallery_metrics"},
         {"protocol", "aethera.gallery.video"},
-        {"version", 4},
-        {"clock_sequence", sample.sample_sequence},
-        {"sampled_frame_count", sampler_state.sampled_frames},
-        {"encoded_frame_count", encoded_frame_count},
+        {"version", 5},
+        {"media_sequence", frame.sequence},
+        {"received_frame_count", received_total},
+        {"processed_frame_count", processed_total},
+        {"encoded_frame_count", encoded_total},
         {"encoder_backend", encoder_backend
             ? video_encoder_backend_name(*encoder_backend)
             : std::string_view{"inactive"}},
         {"profile_level_id", h264_profile_level_id()},
-        {"target_frame_rate_fps", gallery_frame_rate},
         {
-            "clock_delivery_rate_fps",
-            static_cast<double>(clock_total - metric_clock_start) / seconds
+            "received_frame_rate_fps",
+            static_cast<double>(received_total - metric_received_start) /
+                seconds
         },
         {
-            "sampled_frame_rate_fps",
-            static_cast<double>(sampler_state.sampled_frames -
-                metric_sample_start) / seconds
+            "processed_frame_rate_fps",
+            static_cast<double>(processed_total -
+                metric_processed_start) / seconds
         },
         {
             "encoded_frame_rate_fps",
-            static_cast<double>(encoded_frame_count - metric_encoded_start) / seconds
+            static_cast<double>(encoded_total - metric_encoded_start) / seconds
         },
         {"compose_average_ms", compose.average},
         {"compose_p95_ms", compose.p95},
-        {"sample_delay_average_ms", sample_delay.average},
-        {"sample_delay_p95_ms", sample_delay.p95},
+        {"queue_delay_average_ms", queue_delay.average},
+        {"queue_delay_p95_ms", queue_delay.p95},
         {"encode_average_ms", encode.average},
         {"encode_p95_ms", encode.p95},
         {"publish_average_ms", publish_time.average},
@@ -283,57 +293,58 @@ void Gallery_Video_Stream::Private::update_metrics(
         {
             "statistics", {
                 {"compose_ms", statistic_json(compose)},
-                {"sample_delay_ms", statistic_json(sample_delay)},
+                {"queue_delay_ms", statistic_json(queue_delay)},
                 {"encode_ms", statistic_json(encode)},
                 {"publish_ms", statistic_json(publish_time)}
             }
         },
+        {"pending_frame_count", pending_frames.size_approx()},
         {"encoded_bytes", encoded_bytes},
         {"fresh_tiles", composition.fresh_tile_count},
         {"missing_tiles", composition.missing_tile_count},
         {"rejected_frames", composition.rejected_frame_count},
-        {
-            "skipped_sample_ticks",
-            skipped_sample_ticks.load(std::memory_order_relaxed)
-        },
-        {"sampler", {
-            {"accepted_source_frames", sampler_state.accepted_source_frames},
-            {"rejected_source_frames", sampler_state.rejected_source_frames},
-            {"early_ticks", sampler_state.early_ticks},
-            {"missed_periods", sampler_state.missed_periods}}},
         {"sources", std::move(source_metrics)}
     };
     metric_started = now;
-    metric_encoded_start = encoded_frame_count;
-    metric_sample_start = sampler_state.sampled_frames;
-    metric_clock_start = clock_total;
+    metric_received_start = received_total;
+    metric_processed_start = processed_total;
+    metric_encoded_start = encoded_total;
     object->template update_state<&State::diagnostics>(std::move(output));
     object->template publish_state<Base_Tag>();
 }
-void Gallery_Video_Stream::Private::sample_media_frame() {
-    active_sample.reset();
+void Gallery_Video_Stream::Private::begin_media_frame() {
+    active_frame.reset();
     active_video.reset();
     if (stopping.load(std::memory_order_acquire) ||
-        failed.load(std::memory_order_acquire) ||
-        !consumer_accepts()) {
-        skipped_sample_ticks.fetch_add(1, std::memory_order_relaxed);
-        return;
-    }
+        failed.load(std::memory_order_acquire)) return;
+    Pending_Plot_Frame pending;
+    if (!pending_frames.try_dequeue(pending)) return;
+    ++processed_frame_count;
     const auto started = std::chrono::steady_clock::now();
-    active_sample = sampler->sample();
-    if (!active_sample) {
-        skipped_sample_ticks.fetch_add(1, std::memory_order_relaxed);
-        return;
-    }
+    const auto accepted = atlas->accept_frame(
+        pending.slot, std::move(pending.frame));
+    if (accepted ==
+        detail::Gallery_Frame_Atlas::Accept_Frame_Result::invalid_frame)
+        throw std::logic_error(
+            "Plot published an invalid gallery pixel frame");
+    auto presentation_time = std::chrono::duration_cast<
+        std::chrono::microseconds>(started - media_origin);
+    if (presentation_time <= last_presentation_time)
+        presentation_time = last_presentation_time +
+            std::chrono::microseconds{1};
+    last_presentation_time = presentation_time;
+    active_frame = Active_Media_Frame{
+        atlas->compose(), next_media_sequence++, presentation_time,
+        pending.source_time_unix,
+        std::chrono::duration<double, std::milli>(
+            started - pending.enqueued_at).count()};
     static_cast<void>(compose_ms.submit(
         std::chrono::duration<double, std::milli>(
             std::chrono::steady_clock::now() - started).count()));
-    static_cast<void>(sample_delay_ms.submit(
-        active_sample->deadline_delay_ms));
+    static_cast<void>(queue_delay_ms.submit(active_frame->queue_delay_ms));
 }
 void Gallery_Video_Stream::Private::encode_media_frame() {
-    if (!active_sample || !consumer_accepts())
-        return;
+    if (!active_frame) return;
     /*
      * 直接在 FFmpeg.H264.encode Taskflow 节点中调用 FFmpeg。这样实际
      * RGB->YUV420P 与 libx264 send/receive 全部落在同一个业务节点的
@@ -342,7 +353,14 @@ void Gallery_Video_Stream::Private::encode_media_frame() {
      * H264_Encoder 因而仍只有一个执行位置，不需要额外 mutex 或专用线程。
      */
     const auto started = std::chrono::steady_clock::now();
-    if (auto encoded = ffmpeg_transport.encode(*active_sample))
+    if (auto encoded = ffmpeg_transport.encode(
+            active_frame->composition.pixels,
+            active_frame->composition.width,
+            active_frame->composition.height,
+            active_frame->composition.layout,
+            active_frame->sequence,
+            active_frame->presentation_time,
+            active_frame->source_time_unix))
         active_video = std::make_shared<const Encoded_Video_Frame>(
             std::move(*encoded));
     static_cast<void>(encode_ms.submit(
@@ -359,13 +377,13 @@ void Gallery_Video_Stream::Private::publish_video_frame() {
             std::chrono::steady_clock::now() - started).count()));
 }
 void Gallery_Video_Stream::Private::complete_media_frame() {
-    if (!active_sample) return;
+    if (!active_frame) return;
     const auto encoded_bytes = active_video ? active_video->annex_b.size() : 0U;
     const auto encoder_backend = active_video
         ? std::optional{active_video->backend}
         : std::nullopt;
-    update_metrics(*active_sample, encoded_bytes, encoder_backend);
-    active_sample.reset();
+    update_metrics(*active_frame, encoded_bytes, encoder_backend);
+    active_frame.reset();
     active_video.reset();
 }
 std::shared_ptr<Gallery_Video_Stream> Gallery_Video_Stream::create(
@@ -383,28 +401,86 @@ Gallery_Video_Stream::Gallery_Video_Stream() = default;
 Gallery_Video_Stream::~Gallery_Video_Stream() {
     shutdown();
 }
+void Gallery_Video_Stream::Private::arm_media_task(
+    std::weak_ptr<Gallery_Video_Stream> lifetime) {
+    if (stopping.load(std::memory_order_acquire) ||
+        failed.load(std::memory_order_acquire))
+        return;
+    bool idle{};
+    if (!media_task_scheduled.compare_exchange_strong(
+            idle, true, std::memory_order_acq_rel,
+            std::memory_order_acquire))
+        return;
+
+    const auto graph = media_graph;
+    bool trace_reserved = mark_taskflow_trace();
+    auto trace_frame = trace_reserved
+        ? std::make_shared<Render_Frame>(Frame_Identity{
+              received_frame_count.load(std::memory_order_acquire), 0})
+        : std::shared_ptr<Render_Frame>{};
+    bool trace_started{};
+    if (trace_frame) {
+        trace_frame->request_taskflow_trace();
+        trace_started = aethera::detail::begin_taskflow_trace(*trace_frame);
+        if (!trace_started) {
+            restore_taskflow_trace();
+            trace_reserved = false;
+            trace_frame.reset();
+        }
+    }
+
+    try {
+        auto completion = [lifetime, graph, trace_frame, trace_started] {
+            static_cast<void>(graph);
+            if (trace_started)
+                aethera::detail::finish_taskflow_trace(*trace_frame);
+            const auto completed = lifetime.lock();
+            if (!completed) return;
+            auto& data = static_cast<Private&>(*completed->d);
+            if (trace_started)
+                data.store_taskflow_trace(
+                    trace_frame->take_taskflow_trace());
+            data.media_task_scheduled.store(false,
+                                            std::memory_order_release);
+            if (data.received_frame_count.load(std::memory_order_acquire) >
+                data.processed_frame_count.load(std::memory_order_acquire))
+                data.arm_media_task(lifetime);
+        };
+        if (trace_frame)
+            aethera::detail::run_taskflow(
+                *graph, *trace_frame, "gallery.media",
+                std::move(completion));
+        else
+            aethera::detail::run_taskflow(*graph, std::move(completion));
+    }
+    catch (...) {
+        if (trace_started)
+            aethera::detail::finish_taskflow_trace(*trace_frame);
+        if (trace_reserved) restore_taskflow_trace();
+        media_task_scheduled.store(false, std::memory_order_release);
+        fail(std::current_exception());
+    }
+}
 void Gallery_Video_Stream::bind_plots() {
     auto& data = static_cast<Private&>(*d);
     const auto weak = weak_from_this();
     if (data.sources.empty())
         throw std::invalid_argument("gallery video stream requires a Plot");
 
-    /*
-     * 图集媒体有自己的固定采样时钟，不再依赖某一张 Plot 完成后才能运行。
-     * 每个来源只发布 latest；静态画面仍按 30 FPS 编码，上一轮媒体 DAG
-     * 尚未完成时合并本次 tick，不排队也不占用 Worker 等待。
-     */
+    /* 每个 Plot 独立帧策略发布的物理像素帧按到达顺序进入本 DAG。
+     * Gallery 不再拥有采样时钟、准入或丢帧规则；持久 DAG 每次消费一个
+     * 完成帧，串行保护图集像素和 FFmpeg 上下文，完成后异步续跑下一帧。 */
     auto media = std::make_shared<Task_Graph>("gallery.video.frame");
-    auto sample = media->add("gallery.sample.capture", [weak] {
+    auto compose = media->add("gallery.atlas.compose", [weak] {
         if (const auto owner = weak.lock()) {
             auto& owner_data = static_cast<Private&>(*owner->d);
-            try { owner_data.sample_media_frame(); }
+            try { owner_data.begin_media_frame(); }
             catch (...) { owner_data.fail(std::current_exception()); }
         }
     });
-    sample.describe("owner", "gallery")
-          .describe("stage", "30 FPS latest-frame sampling deadline")
-          .describe("frame_rate_fps", "30");
+    compose.describe("owner", "gallery")
+           .describe("stage", "apply one policy-approved Plot frame")
+           .describe("admission", "owned by source Plot frame policy");
     auto encode = media->add("FFmpeg.H264.encode", [weak] {
         if (const auto owner = weak.lock()) {
             auto& owner_data = static_cast<Private&>(*owner->d);
@@ -413,7 +489,7 @@ void Gallery_Video_Stream::bind_plots() {
         }
     });
     encode.describe("owner", "gallery")
-          .describe("stage", "FFmpeg H.264 encode after Plot pixel publish")
+          .describe("stage", "encode every composed Plot completion")
           .describe("backend", "FFmpeg")
           .describe("codec", "H.264")
           .describe("execution", "Taskflow worker");
@@ -427,7 +503,7 @@ void Gallery_Video_Stream::bind_plots() {
     publish_video.describe("owner", "gallery")
                    .describe("transport", "Drogon WebSocket")
                    .describe("stage", "H.264 access unit enqueue");
-    auto complete = media->add("gallery.sample.complete", [weak] {
+    auto complete = media->add("gallery.media.complete", [weak] {
         if (const auto owner = weak.lock()) {
             auto& owner_data = static_cast<Private&>(*owner->d);
             try { owner_data.complete_media_frame(); }
@@ -435,80 +511,11 @@ void Gallery_Video_Stream::bind_plots() {
         }
     });
     complete.describe("owner", "gallery")
-            .describe("stage", "sample metrics and ownership release");
-    sample.precede(encode);
+            .describe("stage", "metrics and frame ownership release");
+    compose.precede(encode);
     encode.precede(publish_video);
     publish_video.precede(complete);
     data.media_graph = media;
-    data.sample_timer = Frame_Scheduler::instance().make_timer(
-        [weak](Frame_Scheduler::Tick) {
-            const auto owner = weak.lock();
-            if (!owner) return;
-            auto& owner_data = static_cast<Private&>(*owner->d);
-            if (owner_data.stopping.load(std::memory_order_acquire) ||
-                owner_data.failed.load(std::memory_order_acquire) ||
-                !owner_data.consumer_accepts())
-                return;
-            owner_data.sampling_clock_ticks.fetch_add(
-                1, std::memory_order_relaxed);
-            bool idle{};
-            if (!owner_data.media_busy.compare_exchange_strong(
-                    idle, true, std::memory_order_acq_rel,
-                    std::memory_order_acquire)) {
-                owner_data.skipped_sample_ticks.fetch_add(
-                    1, std::memory_order_relaxed);
-                return;
-            }
-            const auto graph = owner_data.media_graph;
-            bool trace_reserved = owner_data.mark_taskflow_trace();
-            auto trace_frame = trace_reserved
-                ? std::make_shared<Render_Frame>(Frame_Identity{
-                    owner_data.sampling_clock_ticks.load(
-                        std::memory_order_acquire), 0})
-                : std::shared_ptr<Render_Frame>{};
-            bool trace_started{};
-            if (trace_frame) {
-                trace_frame->request_taskflow_trace();
-                trace_started = aethera::detail::begin_taskflow_trace(
-                    *trace_frame);
-                if (!trace_started) {
-                    owner_data.restore_taskflow_trace();
-                    trace_reserved = false;
-                    trace_frame.reset();
-                }
-            }
-            try {
-                auto completion = [weak, graph, trace_frame, trace_started] {
-                    static_cast<void>(graph);
-                    if (trace_started)
-                        aethera::detail::finish_taskflow_trace(*trace_frame);
-                    if (const auto completed = weak.lock()) {
-                        auto& completed_data =
-                            static_cast<Private&>(*completed->d);
-                        if (trace_started)
-                            completed_data.store_taskflow_trace(
-                                trace_frame->take_taskflow_trace());
-                        completed_data.media_busy.store(
-                            false, std::memory_order_release);
-                    }
-                };
-                if (trace_frame)
-                    aethera::detail::run_taskflow(
-                        *graph, *trace_frame, "gallery.media",
-                        std::move(completion));
-                else
-                    aethera::detail::run_taskflow(
-                        *graph, std::move(completion));
-            }
-            catch (...) {
-                if (trace_started)
-                    aethera::detail::finish_taskflow_trace(*trace_frame);
-                if (trace_reserved) owner_data.restore_taskflow_trace();
-                owner_data.media_busy.store(false,
-                                            std::memory_order_release);
-                owner_data.fail(std::current_exception());
-            }
-        });
 
     for (std::size_t slot = 0; slot < data.sources.size(); ++slot) {
         auto& source = data.sources[slot];
@@ -518,8 +525,8 @@ void Gallery_Video_Stream::bind_plots() {
                 if (!owner) return;
                 auto& owner_data = static_cast<Private&>(*owner->d);
                 if (owner_data.stopping.load(std::memory_order_acquire)) return;
-                /* 每个 Plot 只更新自己槽位的权威最新完成帧。 */
-                owner_data.accept_frame(slot, std::move(frame));
+                try { owner_data.accept_frame(slot, std::move(frame)); }
+                catch (...) { owner_data.fail(std::current_exception()); }
             });
         source.entry.plot->configure_stream(
             source.stream, tile_width, tile_height);
@@ -558,7 +565,6 @@ Gallery_Video_Stream::Stream_Id Gallery_Video_Stream::subscribe(
                     std::memory_order_acquire)) {
                 if (data.stopping.load(std::memory_order_acquire)) {
                     static_cast<void>(data.remove_consumer(id));
-                    data.sample_timer.cancel();
                     throw std::logic_error(
                         "gallery video stream is shutting down");
                 }
@@ -579,10 +585,8 @@ Gallery_Video_Stream::Stream_Id Gallery_Video_Stream::subscribe(
     if (!inserted)
         throw std::runtime_error(
             "gallery video stream consumer capacity is exhausted");
-    data.sample_timer.start_periodic(gallery_frame_rate);
     if (data.stopping.load(std::memory_order_acquire)) {
         static_cast<void>(data.remove_consumer(id));
-        data.sample_timer.cancel();
         throw std::logic_error("gallery video stream is shutting down");
     }
     return id;
@@ -600,7 +604,6 @@ void Gallery_Video_Stream::request_video_key_frame() {
 void Gallery_Video_Stream::shutdown() noexcept {
     auto& data = static_cast<Private&>(*d);
     if (data.stopping.exchange(true, std::memory_order_acq_rel)) return;
-    data.sample_timer.cancel();
     for (auto& source : data.sources) {
         if (source.stream == 0) continue;
         try {
@@ -614,15 +617,14 @@ void Gallery_Video_Stream::shutdown() noexcept {
 }
 std::string Gallery_Video_Stream::layout_description() const {
     const auto& data = static_cast<const Private&>(*d);
-    const auto description = data.sampler->describe();
+    const auto description = data.atlas->describe();
     nlohmann::json plots = nlohmann::json::object();
     for (const auto& source : description.sources) plots[source.id] = {{"column", source.column}, {"row", source.row}};
     return nlohmann::json{
         {"kind", "gallery_layout"},
         {"protocol", "aethera.gallery.video"},
-        {"version", 4},
+        {"version", 5},
         {"transport", "drogon_h264"},
-        {"frame_rate_fps", gallery_frame_rate},
         {"columns", description.columns},
         {"rows", description.rows},
         {"tile_width", description.tile_width},
@@ -636,8 +638,8 @@ std::string Gallery_Video_Stream::layout_description() const {
                 {"profile_level_id", h264_profile_level_id()},
                 {"transport", "Drogon WebSocket binary"},
                 {"bitstream", "Annex B"},
-                {"protocol_version", 1},
-                {"header_bytes", 40}
+                {"protocol_version", 2},
+                {"header_bytes", 48}
             }
         },
         {"plots", std::move(plots)}
@@ -647,13 +649,20 @@ nlohmann::json Gallery_Video_Stream::diagnostics() const {
     nlohmann::json output{
         {"kind", "gallery_metrics"},
         {"protocol", "aethera.gallery.video"},
-        {"version", 4},
+        {"version", 5},
         {"sources", nlohmann::json::object()}
     };
     this->template access_state<Base_Tag>([&output](const State& state) {
         if (!state.diagnostics.empty()) output = state.diagnostics;
     });
     const auto& data = static_cast<const Private&>(*d);
+    output["received_frame_count"] = data.received_frame_count.load(
+        std::memory_order_relaxed);
+    output["processed_frame_count"] = data.processed_frame_count.load(
+        std::memory_order_relaxed);
+    output["encoded_frame_count"] = data.encoded_frame_count.load(
+        std::memory_order_relaxed);
+    output["pending_frame_count"] = data.pending_frames.size_approx();
     std::size_t consumer_count{};
     for (const auto& slot : data.consumers) {
         const auto consumer = slot.load(std::memory_order_acquire);

web_server/src/media/Gallery_Video_Stream.hpp

@@ -1,6 +1,6 @@
 #pragma once
 #include <mcp/core/runtime/Plot.hpp>
-#include "frame_sampling/ffmpeg/FFmpeg_Frame_Transport.hpp"
+#include "Encoded_Video_Frame.hpp"
 #include <render_common.hpp>
 #include <cstdint>
 #include <functional>
@@ -9,7 +9,7 @@
 #include <string>
 #include <vector>
 
-namespace aethera::web {
+namespace aethera::web::media {
 struct Gallery_Stream_Frame {
     std::shared_ptr<const Encoded_Video_Frame> video{}; /* FFmpeg 订阅者共享的 H.264 access unit。 */
     std::string notification{};                /* 仅承载终止错误等低频控制通知。 */

web_server/src/media/Gallery_Video_Stream.ipp

@@ -0,0 +1,116 @@
+#pragma once
+#include "detail/FFmpeg_Frame_Transport.hpp"
+#include "detail/Gallery_Frame_Atlas.hpp"
+#include <concurrentqueue-1.0.5/concurrentqueue.h>
+#include <frame_statistics.hpp>
+#include <ownership.hpp>
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <optional>
+
+namespace aethera::web::media {
+
+struct Gallery_Video_Stream::Private : Prev_Private {
+    using Object = Gallery_Video_Stream;
+
+    struct Source {
+        Plot_Entry entry;          /* 图集槽位关联的实际 Plot。 */
+        Plot::Stream_Id stream{};  /* Plot 完成帧的唯一订阅标识。 */
+    };
+
+    struct Consumer {
+        Stream_Id id{};                                        /* 当前网页媒体连接的订阅身份。 */
+        std::string connection{};                              /* 同一页面重连期间稳定的协议身份。 */
+        std::shared_ptr<const Stream_Handler> handler{};        /* 发布期间保持回调存活。 */
+        std::shared_ptr<const Transport_Readiness> readiness{}; /* 连接和解码器是否可接收媒体。 */
+        std::shared_ptr<const Transport_Diagnostics> diagnostics{}; /* 连接自身的传输诊断来源。 */
+    };
+
+    struct Pending_Plot_Frame {
+        std::size_t slot{};                                  /* 该完成帧所属的稳定图集槽位。 */
+        std::shared_ptr<const Plot_Pixel_Frame> frame{};      /* 已通过所属 Plot 帧策略的不可变像素帧。 */
+        std::chrono::steady_clock::time_point enqueued_at{};  /* 进入媒体串行队列的单调时刻。 */
+        std::chrono::nanoseconds source_time_unix{};          /* 进入媒体串行队列的 Unix 时间。 */
+    };
+
+    struct Active_Media_Frame {
+        detail::Gallery_Atlas_Composition composition{}; /* 应用一个 Plot 完成帧后的完整图集视图。 */
+        std::uint64_t sequence{};                        /* 媒体流严格递增的 access unit 序号。 */
+        std::chrono::microseconds presentation_time{};   /* 媒体流起点以来的单调显示时间。 */
+        std::chrono::nanoseconds source_time_unix{};      /* 对应 Plot 像素进入媒体流水线的 Unix 时间。 */
+        double queue_delay_ms{};                         /* Plot 发布到开始合成的队列时间。 */
+    };
+
+    static constexpr std::size_t maximum_consumers{32};
+
+    Object* object{}; /* Def 最终对象的可空非拥有借用；bind_private_crtp 后有效。 */
+    Sliding_Statistics compose_ms{600};      /* 单个 Plot 帧应用到持久图集的耗时。 */
+    Sliding_Statistics queue_delay_ms{600};  /* Plot 发布到媒体 Taskflow 开始处理的排队时间。 */
+    Sliding_Statistics encode_ms{600};       /* H.264 编码耗时。 */
+    Sliding_Statistics publish_ms{600};      /* H.264 WebSocket 发布调用耗时。 */
+    std::vector<Source> sources{};            /* 已按业务标识排序的稳定图集来源。 */
+    std::unique_ptr<detail::Gallery_Frame_Atlas> atlas{}; /* 图集像素和各槽位进度的唯一所有者。 */
+    moodycamel::ConcurrentQueue<Pending_Plot_Frame> pending_frames{}; /* 各 Plot 已完成像素帧的无损 MPMC 入口。 */
+    std::shared_ptr<Task_Graph> media_graph{}; /* 每次只消费一个 Plot 完成帧的持久异步 DAG。 */
+    std::atomic_bool media_task_scheduled{};   /* 唯一媒体 DAG 准入；不承担丢帧策略。 */
+    std::atomic_uint64_t media_work_generation{}; /* 新完成帧入队后推进，封闭任务退出竞争窗口。 */
+    detail::FFmpeg_Frame_Transport ffmpeg_transport; /* 仅由媒体 DAG 串行访问的编码上下文。 */
+    std::optional<Active_Media_Frame> active_frame{}; /* 当前媒体 DAG 独占的图集帧。 */
+    std::shared_ptr<const Encoded_Video_Frame> active_video{}; /* 当前编码结果的共享所有权。 */
+    std::atomic_uint64_t next_consumer_id{1}; /* 订阅身份生成器。 */
+    std::array<std::atomic<std::shared_ptr<const Consumer>>,
+               maximum_consumers> consumers{}; /* 订阅关系的权威原子槽位。 */
+    std::atomic_bool stopping{};                /* 关闭开始后拒绝新工作。 */
+    std::atomic_bool failed{};                  /* 首次 Unknown Failure 后停止媒体热路径。 */
+    std::optional<std::string> terminal_failure{}; /* 首次终止失败文本，仅故障路径访问。 */
+    std::atomic_uint64_t received_frame_count{}; /* 活跃消费者期间收到的 Plot 像素帧总数。 */
+    std::atomic_uint64_t processed_frame_count{}; /* 媒体 DAG 已按顺序消费的 Plot 像素帧总数。 */
+    std::atomic_uint64_t encoded_frame_count{};   /* 成功产生 H.264 access unit 的累计帧数。 */
+    std::uint64_t next_media_sequence{1};       /* 仅媒体 DAG 访问的 access unit 序号。 */
+    std::chrono::steady_clock::time_point media_origin{std::chrono::steady_clock::now()}; /* 单调 H.264 PTS 原点。 */
+    std::chrono::microseconds last_presentation_time{}; /* 仅媒体 DAG 访问的最近 H.264 PTS。 */
+    std::chrono::steady_clock::time_point metric_started{}; /* 当前统计窗口起点。 */
+    std::uint64_t metric_received_start{};       /* 窗口起点累计收到帧数。 */
+    std::uint64_t metric_processed_start{};      /* 窗口起点累计处理帧数。 */
+    std::uint64_t metric_encoded_start{};        /* 窗口起点累计编码帧数。 */
+    std::vector<std::uint64_t> metric_completion_starts{}; /* 各 Plot 窗口起点逻辑完成数。 */
+    std::vector<std::uint64_t> metric_rendered_starts{};   /* 各 Plot 窗口起点真实画面数。 */
+    static constexpr std::size_t maximum_taskflow_trace_frames{120};
+    std::atomic_size_t taskflow_trace_remaining{}; /* 尚待捕获的媒体 DAG 次数。 */
+    std::atomic_uint64_t taskflow_trace_control{}; /* 高 32 位 requested，低 32 位 captured。 */
+    std::array<std::atomic<std::shared_ptr<const nlohmann::json>>,
+               maximum_taskflow_trace_frames> taskflow_trace_slots{};
+
+    Private();
+    ~Private();
+
+    template <Attached Attached_Object>
+    void bind_private_crtp(Attached_Object* attached) {
+        Prev_Private::bind_private_crtp(attached);
+        object = not_null{static_cast<Object*>(attached)};
+    }
+
+    void initialize(std::vector<Plot_Entry> plots);
+    [[nodiscard]] bool consumer_accepts() const noexcept;
+    [[nodiscard]] bool remove_consumer(Stream_Id stream) noexcept;
+    void publish(std::shared_ptr<const Encoded_Video_Frame> video,
+                 std::string notification) noexcept;
+    void fail(std::exception_ptr failure) noexcept;
+    void accept_frame(std::size_t slot,
+                      std::shared_ptr<const Plot_Stream_Frame> frame);
+    void arm_media_task(std::weak_ptr<Gallery_Video_Stream> lifetime);
+    void begin_media_frame();
+    void encode_media_frame();
+    void publish_video_frame();
+    void complete_media_frame();
+    void update_metrics(const Active_Media_Frame& frame,
+                        std::size_t encoded_bytes,
+                        std::optional<Video_Encoder_Backend> encoder_backend);
+    [[nodiscard]] bool mark_taskflow_trace();
+    void restore_taskflow_trace() noexcept;
+    void store_taskflow_trace(const Taskflow_Frame_Trace& trace);
+    [[nodiscard]] nlohmann::json taskflow_trace_response() const;
+};
+
+}

web_server/src/media/Gallery_Video_WebSocket.cpp

@@ -1,4 +1,4 @@
-#include "Gallery_WebSocket.hpp"
+#include "Gallery_Video_WebSocket.hpp"
 #include <nlohmann/json.hpp>
 #include <algorithm>
 #include <atomic>
@@ -13,10 +13,9 @@
 #include <utility>
 #include <vector>
 
-namespace aethera::web {
+namespace aethera::web::media {
 namespace {
-constexpr std::size_t h264_header_bytes{40};
-constexpr std::size_t maximum_outstanding_frames{3};
+constexpr std::size_t h264_header_bytes{48};
 
 template <class Integer>
 void write_little_endian(std::span<std::byte> output,
@@ -39,24 +38,25 @@ std::vector<std::byte> websocket_packet(const Encoded_Video_Frame& frame) {
     output[2] = static_cast<std::byte>('H');
     output[3] = static_cast<std::byte>('1');
     const auto bytes = std::span{output};
-    write_little_endian(bytes, 4, std::uint16_t{1});
+    write_little_endian(bytes, 4, std::uint16_t{2});
     write_little_endian(bytes, 6,
         static_cast<std::uint16_t>(frame.key_frame ? 1U : 0U));
     write_little_endian(bytes, 8, frame.sequence);
     write_little_endian(bytes, 16, frame.presentation_time.count());
+    write_little_endian(bytes, 24, frame.source_time_unix.count());
     const auto packed_time = std::chrono::duration_cast<
         std::chrono::nanoseconds>(
             std::chrono::system_clock::now().time_since_epoch()).count();
-    write_little_endian(bytes, 24, packed_time);
-    write_little_endian(bytes, 32,
+    write_little_endian(bytes, 32, packed_time);
+    write_little_endian(bytes, 40,
         static_cast<std::uint32_t>(frame.annex_b.size()));
-    write_little_endian(bytes, 36, std::uint32_t{0});
+    write_little_endian(bytes, 44, std::uint32_t{0});
     std::ranges::copy(frame.annex_b, output.begin() + h264_header_bytes);
     return output;
 }
 }
 
-struct Gallery_WebSocket::Private final {
+struct Gallery_Video_WebSocket::Private final {
     std::weak_ptr<drogon::WebSocketConnection> connection;
     std::shared_ptr<Gallery_Video_Stream> stream;
     Gallery_Video_Stream::Stream_Id subscription{};
@@ -71,7 +71,7 @@ struct Gallery_WebSocket::Private final {
     std::atomic_uint64_t rejected_count{};
 };
 
-Gallery_WebSocket::Gallery_WebSocket(
+Gallery_Video_WebSocket::Gallery_Video_WebSocket(
     drogon::WebSocketConnectionPtr connection,
     std::shared_ptr<Gallery_Video_Stream> stream,
     std::string connection_id)
@@ -84,9 +84,9 @@ Gallery_WebSocket::Gallery_WebSocket(
     d->connection_id = std::move(connection_id);
 }
 
-Gallery_WebSocket::~Gallery_WebSocket() { close(); }
+Gallery_Video_WebSocket::~Gallery_Video_WebSocket() { close(); }
 
-void Gallery_WebSocket::start() {
+void Gallery_Video_WebSocket::start() {
     if (d->attached.exchange(true, std::memory_order_acq_rel)) return;
     const auto weak = weak_from_this();
     d->subscription = d->stream->subscribe(
@@ -102,10 +102,7 @@ void Gallery_WebSocket::start() {
                 return false;
             const auto connection = socket->d->connection.lock();
             return connection && connection->connected() &&
-                socket->d->decoder_ready.load(std::memory_order_acquire) &&
-                socket->d->outstanding_frames.load(
-                    std::memory_order_acquire) <
-                    maximum_outstanding_frames;
+                socket->d->decoder_ready.load(std::memory_order_acquire);
         },
         [weak] {
             const auto socket = weak.lock();
@@ -135,7 +132,7 @@ void Gallery_WebSocket::start() {
                          drogon::WebSocketMessageType::Text);
 }
 
-void Gallery_WebSocket::deliver(Gallery_Stream_Frame frame) {
+void Gallery_Video_WebSocket::deliver(Gallery_Stream_Frame frame) {
     if (!d->attached.load(std::memory_order_acquire)) return;
     if (frame.video && !queue_video_frame(std::move(frame.video))) {
         d->rejected_count.fetch_add(1, std::memory_order_relaxed);
@@ -149,7 +146,7 @@ void Gallery_WebSocket::deliver(Gallery_Stream_Frame frame) {
     }
 }
 
-bool Gallery_WebSocket::queue_video_frame(
+bool Gallery_Video_WebSocket::queue_video_frame(
     std::shared_ptr<const Encoded_Video_Frame> frame) {
     if (!frame || !d->attached.load(std::memory_order_acquire))
         return false;
@@ -173,7 +170,7 @@ bool Gallery_WebSocket::queue_video_frame(
     }
 }
 
-void Gallery_WebSocket::acknowledge_video_frame(std::uint64_t sequence) {
+void Gallery_Video_WebSocket::acknowledge_video_frame(std::uint64_t sequence) {
     auto acknowledged = d->latest_acknowledged_sequence.load(
         std::memory_order_acquire);
     while (sequence > acknowledged) {
@@ -193,7 +190,7 @@ void Gallery_WebSocket::acknowledge_video_frame(std::uint64_t sequence) {
     }
 }
 
-void Gallery_WebSocket::receive(std::string_view message) {
+void Gallery_Video_WebSocket::receive(std::string_view message) {
     const auto json = nlohmann::json::parse(message, nullptr, false);
     if (!json.is_object()) return;
     const auto kind = json.value("kind", std::string{});
@@ -215,7 +212,7 @@ void Gallery_WebSocket::receive(std::string_view message) {
         d->stream->request_video_key_frame();
 }
 
-void Gallery_WebSocket::close() noexcept {
+void Gallery_Video_WebSocket::close() noexcept {
     if (!d->attached.exchange(false, std::memory_order_acq_rel)) return;
     try {
         if (d->subscription != 0) d->stream->unsubscribe(d->subscription);
@@ -226,19 +223,19 @@ void Gallery_WebSocket::close() noexcept {
     d->outstanding_frames.store(0, std::memory_order_release);
 }
 
-Gallery_WebSocket_Controller::Gallery_WebSocket_Controller(
+Gallery_Video_WebSocket_Controller::Gallery_Video_WebSocket_Controller(
     Stream_Resolver resolver) : resolve_stream(std::move(resolver)) {
     if (!resolve_stream)
         throw std::invalid_argument(
             "gallery WebSocket requires a stream resolver");
 }
 
-void Gallery_WebSocket_Controller::initPathRouting() {
+void Gallery_Video_WebSocket_Controller::initPathRouting() {
     drogon::app().registerWebSocketController(
         "/ws/gallery", classTypeName());
 }
 
-void Gallery_WebSocket_Controller::handleNewConnection(
+void Gallery_Video_WebSocket_Controller::handleNewConnection(
     const drogon::HttpRequestPtr& request,
     const drogon::WebSocketConnectionPtr& connection) {
     try {
@@ -249,7 +246,7 @@ void Gallery_WebSocket_Controller::handleNewConnection(
                                  "Invalid gallery H.264 request");
             return;
         }
-        auto socket = std::make_shared<Gallery_WebSocket>(
+        auto socket = std::make_shared<Gallery_Video_WebSocket>(
             connection, std::move(stream), connection_id);
         connection->setContext(socket);
         connection->setPingMessage(
@@ -257,7 +254,7 @@ void Gallery_WebSocket_Controller::handleNewConnection(
         socket->start();
     }
     catch (...) {
-        if (const auto socket = connection->getContext<Gallery_WebSocket>())
+        if (const auto socket = connection->getContext<Gallery_Video_WebSocket>())
             socket->close();
         try {
             connection->shutdown(drogon::CloseCode::kViolation,
@@ -267,19 +264,19 @@ void Gallery_WebSocket_Controller::handleNewConnection(
     }
 }
 
-void Gallery_WebSocket_Controller::handleNewMessage(
+void Gallery_Video_WebSocket_Controller::handleNewMessage(
     const drogon::WebSocketConnectionPtr& connection, std::string&& message,
     const drogon::WebSocketMessageType& type) {
     if (type != drogon::WebSocketMessageType::Text ||
         message.size() > 4096U)
         return;
-    if (const auto socket = connection->getContext<Gallery_WebSocket>())
+    if (const auto socket = connection->getContext<Gallery_Video_WebSocket>())
         socket->receive(message);
 }
 
-void Gallery_WebSocket_Controller::handleConnectionClosed(
+void Gallery_Video_WebSocket_Controller::handleConnectionClosed(
     const drogon::WebSocketConnectionPtr& connection) {
-    if (const auto socket = connection->getContext<Gallery_WebSocket>())
+    if (const auto socket = connection->getContext<Gallery_Video_WebSocket>())
         socket->close();
     connection->clearContext();
 }

web_server/src/media/Gallery_Video_WebSocket.hpp

@@ -5,17 +5,17 @@
 #include <memory>
 #include <string_view>
 
-namespace aethera::web {
+namespace aethera::web::media {
 
 /* 同一个 Drogon WebSocket 承载布局/背压控制文本和 H.264 Annex-B 二进制帧。 */
-struct Gallery_WebSocket final
-    : std::enable_shared_from_this<Gallery_WebSocket> {
-    Gallery_WebSocket(drogon::WebSocketConnectionPtr connection,
+struct Gallery_Video_WebSocket final
+    : std::enable_shared_from_this<Gallery_Video_WebSocket> {
+    Gallery_Video_WebSocket(drogon::WebSocketConnectionPtr connection,
                       std::shared_ptr<Gallery_Video_Stream> stream,
                       std::string connection_id);
-    ~Gallery_WebSocket();
-    Gallery_WebSocket(const Gallery_WebSocket&) = delete;
-    Gallery_WebSocket& operator=(const Gallery_WebSocket&) = delete;
+    ~Gallery_Video_WebSocket();
+    Gallery_Video_WebSocket(const Gallery_Video_WebSocket&) = delete;
+    Gallery_Video_WebSocket& operator=(const Gallery_Video_WebSocket&) = delete;
 
     void start();
     void receive(std::string_view message);
@@ -30,11 +30,11 @@ private:
     std::unique_ptr<Private> d;
 };
 
-struct Gallery_WebSocket_Controller final
-    : drogon::WebSocketController<Gallery_WebSocket_Controller, false> {
+struct Gallery_Video_WebSocket_Controller final
+    : drogon::WebSocketController<Gallery_Video_WebSocket_Controller, false> {
     using Stream_Resolver = std::function<
         std::shared_ptr<Gallery_Video_Stream>(std::string_view)>;
-    explicit Gallery_WebSocket_Controller(Stream_Resolver resolver);
+    explicit Gallery_Video_WebSocket_Controller(Stream_Resolver resolver);
     void handleNewMessage(const drogon::WebSocketConnectionPtr& connection,
                           std::string&& message,
                           const drogon::WebSocketMessageType& type) override;

web_server/src/media/detail/FFmpeg_Frame_Transport.cpp

@@ -1,17 +1,14 @@
 #include "FFmpeg_Frame_Transport.hpp"
 #include <atomic>
-#include <cmath>
 
-namespace aethera::web::frame_sampling::ffmpeg {
+namespace aethera::web::media::detail {
 
 struct FFmpeg_Frame_Transport::Private {
     H264_Encoder encoder;                  /* FFmpeg 编码上下文的唯一所有者。 */
-    double frame_rate_fps{};               /* 媒体序号到单调 PTS 的唯一换算基准。 */
     std::atomic_bool key_frame_requested{}; /* 下一次实际编码样本是否强制关键帧。 */
 
     explicit Private(double value_frame_rate_fps)
-        : encoder(value_frame_rate_fps),
-          frame_rate_fps(value_frame_rate_fps) {}
+        : encoder(value_frame_rate_fps) {}
 };
 
 FFmpeg_Frame_Transport::FFmpeg_Frame_Transport(double frame_rate_fps)
@@ -24,20 +21,21 @@ void FFmpeg_Frame_Transport::request_key_frame() noexcept {
 }
 
 std::optional<Encoded_Video_Frame> FFmpeg_Frame_Transport::encode(
-    const Sampled_Gallery_Frame& frame) {
+    std::span<const std::byte> pixels,
+    std::uint32_t width,
+    std::uint32_t height,
+    Plot_Pixel_Layout layout,
+    std::uint64_t sequence,
+    std::chrono::microseconds presentation_time,
+    std::chrono::nanoseconds source_time_unix) {
     if (d->key_frame_requested.exchange(false, std::memory_order_acq_rel))
         d->encoder.request_key_frame();
     return d->encoder.encode(
-        frame.composition.pixels,
-        frame.composition.width,
-        frame.composition.height,
-        frame.composition.layout == Plot_Pixel_Layout::bgra8
+        pixels, width, height,
+        layout == Plot_Pixel_Layout::bgra8
             ? Video_Pixel_Layout::bgra
             : Video_Pixel_Layout::rgba,
-        frame.sample_sequence,
-        std::chrono::microseconds{static_cast<std::int64_t>(std::llround(
-            static_cast<double>(frame.sample_sequence) *
-            (1'000'000.0 / d->frame_rate_fps)))});
+        sequence, presentation_time, source_time_unix);
 }
 
 }

web_server/src/media/detail/FFmpeg_Frame_Transport.hpp

@@ -0,0 +1,36 @@
+#pragma once
+#include "H264_Encoder.hpp"
+#include <mcp/core/runtime/Plot.hpp>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <span>
+
+namespace aethera::web::media::detail {
+
+/* 已通过 Plot 帧策略的图集像素到 H.264 access unit 的单一编码位置。 */
+struct FFmpeg_Frame_Transport final {
+public:
+    explicit FFmpeg_Frame_Transport(double frame_rate_fps);
+    ~FFmpeg_Frame_Transport();
+    FFmpeg_Frame_Transport(const FFmpeg_Frame_Transport&) = delete;
+    FFmpeg_Frame_Transport& operator=(const FFmpeg_Frame_Transport&) = delete;
+
+    void request_key_frame() noexcept;
+    [[nodiscard]] std::optional<Encoded_Video_Frame> encode(
+        std::span<const std::byte> pixels,
+        std::uint32_t width,
+        std::uint32_t height,
+        Plot_Pixel_Layout layout,
+        std::uint64_t sequence,
+        std::chrono::microseconds presentation_time,
+        std::chrono::nanoseconds source_time_unix);
+
+private:
+    struct Private;
+    std::unique_ptr<Private> d;
+};
+
+}

web_server/src/media/detail/Gallery_Frame_Atlas.cpp

@@ -6,7 +6,7 @@
 #include <stdexcept>
 #include <utility>
 
-namespace aethera::web::detail {
+namespace aethera::web::media::detail {
 struct Gallery_Frame_Atlas::Private {
     struct Source {
         std::atomic<std::shared_ptr<const Plot_Pixel_Frame>>
@@ -96,15 +96,10 @@ Gallery_Frame_Atlas::Accept_Frame_Result Gallery_Frame_Atlas::accept_frame(
     }
     auto current = source.latest_completion.load(
         std::memory_order_acquire);
-    bool rendered_advanced{};
+    const bool rendered_advanced = frame->rendered_sequence != 0 &&
+        (!current ||
+         frame->rendered_sequence != current->rendered_sequence);
     for (;;) {
-        if (current && frame->sequence <= current->sequence) {
-            d->rejected_frame_count.fetch_add(1, std::memory_order_relaxed);
-            return Accept_Frame_Result::stale_frame;
-        }
-        rendered_advanced = frame->rendered_sequence != 0 &&
-            (!current ||
-             frame->rendered_sequence != current->rendered_sequence);
         auto desired = frame;
         if (source.latest_completion.compare_exchange_weak(
                 current, std::move(desired), std::memory_order_release,
@@ -115,18 +110,9 @@ Gallery_Frame_Atlas::Accept_Frame_Result Gallery_Frame_Atlas::accept_frame(
     if (rendered_advanced)
         source.rendered_frame_count.fetch_add(
             1, std::memory_order_release);
-    if (frame->pixels && frame->rendered_sequence != 0) {
-        auto pixels = source.latest_pixels.load(std::memory_order_acquire);
-        while (!pixels ||
-               frame->rendered_sequence > pixels->rendered_sequence) {
-            auto desired = frame;
-            if (source.latest_pixels.compare_exchange_weak(
-                    pixels, std::move(desired),
-                    std::memory_order_release,
-                    std::memory_order_acquire))
-                break;
-        }
-    }
+    if (frame->pixels && frame->rendered_sequence != 0)
+        source.latest_pixels.store(std::move(frame),
+                                   std::memory_order_release);
     return Accept_Frame_Result::accepted;
 }
 

web_server/src/media/detail/Gallery_Frame_Atlas.hpp

@@ -7,7 +7,7 @@
 #include <span>
 #include <string>
 #include <vector>
-namespace aethera::web::detail {
+namespace aethera::web::media::detail {
 struct Gallery_Atlas_Source_Description {
     std::string id; /* 图集槽位对应的 Plot 业务标识。 */
     std::size_t slot{}; /* 从左到右、从上到下的稳定槽位序号。 */
@@ -46,8 +46,7 @@ struct Gallery_Frame_Atlas final {
 public:
     enum struct Accept_Frame_Result : std::uint8_t {
         accepted,
-        invalid_frame,
-        stale_frame
+        invalid_frame
     };
     Gallery_Frame_Atlas(std::uint32_t tile_width,
                         std::uint32_t tile_height,

web_server/src/media/detail/H264_Encoder.cpp

@@ -16,7 +16,7 @@ extern "C" {
 #include <libswscale/swscale.h>
 }
 
-namespace aethera::web {
+namespace aethera::web::media {
 namespace {
 constexpr std::string_view high_profile_level_5_1{"640033"};
 
@@ -49,10 +49,13 @@ std::string_view h264_profile_level_id() noexcept {
     return high_profile_level_5_1;
 }
 
+namespace detail {
+
 struct H264_Encoder::Private {
     struct Submitted_Frame {
         std::uint64_t sequence{};
         std::chrono::microseconds presentation_time{};
+        std::chrono::nanoseconds source_time_unix{};
     };
 
     double frame_rate{};
@@ -199,7 +202,8 @@ void H264_Encoder::request_key_frame() {
 std::optional<Encoded_Video_Frame> H264_Encoder::encode(
     std::span<const std::byte> pixels, std::uint32_t width,
     std::uint32_t height, Video_Pixel_Layout layout,
-    std::uint64_t sequence, std::chrono::microseconds presentation_time) {
+    std::uint64_t sequence, std::chrono::microseconds presentation_time,
+    std::chrono::nanoseconds source_time_unix) {
     if (width == 0 || height == 0 || (width & 1U) != 0 ||
         (height & 1U) != 0 ||
         static_cast<std::size_t>(width) >
@@ -218,7 +222,8 @@ std::optional<Encoded_Video_Frame> H264_Encoder::encode(
         ? AV_PICTURE_TYPE_I : AV_PICTURE_TYPE_NONE;
     d->key_frame_requested = false;
 
-    d->submitted_frames.push_back({sequence, presentation_time});
+    d->submitted_frames.push_back(
+        {sequence, presentation_time, source_time_unix});
     const auto submitted = avcodec_send_frame(
         d->codec_context, d->frame);
     if (submitted < 0) {
@@ -247,9 +252,12 @@ std::optional<Encoded_Video_Frame> H264_Encoder::encode(
         reinterpret_cast<const std::byte*>(
             d->packet->data + d->packet->size));
     output.presentation_time = submitted_frame.presentation_time;
+    output.source_time_unix = submitted_frame.source_time_unix;
     output.sequence = submitted_frame.sequence;
     output.backend = Video_Encoder_Backend::libx264;
     output.key_frame = (d->packet->flags & AV_PKT_FLAG_KEY) != 0;
     return output;
 }
+
+}
 }

web_server/src/media/detail/H264_Encoder.hpp

@@ -0,0 +1,26 @@
+#pragma once
+#include "../Encoded_Video_Frame.hpp"
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <span>
+namespace aethera::web::media::detail {
+struct H264_Encoder final {
+public:
+    explicit H264_Encoder(double frame_rate);
+    ~H264_Encoder();
+    H264_Encoder(const H264_Encoder&) = delete;
+    H264_Encoder& operator=(const H264_Encoder&) = delete;
+    void request_key_frame();
+    [[nodiscard]] std::optional<Encoded_Video_Frame> encode(
+        std::span<const std::byte> pixels, std::uint32_t width,
+        std::uint32_t height, Video_Pixel_Layout layout,
+        std::uint64_t sequence, std::chrono::microseconds presentation_time,
+        std::chrono::nanoseconds source_time_unix);
+private:
+    struct Private;
+    std::unique_ptr<Private> d;
+};
+}

web_server/tests/Gallery_Frame_Atlas_Tests.cpp

@@ -1,11 +1,11 @@
 #include <gtest/gtest.h>
-#include <web_server/src/detail/Gallery_Frame_Atlas.hpp>
+#include <web_server/src/media/detail/Gallery_Frame_Atlas.hpp>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <vector>
 
-namespace aethera::web::detail {
+namespace aethera::web::media::detail {
 namespace {
 std::shared_ptr<const Plot_Pixel_Frame> solid_frame(
     std::uint64_t sequence, std::uint32_t width,
@@ -70,20 +70,21 @@ TEST(Gallery_Frame_Atlas,
     EXPECT_EQ(composition.sources[0].rendered_frame_count, 1U);
 }
 
-TEST(Gallery_Frame_Atlas, Rejects_Stale_Completion_Without_Mutating_Progress) {
+TEST(Gallery_Frame_Atlas, Applies_Out_Of_Order_Completions_In_Arrival_Order) {
     Gallery_Frame_Atlas atlas(2, 2, 1, {"alpha"});
     ASSERT_EQ(atlas.accept_frame(
                   0, solid_frame(2, 2, 2, std::byte{31})),
               Gallery_Frame_Atlas::Accept_Frame_Result::accepted);
+    static_cast<void>(atlas.compose());
     ASSERT_EQ(atlas.accept_frame(
-                  0, completion_frame(1, 1, 2, 2)),
-              Gallery_Frame_Atlas::Accept_Frame_Result::stale_frame);
+                  0, solid_frame(1, 2, 2, std::byte{47})),
+              Gallery_Frame_Atlas::Accept_Frame_Result::accepted);
 
     const auto composition = atlas.compose();
     ASSERT_EQ(composition.sources.size(), 1U);
-    EXPECT_EQ(composition.rejected_frame_count, 1U);
-    EXPECT_EQ(composition.sources[0].completion_sequence, 2U);
-    EXPECT_EQ(composition.sources[0].completion_count, 1U);
-    EXPECT_EQ(composition.pixels[0], std::byte{31});
+    EXPECT_EQ(composition.rejected_frame_count, 0U);
+    EXPECT_EQ(composition.sources[0].completion_sequence, 1U);
+    EXPECT_EQ(composition.sources[0].completion_count, 2U);
+    EXPECT_EQ(composition.pixels[0], std::byte{47});
 }
 }

web_server/tests/Gallery_Video_Stream_Tests.cpp

@@ -0,0 +1,122 @@
+#include <gtest/gtest.h>
+#include <mcp/core/runtime/Gallery_Plots.hpp>
+#include <render_common.hpp>
+#include <web_server/src/media/Gallery_Video_Stream.hpp>
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <ranges>
+#include <stdexcept>
+#include <string>
+
+namespace aethera::web::media {
+namespace {
+
+using Clock = std::chrono::steady_clock;
+
+void corun_until_or_throw(std::string_view operation,
+                          const std::function<bool()>& completed) {
+    std::string failure;
+    Task_Graph coordinator{"gallery.video.test.await"};
+    coordinator.add("await", [&] {
+        const auto deadline = Clock::now() + std::chrono::seconds{10};
+        Task_Graph::corun_until([&] {
+            if (completed()) return true;
+            if (Clock::now() < deadline) return false;
+            failure = std::string{operation} + " timed out";
+            return true;
+        });
+    });
+    aethera::detail::run_taskflow(coordinator);
+    if (!failure.empty()) throw std::runtime_error(failure);
+}
+
+void configure_manual_pixel_delivery(const std::shared_ptr<Plot>& plot) {
+    ASSERT_TRUE(plot);
+    ASSERT_TRUE(plot->write_prop(
+        "frame-analysis", "pacing_mode", "manual").value(
+            "success", false));
+    ASSERT_TRUE(plot->write_prop(
+        "frame-analysis", "pixel_delivery_enabled", true).value(
+            "success", false));
+    ASSERT_NO_THROW(corun_until_or_throw("manual Plot policy", [&] {
+        const auto diagnostics = plot->diagnostics();
+        const auto& configuration = diagnostics.at("frame_policy").at(
+            "configuration");
+        return configuration.at("mode") == "manual" &&
+            configuration.at("pixel_delivery_enabled").get<bool>();
+    }));
+}
+
+void schedule_manual_frame(const std::shared_ptr<Plot>& plot,
+                           std::uint64_t correlation_sequence) {
+    const auto now = Clock::now();
+    plot->schedule_render(Plot_Render_Tick{
+        .issued_at = now,
+        .sequence = correlation_sequence,
+        .time_milliseconds =
+            std::chrono::duration<double, std::milli>(
+                now.time_since_epoch()).count(),
+        .width = 720,
+        .height = 420,
+        .source = Frame_Request_Source::immediate
+    });
+}
+
+TEST(Gallery_Video_Stream,
+     Encodes_Each_Plots_Independent_Policy_Frames_Without_Grouping) {
+    const auto definitions = gallery_plot_definitions();
+    const auto found = std::ranges::find_if(
+        definitions, [](const Plot_Definition& definition) {
+            return definition.dimension == Plot_Dimension::two_d;
+        });
+    ASSERT_NE(found, definitions.end());
+    auto plot_a = found->create();
+    auto plot_b = found->create();
+    configure_manual_pixel_delivery(plot_a);
+    configure_manual_pixel_delivery(plot_b);
+
+    auto stream = Gallery_Video_Stream::create({
+        {"plot-a", plot_a}, {"plot-b", plot_b}});
+    ASSERT_TRUE(stream);
+    constexpr std::size_t frame_count{5};
+    std::array<std::atomic_uint64_t, frame_count> sequences{};
+    std::atomic_size_t received{};
+    const auto subscription = stream->subscribe(
+        "gallery-video-test",
+        [&](Gallery_Stream_Frame frame) {
+            if (!frame.video) return;
+            const auto index = received.fetch_add(
+                1, std::memory_order_acq_rel);
+            if (index < sequences.size())
+                sequences[index].store(
+                    frame.video->sequence, std::memory_order_release);
+        },
+        [] { return true; },
+        [] { return nlohmann::json::object(); });
+
+    const std::array<std::shared_ptr<Plot>, frame_count> schedule{
+        plot_a, plot_a, plot_b, plot_a, plot_b};
+    for (std::size_t index = 0; index < schedule.size(); ++index) {
+        schedule_manual_frame(schedule[index], index + 1U);
+        ASSERT_NO_THROW(corun_until_or_throw("encoded Plot frame", [&] {
+            return received.load(std::memory_order_acquire) > index;
+        }));
+    }
+
+    EXPECT_EQ(received.load(std::memory_order_acquire), frame_count);
+    for (std::size_t index = 0; index < frame_count; ++index)
+        EXPECT_EQ(sequences[index].load(std::memory_order_acquire),
+                  index + 1U);
+    const auto diagnostics = stream->diagnostics();
+    EXPECT_EQ(diagnostics.at("received_frame_count"), frame_count);
+    EXPECT_EQ(diagnostics.at("processed_frame_count"), frame_count);
+    EXPECT_EQ(diagnostics.at("encoded_frame_count"), frame_count);
+    EXPECT_EQ(diagnostics.at("pending_frame_count"), 0U);
+
+    stream->unsubscribe(subscription);
+    stream->shutdown();
+}
+
+}
+}

web_server/tests/H264_Encoder_Tests.cpp

@@ -1,12 +1,12 @@
 #include <gtest/gtest.h>
-#include <web_server/src/H264_Encoder.hpp>
+#include <web_server/src/media/detail/H264_Encoder.hpp>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <optional>
 #include <vector>
 
-namespace aethera::web {
+namespace aethera::web::media::detail {
 namespace {
 std::vector<std::byte> test_pattern(std::uint32_t width,
                                     std::uint32_t height) {
@@ -34,13 +34,15 @@ TEST(H264_Encoder, Rejects_Invalid_Input_Before_FFmpeg) {
     const auto pixels = test_pattern(320, 180);
     EXPECT_THROW(static_cast<void>(encoder.encode(
                      pixels, 319, 180, Video_Pixel_Layout::rgba,
-                     1, std::chrono::microseconds{33'333})),
+                     1, std::chrono::microseconds{33'333},
+                     std::chrono::nanoseconds{1})),
                  std::invalid_argument);
     EXPECT_THROW(static_cast<void>(encoder.encode(
                      std::span<const std::byte>{pixels}.first(
                          pixels.size() - 1U),
                      320, 180, Video_Pixel_Layout::rgba,
-                     1, std::chrono::microseconds{33'333})),
+                     1, std::chrono::microseconds{33'333},
+                     std::chrono::nanoseconds{1})),
                  std::invalid_argument);
 }
 
@@ -57,11 +59,15 @@ TEST(H264_Encoder, Libx264_Produces_Ordered_Annex_B_Access_Units) {
         auto encoded = encoder.encode(
             pixels, width, height, Video_Pixel_Layout::rgba, sequence,
             std::chrono::microseconds{
-                static_cast<std::int64_t>(sequence * 33'333)});
+                static_cast<std::int64_t>(sequence * 33'333)},
+            std::chrono::nanoseconds{
+                static_cast<std::int64_t>(sequence * 1'000)});
         if (!encoded) continue;
         if (!first) first = encoded;
         EXPECT_EQ(encoded->sequence, sequence);
         EXPECT_EQ(encoded->backend, Video_Encoder_Backend::libx264);
+        EXPECT_EQ(encoded->source_time_unix,
+                  std::chrono::nanoseconds{sequence * 1'000});
         EXPECT_FALSE(encoded->annex_b.empty());
         ++produced;
     }

webapp_gallery/src/app.tsx

@@ -48,10 +48,11 @@ type Plot_Diagnostics = {protocol: "aethera.plot.diagnostics"; version: 4; dimen
 type Frame_Sample = {sequence: number; generated_at_ms: number; received_at_ms: number; values: Frame_Stage_Values};
 type Video_Presentation_Metrics = {frame_rate_fps: number; presented_frames: number; dropped_frames: number;
     jitter_buffer_ms: number; decode_processing_ms: number; estimated_playout_delay_ms: number;
-    websocket_arrival_ms: number; browser_queue_ms: number; browser_render_ms: number;
+    server_pipeline_ms: number; websocket_arrival_ms: number; source_arrival_ms: number;
+    browser_queue_ms: number; browser_render_ms: number;
     frame_latency_ms: number; freeze_count: number; current_time_seconds: number; ready_state: number};
 type Gallery_Source = {column: number; row: number};
-type Gallery_Layout = {kind: "gallery_layout"; protocol: "aethera.gallery.video"; version: 4;
+type Gallery_Layout = {kind: "gallery_layout"; protocol: "aethera.gallery.video"; version: 5;
     transport: "drogon_h264"; frame_rate_fps?: number;
     columns: number; rows: number; width: number; height: number;
     tile_width: number; tile_height: number;
@@ -62,13 +63,13 @@ type Gallery_Layout = {kind: "gallery_layout"; protocol: "aethera.gallery.video"
 type Gallery_Source_Metrics = {has_rendered_frame: boolean; logical_completion_rate_fps: number; rendered_frame_rate_fps: number;
     logical_completion_count: number; rendered_frame_count: number; latest_completion_sequence: number;
     latest_rendered_sequence: number; latest_rendered_clock_sequence: number; frame_lag: number};
-type Gallery_Video_Metrics = {kind: "gallery_metrics"; protocol: "aethera.gallery.video"; version: 4; clock_sequence: number;
-    sampled_frame_count: number; encoded_frame_count: number;
-    target_frame_rate_fps: number; clock_delivery_rate_fps: number; sampled_frame_rate_fps: number;
-    encoded_frame_rate_fps: number; compose_average_ms: number; compose_p95_ms: number;
-    encode_average_ms: number; encode_p95_ms: number; sample_delay_average_ms: number; sample_delay_p95_ms: number;
+type Gallery_Video_Metrics = {kind: "gallery_metrics"; protocol: "aethera.gallery.video"; version: 5; media_sequence: number;
+    received_frame_count: number; processed_frame_count: number; encoded_frame_count: number;
+    received_frame_rate_fps: number; processed_frame_rate_fps: number; encoded_frame_rate_fps: number;
+    compose_average_ms: number; compose_p95_ms: number; queue_delay_average_ms: number; queue_delay_p95_ms: number;
+    encode_average_ms: number; encode_p95_ms: number;
     publish_average_ms: number; publish_p95_ms: number; encoded_bytes: number; fresh_tiles: number; missing_tiles: number;
-    rejected_frames: number; skipped_sample_ticks: number; encoder_backend: string;
+    rejected_frames: number; pending_frame_count: number; encoder_backend: string;
     sources: Record<string, Gallery_Source_Metrics>};
 type Gallery_Video_State = {status: Stream_Status; layout: Gallery_Layout | null;
     video_group: Gallery_H264_Group | null; presentation: Video_Presentation_Metrics;
@@ -167,7 +168,7 @@ function valid_gallery_layout(value: unknown): value is Gallery_Layout {
     if (!value || typeof value !== "object") return false;
     const layout = value as Partial<Gallery_Layout>;
     return layout.kind === "gallery_layout" &&
-        layout.protocol === "aethera.gallery.video" && layout.version === 4 &&
+        layout.protocol === "aethera.gallery.video" && layout.version === 5 &&
         layout.transport === "drogon_h264" &&
         typeof layout.width === "number" && typeof layout.height === "number" &&
         Boolean(layout.video) &&
@@ -179,9 +180,10 @@ function valid_gallery_metrics(value: unknown): value is Gallery_Video_Metrics {
     if (!value || typeof value !== "object") return false;
     const metrics = value as Partial<Gallery_Video_Metrics>;
     return metrics.kind === "gallery_metrics" &&
-        metrics.protocol === "aethera.gallery.video" && metrics.version === 4 &&
+        metrics.protocol === "aethera.gallery.video" && metrics.version === 5 &&
         typeof metrics.encoded_frame_rate_fps === "number" &&
-        typeof metrics.clock_delivery_rate_fps === "number" &&
+        typeof metrics.received_frame_rate_fps === "number" &&
+        typeof metrics.processed_frame_rate_fps === "number" &&
         Boolean(metrics.sources);
 }
 
@@ -234,7 +236,8 @@ const empty_video_presentation = (): Video_Presentation_Metrics => ({
     frame_rate_fps: 0, presented_frames: 0, dropped_frames: 0,
     jitter_buffer_ms: 0, decode_processing_ms: 0,
     estimated_playout_delay_ms: 0, freeze_count: 0,
-    websocket_arrival_ms: 0, browser_queue_ms: 0, browser_render_ms: 0,
+    server_pipeline_ms: 0, websocket_arrival_ms: 0, source_arrival_ms: 0,
+    browser_queue_ms: 0, browser_render_ms: 0,
     frame_latency_ms: 0,
     current_time_seconds: 0, ready_state: 0
 });
@@ -242,13 +245,14 @@ const empty_video_presentation = (): Video_Presentation_Metrics => ({
 type Gallery_H264_Packet = {
     sequence: number;
     timestamp_microseconds: number;
+    source_unix_milliseconds: number;
     packed_unix_milliseconds: number;
     key_frame: boolean;
     annex_b: Uint8Array;
 };
 
 function parse_gallery_h264_packet(buffer: ArrayBuffer): Gallery_H264_Packet {
-    if (buffer.byteLength < 40) throw new Error("H.264 帧头不完整");
+    if (buffer.byteLength < 48) throw new Error("H.264 帧头不完整");
     const bytes = new Uint8Array(buffer);
     if (bytes[0] !== 0x41 || bytes[1] !== 0x45 ||
         bytes[2] !== 0x48 || bytes[3] !== 0x31)
@@ -256,19 +260,23 @@ function parse_gallery_h264_packet(buffer: ArrayBuffer): Gallery_H264_Packet {
     const view = new DataView(buffer);
     const version = view.getUint16(4, true);
     const flags = view.getUint16(6, true);
-    const payload_bytes = view.getUint32(32, true);
-    if (version !== 1 || payload_bytes === 0 ||
-        buffer.byteLength !== 40 + payload_bytes)
+    const payload_bytes = view.getUint32(40, true);
+    if (version !== 2 || payload_bytes === 0 ||
+        buffer.byteLength !== 48 + payload_bytes)
         throw new Error("H.264 帧协议字段无效");
-    const packed_nanoseconds = view.getBigInt64(24, true);
+    const source_nanoseconds = view.getBigInt64(24, true);
+    const packed_nanoseconds = view.getBigInt64(32, true);
     return {
         sequence: Number(view.getBigUint64(8, true)),
         timestamp_microseconds: Number(view.getBigInt64(16, true)),
+        source_unix_milliseconds:
+            Number(source_nanoseconds / 1_000_000n) +
+            Number(source_nanoseconds % 1_000_000n) / 1_000_000,
         packed_unix_milliseconds:
             Number(packed_nanoseconds / 1_000_000n) +
             Number(packed_nanoseconds % 1_000_000n) / 1_000_000,
         key_frame: (flags & 1) !== 0,
-        annex_b: new Uint8Array(buffer, 40, payload_bytes)
+        annex_b: new Uint8Array(buffer, 48, payload_bytes)
     };
 }
 
@@ -281,16 +289,25 @@ type Pending_H264_Frame = {
     sequence: number;
     received_performance_ms: number;
     received_unix_ms: number;
+    source_unix_milliseconds: number;
     packed_unix_milliseconds: number;
 };
 
+type Decoded_H264_Frame = Pending_H264_Frame & {
+    decoded_performance_ms: number;
+};
+
 class Gallery_H264_Group {
     private layout: Gallery_Layout | null = null;
     private decoder: VideoDecoder | null = null;
     private disposed = false;
     private readonly surfaces = new Map<string, Gallery_Video_Surface>();
     private readonly pending = new Map<number, Pending_H264_Frame>();
-    private latest_frame: VideoFrame | null = null;
+    private latest_frame: {
+        video: VideoFrame;
+        metadata: Decoded_H264_Frame;
+    } | null = null;
+    private readonly presentation_acknowledgements: number[] = [];
     private animation_frame = 0;
     private ready = false;
     private presented_frames = 0;
@@ -298,7 +315,9 @@ class Gallery_H264_Group {
     private freeze_count = 0;
     private started_performance_ms = 0;
     private last_presented_performance_ms = 0;
+    private server_pipeline_ms = 0;
     private websocket_arrival_ms = 0;
+    private source_arrival_ms = 0;
     private browser_queue_ms = 0;
     private browser_render_ms = 0;
     private frame_latency_ms = 0;
@@ -320,6 +339,12 @@ class Gallery_H264_Group {
         this.readiness(ready);
     }
 
+    private acknowledge_presentable_frames() {
+        for (const sequence of this.presentation_acknowledgements)
+            this.acknowledge(sequence);
+        this.presentation_acknowledgements.length = 0;
+    }
+
     private reset_decoder() {
         if (this.disposed) return;
         if (this.decoder) {
@@ -328,6 +353,9 @@ class Gallery_H264_Group {
         for (const frame of this.pending.values())
             this.acknowledge(frame.sequence);
         this.pending.clear();
+        this.acknowledge_presentable_frames();
+        this.latest_frame?.video.close();
+        this.latest_frame = null;
         this.decoder = null;
         if (!this.layout || typeof VideoDecoder === "undefined") {
             this.update_readiness();
@@ -403,14 +431,13 @@ class Gallery_H264_Group {
             sequence: packet.sequence,
             received_performance_ms,
             received_unix_ms,
+            source_unix_milliseconds: packet.source_unix_milliseconds,
             packed_unix_milliseconds: packet.packed_unix_milliseconds
         });
         try {
             this.decoder.decode(new EncodedVideoChunk({
                 type: packet.key_frame ? "key" : "delta",
                 timestamp: packet.timestamp_microseconds,
-                duration: Math.round(1_000_000 /
-                    (this.layout?.frame_rate_fps ?? 30)),
                 data: packet.annex_b
             }));
         } catch (failure) {
@@ -423,31 +450,44 @@ class Gallery_H264_Group {
 
     private accept_decoded_frame(frame: VideoFrame) {
         const metadata = this.pending.get(frame.timestamp);
-        if (metadata) {
-            this.pending.delete(frame.timestamp);
-            this.acknowledge(metadata.sequence);
-            this.websocket_arrival_ms = Math.max(0,
-                metadata.received_unix_ms -
-                metadata.packed_unix_milliseconds);
-            this.decode_processing_ms = Math.max(0,
-                performance.now() - metadata.received_performance_ms);
+        if (!metadata) {
+            frame.close();
+            return;
         }
+        this.pending.delete(frame.timestamp);
+        const decoded_performance_ms = performance.now();
+        this.presentation_acknowledgements.push(metadata.sequence);
+        this.server_pipeline_ms = Math.max(0,
+            metadata.packed_unix_milliseconds -
+            metadata.source_unix_milliseconds);
+        this.websocket_arrival_ms = Math.max(0,
+            metadata.received_unix_ms -
+            metadata.packed_unix_milliseconds);
+        this.source_arrival_ms = Math.max(0,
+            metadata.received_unix_ms -
+            metadata.source_unix_milliseconds);
+        this.decode_processing_ms = Math.max(0,
+            decoded_performance_ms - metadata.received_performance_ms);
         if (this.latest_frame) {
-            this.latest_frame.close();
+            this.latest_frame.video.close();
             ++this.dropped_frames;
         }
-        this.latest_frame = frame;
+        this.latest_frame = {
+            video: frame,
+            metadata: {...metadata, decoded_performance_ms}
+        };
         if (this.animation_frame === 0)
             this.animation_frame = requestAnimationFrame(() => this.present());
     }
 
     private present() {
         this.animation_frame = 0;
-        const frame = this.latest_frame;
+        const decoded = this.latest_frame;
         this.latest_frame = null;
         const layout = this.layout;
-        if (!frame || !layout) {
-            frame?.close();
+        if (!decoded || !layout) {
+            decoded?.video.close();
+            this.acknowledge_presentable_frames();
             return;
         }
         const started = performance.now();
@@ -455,20 +495,22 @@ class Gallery_H264_Group {
             const source = layout.plots[plot_id];
             if (!source) continue;
             surface.context.drawImage(
-                frame,
+                decoded.video,
                 source.column * layout.tile_width,
                 source.row * layout.tile_height,
                 layout.tile_width,
                 layout.tile_height,
                 0, 0, layout.tile_width, layout.tile_height);
         }
-        frame.close();
+        decoded.video.close();
         const completed = performance.now();
         this.browser_render_ms = completed - started;
         this.browser_queue_ms = Math.max(0,
-            started - (this.last_presented_performance_ms || started));
-        this.frame_latency_ms = this.websocket_arrival_ms +
-            this.decode_processing_ms + this.browser_render_ms;
+            started - decoded.metadata.decoded_performance_ms);
+        this.frame_latency_ms = this.source_arrival_ms +
+            this.decode_processing_ms + this.browser_queue_ms +
+            this.browser_render_ms;
+        this.acknowledge_presentable_frames();
         if (this.started_performance_ms === 0)
             this.started_performance_ms = completed;
         if (this.last_presented_performance_ms !== 0 &&
@@ -491,7 +533,9 @@ class Gallery_H264_Group {
             presented_frames: this.presented_frames,
             dropped_frames: this.dropped_frames,
             decode_processing_ms: this.decode_processing_ms,
+            server_pipeline_ms: this.server_pipeline_ms,
             websocket_arrival_ms: this.websocket_arrival_ms,
+            source_arrival_ms: this.source_arrival_ms,
             browser_queue_ms: this.browser_queue_ms,
             browser_render_ms: this.browser_render_ms,
             frame_latency_ms: this.frame_latency_ms,
@@ -508,11 +552,12 @@ class Gallery_H264_Group {
         if (this.animation_frame !== 0)
             cancelAnimationFrame(this.animation_frame);
         this.animation_frame = 0;
-        this.latest_frame?.close();
+        this.latest_frame?.video.close();
         this.latest_frame = null;
         for (const frame of this.pending.values())
             this.acknowledge(frame.sequence);
         this.pending.clear();
+        this.acknowledge_presentable_frames();
         if (this.decoder) {
             try { this.decoder.close(); } catch { /* already closed */ }
         }
@@ -2728,11 +2773,30 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p
     const metrics = graph.metrics;
     const backend_policy = graph.server_diagnostics?.frame_policy ?? null;
     const source_transport = gallery.transport?.sources[plot.id] ?? null;
+    const event_statistics = Object.values(
+        graph.server_diagnostics?.input_statistics ?? {});
+    const event_statistic_maximum = (
+        key: "queue_wait_ms" | "dispatch_ms" | "total_ms",
+        member: "latest" | "p95") => event_statistics.reduce(
+            (maximum, event) => Math.max(maximum,
+                event[key]?.[member] ?? 0), 0);
+    const browser_inputs = Object.values(graph.browser_input_statistics);
+    const browser_input_latest_ms = browser_inputs.reduce(
+        (maximum, input) => Math.max(maximum, input.latest_ms), 0);
+    const browser_input_buffered_bytes = browser_inputs.reduce(
+        (maximum, input) => Math.max(
+            maximum, input.websocket_buffered_bytes), 0);
+    const event_queue_p95_ms = event_statistic_maximum(
+        "queue_wait_ms", "p95");
+    const event_dispatch_p95_ms = event_statistic_maximum(
+        "dispatch_ms", "p95");
     useEffect(() => {
         const canvas = pixel_canvas_ref.current;
-        if (!gallery.video_group || !source || !canvas) return;
+        if (!policy.visible || !gallery.video_group || !source || !canvas)
+            return;
         return gallery.video_group.register(plot.id, canvas);
-    }, [gallery.video_group, plot.id, source?.column, source?.row]);
+    }, [policy.visible, gallery.video_group, plot.id,
+        source?.column, source?.row]);
     const generated_time = metrics ? new Date(metrics.generated_time_unix_ms).toLocaleTimeString("zh-CN", {hour: "2-digit", minute: "2-digit", second: "2-digit", hour12: false, fractionalSecondDigits: 3}) : "尚未生成帧";
     return <article ref={card_ref} className={`card${selected ? " selected" : ""}`} data-plot-id={plot.id}
         aria-current={selected ? "true" : undefined}
@@ -2746,12 +2810,16 @@ const Plot_Card = memo(function Plot_Card({plot, selected, policy, gallery, on_p
                     <span title="每个 Plot 的逻辑完成吞吐，不是 Gallery 采样率；选中后使用该 Plot 的权威帧策略值。">后台完成 {backend_policy ? backend_policy.throughput.completion_rate_fps.toFixed(1) : source_transport ? source_transport.logical_completion_rate_fps.toFixed(1) : "--.-"} FPS</span>
                     <span title={backend_policy ? `平均 ${backend_policy.latency.average_completion_ms.toFixed(1)} ms · 最大 ${backend_policy.latency.maximum_completion_ms.toFixed(1)} ms` : "选中该图后读取权威后台完成延迟"}>后台延迟 {backend_policy ? backend_policy.latency.latest_completion_ms.toFixed(1) : "--.-"} ms</span>
                     <span>画面产出 {source_transport ? source_transport.rendered_frame_rate_fps.toFixed(1) : metrics ? metrics.frame_rate_fps.toFixed(1) : "--.-"} FPS</span>
-                    <span>采样 {gallery.transport ? gallery.transport.sampled_frame_rate_fps.toFixed(1) : "--.-"} FPS</span>
+                    <span title="各 Plot 自己的帧策略完成后进入媒体流水线的总速率；Gallery 不再二次采样。">策略后像素 {gallery.transport ? gallery.transport.received_frame_rate_fps.toFixed(1) : "--.-"} FPS</span>
+                    <span title={gallery.transport ? `串行媒体处理 ${gallery.transport.processed_frame_rate_fps.toFixed(1)} FPS · 当前排队 ${gallery.transport.pending_frame_count} 帧 · 排队 P95 ${gallery.transport.queue_delay_p95_ms.toFixed(1)} ms` : "等待媒体流水线指标"}>媒体处理 {gallery.transport ? gallery.transport.processed_frame_rate_fps.toFixed(1) : "--.-"} FPS</span>
                     <span>WebCodecs {gallery.presentation.frame_rate_fps.toFixed(1)} FPS</span>
-                    <span title="服务端封包 Unix 时间到浏览器 WebSocket onmessage；包含发送、网络和浏览器消息排队。">到达帧龄 {gallery.presentation.websocket_arrival_ms.toFixed(1)} ms</span>
-                    <span title={`WebCodecs 解码 ${gallery.presentation.decode_processing_ms.toFixed(1)} ms · Canvas 裁剪提交 ${gallery.presentation.browser_render_ms.toFixed(1)} ms`}>浏览器帧龄 {gallery.presentation.frame_latency_ms.toFixed(1)} ms</span>
+                    <span title={`Plot 像素发布到服务端封包 ${gallery.presentation.server_pipeline_ms.toFixed(1)} ms · 封包到浏览器 onmessage ${gallery.presentation.websocket_arrival_ms.toFixed(1)} ms`}>像素到达 {gallery.presentation.source_arrival_ms.toFixed(1)} ms</span>
+                    <span title={`从 Plot 像素发布开始：服务端媒体 ${gallery.presentation.server_pipeline_ms.toFixed(1)} ms · WebSocket 到达 ${gallery.presentation.websocket_arrival_ms.toFixed(1)} ms · WebCodecs 解码 ${gallery.presentation.decode_processing_ms.toFixed(1)} ms · 等待 Canvas ${gallery.presentation.browser_queue_ms.toFixed(1)} ms · Canvas 裁剪提交 ${gallery.presentation.browser_render_ms.toFixed(1)} ms`}>像素到 Canvas {gallery.presentation.frame_latency_ms.toFixed(1)} ms</span>
                     <span title={gallery.transport ? `图集合成 ${gallery.transport.compose_average_ms.toFixed(2)} ms · 发布调用 ${gallery.transport.publish_average_ms.toFixed(2)} ms · 后端 ${gallery.transport.encoder_backend}` : "等待 H.264 流水线指标"}>H.264 编码 {gallery.transport ? gallery.transport.encode_average_ms.toFixed(1) : "--.-"} ms</span>
                     <span title="本 Plot 最新逻辑完成序号减去最新真实画面序号；不再跨 Plot 比较全局时间轮 tick。">画面落后 {source_transport?.has_rendered_frame ? source_transport.frame_lag : "--"} 帧</span>
+                    <span title={`DOM 事件产生到控制 WebSocket send 返回；不包含网络和 Scene。WebSocket 历史发送缓冲峰值 ${frame_bytes(browser_input_buffered_bytes)}。`}>浏览器输入 {browser_inputs.length ? browser_input_latest_ms.toFixed(1) : "--.-"} ms</span>
+                    <span title="事件对象进入 Scene 后，到下一次 Renderable 分发开始的 P95；高值通常表示帧准入或 Worker 排队。">事件排队 P95 {event_statistics.length ? event_queue_p95_ms.toFixed(1) : "--.-"} ms</span>
+                    <span title="Renderable 处理链实际执行事件的 P95；不包含浏览器、网络和 Scene 排队。">事件分发 P95 {event_statistics.length ? event_dispatch_p95_ms.toFixed(1) : "--.-"} ms</span>
                 </div>
             </div>
         </header>