// RootNode::register_node — called from any thread
void RootNode::register_node(const std::shared_ptr<Node>& node)
{
    if (m_is_processing.load(std::memory_order_acquire)) {

        // Try to claim a pending slot via CAS
        for (auto& m_pending_op : m_pending_ops) {
            bool expected = false;
            if (m_pending_op.active.compare_exchange_strong(
                    expected, true,
                    std::memory_order_acquire,
                    std::memory_order_relaxed)) {
                m_pending_op.node = node;
                atomic_add_flag(node->m_state, NodeState::INACTIVE);
                m_pending_count.fetch_add(1, std::memory_order_relaxed);
                return;
            }
        }

        // All slots full — wait for processing to finish
        while (m_is_processing.load(std::memory_order_acquire)) {
            m_is_processing.wait(true, std::memory_order_acquire);
        }
    }

    m_Nodes.push_back(node);
    atomic_add_flag(node->m_state, NodeState::ACTIVE);
}

void atomic_add_flag(std::atomic<NodeState>& state, NodeState flag)
{
    auto current = state.load();
    NodeState desired;
    do {
        desired = static_cast<NodeState>(current | flag);
    } while (!state.compare_exchange_weak(current, desired,
        std::memory_order_acq_rel,
        std::memory_order_acquire));
}

bool Node::try_claim_snapshot_context(uint64_t context_id)
{
    uint64_t expected = 0;
    return m_snapshot_context_id.compare_exchange_strong(
        expected, context_id,
        std::memory_order_acq_rel,
        std::memory_order_acquire);
}

bool RootNode::preprocess()
{
    bool expected = false;
    if (!m_is_processing.compare_exchange_strong(expected, true,
            std::memory_order_acquire, std::memory_order_relaxed)) {
        return false;
    }

    if (m_pending_count.load(std::memory_order_relaxed) > 0) {
        process_pending_operations();
    }
    return true;
}

void RootNode::postprocess()
{
    for (auto& node : m_Nodes) {
        node->request_reset_from_channel(m_channel);
    }
    m_is_processing.store(false, std::memory_order_release);
    m_is_processing.notify_all();
}

struct SampleDelay {
    using promise_handle = Vruta::audio_promise;
    uint64_t samples_to_wait;

    [[nodiscard]] bool await_ready() const { return samples_to_wait == 0; }
    void await_resume() { }

    void await_suspend(std::coroutine_handle<Vruta::promise_type> h)
    {
        auto& promise = h.promise();
        promise.next_sample += samples_to_wait;
    }
};

scheduler->process_token(Vruta::ProcessingToken::SAMPLE_ACCURATE, 1024);
scheduler->process_token(Vruta::ProcessingToken::FRAME_ACCURATE, 1);

auto routine = [](Vruta::TaskScheduler& scheduler) -> Vruta::SoundRoutine {
    auto& promise = co_await Kriya::GetAudioPromise{};

    while (true) {
        if (promise.should_terminate) break;

        // Gate: suspend until a logic node outputs true
        co_await Kriya::Gate{scheduler, []() {
            schedule_another_clock();
        }, logic_node, true};

        // Dynamic timing from musical state
        float wait_time = calculate_musical_timing();
        co_await Kriya::SampleDelay{scheduler.seconds_to_samples(wait_time)};

        // Trigger: fire callback and synchronize
        co_await Kriya::Trigger{scheduler, true, []() {
            sync_frame_clock();
        }, sync_node};
    }
};

MayaFlux::schedule_metro(2.0, []() { modulate_filter_cutoff(); }, "main_clock");

MayaFlux::schedule_pattern([](uint64_t beat) {
    return beat % 8 == 0;
}, []() { change_distribution(); }, 1.0, "pattern_trigger");

auto shape = vega.Polynomial({0.1, 0.5, 2.f});
shape >> Time(2.f);  // Creates coroutine, registers with TaskScheduler

template <typename T>
concept ArithmeticData = std::is_integral_v<T> || std::is_floating_point_v<T>;

