#include "graph_manager.h"

#include <fstream>
#include <iostream>
#include <algorithm>
#include <cstdlib>
#include <map>
#include <set>
#include <deque>
#include <thread>
#include <chrono>
#include <unordered_map>
#include <sstream>

#include "utils/config_expand.h"
#include "utils/config_schema.h"
#include "utils/logger.h"
#include "utils/simple_json_writer.h"
#include "utils/thread_affinity.h"
#include "hw/hw_factory.h"

namespace rk3588 {

namespace {

int GetEnvInt(const char* name, int def) {
    if (!name) return def;
    const char* v = std::getenv(name);
    if (!v || !*v) return def;
    try {
        return std::stoi(v);
    } catch (...) {
        return def;
    }
}

int DefaultGraphPoolThreads() {
    // Override via env: RK3588_GRAPH_POOL_THREADS
    const int v = GetEnvInt("RK3588_GRAPH_POOL_THREADS", 0);
    if (v > 0 && v <= 64) return v;
    const unsigned hc = std::thread::hardware_concurrency();
    const int d = (hc == 0) ? 2 : static_cast<int>(hc);
    return std::max(1, std::min(4, d));
}

size_t DefaultGraphBatchSize() {
    // Override via env: RK3588_GRAPH_BATCH_SIZE
    const int v = GetEnvInt("RK3588_GRAPH_BATCH_SIZE", 0);
    if (v > 0 && v <= 1024) return static_cast<size_t>(v);
    return 8;
}

size_t DefaultGraphRunBudget() {
    // Max frames processed per scheduled run before yielding.
    const int v = GetEnvInt("RK3588_GRAPH_RUN_BUDGET", 0);
    if (v > 0 && v <= 4096) return static_cast<size_t>(v);
    return 64;
}

std::string AffinityKey(std::vector<int> cpus) {
    if (cpus.empty()) return "";
    std::sort(cpus.begin(), cpus.end());
    cpus.erase(std::unique(cpus.begin(), cpus.end()), cpus.end());
    std::ostringstream oss;
    for (size_t i = 0; i < cpus.size(); ++i) {
        if (i) oss << ',';
        oss << cpus[i];
    }
    return oss.str();
}

QueueDropStrategy ParseDropStrategy(const std::string& s, QueueDropStrategy def) {
    if (s == "drop_oldest") return QueueDropStrategy::DropOldest;
    if (s == "drop_newest") return QueueDropStrategy::DropNewest;
    if (s == "block") return QueueDropStrategy::Block;
    return def;
}

void ApplyQueueConfig(const SimpleJson* queue_cfg, size_t default_queue_size,
                      QueueDropStrategy default_strategy, size_t& out_size,
                      QueueDropStrategy& out_strategy) {
    out_size = default_queue_size;
    out_strategy = default_strategy;
    if (!queue_cfg || !queue_cfg->IsObject()) return;

    out_size = static_cast<size_t>(queue_cfg->ValueOr<int>("size", static_cast<int>(default_queue_size)));
    std::string policy = queue_cfg->ValueOr<std::string>("policy", "");
    std::string strat = queue_cfg->ValueOr<std::string>("strategy", "");
    std::string final_policy = policy.empty() ? strat : policy;
    if (!final_policy.empty()) {
        out_strategy = ParseDropStrategy(final_policy, default_strategy);
    }
}

struct ParsedEdge {
    std::string from;
    std::string to;
    const SimpleJson* queue_cfg = nullptr;   // can be null
    SimpleJson queue_value;                  // normalized copy (null if none)
};

bool ParseEdges(const SimpleJson& edges_json, std::vector<ParsedEdge>& out, std::string& err) {
    out.clear();
    if (!edges_json.IsArray()) {
        err = "edges must be array";
        return false;
    }
    for (const auto& edge_val : edges_json.AsArray()) {
        ParsedEdge e;
        if (edge_val.IsArray()) {
            const auto& edge_arr = edge_val.AsArray();
            if (edge_arr.size() < 2) {
                err = "Edge array must be [from, to] or [from, to, {...}]";
                return false;
            }
            e.from = edge_arr[0].AsString("");
            e.to = edge_arr[1].AsString("");
            if (edge_arr.size() >= 3 && edge_arr[2].IsObject()) {
                const SimpleJson* q = edge_arr[2].Find("queue");
                e.queue_cfg = q ? q : &edge_arr[2];
            }
        } else if (edge_val.IsObject()) {
            e.from = edge_val.ValueOr<std::string>("from", "");
            e.to = edge_val.ValueOr<std::string>("to", "");
            e.queue_cfg = edge_val.Find("queue");
        } else {
            err = "Edge must be an array or object";
            return false;
        }
        if (e.from.empty() || e.to.empty()) {
            err = "Edge has empty endpoint";
            return false;
        }
        if (e.queue_cfg) e.queue_value = *e.queue_cfg;
        out.push_back(std::move(e));
    }
    return true;
}

struct EdgeKey {
    std::string from;
    std::string to;
    bool operator<(const EdgeKey& o) const {
        if (from != o.from) return from < o.from;
        return to < o.to;
    }
};

bool EffectiveQueueForEdge(const ParsedEdge& e, const SimpleJson* from_node_queue_cfg,
                           size_t default_queue_size, QueueDropStrategy default_strategy,
                           size_t& out_size, QueueDropStrategy& out_strategy) {
    const SimpleJson* chosen = e.queue_cfg ? e.queue_cfg : from_node_queue_cfg;
    ApplyQueueConfig(chosen, default_queue_size, default_strategy, out_size, out_strategy);
    return true;
}

}  // namespace

class Graph::Executor {
public:
    explicit Executor(Graph& g)
        : g_(g), batch_size_(DefaultGraphBatchSize()), run_budget_(DefaultGraphRunBudget()) {
        if (const SimpleJson* exec = g_.graph_cfg_.Find("executor"); exec && exec->IsObject()) {
            const int batch = exec->ValueOr<int>("batch_size", static_cast<int>(batch_size_));
            const int budget = exec->ValueOr<int>("run_budget", static_cast<int>(run_budget_));
            if (batch > 0 && batch <= 1024) batch_size_ = static_cast<size_t>(batch);
            if (budget > 0 && budget <= 4096) run_budget_ = static_cast<size_t>(budget);
        }
    }

