优化了 还是卡

kernel/src/kernel/Task_Graph.cpp

@@ -158,6 +158,8 @@ Task_Node Task_Graph::add(std::string task_name,
         });
     const auto node_id = d->unique_node_id(task_name);
     task.name(node_id);
+    detail::register_taskflow_node(
+        static_cast<std::uint64_t>(task.hash_value()), node_id);
     d->nodes.push_back({task, std::move(task_name), node_id, nullptr});
     return Task_Node{std::make_shared<Task_Node::Private>(
         Task_Node::Private{d, d->nodes.size() - 1, d->generation})};
@@ -182,6 +184,8 @@ Task_Node Task_Graph::add_condition(std::string task_name,
         });
     const auto node_id = d->unique_node_id(task_name);
     task.name(node_id);
+    detail::register_taskflow_node(
+        static_cast<std::uint64_t>(task.hash_value()), node_id);
     d->nodes.push_back({task, std::move(task_name), node_id, nullptr});
     return Task_Node{std::make_shared<Task_Node::Private>(
         Task_Node::Private{d, d->nodes.size() - 1, d->generation})};
@@ -191,6 +195,8 @@ Task_Node Task_Graph::compose(std::string task_name, Task_Graph& child) {
     auto task = d->taskflow.composed_of(child.d->taskflow);
     const auto node_id = d->unique_node_id(task_name);
     task.name(node_id);
+    detail::register_taskflow_node(
+        static_cast<std::uint64_t>(task.hash_value()), node_id);
     d->nodes.push_back({task, std::move(task_name), node_id, child.d});
     return Task_Node{std::make_shared<Task_Node::Private>(
         Task_Node::Private{d, d->nodes.size() - 1, d->generation})};
@@ -209,4 +215,9 @@ std::size_t Task_Graph::size() const noexcept {
 const std::string& Task_Graph::name() const noexcept {
     return d->taskflow.name();
 }
+void Task_Graph::corun_until(std::function<bool()> predicate) {
+    if (!predicate)
+        throw std::invalid_argument("Task graph cooperative predicate is empty");
+    detail::corun_taskflow_until(std::move(predicate));
+}
 }

kernel/src/kernel/Task_Graph.hpp

@@ -48,6 +48,12 @@ public:
     [[nodiscard]] bool empty() const noexcept;
     [[nodiscard]] std::size_t size() const noexcept;
     [[nodiscard]] const std::string& name() const noexcept;
+    /*
+     * 当前 Taskflow Worker 在异步业务完成前继续执行同一 Executor 的其他任务。
+     * 只能从本 Kernel Executor 的任务节点调用；predicate 为 true 后立即返回。
+     * Taskflow 原生 Executor/Worker 类型只存在于实现单元，不进入公共接口。
+     */
+    static void corun_until(std::function<bool()> predicate);
 private:
     friend struct Task_Node::Private;
     friend struct detail::Task_Graph_Access;

kernel/src/kernel/Task_Graph_Internal.hpp

@@ -1,6 +1,10 @@
 #pragma once
 #include "Task_Graph.hpp"
 #include "frame.hpp"
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <string_view>
 
 namespace aethera::detail {
 /* 仅供 Kernel 实现单元把不透明业务图交给原生 Executor 和帧观测器。 */
@@ -9,4 +13,11 @@ struct Task_Graph_Access {
     [[nodiscard]] static std::vector<Taskflow_Graph_Trace::Node> nodes(
         const Task_Graph& graph);
 };
+/* Task_Graph.cpp 到全局 Kernel Executor 的内部桥；不得在模块公共头文件中声明。 */
+void corun_taskflow_until(std::function<bool()> predicate);
+/* 构图期登记稳定业务名；只供 watchdog 故障路径按原生节点 ID 解析。 */
+void register_taskflow_node(std::uint64_t native_id,
+                            std::string_view name);
+[[nodiscard]] std::shared_ptr<const std::string> taskflow_node_name(
+    std::uint64_t native_id) noexcept;
 }

kernel/src/kernel/render_common.cpp

@@ -1,8 +1,12 @@
 #include "render_common.hpp"
 #include "Taskflow_Frame_Access.hpp"
 #include "Task_Graph_Internal.hpp"
+#include <algorithm>
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
+#include <cstdio>
+#include <cstdlib>
 #include <limits>
 #include <mutex>
 #include <optional>
@@ -10,6 +14,7 @@
 #include <stdexcept>
 #include <taskflow/observer/interface.hpp>
 #include <taskflow/taskflow.hpp>
+#include <unordered_map>
 #if defined(_WIN32)
 #define WIN32_LEAN_AND_MEAN
 #define NOMINMAX
@@ -19,15 +24,27 @@
 #endif
 namespace aethera {
 namespace {
+struct Task_Name_Registry {
+    std::mutex mutex;
+    std::unordered_map<std::uint64_t,
+                       std::shared_ptr<const std::string>> names;
+};
+
+Task_Name_Registry& task_name_registry() {
+    static Task_Name_Registry value;
+    return value;
+}
+
 tf::Taskflow& native_taskflow(Task_Graph& graph) noexcept {
     return *static_cast<tf::Taskflow*>(
         detail::Task_Graph_Access::native_storage(graph));
 }
 
-std::uint64_t current_thread_cpu_ns() noexcept {
 #if defined(_WIN32)
+std::uint64_t thread_cpu_ns(HANDLE thread) noexcept {
     FILETIME created{}, exited{}, kernel{}, user{};
-    if (!GetThreadTimes(GetCurrentThread(), &created, &exited, &kernel, &user))
+    if (!thread ||
+        !GetThreadTimes(thread, &created, &exited, &kernel, &user))
         return 0;
     const auto ticks = [](FILETIME value) noexcept {
         ULARGE_INTEGER result{};
@@ -36,6 +53,18 @@ std::uint64_t current_thread_cpu_ns() noexcept {
         return result.QuadPart;
     };
     return (ticks(kernel) + ticks(user)) * 100ULL;
+}
+
+std::uint64_t thread_cpu_cycles(HANDLE thread) noexcept {
+    ULONG64 cycles{};
+    return thread && QueryThreadCycleTime(thread, &cycles)
+        ? static_cast<std::uint64_t>(cycles) : 0;
+}
+#endif
+
+std::uint64_t current_thread_cpu_ns() noexcept {
+#if defined(_WIN32)
+    return thread_cpu_ns(GetCurrentThread());
 #elif defined(CLOCK_THREAD_CPUTIME_ID)
     timespec value{};
     if (clock_gettime(CLOCK_THREAD_CPUTIME_ID, &value) != 0) return 0;
@@ -55,24 +84,48 @@ private:
         std::atomic_uint64_t min_time_ns{std::numeric_limits<std::uint64_t>::max()};
         std::atomic_uint64_t max_time_ns{};
     };
-    struct Worker_Statistics {
+    struct alignas(64) Worker_Statistics {
+        std::array<Task_Statistics, tf::TASK_TYPES.size()> task_types{};
         std::atomic_size_t task_count{};
+        std::atomic_size_t named_task_count{};
+        std::atomic_size_t active_depth{};
+        std::atomic_size_t peak_active_depth{};
         std::atomic_size_t current_queue_size{};
         std::atomic_size_t current_queue_capacity{};
         std::atomic_size_t peak_queue_size{};
         std::atomic_size_t max_queue_capacity{};
         std::atomic_uint64_t active_task_hash{};
         std::atomic_uint64_t active_task_started_ns{};
+        std::atomic_uint64_t active_segment_started_ns{};
+        std::atomic_uint64_t active_segment_cpu_started_ns{};
+#if defined(_WIN32)
+        std::atomic_uintptr_t native_thread_handle{};        /* Watchdog 只读的真实 Worker 线程句柄。 */
+        std::atomic_uint64_t active_cpu_cycles{};            /* 连续片段最近一次采样的 Worker CPU 周期。 */
+        std::atomic_uint64_t active_cpu_progress_ns{};       /* CPU 周期最后前进的墙钟时刻。 */
+#endif
         std::atomic<tf::TaskType> active_task_type{tf::TaskType::UNDEFINED};
+        std::atomic_bool active_task_reported{};
         std::atomic_uint64_t task_time_ns{};
         std::atomic_uint64_t busy_time_ns{};
         std::atomic_uint64_t cpu_time_ns{};
         std::atomic_uint64_t min_task_time_ns{std::numeric_limits<std::uint64_t>::max()};
         std::atomic_uint64_t max_task_time_ns{};
+        std::atomic_size_t max_predecessors{};
+        std::atomic_size_t max_successors{};
+        std::atomic_size_t max_strong_dependencies{};
+        std::atomic_size_t max_weak_dependencies{};
+        std::atomic_uint64_t first_task_time_ns{};
+        std::atomic_uint64_t last_task_time_ns{};
+        std::atomic_uint64_t longest_task_time_ns{};
+        std::atomic_size_t longest_task_hash{};
+        std::atomic<tf::TaskType> longest_task_type{tf::TaskType::UNDEFINED};
+        std::atomic<std::shared_ptr<const std::string>> longest_task_name{};
     };
     struct Start_Record {
         Clock::time_point entered{};                         /* Observer on_entry 进入时间。 */
         Clock::time_point started{};                         /* on_entry 完成、任务体即将执行的时间。 */
+        Clock::time_point segment_started{};                 /* 当前连续独占 Worker 片段的起点。 */
+        std::uint64_t maximum_segment_ns{};                  /* 已结束连续独占片段的最大墙钟。 */
         std::uint64_t cpu_entered_ns{};                      /* on_entry 进入时的 worker CPU 时间。 */
         std::uint64_t cpu_started_ns{};                      /* 任务体开始前的 worker CPU 时间。 */
         Render_Frame* frame{};                              /* 进入任务时唯一活动的按帧捕获。 */
@@ -80,36 +133,21 @@ private:
         std::size_t queue_capacity{};                       /* 进入任务时 worker 队列容量。 */
         std::uint64_t native_id{};                          /* 嵌套 corun 返回外层任务时恢复其原生身份。 */
         tf::TaskType type{tf::TaskType::UNDEFINED};         /* 嵌套 corun 返回外层任务时恢复其原生类型。 */
+        bool cooperatively_suspended{};                     /* 外层任务是否主动让出 Worker 执行子图。 */
     };
     std::vector<std::vector<Start_Record>> starts;
     std::vector<Clock::time_point> worker_busy_starts;
     std::vector<std::uint64_t> worker_cpu_starts;
     std::unique_ptr<Worker_Statistics[]> worker_statistics;
     std::size_t worker_statistics_count{};
-    std::array<Task_Statistics, tf::TASK_TYPES.size()> task_types;
-    std::atomic_size_t active_tasks{};
-    std::atomic_size_t peak_active_tasks{};
-    std::atomic_size_t active_workers{};
-    std::atomic_size_t peak_active_workers{};
-    std::atomic_size_t completed_tasks{};
-    std::atomic_size_t named_tasks{};
-    std::atomic_size_t peak_queue_size{};
-    std::atomic_size_t max_queue_capacity{};
-    std::atomic_size_t max_predecessors{};
-    std::atomic_size_t max_successors{};
-    std::atomic_size_t max_strong_dependencies{};
-    std::atomic_size_t max_weak_dependencies{};
-    std::atomic_uint64_t total_execution_time_ns{};
-    std::atomic_uint64_t worker_busy_time_ns{};
-    std::atomic_uint64_t worker_cpu_time_ns{};
-    std::atomic_uint64_t first_task_time_ns{};
-    std::atomic_uint64_t last_task_time_ns{};
-    std::atomic_uint64_t longest_task_time_ns{};
-    std::atomic_size_t longest_task_hash{};
-    std::atomic<tf::TaskType> longest_task_type{tf::TaskType::UNDEFINED};
-    std::atomic<std::shared_ptr<const std::string>> longest_task_name{};
     std::atomic<Render_Frame*> trace_frame{};
     std::shared_mutex trace_mutex{};                        /* 仅按需捕获时保护 Frame* 获取与关闭。 */
+    std::uint64_t worker_occupation_limit_ns{};              /* 单节点连续非 CPU 等待 Worker 的上限。 */
+    Task_Overrun_Action worker_overrun_action{};             /* 节点超过占用上限后的处置策略。 */
+    std::atomic_bool watchdog_stopping{};                    /* Watchdog 生命周期停止标志。 */
+    std::mutex watchdog_mutex{};                             /* 仅用于 Watchdog 条件休眠。 */
+    std::condition_variable watchdog_wake{};                 /* Observer 销毁时唤醒 Watchdog。 */
+    std::thread watchdog_thread{};                           /* 不占用 Executor Worker 的超时检查线程。 */
     static std::uint64_t clock_ns(Clock::time_point value) noexcept {
         return static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::nanoseconds>(value.time_since_epoch()).count());
     }
@@ -129,7 +167,157 @@ private:
     static std::size_t task_type_index(tf::TaskType type) noexcept {
         return static_cast<std::size_t>(type);
     }
+    void report_active_overrun(std::size_t worker,
+                               const Worker_Statistics& state,
+                               std::uint64_t elapsed_ns,
+                               std::uint64_t stalled_ns) const noexcept {
+        const auto type = state.active_task_type.load(
+            std::memory_order_relaxed);
+        const auto native_id = state.active_task_hash.load(
+            std::memory_order_relaxed);
+        const auto name = detail::taskflow_node_name(native_id);
+        std::fprintf(
+            stderr,
+            "Taskflow worker occupation limit exceeded while node is active\n"
+            "  node: %s\n"
+            "  hash: %llu\n"
+            "  type: %s\n"
+            "  worker: %zu\n"
+            "  segment-wall: %.3f ms\n"
+            "  no-cpu-progress: %.3f ms\n"
+            "  limit: %.3f ms\n"
+            "  queue: %zu/%zu\n",
+            name ? name->c_str() : "<unregistered task>",
+            static_cast<unsigned long long>(native_id),
+            tf::to_string(type), worker,
+            static_cast<double>(elapsed_ns) / 1'000'000.0,
+            static_cast<double>(stalled_ns) / 1'000'000.0,
+            static_cast<double>(worker_occupation_limit_ns) / 1'000'000.0,
+            state.current_queue_size.load(std::memory_order_relaxed),
+            state.current_queue_capacity.load(std::memory_order_relaxed));
+        std::fflush(stderr);
+        if (worker_overrun_action == Task_Overrun_Action::fast_fail)
+            std::abort();
+    }
+    void run_watchdog() noexcept {
+        const auto interval = std::chrono::nanoseconds(
+            std::min<std::uint64_t>(worker_occupation_limit_ns / 4,
+                                    10'000'000ULL));
+        for (;;) {
+            {
+                std::unique_lock lock(watchdog_mutex);
+                watchdog_wake.wait_for(lock, interval, [this] {
+                    return watchdog_stopping.load(std::memory_order_acquire);
+                });
+            }
+            if (watchdog_stopping.load(std::memory_order_acquire)) return;
+            const auto now = clock_ns(Clock::now());
+            for (std::size_t worker = 0;
+                 worker < worker_statistics_count; ++worker) {
+                auto& state = worker_statistics[worker];
+                const auto started = state.active_segment_started_ns.load(
+                    std::memory_order_acquire);
+                if (!started || now <= started ||
+                    now - started <= worker_occupation_limit_ns)
+                    continue;
+#if defined(_WIN32)
+                const auto handle = state.native_thread_handle.load(
+                    std::memory_order_acquire);
+                const auto wall = now - started;
+                const auto cycles = thread_cpu_cycles(
+                    reinterpret_cast<HANDLE>(handle));
+                if (!cycles) continue;
+                const auto previous = state.active_cpu_cycles.exchange(
+                    cycles, std::memory_order_acq_rel);
+                if (cycles != previous) {
+                    state.active_cpu_progress_ns.store(
+                        now, std::memory_order_release);
+                    continue;
+                }
+                const auto progress = state.active_cpu_progress_ns.load(
+                    std::memory_order_acquire);
+                if (!progress || now <= progress ||
+                    now - progress <= worker_occupation_limit_ns)
+                    continue;
+                const auto stalled = now - progress;
+#else
+                const auto wall = now - started;
+                const auto stalled = wall;
+#endif
+                if (state.active_task_reported.exchange(
+                        true, std::memory_order_acq_rel))
+                    continue;
+                report_active_overrun(worker, state, wall, stalled);
+            }
+        }
+    }
+    void pause_worker(std::size_t worker) noexcept {
+        if (worker >= starts.size() || starts[worker].empty()) return;
+        const auto now = Clock::now();
+        auto& active = starts[worker].back();
+        if (active.segment_started != Clock::time_point{}) {
+            const auto elapsed = static_cast<std::uint64_t>(
+                std::chrono::duration_cast<std::chrono::nanoseconds>(
+                    now - active.segment_started).count());
+            active.maximum_segment_ns = std::max(
+                active.maximum_segment_ns, elapsed);
+        }
+        active.segment_started = {};
+        active.cooperatively_suspended = true;
+        worker_statistics[worker].active_segment_started_ns.store(
+            0, std::memory_order_release);
+        worker_statistics[worker].active_segment_cpu_started_ns.store(
+            0, std::memory_order_release);
+#if defined(_WIN32)
+        worker_statistics[worker].active_cpu_cycles.store(
+            0, std::memory_order_release);
+        worker_statistics[worker].active_cpu_progress_ns.store(
+            0, std::memory_order_release);
+#endif
+    }
+    void resume_worker(std::size_t worker) noexcept {
+        if (worker >= starts.size() || starts[worker].empty()) return;
+        const auto now = Clock::now();
+        auto& active = starts[worker].back();
+        active.cooperatively_suspended = false;
+        active.segment_started = now;
+        worker_statistics[worker].active_segment_started_ns.store(
+            clock_ns(now), std::memory_order_release);
+        worker_statistics[worker].active_segment_cpu_started_ns.store(
+            current_thread_cpu_ns(), std::memory_order_release);
+#if defined(_WIN32)
+        const auto handle = worker_statistics[worker].native_thread_handle.load(
+            std::memory_order_acquire);
+        worker_statistics[worker].active_cpu_cycles.store(
+            thread_cpu_cycles(reinterpret_cast<HANDLE>(handle)),
+            std::memory_order_release);
+        worker_statistics[worker].active_cpu_progress_ns.store(
+            clock_ns(now), std::memory_order_release);
+#endif
+        worker_statistics[worker].active_task_reported.store(
+            false, std::memory_order_relaxed);
+    }
+    friend class Task_Resource;
 public:
+    Task_Observer(std::chrono::milliseconds worker_occupation_limit,
+                  Task_Overrun_Action worker_overrun_action_value)
+        : worker_occupation_limit_ns(static_cast<std::uint64_t>(
+              std::chrono::duration_cast<std::chrono::nanoseconds>(
+                  worker_occupation_limit).count())),
+          worker_overrun_action(worker_overrun_action_value) {}
+    ~Task_Observer() override {
+        watchdog_stopping.store(true, std::memory_order_release);
+        watchdog_wake.notify_one();
+        if (watchdog_thread.joinable()) watchdog_thread.join();
+#if defined(_WIN32)
+        for (std::size_t worker = 0;
+             worker < worker_statistics_count; ++worker) {
+            const auto handle = worker_statistics[worker]
+                .native_thread_handle.exchange(0, std::memory_order_acq_rel);
+            if (handle) CloseHandle(reinterpret_cast<HANDLE>(handle));
+        }
+#endif
+    }
     void set_up(std::size_t workers) override {
         starts.resize(workers);
         worker_busy_starts.resize(workers);
@@ -137,20 +325,48 @@ public:
         worker_statistics = std::make_unique<Worker_Statistics[]>(workers);
         worker_statistics_count = workers;
         for (auto& worker : starts) worker.reserve(8);
+        watchdog_thread = std::thread([this] { run_watchdog(); });
     }
     void on_entry(tf::WorkerView worker, tf::TaskView task) override {
         auto now = Clock::now();
         auto& worker_starts = starts[worker.id()];
+        auto& worker_state = worker_statistics[worker.id()];
+#if defined(_WIN32)
+        if (worker_state.native_thread_handle.load(
+                std::memory_order_acquire) == 0) {
+            HANDLE duplicated{};
+            if (DuplicateHandle(
+                    GetCurrentProcess(), GetCurrentThread(),
+                    GetCurrentProcess(), &duplicated, 0, FALSE,
+                    DUPLICATE_SAME_ACCESS)) {
+                std::uintptr_t expected{};
+                if (!worker_state.native_thread_handle.compare_exchange_strong(
+                        expected, reinterpret_cast<std::uintptr_t>(duplicated),
+                        std::memory_order_release,
+                        std::memory_order_relaxed))
+                    CloseHandle(duplicated);
+            }
+        }
+#endif
+        if (!worker_starts.empty()) {
+            auto& parent = worker_starts.back();
+            if (parent.segment_started != Clock::time_point{}) {
+                const auto elapsed = static_cast<std::uint64_t>(
+                    std::chrono::duration_cast<std::chrono::nanoseconds>(
+                        now - parent.segment_started).count());
+                parent.maximum_segment_ns = std::max(
+                    parent.maximum_segment_ns, elapsed);
+                parent.segment_started = {};
+            }
+        }
         if (worker_starts.empty()) {
             worker_busy_starts[worker.id()] = now;
             worker_cpu_starts[worker.id()] = current_thread_cpu_ns();
-            auto active = active_workers.fetch_add(1, std::memory_order_relaxed) + 1;
-            update_max(peak_active_workers, active);
         }
         worker_starts.push_back(Start_Record{
-            now, {}, current_thread_cpu_ns(), 0, nullptr,
+            now, {}, {}, 0, current_thread_cpu_ns(), 0, nullptr,
             worker.queue_size(), worker.queue_capacity(),
-            static_cast<std::uint64_t>(task.hash_value()), task.type()});
+            static_cast<std::uint64_t>(task.hash_value()), task.type(), false});
         auto* frame = trace_frame.load(std::memory_order_acquire);
         /*
          * 帧租约从 on_entry 持续到对应 on_exit。只在退出时登记写入者会留下
@@ -169,26 +385,42 @@ public:
             else frame = nullptr;
         }
         worker_starts.back().frame = frame;
-        auto active = active_tasks.fetch_add(1, std::memory_order_relaxed) + 1;
-        update_max(peak_active_tasks, active);
-        update_max(peak_queue_size, worker.queue_size());
-        update_max(max_queue_capacity, worker.queue_capacity());
-        auto& worker_state = worker_statistics[worker.id()];
+        const auto active_depth = worker_state.active_depth.fetch_add(
+            1, std::memory_order_relaxed) + 1;
+        update_max(worker_state.peak_active_depth, active_depth);
         worker_state.current_queue_size.store(worker.queue_size(), std::memory_order_relaxed);
         worker_state.current_queue_capacity.store(worker.queue_capacity(), std::memory_order_relaxed);
         worker_state.active_task_hash.store(task.hash_value(), std::memory_order_relaxed);
         worker_state.active_task_started_ns.store(clock_ns(now), std::memory_order_relaxed);
+        worker_state.active_segment_started_ns.store(clock_ns(now),
+                                                     std::memory_order_release);
+        worker_state.active_segment_cpu_started_ns.store(
+            current_thread_cpu_ns(), std::memory_order_release);
+#if defined(_WIN32)
+        const auto native_handle = worker_state.native_thread_handle.load(
+            std::memory_order_acquire);
+        worker_state.active_cpu_cycles.store(
+            thread_cpu_cycles(reinterpret_cast<HANDLE>(native_handle)),
+            std::memory_order_release);
+        worker_state.active_cpu_progress_ns.store(
+            clock_ns(now), std::memory_order_release);
+#endif
         worker_state.active_task_type.store(task.type(), std::memory_order_relaxed);
+        worker_state.active_task_reported.store(false, std::memory_order_relaxed);
         update_max(worker_state.peak_queue_size, worker.queue_size());
         update_max(worker_state.max_queue_capacity, worker.queue_capacity());
-        update_max(max_predecessors, task.num_predecessors());
-        update_max(max_successors, task.num_successors());
-        update_max(max_strong_dependencies, task.num_strong_dependencies());
-        update_max(max_weak_dependencies, task.num_weak_dependencies());
-        if (!task.name().empty()) named_tasks.fetch_add(1, std::memory_order_relaxed);
-        update_first(first_task_time_ns, clock_ns(now));
+        update_max(worker_state.max_predecessors, task.num_predecessors());
+        update_max(worker_state.max_successors, task.num_successors());
+        update_max(worker_state.max_strong_dependencies,
+                   task.num_strong_dependencies());
+        update_max(worker_state.max_weak_dependencies,
+                   task.num_weak_dependencies());
+        if (!task.name().empty())
+            worker_state.named_task_count.fetch_add(1, std::memory_order_relaxed);
+        update_first(worker_state.first_task_time_ns, clock_ns(now));
         worker_starts.back().cpu_started_ns = current_thread_cpu_ns();
         worker_starts.back().started = Clock::now();
+        worker_starts.back().segment_started = worker_starts.back().started;
     }
     void on_exit(tf::WorkerView worker, tf::TaskView task) override {
         const auto finished = Clock::now();
@@ -199,31 +431,39 @@ public:
         const auto elapsed = static_cast<std::uint64_t>(
             std::chrono::duration_cast<std::chrono::nanoseconds>(
                 finished - start.started).count());
-        total_execution_time_ns.fetch_add(elapsed, std::memory_order_relaxed);
-        completed_tasks.fetch_add(1, std::memory_order_relaxed);
+        if (start.segment_started != Clock::time_point{}) {
+            const auto segment = static_cast<std::uint64_t>(
+                std::chrono::duration_cast<std::chrono::nanoseconds>(
+                    finished - start.segment_started).count());
+            start.maximum_segment_ns = std::max(
+                start.maximum_segment_ns, segment);
+        }
         auto& worker_state = worker_statistics[worker.id()];
+        worker_state.active_task_reported.store(true,
+                                                std::memory_order_release);
         worker_state.task_count.fetch_add(1, std::memory_order_relaxed);
         worker_state.task_time_ns.fetch_add(elapsed, std::memory_order_relaxed);
         update_min(worker_state.min_task_time_ns, elapsed);
         update_max(worker_state.max_task_time_ns, elapsed);
         auto type_index = task_type_index(task.type());
-        if (type_index < task_types.size()) {
-            auto& type = task_types[type_index];
+        if (type_index < worker_state.task_types.size()) {
+            auto& type = worker_state.task_types[type_index];
             type.count.fetch_add(1, std::memory_order_relaxed);
             type.total_time_ns.fetch_add(elapsed, std::memory_order_relaxed);
             update_min(type.min_time_ns, elapsed);
             update_max(type.max_time_ns, elapsed);
         }
-        auto longest = longest_task_time_ns.load(std::memory_order_relaxed);
-        if (longest < elapsed && longest_task_time_ns.compare_exchange_strong(
+        auto longest = worker_state.longest_task_time_ns.load(
+            std::memory_order_relaxed);
+        if (longest < elapsed && worker_state.longest_task_time_ns.compare_exchange_strong(
                 longest, elapsed, std::memory_order_relaxed)) {
-            longest_task_hash.store(task.hash_value(), std::memory_order_relaxed);
-            longest_task_type.store(task.type(), std::memory_order_relaxed);
-            longest_task_name.store(
+            worker_state.longest_task_hash.store(task.hash_value(), std::memory_order_relaxed);
+            worker_state.longest_task_type.store(task.type(), std::memory_order_relaxed);
+            worker_state.longest_task_name.store(
                 std::make_shared<const std::string>(task.name()),
                 std::memory_order_release);
         }
-        active_tasks.fetch_sub(1, std::memory_order_relaxed);
+        worker_state.active_depth.fetch_sub(1, std::memory_order_relaxed);
         std::optional<std::size_t> trace_task;
         if (start.frame) {
             try {
@@ -251,33 +491,70 @@ public:
             auto busy = static_cast<std::uint64_t>(
                 std::chrono::duration_cast<std::chrono::nanoseconds>(
                     completed - worker_busy_starts[worker.id()]).count());
-            worker_busy_time_ns.fetch_add(busy, std::memory_order_relaxed);
             worker_state.busy_time_ns.fetch_add(busy, std::memory_order_relaxed);
             const auto cpu_started = worker_cpu_starts[worker.id()];
             const auto cpu = cpu_completed_ns >= cpu_started
                 ? cpu_completed_ns - cpu_started : 0;
-            worker_cpu_time_ns.fetch_add(cpu, std::memory_order_relaxed);
             worker_state.cpu_time_ns.fetch_add(cpu, std::memory_order_relaxed);
-            active_workers.fetch_sub(1, std::memory_order_relaxed);
             worker_state.active_task_hash.store(0, std::memory_order_relaxed);
             worker_state.active_task_started_ns.store(0, std::memory_order_relaxed);
+            worker_state.active_segment_started_ns.store(0,
+                                                         std::memory_order_release);
+            worker_state.active_segment_cpu_started_ns.store(
+                0, std::memory_order_release);
+#if defined(_WIN32)
+            worker_state.active_cpu_cycles.store(0, std::memory_order_release);
+            worker_state.active_cpu_progress_ns.store(0,
+                                                      std::memory_order_release);
+#endif
             worker_state.active_task_type.store(tf::TaskType::UNDEFINED,
                                                 std::memory_order_relaxed);
             worker_state.current_queue_size.store(0, std::memory_order_relaxed);
             worker_state.current_queue_capacity.store(0, std::memory_order_relaxed);
         }
         else {
-            const auto& parent = worker_starts.back();
+            auto& parent = worker_starts.back();
             worker_state.active_task_hash.store(parent.native_id, std::memory_order_relaxed);
             worker_state.active_task_started_ns.store(clock_ns(parent.entered),
                                                       std::memory_order_relaxed);
+            if (parent.cooperatively_suspended) {
+                worker_state.active_segment_started_ns.store(
+                    0, std::memory_order_release);
+                worker_state.active_segment_cpu_started_ns.store(
+                    0, std::memory_order_release);
+#if defined(_WIN32)
+                worker_state.active_cpu_cycles.store(0,
+                                                      std::memory_order_release);
+                worker_state.active_cpu_progress_ns.store(0,
+                                                           std::memory_order_release);
+#endif
+            }
+            else {
+                parent.segment_started = completed;
+                worker_state.active_segment_started_ns.store(
+                    clock_ns(completed), std::memory_order_release);
+                worker_state.active_segment_cpu_started_ns.store(
+                    cpu_completed_ns, std::memory_order_release);
+#if defined(_WIN32)
+                const auto native_handle =
+                    worker_state.native_thread_handle.load(
+                        std::memory_order_acquire);
+                worker_state.active_cpu_cycles.store(
+                    thread_cpu_cycles(reinterpret_cast<HANDLE>(native_handle)),
+                    std::memory_order_release);
+                worker_state.active_cpu_progress_ns.store(
+                    clock_ns(completed), std::memory_order_release);
+#endif
+            }
             worker_state.active_task_type.store(parent.type, std::memory_order_relaxed);
+            worker_state.active_task_reported.store(false,
+                                                    std::memory_order_relaxed);
             worker_state.current_queue_size.store(parent.queue_size,
                                                   std::memory_order_relaxed);
             worker_state.current_queue_capacity.store(parent.queue_capacity,
                                                   std::memory_order_relaxed);
         }
-        update_max(last_task_time_ns, clock_ns(completed));
+        update_max(worker_state.last_task_time_ns, clock_ns(completed));
     }
     bool begin_trace(Render_Frame& frame, std::size_t workers) {
         std::unique_lock guard(trace_mutex);
@@ -296,41 +573,30 @@ public:
     void write_state(Task_Runtime_State& state, std::size_t workers, std::size_t active_topologies) const {
         state.worker_count = workers;
         state.active_topology_count = active_topologies;
-        state.active_task_count = active_tasks.load(std::memory_order_relaxed);
-        state.peak_active_task_count = peak_active_tasks.load(std::memory_order_relaxed);
-        state.active_worker_count = active_workers.load(std::memory_order_relaxed);
-        state.peak_active_worker_count = peak_active_workers.load(std::memory_order_relaxed);
-        state.observed_task_count = completed_tasks.load(std::memory_order_relaxed);
-        state.named_task_count = named_tasks.load(std::memory_order_relaxed);
-        state.peak_observed_worker_queue_size = peak_queue_size.load(std::memory_order_relaxed);
-        state.max_observed_worker_queue_capacity = max_queue_capacity.load(std::memory_order_relaxed);
-        state.max_predecessors = max_predecessors.load(std::memory_order_relaxed);
-        state.max_successors = max_successors.load(std::memory_order_relaxed);
-        state.max_strong_dependencies = max_strong_dependencies.load(std::memory_order_relaxed);
-        state.max_weak_dependencies = max_weak_dependencies.load(std::memory_order_relaxed);
-        state.total_task_time_ns = total_execution_time_ns.load(std::memory_order_relaxed);
-        state.worker_busy_time_ns = worker_busy_time_ns.load(std::memory_order_relaxed);
-        state.worker_cpu_time_ns = worker_cpu_time_ns.load(std::memory_order_relaxed);
-        auto first = first_task_time_ns.load(std::memory_order_relaxed);
-        auto last = last_task_time_ns.load(std::memory_order_relaxed);
-        state.observed_wall_time_ns = first && last >= first ? last - first : 0;
-        state.worker_utilization = workers && state.observed_wall_time_ns ? static_cast<double>(state.worker_busy_time_ns) * 100.0 / static_cast<double>(state.observed_wall_time_ns) / static_cast<double>(workers) : 0.0;
-        state.worker_cpu_utilization = workers && state.observed_wall_time_ns
-            ? static_cast<double>(state.worker_cpu_time_ns) * 100.0 /
-              static_cast<double>(state.observed_wall_time_ns) /
-              static_cast<double>(workers)
-            : 0.0;
-        state.task_types.resize(task_types.size());
-        for (std::size_t i = 0; i < task_types.size(); ++i) {
-            const auto& source = task_types[i];
-            auto& target = state.task_types[i];
-            target.name = std::string(tf::to_string(tf::TASK_TYPES[i]));
-            target.count = source.count.load(std::memory_order_relaxed);
-            target.total_time_ns = source.total_time_ns.load(std::memory_order_relaxed);
-            auto min = source.min_time_ns.load(std::memory_order_relaxed);
-            target.min_time_ns = target.count ? min : 0;
-            target.max_time_ns = source.max_time_ns.load(std::memory_order_relaxed);
-        }
+        state.active_task_count = 0;
+        state.peak_active_task_count = 0;
+        state.active_worker_count = 0;
+        state.observed_task_count = 0;
+        state.named_task_count = 0;
+        state.peak_observed_worker_queue_size = 0;
+        state.max_observed_worker_queue_capacity = 0;
+        state.max_predecessors = 0;
+        state.max_successors = 0;
+        state.max_strong_dependencies = 0;
+        state.max_weak_dependencies = 0;
+        state.total_task_time_ns = 0;
+        state.worker_busy_time_ns = 0;
+        state.worker_cpu_time_ns = 0;
+        state.longest_task_time_ns = 0;
+        state.longest_task_hash = 0;
+        state.longest_task_name.clear();
+        state.longest_task_type.clear();
+        std::uint64_t first{};
+        std::uint64_t last{};
+        state.task_types.assign(tf::TASK_TYPES.size(), {});
+        for (std::size_t type = 0; type < tf::TASK_TYPES.size(); ++type)
+            state.task_types[type].name =
+                std::string(tf::to_string(tf::TASK_TYPES[type]));
         state.workers.resize(worker_statistics_count);
         const auto read_time_ns = clock_ns(Clock::now());
         for (std::size_t i = 0; i < worker_statistics_count; ++i) {
@@ -363,13 +629,91 @@ public:
                 ? static_cast<double>(target.cpu_time_ns) * 100.0 /
                   static_cast<double>(state.observed_wall_time_ns)
                 : 0.0;
+            const auto active_depth = source.active_depth.load(
+                std::memory_order_relaxed);
+            state.active_task_count += active_depth;
+            state.peak_active_task_count += source.peak_active_depth.load(
+                std::memory_order_relaxed);
+            state.active_worker_count += active_depth != 0;
+            state.observed_task_count += target.task_count;
+            state.named_task_count += source.named_task_count.load(
+                std::memory_order_relaxed);
+            state.peak_observed_worker_queue_size = std::max(
+                state.peak_observed_worker_queue_size,
+                target.peak_observed_queue_size);
+            state.max_observed_worker_queue_capacity = std::max(
+                state.max_observed_worker_queue_capacity,
+                target.max_observed_queue_capacity);
+            state.max_predecessors = std::max(state.max_predecessors,
+                source.max_predecessors.load(std::memory_order_relaxed));
+            state.max_successors = std::max(state.max_successors,
+                source.max_successors.load(std::memory_order_relaxed));
+            state.max_strong_dependencies = std::max(
+                state.max_strong_dependencies,
+                source.max_strong_dependencies.load(std::memory_order_relaxed));
+            state.max_weak_dependencies = std::max(
+                state.max_weak_dependencies,
+                source.max_weak_dependencies.load(std::memory_order_relaxed));
+            state.total_task_time_ns += target.task_time_ns;
+            state.worker_busy_time_ns += target.busy_time_ns;
+            state.worker_cpu_time_ns += target.cpu_time_ns;
+            const auto worker_first = source.first_task_time_ns.load(
+                std::memory_order_relaxed);
+            if (worker_first && (!first || worker_first < first))
+                first = worker_first;
+            last = std::max(last, source.last_task_time_ns.load(
+                std::memory_order_relaxed));
+            for (std::size_t type = 0; type < tf::TASK_TYPES.size(); ++type) {
+                const auto& source_type = source.task_types[type];
+                auto& target_type = state.task_types[type];
+                const auto count = source_type.count.load(
+                    std::memory_order_relaxed);
+                target_type.count += count;
+                target_type.total_time_ns += source_type.total_time_ns.load(
+                    std::memory_order_relaxed);
+                const auto minimum = source_type.min_time_ns.load(
+                    std::memory_order_relaxed);
+                if (count && (!target_type.min_time_ns ||
+                              minimum < target_type.min_time_ns))
+                    target_type.min_time_ns = minimum;
+                target_type.max_time_ns = std::max(
+                    target_type.max_time_ns,
+                    source_type.max_time_ns.load(std::memory_order_relaxed));
+            }
+            const auto longest = source.longest_task_time_ns.load(
+                std::memory_order_relaxed);
+            if (longest > state.longest_task_time_ns) {
+                state.longest_task_time_ns = longest;
+                state.longest_task_hash = source.longest_task_hash.load(
+                    std::memory_order_relaxed);
+                const auto name = source.longest_task_name.load(
+                    std::memory_order_acquire);
+                state.longest_task_name = name ? *name : std::string{};
+                state.longest_task_type = std::string(tf::to_string(
+                    source.longest_task_type.load(std::memory_order_relaxed)));
+            }
+        }
+        state.observed_wall_time_ns = first && last >= first ? last - first : 0;
+        state.peak_active_worker_count = std::max(
+            state.peak_active_worker_count, state.active_worker_count);
+        state.worker_utilization = workers && state.observed_wall_time_ns
+            ? static_cast<double>(state.worker_busy_time_ns) * 100.0 /
+              static_cast<double>(state.observed_wall_time_ns) /
+              static_cast<double>(workers) : 0.0;
+        state.worker_cpu_utilization = workers && state.observed_wall_time_ns
+            ? static_cast<double>(state.worker_cpu_time_ns) * 100.0 /
+              static_cast<double>(state.observed_wall_time_ns) /
+              static_cast<double>(workers) : 0.0;
+        for (auto& target : state.workers) {
+            target.idle_time_ns = state.observed_wall_time_ns > target.busy_time_ns
+                ? state.observed_wall_time_ns - target.busy_time_ns : 0;
+            target.utilization = state.observed_wall_time_ns
+                ? static_cast<double>(target.busy_time_ns) * 100.0 /
+                  static_cast<double>(state.observed_wall_time_ns) : 0.0;
+            target.cpu_utilization = state.observed_wall_time_ns
+                ? static_cast<double>(target.cpu_time_ns) * 100.0 /
+                  static_cast<double>(state.observed_wall_time_ns) : 0.0;
         }
-        state.longest_task_time_ns = longest_task_time_ns.load(std::memory_order_relaxed);
-        state.longest_task_hash = longest_task_hash.load(std::memory_order_relaxed);
-        const auto longest_name = longest_task_name.load(std::memory_order_acquire);
-        state.longest_task_name = longest_name ? *longest_name : std::string{};
-        state.longest_task_type = std::string(tf::to_string(
-            longest_task_type.load(std::memory_order_relaxed)));
     }
 };
 class Task_Resource : Pinned {
@@ -386,10 +730,13 @@ private:
     std::recursive_mutex state_mutex;
     std::atomic_bool state_callback_enabled{};
     std::atomic_bool executor_ready{};
+    Task_Runtime_Configuration configuration{};
     void create_executor(std::size_t workers, std::shared_ptr<tf::WorkerInterface> worker_interface) {
         executor_ready.store(false, std::memory_order_relaxed);
         executor = std::make_unique<tf::Executor>(workers, std::move(worker_interface));
-        observer = executor->make_observer<Task_Observer>();
+        observer = executor->make_observer<Task_Observer>(
+            configuration.worker_occupation_limit,
+            configuration.worker_overrun_action);
         state = {};
         executor_ready.store(true, std::memory_order_release);
     }
@@ -399,7 +746,9 @@ private:
         std::lock_guard guard(state_mutex);
         if (executor) return;
         executor = std::make_unique<tf::Executor>();
-        observer = executor->make_observer<Task_Observer>();
+        observer = executor->make_observer<Task_Observer>(
+            configuration.worker_occupation_limit,
+            configuration.worker_overrun_action);
         executor_ready.store(true, std::memory_order_release);
     }
     void publish_state() {
@@ -418,10 +767,18 @@ public:
         static Task_Resource value;
         return value;
     }
-    void initialize(std::size_t workers, std::shared_ptr<tf::WorkerInterface> worker_interface, Pmr pmr) {
+    void initialize(Task_Runtime_Configuration value,
+                    std::shared_ptr<tf::WorkerInterface> worker_interface) {
+        if (!value.workers)
+            throw std::invalid_argument(
+                "Taskflow runtime requires at least one worker");
+        if (value.worker_occupation_limit <= std::chrono::milliseconds::zero())
+            throw std::invalid_argument(
+                "Taskflow worker occupation limit must be positive");
         std::lock_guard guard(state_mutex);
-        memory = pmr.resource();
-        create_executor(workers, std::move(worker_interface));
+        configuration = value;
+        memory = configuration.pmr.resource();
+        create_executor(configuration.workers, std::move(worker_interface));
         active_taskflows.store(0, std::memory_order_relaxed);
         peak_active_taskflows.store(0, std::memory_order_relaxed);
         completed_taskflows.store(0, std::memory_order_relaxed);
@@ -434,8 +791,17 @@ public:
         while (peak < active && !peak_active_taskflows.compare_exchange_weak(peak, active, std::memory_order_relaxed)) {}
         auto start = std::chrono::steady_clock::now();
         try {
-            if (executor->this_worker())
-                executor->corun(native_taskflow(taskflow));
+            if (const auto* worker = executor->this_worker()) {
+                observer->pause_worker(worker->id());
+                try {
+                    executor->corun(native_taskflow(taskflow));
+                }
+                catch (...) {
+                    observer->resume_worker(worker->id());
+                    throw;
+                }
+                observer->resume_worker(worker->id());
+            }
             else
                 executor->run(native_taskflow(taskflow)).get();
         }
@@ -494,6 +860,22 @@ public:
         ensure_executor();
         executor->silent_async(std::move(name), std::move(task));
     }
+    void corun_until(std::function<bool()> predicate) {
+        ensure_executor();
+        const auto* worker = executor->this_worker();
+        if (!worker)
+            throw std::logic_error(
+                "Task graph cooperative wait requires a Kernel Executor worker");
+        observer->pause_worker(worker->id());
+        try {
+            executor->corun_until(std::move(predicate));
+        }
+        catch (...) {
+            observer->resume_worker(worker->id());
+            throw;
+        }
+        observer->resume_worker(worker->id());
+    }
     std::pmr::memory_resource* memory_resource() const noexcept {
         return memory;
     }
@@ -528,8 +910,37 @@ public:
     }
 };
 }
-void initialize_runtime(std::size_t workers, Pmr pmr) {
-    Task_Resource::instance().initialize(workers, nullptr, pmr);
+
+namespace detail {
+void register_taskflow_node(std::uint64_t native_id,
+                            std::string_view name) {
+    if (!native_id || name.empty()) return;
+    auto& registry = task_name_registry();
+    std::lock_guard lock(registry.mutex);
+    registry.names.insert_or_assign(
+        native_id, std::make_shared<const std::string>(name));
+}
+
+std::shared_ptr<const std::string> taskflow_node_name(
+    std::uint64_t native_id) noexcept {
+    try {
+        auto& registry = task_name_registry();
+        std::lock_guard lock(registry.mutex);
+        const auto found = registry.names.find(native_id);
+        return found == registry.names.end() ? nullptr : found->second;
+    }
+    catch (...) {
+        return {};
+    }
+}
+
+void corun_taskflow_until(std::function<bool()> predicate) {
+    Task_Resource::instance().corun_until(std::move(predicate));
+}
+}
+
+void initialize_runtime(Task_Runtime_Configuration configuration) {
+    Task_Resource::instance().initialize(std::move(configuration), nullptr);
 }
 void schedule_task(std::string name, std::function<void()> task) {
     Task_Resource::instance().schedule(std::move(name), std::move(task));

kernel/src/kernel/render_common.hpp

@@ -3,6 +3,7 @@
 #include "frame.hpp"
 #include "Task_Graph.hpp"
 #include <array>
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <functional>
@@ -43,17 +44,17 @@ struct Task_Type_State {
 struct Task_Worker_State {
     std::size_t id{};
     std::size_t task_count{};
-    std::size_t current_queue_size{};                       /* 当前活跃任务进入时观察到的 Worker 队列深度。 */
-    std::size_t current_queue_capacity{};                   /* 当前活跃任务进入时观察到的 Worker 队列容量。 */
+    std::size_t current_queue_size{}; /* 当前活跃任务进入时观察到的 Worker 队列深度。 */
+    std::size_t current_queue_capacity{}; /* 当前活跃任务进入时观察到的 Worker 队列容量。 */
     std::size_t peak_observed_queue_size{};
     std::size_t max_observed_queue_capacity{};
-    std::uint64_t active_task_hash{};                       /* 当前最内层原生任务身份；空闲时为 0。 */
-    std::uint64_t active_task_time_ns{};                    /* 当前任务从 on_entry 到本次读取已经持续的时间。 */
-    std::string active_task_type{};                         /* 当前任务的 Taskflow 原生 TaskType；空闲时为空。 */
+    std::uint64_t active_task_hash{}; /* 当前最内层原生任务身份；空闲时为 0。 */
+    std::uint64_t active_task_time_ns{}; /* 当前任务从 on_entry 到本次读取已经持续的时间。 */
+    std::string active_task_type{}; /* 当前任务的 Taskflow 原生 TaskType；空闲时为空。 */
     std::uint64_t task_time_ns{};
     std::uint64_t busy_time_ns{};
-    std::uint64_t cpu_time_ns{};                            /* Worker最外层任务活跃区间内累计的线程 CPU 时间。 */
-    std::uint64_t non_cpu_time_ns{};                        /* busy_time_ns 减去 cpu_time_ns；只表示未计费墙钟，不推断锁或抢占。 */
+    std::uint64_t cpu_time_ns{}; /* Worker最外层任务活跃区间内累计的线程 CPU 时间。 */
+    std::uint64_t non_cpu_time_ns{}; /* busy_time_ns 减去 cpu_time_ns；只表示未计费墙钟，不推断锁或抢占。 */
     std::uint64_t idle_time_ns{};
     std::uint64_t min_task_time_ns{};
     std::uint64_t max_task_time_ns{};
@@ -98,13 +99,25 @@ struct Task_Runtime_State : State_Type<Task_Runtime_State_Tag> {
     std::vector<Task_Worker_State> workers;
     bool operator==(const Task_Runtime_State&) const = default;
 };
+enum class Task_Overrun_Action : std::uint8_t {
+    warning,
+    fast_fail
+};
+struct Task_Runtime_Configuration {
+    std::size_t workers{std::thread::hardware_concurrency()}; /* 全局 Executor 的 Worker 数量。 */
+    std::chrono::milliseconds worker_occupation_limit{100}; /* 单个节点连续非 CPU 等待 Worker 的上限；协作让出区间不计。 */
+    Task_Overrun_Action worker_overrun_action{
+        /* 节点超过占用上限后的处置策略。 */
+        Task_Overrun_Action::fast_fail
+    };
+    Pmr pmr{}; /* Task Graph 临时结构使用的内存资源。 */
+};
 /*
  * 初始化全局 Taskflow 运行时。
  * workers 为 Executor Worker 数；worker_interface 可自定义 Worker；pmr 作为渲染内核临时图结构的统一内存资源。
  * 未主动调用时运行时会在第一次执行 Taskflow 时按 Taskflow 默认配置惰性创建 Executor。
  */
-void initialize_runtime(std::size_t workers = std::thread::hardware_concurrency(),
-                        Pmr pmr = {});
+void initialize_runtime(Task_Runtime_Configuration configuration = {});
 /* 把具名独立业务任务提交给全局 Taskflow worker；任务不得执行阻塞式设备等待。 */
 void schedule_task(std::string name, std::function<void()> task);
 /* 读取全局 Executor 的累计状态；计算只读取原子计数，不触发逐任务导出。 */

kernel/src/kernel/scene.ipp

@@ -121,6 +121,11 @@ void Scene::Private::after_advance(Object* object,
         Task_Node prepare_exit; /* Prepare 完成任务，作为该阶段依赖出口。 */
     };
     std::pmr::unordered_map<Root*, Stage_Tasks> stage_tasks{resource};
+    /*
+     * Prepare 的并行关系只有一个权威来源：Prepare_Data_Tag。
+     * 每个 Renderable 先建立独立的条件、执行和完成节点；下面只为显式
+     * 依赖连边，不得按遍历顺序增加 previous 链或其他隐式串行关系。
+     */
     for (auto* renderable : renderables) {
         Root* root = renderable;
         auto* data = prepare_dependencies.private_data(root);

kernel/src/test/render_test.cpp

@@ -91,7 +91,7 @@ void add_renderable(Scene& scene, Renderable* renderable) {
 }
 }
 TEST(renderable_capability, direct_stages_do_not_allocate_subgraphs) {
-    aethera::initialize_runtime(2);
+    aethera::initialize_runtime({.workers = 2});
     auto renderable = build_object<Direct>();
     auto scene = build_object<Scene>();
     add_renderable(*scene, renderable.get());
@@ -112,7 +112,7 @@ TEST(renderable_capability, direct_stages_do_not_allocate_subgraphs) {
     EXPECT_EQ(prepare_extension_calls, 1);
 }
 TEST(renderable_capability, graph_stage_builds_lazily_and_rebuilds_inside_condition) {
-    aethera::initialize_runtime(2);
+    aethera::initialize_runtime({.workers = 2});
     auto renderable = build_object<Graph>();
     auto scene = build_object<Scene>();
     add_renderable(*scene, renderable.get());
@@ -133,7 +133,7 @@ TEST(renderable_capability, graph_stage_builds_lazily_and_rebuilds_inside_condit
     EXPECT_EQ(data.prepare_calls, 2);
 }
 TEST(renderable_state, scene_and_renderable_callbacks_publish_at_stage_boundaries) {
-    aethera::initialize_runtime(2);
+    aethera::initialize_runtime({.workers = 2});
     auto renderable = build_object<Direct>();
     auto scene = build_object<Scene>();
     add_renderable(*scene, renderable.get());
@@ -161,7 +161,7 @@ TEST(renderable_state, scene_and_renderable_callbacks_publish_at_stage_boundarie
     aethera::clear_runtime_state_callback<aethera::Task_Runtime_State_Tag>();
 }
 TEST(scene_state, structural_statistics_survive_a_process_without_rebuild) {
-    aethera::initialize_runtime(2);
+    aethera::initialize_runtime({.workers = 2});
     auto renderable = build_object<Direct>();
     auto scene = build_object<Scene>();
     add_renderable(*scene, renderable.get());
@@ -188,7 +188,7 @@ TEST(scene_state, structural_statistics_survive_a_process_without_rebuild) {
     EXPECT_EQ(updates, 2);
 }
 TEST(scene_condition, upstream_change_makes_downstream_run_in_same_taskflow) {
-    aethera::initialize_runtime(2);
+    aethera::initialize_runtime({.workers = 2});
     auto source = build_object<Dependency>();
     auto target = build_object<Dependency>();
     auto scene = build_object<Scene>();
@@ -215,7 +215,7 @@ TEST(scene_condition, upstream_change_makes_downstream_run_in_same_taskflow) {
 }
 
 TEST(task_graph_observer, business_dag_and_native_execution_share_node_identity) {
-    aethera::initialize_runtime(2);
+    aethera::initialize_runtime({.workers = 2});
     std::atomic_int completed{};
     aethera::Task_Graph child{"test.visual"};
     auto prepare = child.add("prepare.samples", [&] { completed.fetch_add(1); });