template <typename T>
concept ComplexData = requires {
    typename T::value_type;
    std::is_same_v<T, std::complex<float>> || std::is_same_v<T, std::complex<double>>;
};

template <typename T>
concept ContiguousContainer = requires(T t) {
    { t.data() } -> std::convertible_to<typename T::value_type*>;
    { t.size() } -> std::convertible_to<std::size_t>;
    typename T::value_type;
};

template <typename From, typename To>
concept SafeArithmeticConversion = ArithmeticData<From> && ArithmeticData<To>
    && (SafeIntegerConversion<From, To>
     || SafeDecimalConversion<From, To>
     || (IntegerData<From> && DecimalData<To>));

using DataVariant = std::variant<
    std::vector<double>,
    std::vector<float>,
    std::vector<uint8_t>,
    std::vector<uint16_t>,
    std::vector<uint32_t>,
    std::vector<std::complex<float>>,
    std::vector<std::complex<double>>,
    std::vector<glm::vec2>,
    std::vector<glm::vec3>,
    std::vector<glm::vec4>,
    std::vector<glm::mat4>
>;

DataAccess access(variant, dimensions, DataModality::AUDIO_1D);

auto audio_view = access.view<double>();      // span<double>
auto gpu_info   = access.gpu_buffer();        // (void*, byte_count, format_hint)
auto elems      = access.element_count();
auto comps      = access.component_count();   // 1 for scalar, 3 for vec3, etc.

template <> struct extraction_traits_d<Kakshya::DataVariant> {
    static constexpr bool is_multi_variant = false;
    static constexpr bool requires_container = false;
    static constexpr bool is_region_like = false;
    using result_type = std::span<double>;
};

template <> struct extraction_traits_d<Kakshya::Region> {
    static constexpr bool is_multi_variant = true;
    static constexpr bool requires_container = true;
    static constexpr bool is_region_like = true;
    using result_type = std::vector<std::span<double>>;
};

template <typename T>
concept ComputeData =
    std::same_as<T, Kakshya::DataVariant> ||
    std::same_as<T, std::vector<Kakshya::DataVariant>> ||
    std::same_as<T, std::shared_ptr<Kakshya::SignalSourceContainer>> ||
    std::same_as<T, Kakshya::Region> ||
    std::same_as<T, Kakshya::RegionGroup> ||
    std::same_as<T, std::vector<Kakshya::RegionSegment>> ||
    std::is_base_of_v<Eigen::MatrixBase<T>, T> ||
    VariantVector<T> ||
    std::constructible_from<Kakshya::DataVariant, T>;

template <typename To, typename From>
CastResult<To> try_convert(const From& value)
{
    CastResult<To> result;
    if constexpr (SafeArithmeticConversion<From, To>) {
        result.value = static_cast<To>(value);
        result.precision_loss = (sizeof(From) > sizeof(To));
    } else if constexpr (ComplexData<From> && ArithmeticData<To>) {
        result.value = static_cast<To>(std::abs(value));
        result.precision_loss = (std::imag(value) != 0);
    }
    // ...
    return result;
}

enum Domain : uint64_t {
    AUDIO = (uint64_t(Nodes::ProcessingToken::AUDIO_RATE) << 32)
          | (uint64_t(Buffers::ProcessingToken::AUDIO_BACKEND) << 16)
          | (uint64_t(Vruta::ProcessingToken::SAMPLE_ACCURATE)),

    GRAPHICS = (uint64_t(Nodes::ProcessingToken::VISUAL_RATE) << 32)
             | (uint64_t(Buffers::ProcessingToken::GRAPHICS_BACKEND) << 16)
             | (uint64_t(Vruta::ProcessingToken::FRAME_ACCURATE)),
};

template <typename T, typename StoragePolicy,
          typename ConcurrencyPolicy = SingleThreadedPolicy,
          typename AccessPattern = QueueAccess>
class RingBuffer;