    bool Start() {
        Stop();
        stop_.store(false);

        // Build pools: default pool (no affinity) + pools for explicit cpu_affinity.
        pools_.clear();
        key_to_pool_.clear();

        auto add_pool = [&](const std::string& key, std::vector<int> cpus, int threads) -> uint32_t {
            auto p = std::make_unique<Pool>();
            p->key = key;
            p->cpus = std::move(cpus);
            p->thread_count = std::max(1, threads);
            pools_.push_back(std::move(p));
            const uint32_t idx = static_cast<uint32_t>(pools_.size() - 1);
            key_to_pool_[key] = idx;
            return idx;
        };

        default_pool_index_ = add_pool("", {}, DefaultGraphPoolThreads());

        for (auto& n : g_.nodes_) {
            if (!n.enabled || !n.node) continue;
            if (n.role == "source" || !n.context.input_queue) continue;

            n.cpu_affinity = ParseCpuAffinity(n.config);
            const std::string key = AffinityKey(n.cpu_affinity);
            if (!key.empty() && key_to_pool_.find(key) == key_to_pool_.end()) {
                add_pool(key, n.cpu_affinity, 1);
            }
        }

        // Assign nodes to pools.
        for (auto& n : g_.nodes_) {
            if (!n.enabled || !n.node) continue;
            if (n.role == "source" || !n.context.input_queue) continue;

            const std::string key = AffinityKey(n.cpu_affinity);
            auto it = key_to_pool_.find(key);
            n.pool_index = (it != key_to_pool_.end()) ? it->second : default_pool_index_;
            n.sched_state.store(0);
        }

        // Start pool threads.
        for (auto& up : pools_) {
            Pool* p = up.get();
            if (!p) continue;
            for (int i = 0; i < p->thread_count; ++i) {
                p->threads.emplace_back([this, p]() { WorkerLoop(*p); });
            }
        }

        started_ = true;
        return true;
    }

    void Stop() {
        if (!started_) return;
        stop_.store(true);

        for (auto& up : pools_) {
            if (!up) continue;
            Pool& p = *up;
            {
                std::lock_guard<std::mutex> lock(p.mu);
                p.ready.clear();
            }
            p.cv.notify_all();
        }

        for (auto& up : pools_) {
            if (!up) continue;
            Pool& p = *up;
            for (auto& t : p.threads) {
                if (t.joinable()) t.join();
            }
            p.threads.clear();
            {
                std::lock_guard<std::mutex> lock(p.mu);
                p.ready.clear();
            }
        }
        pools_.clear();
        key_to_pool_.clear();
        started_ = false;
    }

    void AttachQueueCallbacks() {
        for (auto& n : g_.nodes_) {
            if (!n.enabled || !n.node) continue;
            if (n.role == "source" || !n.context.input_queue) continue;
            NodeEntry* ptr = &n;
            n.context.input_queue->SetOnDataAvailable([this, ptr]() { Schedule(ptr); });
        }
    }

    void DetachQueueCallbacks() {
        for (auto& n : g_.nodes_) {
            if (!n.enabled || !n.node) continue;
            if (n.role == "source" || !n.context.input_queue) continue;
            n.context.input_queue->SetOnDataAvailable(nullptr);
        }
    }

    void Schedule(NodeEntry* n) {
        if (!n) return;
        if (!n->enabled || !n->node || !n->context.input_queue) return;

        constexpr uint32_t kQueued = 1u;
        constexpr uint32_t kRunning = 2u;

        const uint32_t prev = n->sched_state.fetch_or(kQueued, std::memory_order_acq_rel);
        if (prev & kQueued) return;
        if (prev & kRunning) return;

        const uint32_t idx = (n->pool_index < pools_.size()) ? n->pool_index : default_pool_index_;
        if (idx >= pools_.size() || !pools_[idx]) return;
        Pool& p = *pools_[idx];
        {
            std::lock_guard<std::mutex> lock(p.mu);
            p.ready.push_back(n);
        }
        p.cv.notify_one();
    }

private:
    struct Pool {
        std::string key;
        std::vector<int> cpus;
        int thread_count = 1;
        std::mutex mu;
        std::condition_variable cv;
        std::deque<NodeEntry*> ready;
        std::vector<std::thread> threads;
    };

    void WorkerLoop(Pool& p) {
        if (!p.cpus.empty()) {
            std::string aerr;
            if (!SetCurrentThreadAffinity(p.cpus, aerr)) {
                LogWarn("[Graph] SetCurrentThreadAffinity failed for pool: " + aerr);
            }
        }

        while (true) {
            NodeEntry* n = nullptr;
            {
                std::unique_lock<std::mutex> lock(p.mu);
                p.cv.wait(lock, [&]() { return stop_.load() || !p.ready.empty(); });
                if (stop_.load() && p.ready.empty()) {
                    return;
                }
                if (!p.ready.empty()) {
                    n = p.ready.front();
                    p.ready.pop_front();
                }
            }

            if (n) {
                RunNode(*n, p);
            }
        }
    }

    void RunNode(NodeEntry& entry, Pool& p) {
        if (!entry.enabled || !entry.node || !entry.context.input_queue) return;

        constexpr uint32_t kQueued = 1u;
        constexpr uint32_t kRunning = 2u;

        // Mark running.
        entry.sched_state.fetch_or(kRunning, std::memory_order_acq_rel);

        std::vector<FramePtr> batch;
        batch.reserve(batch_size_);

        while (true) {
            size_t processed = 0;

            // Drain up to run_budget_ frames to avoid starving other nodes.
            while (processed < run_budget_ && entry.context.input_queue->TryPopBatch(batch, batch_size_)) {
                for (auto& frame : batch) {
                    if (!frame) continue;

                    const auto t0 = std::chrono::steady_clock::now();
                    NodeStatus st = entry.node->Process(frame);
                    const auto t1 = std::chrono::steady_clock::now();
                    const auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
                    if (entry.metrics) {
                        if (ns > 0) {
                            entry.metrics->process_time_ns_total.fetch_add(static_cast<uint64_t>(ns),
                                                                           std::memory_order_relaxed);
                        }
                        if (st == NodeStatus::OK) {
                            entry.metrics->ok_total.fetch_add(1, std::memory_order_relaxed);
                        } else if (st == NodeStatus::DROP) {
                            entry.metrics->drop_total.fetch_add(1, std::memory_order_relaxed);
                        } else {
                            entry.metrics->error_total.fetch_add(1, std::memory_order_relaxed);
                        }
                    }

                    ++processed;
                    if (processed >= run_budget_) break;
                }
            }

            // If we hit budget and there's still work, yield by re-enqueueing.
            if (processed >= run_budget_ && entry.context.input_queue->Size() > 0) {
                entry.sched_state.fetch_or(kQueued, std::memory_order_acq_rel);
                entry.sched_state.fetch_and(~kRunning, std::memory_order_acq_rel);
                {
                    std::lock_guard<std::mutex> lock(p.mu);
                    p.ready.push_back(&entry);
                }
                p.cv.notify_one();
                return;
            }

            // No immediate items; if someone scheduled while we were running, continue.
            uint32_t st = entry.sched_state.load(std::memory_order_acquire);
            if (st & kQueued) {
                entry.sched_state.fetch_and(~kQueued, std::memory_order_acq_rel);
                continue;
            }

            // Try to transition RUNNING -> 0. If it fails, a concurrent schedule happened; loop again.
            uint32_t expected = kRunning;
            if (entry.sched_state.compare_exchange_strong(expected, 0u, std::memory_order_acq_rel)) {
                return;
            }
        }
    }