// Lock-free input queue: real-time thread → worker thread
using InputQueue = RingBuffer<InputValue,
    FixedStorage<InputValue, 4096>, LockFreePolicy, QueueAccess>;

// Audio delay line: single-threaded DSP
using AudioDelay = RingBuffer<double,
    DynamicStorage<double>, SingleThreadedPolicy, HistoryBufferAccess>;

// Real-time logging: audio thread → disk writer
using LogBuffer = RingBuffer<RealtimeEntry,
    FixedStorage<RealtimeEntry, 8192>, LockFreePolicy, QueueAccess>;

void compose() {
    // Audio domain: modal synthesis driven by logic
    auto bell = vega.ModalNetwork(12, 220.0,
        ModalNetwork::Spectrum::INHARMONIC)[0] | Audio;

    auto osc = vega.Sine(0.2, 1.0f);
    auto logic = vega.Logic([](double input) {
        static double last = 0.0;
        bool crossed = (last < 0.0) && (input >= 0.0);
        last = input;
        return crossed;
    });
    osc >> logic;

    logic->on_change_to(true, [bell](auto& ctx) {
        bell->excite(get_uniform_random(0.5f, 0.9f));
        bell->set_fundamental(get_uniform_random(220.0f, 1000.0f));
    });

    // Graphics domain: Vulkan point collection
    auto window = MayaFlux::create_window({"Bell Visualization", 1280, 720});
    auto points = vega.PointCollectionNode(500) | Graphics;
    auto geom = vega.GeometryBuffer(points) | Graphics;
    geom->setup_rendering({.target_window = window});
    window->show();

    // Cross-domain coordination via metro coroutine
    MayaFlux::schedule_metro(0.016, [points]() {
        // Runs at ~60Hz, reads audio state, writes to graphics node
        float x = std::cos(angle) * radius;
        float y = std::sin(angle) * radius;
        points->add_point(Nodes::GpuSync::PointVertex{
            .position = glm::vec3(x, y, 0.0f),
            .color = glm::vec3(brightness, brightness * 0.8f, 1.0f),
            .size = 8.0f + radius * 4.0f
        });
    });
}

void Engine::Init(const GlobalStreamInfo& streamInfo,
                  const GlobalGraphicsConfig& graphics_config,
                  const GlobalInputConfig& input_config)
{
    m_scheduler = std::make_shared<Vruta::TaskScheduler>(streamInfo.sample_rate);
    m_buffer_manager = std::make_shared<Buffers::BufferManager>(
        streamInfo.output.channels, /* ... */ streamInfo.buffer_size);
    m_node_graph_manager = std::make_shared<Nodes::NodeGraphManager>(
        streamInfo.sample_rate, streamInfo.buffer_size);
    m_window_manager = std::make_shared<WindowManager>(graphics_config);
    m_input_manager = std::make_shared<InputManager>();

    m_subsystem_manager = std::make_shared<SubsystemManager>(
        m_node_graph_manager, m_buffer_manager, m_scheduler,
        m_window_manager, m_input_manager);

    m_subsystem_manager->create_audio_subsystem(streamInfo);
    m_subsystem_manager->create_graphics_subsystem(graphics_config);
    m_subsystem_manager->create_input_subsystem(input_config);
}

int AudioSubsystem::process_output(double* output_buffer, unsigned int num_frames)
{
    m_callback_active.fetch_add(1, std::memory_order_acquire);

    // 1. Buffer-cycle tasks (metro callbacks, scheduled events)
    m_handle->tasks.process_buffer_cycle();

    // 2. Per-channel: process buffer chains, then collect network outputs
    for (uint32_t channel = 0; channel < num_channels; channel++) {
        m_handle->buffers.process_channel(channel, num_frames);
        all_network_outputs[channel] =
            m_handle->nodes.process_audio_networks(num_frames, channel);
        buffer_data[channel] = m_handle->buffers.read_channel_data(channel);
    }

    // 3. Per-sample: advance scheduler, sum node + buffer + network outputs
    for (size_t i = 0; i < num_frames; ++i) {
        m_handle->tasks.process(1);  // Advance coroutines by 1 sample

        for (size_t j = 0; j < num_channels; ++j) {
            double sample = m_handle->nodes.process_sample(j)
                          + buffer_data[j][i];

            for (const auto& net_buf : all_network_outputs[j]) {
                if (i < net_buf.size()) sample += net_buf[i];
            }

            output_span[i * num_channels + j] = std::clamp(sample, -1., 1.);
        }
    }

    // 4. Drain deferred operations (CAS pending slots)
    m_handle->nodes.cleanup_completed_routing();
    m_handle->buffers.cleanup_completed_routing();

    m_callback_active.fetch_sub(1, std::memory_order_release);
    return 0;
}

void GraphicsSubsystem::start()
{
    m_running.store(true);
    m_frame_clock->reset();

    m_graphics_thread = std::thread([this]() {
        m_graphics_thread_id = std::this_thread::get_id();
        graphics_thread_loop();
    });
}

void GraphicsSubsystem::graphics_thread_loop()
{
    while (m_running.load(std::memory_order_acquire)) {
        if (m_paused.load(std::memory_order_acquire)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(16));
            continue;
        }

        m_frame_clock->tick();       // Wall-clock driven, not hardware-driven

        process();                   // Same handle pattern as AudioSubsystem

        m_frame_clock->wait_for_next_frame();

        if (m_frame_clock->is_frame_late()) {
            uint64_t lag = m_frame_clock->get_frame_lag();
            if (lag > 2) {
                MF_RT_WARN(/* ... */ "Frame lag: {} frames behind (FPS: {:.1f})",
                    lag, m_frame_clock->get_measured_fps());
            }
        }
    }
}

void GraphicsSubsystem::process()
{
    for (auto& [name, hook] : m_handle->pre_process_hooks) { hook(1); }

    m_handle->tasks.process(1);      // FRAME_ACCURATE coroutines
    m_handle->nodes.process(1);      // VISUAL_RATE nodes
    m_handle->buffers.process(1);    // GRAPHICS_BACKEND buffers

    register_windows_for_processing();
    m_backend->handle_window_resize();
    render_all_windows();            // Vulkan: acquire → record → submit → present
    m_handle->windows.process();
    cleanup_closed_windows();

    for (auto& [name, hook] : m_handle->post_process_hooks) { hook(1); }
}

InputSubsystem::InputSubsystem(GlobalInputConfig& config)
    : m_config(config)
    , m_tokens {
        .Buffer = Buffers::ProcessingToken::INPUT_BACKEND,
        .Node = Nodes::ProcessingToken::EVENT_RATE,
        .Task = Vruta::ProcessingToken::EVENT_DRIVEN
    }
{}

void InputSubsystem::register_callbacks()
{
    // Input subsystem doesn't register timing callbacks like audio/graphics.
    // Backends push to InputManager's queue, which has its own thread.
}

void HIDBackend::poll_thread_func()
{
    while (!m_stop_requested.load()) {
        {
            std::lock_guard lock(m_devices_mutex);
            for (auto& [id, state] : m_open_devices) {
                if (state->active.load() && state->handle) {
                    poll_device(id, *state);
                }
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
}

double InputNode::apply_smoothing(double target, double current) const
{
    switch (m_config.smoothing) {
    case SmoothingMode::NONE:       return target;
    case SmoothingMode::LINEAR:     return current + (target - current) * m_config.smoothing_factor;
    case SmoothingMode::EXPONENTIAL: return m_config.smoothing_factor * target
                                          + (1.0 - m_config.smoothing_factor) * current;
    case SmoothingMode::SLEW: {
        double diff = target - current;
        return (std::abs(diff) <= m_config.slew_rate)
            ? target : current + (diff > 0 ? m_config.slew_rate : -m_config.slew_rate);
    }
    }
}