    Graph& g_;
    size_t batch_size_ = 8;
    size_t run_budget_ = 64;
    std::atomic<bool> stop_{false};
    bool started_ = false;
    uint32_t default_pool_index_ = 0;
    std::vector<std::unique_ptr<Pool>> pools_;
    std::unordered_map<std::string, uint32_t> key_to_pool_;
};

Graph::Graph(std::string name) : name_(std::move(name)) {}

Graph::~Graph() { Stop(); }

bool Graph::Build(const SimpleJson& graph_cfg, PluginLoader& loader, size_t default_queue_size,
                  QueueDropStrategy default_strategy, std::string& err) {
    graph_cfg_ = graph_cfg;
    built_default_queue_size_ = default_queue_size;
    built_default_strategy_ = default_strategy;

    nodes_.clear();
    edges_.clear();
    executor_.reset();
    {
        std::lock_guard<std::mutex> lock(rate_mu_);
        last_rate_tp_ = {};
        last_node_in_popped_.clear();
        last_node_out_pushed_.clear();
        last_edge_pushed_.clear();
        last_edge_popped_.clear();
    }

    const auto& obj = graph_cfg.AsObject();
    auto name_it = obj.find("name");
    if (name_it != obj.end() && name_it->second.IsString()) {
        name_ = name_it->second.AsString(name_);
    }

    infer_backend_ = HwFactory::CreateInferBackend(graph_cfg_);

    // Parse nodes
    auto nodes_it = obj.find("nodes");
    if (nodes_it == obj.end() || !nodes_it->second.IsArray()) {
        err = "Graph missing 'nodes' array";
        return false;
    }

    for (const auto& node_val : nodes_it->second.AsArray()) {
        if (!node_val.IsObject()) {
            err = "Node entry is not object";
            return false;
        }
        NodeEntry entry;
        entry.config = node_val;
        entry.id = node_val.ValueOr<std::string>("id", "");
        entry.type = node_val.ValueOr<std::string>("type", "");
        entry.role = node_val.ValueOr<std::string>("role", "");
        entry.enabled = node_val.ValueOr<bool>("enable", true);
        entry.metrics = std::make_shared<NodeMetrics>();
        entry.context.graph_name = name_;
        entry.context.infer_backend = infer_backend_;
        if (entry.type == "preprocess" || entry.type == "ai_shoe_det" || entry.type == "ai_yolo") {
            entry.context.image_processor = HwFactory::CreateImageProcessor(entry.config);
        }
        
        if (entry.id.empty() || entry.type.empty()) {
            err = "Node missing id or type";
            return false;
        }
        nodes_.push_back(std::move(entry));
    }

    // Parse edges
    auto edges_it = obj.find("edges");
    if (edges_it == obj.end() || !edges_it->second.IsArray()) {
        err = "Graph missing 'edges' array";
        return false;
    }

    std::map<std::string, NodeEntry*> id_to_node;
    for (auto& n : nodes_) {
        if (n.enabled) {
            id_to_node[n.id] = &n;
        }
    }

    std::vector<ParsedEdge> parsed_edges;
    if (!ParseEdges(edges_it->second, parsed_edges, err)) {
        return false;
    }

    // Track enabled-edge connectivity for validation.
    std::vector<std::pair<std::string, std::string>> enabled_edges;

    for (const auto& e : parsed_edges) {
        auto from_it = id_to_node.find(e.from);
        auto to_it = id_to_node.find(e.to);
        
        if (from_it == id_to_node.end() || to_it == id_to_node.end()) {
             // Check if nodes exist but are disabled
             bool from_exists = false;
             bool to_exists = false;
             for(const auto& n : nodes_) { 
                 if(n.id == e.from) from_exists = true; 
                 if(n.id == e.to) to_exists = true; 
             }
             
             if (!from_exists || !to_exists) {
                 err = "Edge references unknown node: " + e.from + " -> " + e.to;
                 return false;
             }
             // At least one is disabled, skip edge
             continue;
        }

        const SimpleJson* from_node_queue = from_it->second->config.Find("queue");
        size_t qsize = default_queue_size;
        QueueDropStrategy strategy = default_strategy;
        EffectiveQueueForEdge(e, from_node_queue, default_queue_size, default_strategy, qsize, strategy);
        auto queue = std::make_shared<SpscQueue<FramePtr>>(qsize, strategy);
        from_it->second->context.output_queues.push_back(queue);
        
        if (!to_it->second->context.input_queue) {
            to_it->second->context.input_queue = queue;
        } else {
             err = "Node " + e.to + " has multiple inputs; only 1 is supported";
             return false;
        }

        enabled_edges.emplace_back(e.from, e.to);
        edges_.push_back(EdgeEntry{e.from, e.to, queue});
    }

    // Role validation
    for (auto& entry : nodes_) {
        if (!entry.enabled) continue;

        if (entry.role == "source" && entry.context.input_queue) {
            err = "Source node " + entry.id + " cannot have input";
            return false;
        }
        if (entry.role == "sink" && !entry.context.output_queues.empty()) {
            err = "Sink node " + entry.id + " cannot have output";
            return false;
        }
        if ((entry.role == "filter" || entry.role == "sink") && !entry.context.input_queue) {
            err = "Node " + entry.id + " role=" + entry.role + " must have input";
            return false;
        }
    }

    // Cycle detection on enabled nodes.
    {
        std::map<std::string, int> indeg;
        std::map<std::string, std::vector<std::string>> adj;
        for (const auto& n : nodes_) {
            if (!n.enabled) continue;
            indeg[n.id] = 0;
        }
        for (const auto& pr : enabled_edges) {
            adj[pr.first].push_back(pr.second);
            indeg[pr.second] += 1;
        }
        std::deque<std::string> q;
        for (const auto& kv : indeg) {
            if (kv.second == 0) q.push_back(kv.first);
        }
        size_t visited = 0;
        while (!q.empty()) {
            auto cur = q.front();
            q.pop_front();
            ++visited;
            for (const auto& nxt : adj[cur]) {
                auto it = indeg.find(nxt);
                if (it == indeg.end()) continue;
                if (--it->second == 0) q.push_back(nxt);
            }
        }
        if (visited != indeg.size()) {
            err = "Graph contains a cycle (DAG required)";
            return false;
        }

    }


    // Instantiation
    for (auto& entry : nodes_) {
        if (!entry.enabled) continue;
        std::string load_err;
        entry.node = loader.Create(entry.type, load_err);
        if (!entry.node) {
            err = load_err;
            return false;
        }
        if (!entry.node->Init(entry.config, entry.context)) {
            err = "Init failed for node " + entry.id;
            return false;
        }
    }

    return true;
}

bool Graph::TryUpdateInPlace(const SimpleJson& new_graph_cfg, size_t default_queue_size,
                             QueueDropStrategy default_strategy, std::string& err) {
    err.clear();
    if (!new_graph_cfg.IsObject()) {
        err = "new graph config is not object";
        return false;
    }

    const SimpleJson* new_nodes = new_graph_cfg.Find("nodes");
    const SimpleJson* new_edges = new_graph_cfg.Find("edges");
    const SimpleJson* old_nodes = graph_cfg_.Find("nodes");
    const SimpleJson* old_edges = graph_cfg_.Find("edges");
    if (!new_nodes || !new_edges || !old_nodes || !old_edges) {
        err = "graph missing nodes/edges";
        return false;
    }

    // If topology changed, require rebuild.
    std::vector<ParsedEdge> pe_new;
    std::vector<ParsedEdge> pe_old;
    std::string parse_err;
    if (!ParseEdges(*new_edges, pe_new, parse_err) || !ParseEdges(*old_edges, pe_old, parse_err)) {
        err = parse_err;
        return false;
    }

    auto build_edge_map = [](const std::vector<ParsedEdge>& pes, std::map<EdgeKey, ParsedEdge>& out) {
        out.clear();
        for (const auto& e : pes) {
            out[{e.from, e.to}] = e;
        }
    };
    std::map<EdgeKey, ParsedEdge> m_new;
    std::map<EdgeKey, ParsedEdge> m_old;
    build_edge_map(pe_new, m_new);
    build_edge_map(pe_old, m_old);
    if (m_new.size() != m_old.size()) {
        return false; // rebuild
    }

    // Build node config maps by id.
    std::map<std::string, SimpleJson> new_node_cfg;
    std::map<std::string, SimpleJson> old_node_cfg;
    std::map<std::string, bool> new_enabled;
    std::map<std::string, bool> old_enabled;
    std::map<std::string, std::string> new_type;
    std::map<std::string, std::string> old_type;

    for (const auto& nv : new_nodes->AsArray()) {
        if (!nv.IsObject()) continue;
        std::string id = nv.ValueOr<std::string>("id", "");
        if (id.empty()) continue;
        new_node_cfg[id] = nv;
        new_enabled[id] = nv.ValueOr<bool>("enable", true);
        new_type[id] = nv.ValueOr<std::string>("type", "");
    }
    for (const auto& ov : old_nodes->AsArray()) {
        if (!ov.IsObject()) continue;
        std::string id = ov.ValueOr<std::string>("id", "");
        if (id.empty()) continue;
        old_node_cfg[id] = ov;
        old_enabled[id] = ov.ValueOr<bool>("enable", true);
        old_type[id] = ov.ValueOr<std::string>("type", "");
    }

    if (new_node_cfg.size() != old_node_cfg.size()) {
        return false; // rebuild
    }
    for (const auto& kv : old_node_cfg) {
        auto it = new_node_cfg.find(kv.first);
        if (it == new_node_cfg.end()) return false; // rebuild
        if (new_type[kv.first] != old_type[kv.first]) return false; // rebuild
        if (new_enabled[kv.first] != old_enabled[kv.first]) return false; // rebuild
    }

    // Compare effective queue specs.
    for (const auto& kv : m_old) {
        auto itn = m_new.find(kv.first);
        if (itn == m_new.end()) return false;

        const auto& eo = kv.second;
        const auto& en = itn->second;

        const SimpleJson* from_old_queue = nullptr;
        const SimpleJson* from_new_queue = nullptr;
        if (auto fn = old_node_cfg.find(eo.from); fn != old_node_cfg.end()) {
            from_old_queue = fn->second.Find("queue");
        }
        if (auto fn = new_node_cfg.find(en.from); fn != new_node_cfg.end()) {
            from_new_queue = fn->second.Find("queue");
        }

        size_t osz = 0, nsz = 0;
        QueueDropStrategy ostrat = QueueDropStrategy::DropOldest;
        QueueDropStrategy nstrat = QueueDropStrategy::DropOldest;
        EffectiveQueueForEdge(eo, from_old_queue, built_default_queue_size_, built_default_strategy_, osz, ostrat);
        EffectiveQueueForEdge(en, from_new_queue, default_queue_size, default_strategy, nsz, nstrat);
        if (osz != nsz || ostrat != nstrat) {
            return false; // rebuild
        }
    }

    // In-place update: only apply for nodes whose full config changed.
    for (auto& entry : nodes_) {
        if (!entry.enabled || !entry.node) continue;
        auto it_new = new_node_cfg.find(entry.id);
        auto it_old = old_node_cfg.find(entry.id);
        if (it_new == new_node_cfg.end() || it_old == old_node_cfg.end()) return false;

        if (!JsonDeepEqual(it_new->second, it_old->second)) {
            if (!entry.node->UpdateConfig(it_new->second)) {
                // If any node cannot update in place, request rebuild.
                return false;
            }
            entry.config = it_new->second;
        }
    }

    graph_cfg_ = new_graph_cfg;
    built_default_queue_size_ = default_queue_size;
    built_default_strategy_ = default_strategy;
    return true;
}

namespace {

SimpleJson ReplaceNodeInGraphCfg(const SimpleJson& graph_cfg, const std::string& node_id,
                                const SimpleJson& new_node_cfg) {
    if (!graph_cfg.IsObject()) return graph_cfg;
    SimpleJson::Object obj = graph_cfg.AsObject();
    const SimpleJson* nodes = graph_cfg.Find("nodes");
    if (!nodes || !nodes->IsArray()) return graph_cfg;
    SimpleJson::Array out_nodes;
    out_nodes.reserve(nodes->AsArray().size());
    for (const auto& n : nodes->AsArray()) {
        if (n.IsObject() && n.ValueOr<std::string>("id", "") == node_id) {
            out_nodes.push_back(new_node_cfg);
        } else {
            out_nodes.push_back(n);
        }
    }
    obj["nodes"] = SimpleJson(std::move(out_nodes));
    return SimpleJson(std::move(obj));
}

}  // namespace

bool Graph::UpdateNodeConfig(const std::string& node_id, const SimpleJson& new_node_cfg, std::string& err) {
    err.clear();
    for (auto& entry : nodes_) {
        if (entry.id != node_id) continue;
        if (!entry.enabled || !entry.node) {
            err = "node not running or disabled: " + node_id;
            return false;
        }
        if (!entry.node->UpdateConfig(new_node_cfg)) {
            err = "UpdateConfig returned false";
            return false;
        }
        entry.config = new_node_cfg;
        graph_cfg_ = ReplaceNodeInGraphCfg(graph_cfg_, node_id, new_node_cfg);
        return true;
    }
    err = "node not found: " + node_id;
    return false;
}

bool Graph::Start() {
    bool expected = false;
    if (!running_.compare_exchange_strong(expected, true)) {
        return true; // Already running
    }

    stop_requested_.store(false);

    auto is_source_like = [](const NodeEntry& n) {
        return n.role == "source" || !n.context.input_queue;
    };

    // 1) Start non-source nodes first (they should only allocate resources).
    for (auto& entry : nodes_) {
        if (!entry.enabled || !entry.node) continue;
        if (is_source_like(entry)) continue;
        if (!entry.node->Start()) {
            LogError("[Graph] failed to start node: " + entry.id);
            Stop();
            return false;
        }
    }

    // 2) Start executor + attach callbacks before any source can push frames.
    executor_ = std::make_unique<Executor>(*this);
    if (!executor_->Start()) {
        LogError("[Graph] failed to start executor");
        Stop();
        return false;
    }
    executor_->AttachQueueCallbacks();

    // 3) Start sources last.
    for (auto& entry : nodes_) {
        if (!entry.enabled || !entry.node) continue;
        if (!is_source_like(entry)) continue;
        if (!entry.node->Start()) {
            LogError("[Graph] failed to start node: " + entry.id);
            Stop();
            return false;
        }
    }

    LogInfo("[Graph] started graph " + name_ + " with " + std::to_string(nodes_.size()) + " nodes");
    return true;
}

void Graph::Stop() {
    if (!running_.load()) {
        return;
    }

    stop_requested_.store(true);

    auto is_source_like = [](const NodeEntry& n) {
        return n.role == "source" || !n.context.input_queue;
    };

    // 1) Stop sources first to ensure no new data is produced.
    for (auto& n : nodes_) {
        if (!n.enabled || !n.node) continue;
        if (!is_source_like(n)) continue;

        n.node->Drain();
        n.node->Stop();
        for (auto& q : n.context.output_queues) q->Stop();
    }

    // 2) Stop all edge queues to unblock downstream nodes.
    for (auto& e : edges_) {
        if (e.queue) e.queue->Stop();
    }

    // 3) Stop executor (no concurrent Process() beyond this point).
    if (executor_) {
        executor_->DetachQueueCallbacks();
        executor_->Stop();
        executor_.reset();
    }

    // 4) Drain non-source nodes.
    for (auto& n : nodes_) {
        if (!n.enabled || !n.node) continue;
        if (is_source_like(n)) continue;
        n.node->Drain();
    }

    // 5) Stop remaining nodes (reverse order).
    for (auto it = nodes_.rbegin(); it != nodes_.rend(); ++it) {
        if (!it->enabled || !it->node) continue;
        if (is_source_like(*it)) continue;
        it->node->Stop();
    }

    running_.store(false);
}

namespace {

uint64_t NowEpochMs() {
    using namespace std::chrono;
    return static_cast<uint64_t>(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
}

QueueSnapshot ToQueueSnapshot(const SpscQueue<FramePtr>::Stats& st) {
    QueueSnapshot q;
    q.size = st.size;
    q.capacity = st.capacity;
    q.dropped_total = static_cast<uint64_t>(st.dropped);
    q.pushed_total = static_cast<uint64_t>(st.pushed);
    q.popped_total = static_cast<uint64_t>(st.popped);
    q.stopped = st.stopped;
    return q;
}

}  // namespace

GraphSnapshot Graph::Snapshot() const {
    GraphSnapshot snap;
    snap.name = name_;
    snap.running = running_.load();
    snap.timestamp_ms = NowEpochMs();

    const auto now_tp = std::chrono::steady_clock::now();
    double dt_sec = 0.0;
    {
        std::lock_guard<std::mutex> lock(rate_mu_);
        if (last_rate_tp_.time_since_epoch().count() != 0) {
            dt_sec = std::chrono::duration_cast<std::chrono::duration<double>>(now_tp - last_rate_tp_).count();
        }
        last_rate_tp_ = now_tp;
    }

    double total_fps = 0.0;
    for (const auto& n : nodes_) {
        if (!n.enabled || !n.node) continue;

        NodeSnapshot ns;
        ns.graph = name_;
        ns.id = n.id;
        ns.type = n.type;
        ns.role = n.role;
        ns.enabled = n.enabled;

        uint64_t in_popped = 0;
        if (n.context.input_queue) {
            auto st = n.context.input_queue->GetStats();
            ns.input_queue = ToQueueSnapshot(st);
            in_popped = ns.input_queue.popped_total;
        }

        uint64_t out_pushed = 0;
        for (const auto& oq : n.context.output_queues) {
            if (!oq) continue;
            out_pushed += static_cast<uint64_t>(oq->PushedCount());
        }

        {
            std::lock_guard<std::mutex> lock(rate_mu_);
            const uint64_t last_in = last_node_in_popped_[n.id];
            const uint64_t last_out = last_node_out_pushed_[n.id];
            if (dt_sec > 0.0) {
                if (in_popped >= last_in) ns.input_fps = static_cast<double>(in_popped - last_in) / dt_sec;
                if (out_pushed >= last_out) ns.output_fps = static_cast<double>(out_pushed - last_out) / dt_sec;
            }
            last_node_in_popped_[n.id] = in_popped;
            last_node_out_pushed_[n.id] = out_pushed;
        }

        if (n.metrics) {
            ns.ok_total = n.metrics->ok_total.load(std::memory_order_relaxed);
            ns.drop_total = n.metrics->drop_total.load(std::memory_order_relaxed);
            ns.error_total = n.metrics->error_total.load(std::memory_order_relaxed);
        }
        const uint64_t proc_cnt = ns.ok_total + ns.drop_total + ns.error_total;
        const uint64_t ns_total = n.metrics ? n.metrics->process_time_ns_total.load(std::memory_order_relaxed) : 0;
        if (proc_cnt > 0) {
            ns.avg_process_time_ms = (static_cast<double>(ns_total) / 1e6) / static_cast<double>(proc_cnt);
        }

        {
            SimpleJson cm;
            if (n.node && n.node->GetCustomMetrics(cm)) {
                ns.custom_metrics = cm;
                if (cm.IsObject()) {
                    if (const SimpleJson* a = cm.Find("alarm_total"); a && a->IsNumber()) {
                        snap.alarm_total += static_cast<uint64_t>(a->AsNumber(0.0));
                    }
                    if (const SimpleJson* c = cm.Find("clients"); c && c->IsNumber()) {
                        snap.publish_clients += static_cast<uint64_t>(c->AsNumber(0.0));
                    }
                }
            }
        }

        const bool source_like = (n.role == "source") || !n.context.input_queue;
        if (source_like) {
            total_fps += ns.output_fps;
        }

        snap.nodes.push_back(std::move(ns));
    }

    for (const auto& e : edges_) {
        if (!e.queue) continue;
        EdgeSnapshot es;
        es.from = e.from;
        es.to = e.to;
        const auto st = e.queue->GetStats();
        es.queue = ToQueueSnapshot(st);
        const std::string key = e.from + "->" + e.to;

        {
            std::lock_guard<std::mutex> lock(rate_mu_);
            const uint64_t last_pushed = last_edge_pushed_[key];
            const uint64_t last_popped = last_edge_popped_[key];
            if (dt_sec > 0.0) {
                if (es.queue.pushed_total >= last_pushed) {
                    es.queue.pushed_fps = static_cast<double>(es.queue.pushed_total - last_pushed) / dt_sec;
                }
                if (es.queue.popped_total >= last_popped) {
                    es.queue.popped_fps = static_cast<double>(es.queue.popped_total - last_popped) / dt_sec;
                }
            }
            last_edge_pushed_[key] = es.queue.pushed_total;
            last_edge_popped_[key] = es.queue.popped_total;
        }

        snap.edges.push_back(std::move(es));
    }

    snap.total_fps = total_fps;
    return snap;
}

bool Graph::FindNodeSnapshotById(const std::string& node_id, NodeSnapshot& out) const {
    auto snap = Snapshot();
    for (auto& n : snap.nodes) {
        if (n.id == node_id) {
            out = std::move(n);
            return true;
        }
    }
    return false;
}

GraphManager::GraphManager(std::string plugin_dir)
    : loader_(std::move(plugin_dir)) {}

GraphManager::~GraphManager() { StopAll(); }

bool GraphManager::LoadConfigFile(const std::string& path, SimpleJson& out, std::string& err) {
    std::ifstream ifs(path);
    if (!ifs.is_open()) {
        err = "Failed to open config: " + path;
        return false;
    }
    std::string content((std::istreambuf_iterator<char>(ifs)), std::istreambuf_iterator<char>());
    return ParseSimpleJson(content, out, err);
}

bool GraphManager::Build(const SimpleJson& root_cfg, std::string& err) {
    SimpleJson expanded;
    if (!ExpandRootConfig(root_cfg, expanded, err)) {
        return false;
    }

    if (!ValidateExpandedRootConfig(expanded, err)) {
        return false;
    }

    auto graphs_it = expanded.AsObject().find("graphs");
    if (graphs_it == expanded.AsObject().end() || !graphs_it->second.IsArray()) {
        err = "Root config missing 'graphs' array";
        return false;
    }

    size_t default_queue_size = 8;
    QueueDropStrategy default_strategy = QueueDropStrategy::DropOldest;
    if (const auto* queue_cfg = expanded.Find("queue")) {
        if (queue_cfg->IsObject()) {
            default_queue_size = static_cast<size_t>(queue_cfg->ValueOr<int>("size", 8));
            std::string strategy = queue_cfg->ValueOr<std::string>("strategy", "drop_oldest");
            default_strategy = ParseDropStrategy(strategy, default_strategy);
        }
    }

    // Apply plugin_path only on initial Build (safe: no nodes exist yet).
    if (const SimpleJson* g = expanded.Find("global")) {
        const std::string plugin_path = g->ValueOr<std::string>("plugin_path", "");
        if (!plugin_path.empty()) {
            loader_.SetPluginDir(plugin_path);
        }

        const std::string log_level = g->ValueOr<std::string>("log_level", "");
        if (!log_level.empty()) {
            LogLevel lvl;
            if (ParseLogLevel(log_level, lvl)) {
                Logger::Instance().SetLevel(lvl);
            }
        }
    }

    std::lock_guard<std::mutex> lock(graphs_mu_);
    graphs_.clear();

    for (const auto& graph_val : graphs_it->second.AsArray()) {
        if (!graph_val.IsObject()) {
            err = "Graph entry is not object";
            return false;
        }
        std::string name = graph_val.ValueOr<std::string>("name", "noname");
        auto graph = std::make_unique<Graph>(name);
        if (!graph->Build(graph_val, loader_, default_queue_size, default_strategy, err)) {
            return false;
        }
        graphs_.push_back(std::move(graph));
    }

    last_good_source_root_ = root_cfg;
    last_good_expanded_root_ = expanded;
    default_queue_size_ = default_queue_size;
    default_strategy_ = default_strategy;

    return true;
}

bool GraphManager::BuildFromFile(const std::string& path, std::string& err) {
    config_path_ = path;
    SimpleJson root_cfg;
    if (!LoadConfigFile(path, root_cfg, err)) {
        return false;
    }
    return Build(root_cfg, err);
}

bool GraphManager::StartAll() {
    std::scoped_lock lock(mu_, graphs_mu_);
    if (running_) return true;

    // Start all graphs; on failure, stop any already started graphs so we never leave a partially-running state.
    std::vector<Graph*> started;
    started.reserve(graphs_.size());
    for (auto& g : graphs_) {
        if (!g) continue;
        if (!g->Start()) {
            for (auto* sg : started) {
                if (sg) sg->Stop();
            }
            return false;
        }
        started.push_back(g.get());
    }

    running_ = true;
    return true;
}

void GraphManager::StopAll() {
    {
        std::scoped_lock lock(mu_, graphs_mu_);
        if (!running_) return;
        running_ = false;
        for (auto& g : graphs_) {
            if (g) g->Stop();
        }
    }
    cv_.notify_all();
}

void GraphManager::RequestStop() {
    StopAll();
}

void GraphManager::BlockUntilStop() {
    std::unique_lock<std::mutex> lock(mu_);
    cv_.wait(lock, [&] { return !running_; });
}

bool GraphManager::ReloadFromFile(const std::string& path, std::string& err) {
    if (config_path_.empty()) {
        config_path_ = path;
    }

    SimpleJson root_cfg;
    if (!LoadConfigFile(path, root_cfg, err)) {
        return false;
    }

    // Keep the original source config (templates/instances preserved).
    const SimpleJson source_root = root_cfg;

    SimpleJson expanded;
    if (!ExpandRootConfig(root_cfg, expanded, err)) {
        return false;
    }

    if (!ValidateExpandedRootConfig(expanded, err)) {
        return false;
    }

    auto graphs_it = expanded.AsObject().find("graphs");
    if (graphs_it == expanded.AsObject().end() || !graphs_it->second.IsArray()) {
        err = "Root config missing 'graphs' array";
        return false;
    }

    size_t new_default_queue_size = 8;
    QueueDropStrategy new_default_strategy = QueueDropStrategy::DropOldest;
    if (const auto* queue_cfg = expanded.Find("queue")) {
        if (queue_cfg->IsObject()) {
            new_default_queue_size = static_cast<size_t>(queue_cfg->ValueOr<int>("size", 8));
            std::string strategy = queue_cfg->ValueOr<std::string>("strategy", "drop_oldest");
            new_default_strategy = ParseDropStrategy(strategy, new_default_strategy);
        }
    }

    const std::string new_plugin_path = [&]() -> std::string {
        if (const SimpleJson* g = expanded.Find("global")) {
            return g->ValueOr<std::string>("plugin_path", "");
        }
        return "";
    }();

    std::optional<LogLevel> new_log_level;
    if (const SimpleJson* g = expanded.Find("global")) {
        const std::string log_level = g->ValueOr<std::string>("log_level", "");
        if (!log_level.empty()) {
            LogLevel lvl;
            if (ParseLogLevel(log_level, lvl)) {
                new_log_level = lvl;
            }
        }
    }

    std::lock_guard<std::mutex> lock(graphs_mu_);

    const SimpleJson prev_last_good = last_good_expanded_root_;
    const size_t prev_default_queue_size = default_queue_size_;
    const QueueDropStrategy prev_default_strategy = default_strategy_;
    const std::string prev_plugin_dir = loader_.PluginDir();

    auto build_graphs_locked = [&](const SimpleJson& expanded_root, PluginLoader& loader,
                                   size_t def_q, QueueDropStrategy def_s,
                                   std::vector<std::unique_ptr<Graph>>& out_graphs,
                                   std::string& build_err) -> bool {
        out_graphs.clear();
        const SimpleJson* graphs = expanded_root.Find("graphs");
        if (!graphs || !graphs->IsArray()) {
            build_err = "Root config missing 'graphs' array";
            return false;
        }
        out_graphs.reserve(graphs->AsArray().size());
        for (const auto& gv : graphs->AsArray()) {
            if (!gv.IsObject()) {
                build_err = "Graph entry is not object";
                return false;
            }
            std::string name = gv.ValueOr<std::string>("name", "noname");
            auto graph = std::make_unique<Graph>(name);
            if (!graph->Build(gv, loader, def_q, def_s, build_err)) {
                return false;
            }
            out_graphs.push_back(std::move(graph));
        }
        return true;
    };

    auto start_graphs_locked = [&](std::vector<std::unique_ptr<Graph>>& gs, std::string& start_err) -> bool {
        for (auto& g : gs) {
            if (!g) continue;
            if (!g->Start()) {
                start_err = "Failed to start graph: " + g->Name();
                return false;
            }
        }
        return true;
    };

    auto stop_all_locked = [&]() {
        for (auto& g : graphs_) {
            if (g) g->Stop();
        }
    };

    auto recover_locked = [&](std::string& recover_err) -> bool {
        stop_all_locked();
        graphs_.clear();

        if (!prev_plugin_dir.empty() && prev_plugin_dir != loader_.PluginDir()) {
            loader_.SetPluginDir(prev_plugin_dir);
        }

        std::vector<std::unique_ptr<Graph>> recovered;
        std::string berr;
        if (!build_graphs_locked(prev_last_good, loader_, prev_default_queue_size, prev_default_strategy, recovered,
                                 berr)) {
            recover_err = "Recovery build failed: " + berr;
            return false;
        }
        std::string serr;
        if (!start_graphs_locked(recovered, serr)) {
            recover_err = "Recovery start failed: " + serr;
            for (auto& gg : recovered) {
                if (gg) gg->Stop();
            }
            return false;
        }
        graphs_ = std::move(recovered);
        default_queue_size_ = prev_default_queue_size;
        default_strategy_ = prev_default_strategy;
        // last_good_source_root_/last_good_expanded_root_ remain unchanged.
        return true;
    };

    const bool plugin_dir_change = (!new_plugin_path.empty() && new_plugin_path != loader_.PluginDir());
    if (plugin_dir_change) {
        PluginLoader staged_loader(new_plugin_path);
        std::vector<std::unique_ptr<Graph>> staged_graphs;
        std::string berr;
        if (!build_graphs_locked(expanded, staged_loader, new_default_queue_size, new_default_strategy, staged_graphs,
                                 berr)) {
            err = berr;
            return false;
        }

        // Switch window: allow short downtime, but must be recoverable.
        stop_all_locked();
        graphs_.clear();

        PluginLoader old_loader = std::move(loader_);
        loader_ = std::move(staged_loader);
        graphs_ = std::move(staged_graphs);

        std::string serr;
        if (!start_graphs_locked(graphs_, serr)) {
            err = "Failed to start after plugin_path switch: " + serr;

            // Stop partially started graphs before recovery.
            stop_all_locked();
            graphs_.clear();

            loader_ = std::move(old_loader);
            std::string rerr;
            if (!recover_locked(rerr)) {
                err += "; recovery failed: " + rerr;
                return false;
            }
            return false;
        }

        last_good_source_root_ = source_root;
        last_good_expanded_root_ = expanded;
        default_queue_size_ = new_default_queue_size;
        default_strategy_ = new_default_strategy;
        if (new_log_level) {
            Logger::Instance().SetLevel(*new_log_level);
        }

        return true;
    }

    auto find_graph_index_locked = [&](const std::string& name, size_t& out_idx) -> bool {
        for (size_t i = 0; i < graphs_.size(); ++i) {
            if (graphs_[i] && graphs_[i]->Name() == name) {
                out_idx = i;
                return true;
            }
        }
        return false;
    };

    // Track graphs referenced by new config.
    std::set<std::string> seen;

    // Stage graphs that require rebuild or are newly added.
    std::map<std::string, std::unique_ptr<Graph>> staged;

    for (const auto& graph_val : graphs_it->second.AsArray()) {
        if (!graph_val.IsObject()) {
            err = "Graph entry is not object";
            return false;
        }
        std::string name = graph_val.ValueOr<std::string>("name", "noname");
        seen.insert(name);

        size_t idx = 0;
        if (!find_graph_index_locked(name, idx)) {
            // New graph: stage build (do not start until we have stopped removed graphs).
            auto graph = std::make_unique<Graph>(name);
            if (!graph->Build(graph_val, loader_, new_default_queue_size, new_default_strategy, err)) {
                return false;
            }
            staged[name] = std::move(graph);
            continue;
        }

        auto& g = graphs_[idx];
        std::string upd_err;
        if (g->TryUpdateInPlace(graph_val, new_default_queue_size, new_default_strategy, upd_err)) {
            continue;
        }
        if (!upd_err.empty()) {
            err = "UpdateConfig failed for graph " + name + ": " + upd_err;
            return false;
        }

        auto graph = std::make_unique<Graph>(name);
        if (!graph->Build(graph_val, loader_, new_default_queue_size, new_default_strategy, err)) {
            return false;
        }
        staged[name] = std::move(graph);
    }

    // Stop and remove graphs not present anymore (may free resources needed by staged graphs).
    for (auto itg = graphs_.begin(); itg != graphs_.end();) {
        if (*itg && !seen.count((*itg)->Name())) {
            (*itg)->Stop();
            itg = graphs_.erase(itg);
        } else {
            ++itg;
        }
    }

    // Apply staged graphs.
    for (auto& kv : staged) {
        const std::string& name = kv.first;
        auto& new_g = kv.second;
        if (!new_g) continue;

        size_t idx = 0;
        if (find_graph_index_locked(name, idx)) {
            graphs_[idx]->Stop();
            if (!new_g->Start()) {
                err = "Failed to start rebuilt graph: " + name;
                std::string rerr;
                if (!recover_locked(rerr)) {
                    err += "; recovery failed: " + rerr;
                    return false;
                }
                return false;
            }
            graphs_[idx] = std::move(new_g);
        } else {
            if (!new_g->Start()) {
                err = "Failed to start new graph: " + name;
                std::string rerr;
                if (!recover_locked(rerr)) {
                    err += "; recovery failed: " + rerr;
                    return false;
                }
                return false;
            }
            graphs_.push_back(std::move(new_g));
        }
    }

    last_good_source_root_ = source_root;
    last_good_expanded_root_ = expanded;
    default_queue_size_ = new_default_queue_size;
    default_strategy_ = new_default_strategy;

    if (new_log_level) {
        Logger::Instance().SetLevel(*new_log_level);
    }

    return true;
}

bool GraphManager::UpdateNodeConfig(const std::string& node_id, const std::optional<std::string>& graph,
                                   const SimpleJson& new_node_cfg, std::string& err) {
    std::lock_guard<std::mutex> lock(graphs_mu_);
    if (graph && !graph->empty()) {
        for (const auto& g : graphs_) {
            if (!g || g->Name() != *graph) continue;
            return g->UpdateNodeConfig(node_id, new_node_cfg, err);
        }
        err = "graph not found: " + *graph;
        return false;
    }

    size_t hits = 0;
    Graph* found = nullptr;
    for (const auto& g : graphs_) {
        if (!g) continue;
        NodeSnapshot tmp;
        if (g->FindNodeSnapshotById(node_id, tmp)) {
            found = g.get();
            ++hits;
            if (hits > 1) break;
        }
    }
    if (hits == 0 || !found) {
        err = "node not found: " + node_id;
        return false;
    }
    if (hits > 1) {
        err = "node id not unique, specify ?graph=<name>";
        return false;
    }
    return found->UpdateNodeConfig(node_id, new_node_cfg, err);
}

std::vector<GraphSnapshot> GraphManager::ListGraphSnapshots() {
    std::vector<GraphSnapshot> out;
    std::lock_guard<std::mutex> lock(graphs_mu_);
    out.reserve(graphs_.size());
    for (const auto& g : graphs_) {
        if (!g) continue;
        out.push_back(g->Snapshot());
    }
    return out;
}

bool GraphManager::GetGraphSnapshot(const std::string& name, GraphSnapshot& out, std::string& err) {
    std::lock_guard<std::mutex> lock(graphs_mu_);
    for (const auto& g : graphs_) {
        if (g && g->Name() == name) {
            out = g->Snapshot();
            return true;
        }
    }
    err = "graph not found: " + name;
    return false;
}

bool GraphManager::GetNodeSnapshot(const std::string& node_id, const std::optional<std::string>& graph,
                                  NodeSnapshot& out, std::string& err) {
    std::lock_guard<std::mutex> lock(graphs_mu_);
    if (graph && !graph->empty()) {
        for (const auto& g : graphs_) {
            if (!g || g->Name() != *graph) continue;
            if (g->FindNodeSnapshotById(node_id, out)) return true;
            err = "node not found: " + node_id + " in graph " + *graph;
            return false;
        }
        err = "graph not found: " + *graph;
        return false;
    }

    // Auto-match when unique.
    size_t hits = 0;
    for (const auto& g : graphs_) {
        if (!g) continue;
        NodeSnapshot tmp;
        if (g->FindNodeSnapshotById(node_id, tmp)) {
            out = std::move(tmp);
            ++hits;
            if (hits > 1) break;
        }
    }
    if (hits == 1) return true;
    if (hits == 0) {
        err = "node not found: " + node_id;
        return false;
    }
    err = "node id is not unique; specify ?graph=<name>";
    return false;
}

// New Result-based API implementations

Result<SimpleJson> GraphManager::LoadConfig(const std::string& path) {
    SimpleJson out;
    std::string err;
    if (!LoadConfigFile(path, out, err)) {
        return Error(err);
    }
    return out;
}

Status GraphManager::BuildFromConfig(const SimpleJson& root_cfg) {
    std::string err;
    if (!Build(root_cfg, err)) {
        return Status::Fail(err);
    }
    return Status::Ok();
}

Status GraphManager::BuildFromPath(const std::string& path) {
    std::string err;
    if (!BuildFromFile(path, err)) {
        return Status::Fail(err);
    }
    return Status::Ok();
}

Status GraphManager::Reload(const std::string& path) {
    std::string err;
    if (!ReloadFromFile(path, err)) {
        return Status::Fail(err);
    }
    return Status::Ok();
}

Status GraphManager::SetNodeConfig(const std::string& node_id, const SimpleJson& new_node_cfg,
                                   const std::optional<std::string>& graph) {
    std::string err;
    if (!UpdateNodeConfig(node_id, graph, new_node_cfg, err)) {
        return Status::Fail(err);
    }
    return Status::Ok();
}

Result<GraphSnapshot> GraphManager::GetGraph(const std::string& name) {
    GraphSnapshot out;
    std::string err;
    if (!GetGraphSnapshot(name, out, err)) {
        return Error(err);
    }
    return out;
}

Result<NodeSnapshot> GraphManager::GetNode(const std::string& node_id,
                                           const std::optional<std::string>& graph) {
    NodeSnapshot out;
    std::string err;
    if (!GetNodeSnapshot(node_id, graph, out, err)) {
        return Error(err);
    }
    return out;
}

}  // namespace rk3588