fix: correct HTTP request in fetch_target_info, add socket timeouts and debug logging

- Content-Length was wrong (20 vs actual 18 bytes)
- Removed running_ check from loop (was false during startup)
- Added SO_RCVTIMEO/SO_SNDTIMEO to prevent infinite recv hangs
- Added response size logging for debugging

CMakeLists.txt

@@ -12,7 +12,21 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
 
 option(DMF_BUILD_TESTS "Build tests" ON)
-option(DMF_BUILD_MXL_TOOLS "Build MXL tools (testsrc, sink)" OFF)
+option(DMF_BUILD_MXL_TOOLS "Build MXL tools (testsrc, sink)" ON)
+option(DMF_BUILD_DECKLINK "Build DeckLink I/O nodes" ON)
+option(DMF_BUILD_FABRICS "Build Fabrics bridge nodes" OFF)
+
+set(DECKLINK_SDK_DIR "" CACHE PATH "Path to Blackmagic DeckLink SDK root")
+
+if(DMF_BUILD_DECKLINK AND DECKLINK_SDK_DIR)
+    add_subdirectory(nodes/decklink-in)
+    add_subdirectory(nodes/decklink-out)
+endif()
+
+if(DMF_BUILD_FABRICS)
+    set(MXL_ENABLE_FABRICS_OFI ON CACHE BOOL "" FORCE)
+    add_subdirectory(nodes/fabric-bridge)
+endif()
 
 find_package(fmt CONFIG REQUIRED)
 find_package(spdlog CONFIG REQUIRED)
@@ -20,6 +34,13 @@ find_package(nlohmann_json CONFIG REQUIRED)
 find_package(Libwebsockets CONFIG REQUIRED)
 find_package(Catch2 CONFIG QUIET)
 
+set(BUILD_TESTS OFF CACHE BOOL "" FORCE)
+
+if(DMF_BUILD_MXL_TOOLS)
+    set(BUILD_TOOLS ON CACHE BOOL "" FORCE)
+    set(BUILD_UTILS ON CACHE BOOL "" FORCE)
+endif()
+
 add_subdirectory(extern/mxl)
 add_subdirectory(libs/dmf-node)
 add_subdirectory(libs/dmf-engine)

decklink_test.sh

@@ -0,0 +1,66 @@
+#!/bin/bash
+set -e
+
+ENGINE_PORT=${ENGINE_PORT:-9000}
+BASE_URL="http://127.0.0.1:${ENGINE_PORT}"
+DEVICE_INDEX=${DEVICE_INDEX:-0}
+MODE=${MODE:-1080i50}
+
+log() { echo "=== $1 ==="; }
+
+log "Adding decklink-in node (device=${DEVICE_INDEX}, mode=${MODE})"
+curl -s -X POST "${BASE_URL}/api/graph/nodes" \
+  -H "Content-Type: application/json" \
+  -d "{\"type\":\"decklink-in\",\"id\":\"sdi_in\",\"config\":{\"device_index\":${DEVICE_INDEX},\"mode\":\"${MODE}\"}}" | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Adding passthrough node"
+curl -s -X POST "${BASE_URL}/api/graph/nodes" \
+  -H "Content-Type: application/json" \
+  -d '{"type":"passthrough","id":"pass1"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Adding decklink-out node (device=${DEVICE_INDEX}, mode=${MODE})"
+curl -s -X POST "${BASE_URL}/api/graph/nodes" \
+  -H "Content-Type: application/json" \
+  -d "{\"type\":\"decklink-out\",\"id\":\"sdi_out\",\"config\":{\"device_index\":${DEVICE_INDEX},\"mode\":\"${MODE}\"}}" | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Connecting sdi_in → pass1"
+curl -s -X POST "${BASE_URL}/api/graph/edges" \
+  -H "Content-Type: application/json" \
+  -d '{"from_node":"sdi_in","from_port":"video_out","to_node":"pass1","to_port":"video_in"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Connecting pass1 → sdi_out"
+curl -s -X POST "${BASE_URL}/api/graph/edges" \
+  -H "Content-Type: application/json" \
+  -d '{"from_node":"pass1","from_port":"video_out","to_node":"sdi_out","to_port":"video_in"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Starting graph"
+curl -s -X POST "${BASE_URL}/api/graph/start" | python3 -m json.tool 2>/dev/null || echo ""
+
+sleep 3
+
+log "Status: sdi_in"
+curl -s -X POST "${BASE_URL}/api/graph/nodes/sdi_in/command" \
+  -H "Content-Type: application/json" \
+  -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Status: pass1"
+curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/command" \
+  -H "Content-Type: application/json" \
+  -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Status: sdi_out"
+curl -s -X POST "${BASE_URL}/api/graph/nodes/sdi_out/command" \
+  -H "Content-Type: application/json" \
+  -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Graph state"
+curl -s "${BASE_URL}/api/graph" | python3 -m json.tool 2>/dev/null || echo ""
+
+echo ""
+echo "=== SDI pipeline running. Press Enter to stop. ==="
+read
+
+log "Stopping graph"
+curl -s -X POST "${BASE_URL}/api/graph/stop" | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Done"

extern/mxl

@@ -1 +1 @@
-/home/itten/DMF/mxl
+../mxl

libs/dmf-engine/include/dmf-engine/node_control_client.hpp

@@ -9,6 +9,7 @@ namespace dmf_engine {
 class NodeControlClient {
 public:
     bool send_command(uint16_t port, const std::string& json_cmd);
+    std::string send_command_with_response(uint16_t port, const std::string& json_cmd);
 
     void register_node(const std::string& node_id, uint16_t port);
     void unregister_node(const std::string& node_id);

libs/dmf-engine/src/api_server.cpp

@@ -12,6 +12,7 @@
 
 #include <cstring>
 #include <string>
+#include <thread>
 
 namespace dmf_engine {
 
@@ -178,7 +179,7 @@ static void handle_request(const std::string& method, const std::string& path,
             }
         } else if (path == "/api/graph/start" && method == "POST") {
             auto nodes = graph.get_nodes();
-            uint16_t port = 9100;
+            uint16_t port = g_impl->port + 100;
             for (auto& node : nodes) {
                 pm.start_node(const_cast<GraphNode&>(node), g_impl->mxl_domain, port);
                 cc.register_node(node.id, port);
@@ -254,8 +255,29 @@ static void handle_request(const std::string& method, const std::string& path,
                 return;
             }
 
+            int fps_num = 50, fps_den = 1;
+            int width = 1920, height = 1080;
+
+            nlohmann::json status_cmd;
+            status_cmd["cmd"] = "status";
+            auto status_resp = cc.send_command_with_response(port_num, status_cmd.dump());
+            if (!status_resp.empty()) {
+                try {
+                    auto sr = nlohmann::json::parse(status_resp);
+                    if (sr.contains("data")) {
+                        auto& d = sr["data"];
+                        if (d.contains("grain_rate")) {
+                            fps_num = d["grain_rate"].value("numerator", fps_num);
+                            fps_den = d["grain_rate"].value("denominator", fps_den);
+                        }
+                        width = d.value("width", width);
+                        height = d.value("height", height);
+                    }
+                } catch (...) {}
+            }
+
             auto flow_id = fm.create_flow_id();
-            auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1);
+            auto flow_def = fm.create_v210_flow_def(flow_id, width, height, fps_num, fps_den);
 
             nlohmann::json cmd;
             cmd["cmd"] = "add_writer";
@@ -299,6 +321,172 @@ static void handle_request(const std::string& method, const std::string& path,
             } else {
                 error_resp(500, "Failed to send command to node");
             }
+        } else if (path.find("/api/graph/nodes/") == 0 && path.find("/register") != std::string::npos && method == "POST") {
+            auto prefix = std::string("/api/graph/nodes/");
+            auto suffix_start = path.find("/register");
+            auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
+
+            if (!req_body.contains("port")) {
+                error_resp(400, "Missing port");
+                return;
+            }
+
+            auto control_port = static_cast<uint16_t>(req_body["port"].get<int>());
+            cc.register_node(node_id, control_port);
+
+            auto* node = graph.get_node_mut(node_id);
+            if (node) {
+                node->control_port = control_port;
+                node->state = NodeState::Running;
+            }
+
+            ok({{"node_id", node_id}, {"control_port", control_port}});
+        } else if (path == "/api/fabric/create-target" && method == "POST") {
+            if (!req_body.contains("node_id")) {
+                error_resp(400, "Missing node_id");
+                return;
+            }
+
+            auto node_id = req_body["node_id"].get<std::string>();
+            auto port_id = req_body.value("port_id", "video_out");
+            auto flow_id = req_body.value("flow_id", fm.create_flow_id());
+
+            auto port_num = cc.get_port(node_id);
+            if (port_num == 0) {
+                error_resp(404, "Node not running: " + node_id);
+                return;
+            }
+
+            int fps_num = 25, fps_den = 1;
+            int width = 1920, height = 1080;
+
+            nlohmann::json status_cmd;
+            status_cmd["cmd"] = "status";
+            auto status_resp = cc.send_command_with_response(port_num, status_cmd.dump());
+            if (!status_resp.empty()) {
+                try {
+                    auto sr = nlohmann::json::parse(status_resp);
+                    if (sr.contains("data")) {
+                        auto& d = sr["data"];
+                        if (d.contains("grain_rate")) {
+                            fps_num = d["grain_rate"].value("numerator", fps_num);
+                            fps_den = d["grain_rate"].value("denominator", fps_den);
+                        }
+                        width = d.value("width", width);
+                        height = d.value("height", height);
+                    }
+                } catch (...) {}
+            }
+
+            auto flow_def = fm.create_v210_flow_def(flow_id, width, height, fps_num, fps_den);
+
+            nlohmann::json cmd;
+            cmd["cmd"] = "add_writer";
+            cmd["port_id"] = port_id;
+            cmd["flow_id"] = flow_id;
+            cmd["flow_def"] = flow_def;
+            if (!cc.send_command(port_num, cmd.dump())) {
+                error_resp(500, "Failed to add writer to target node");
+                return;
+            }
+
+            std::string target_info;
+            constexpr int max_poll_attempts = 30;
+            for (int i = 0; i < max_poll_attempts; ++i) {
+                auto resp = cc.send_command_with_response(port_num, status_cmd.dump());
+                if (!resp.empty()) {
+                    try {
+                        auto sr = nlohmann::json::parse(resp);
+                        if (sr.contains("data") && sr["data"].contains("target_info") && !sr["data"]["target_info"].get<std::string>().empty()) {
+                            target_info = sr["data"]["target_info"].get<std::string>();
+                            break;
+                        }
+                    } catch (...) {}
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            }
+
+            if (target_info.empty()) {
+                error_resp(504, "Target node did not produce target_info in time");
+                return;
+            }
+
+            ok({
+                {"node_id", node_id},
+                {"flow_id", flow_id},
+                {"target_info", target_info}
+            });
+        } else if (path == "/api/fabric/connect" && method == "POST") {
+            if (!req_body.contains("target_node_id") || !req_body.contains("initiator_node_id")) {
+                error_resp(400, "Missing target_node_id/initiator_node_id");
+                return;
+            }
+
+            auto target_node_id = req_body["target_node_id"].get<std::string>();
+            auto initiator_node_id = req_body["initiator_node_id"].get<std::string>();
+
+            auto target_port = cc.get_port(target_node_id);
+            auto initiator_port = cc.get_port(initiator_node_id);
+
+            if (target_port == 0) {
+                error_resp(404, "Target node not running: " + target_node_id);
+                return;
+            }
+            if (initiator_port == 0) {
+                error_resp(404, "Initiator node not running: " + initiator_node_id);
+                return;
+            }
+
+            nlohmann::json status_cmd;
+            status_cmd["cmd"] = "status";
+            std::string target_info;
+            constexpr int max_poll_attempts = 30;
+            for (int i = 0; i < max_poll_attempts; ++i) {
+                auto resp = cc.send_command_with_response(target_port, status_cmd.dump());
+                if (!resp.empty()) {
+                    try {
+                        auto sr = nlohmann::json::parse(resp);
+                        if (sr.contains("data") && sr["data"].contains("target_info") && !sr["data"]["target_info"].get<std::string>().empty()) {
+                            target_info = sr["data"]["target_info"].get<std::string>();
+                            break;
+                        }
+                    } catch (...) {}
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            }
+
+            if (target_info.empty()) {
+                error_resp(504, "Target node did not produce target_info in time");
+                return;
+            }
+
+            nlohmann::json configure_cmd;
+            configure_cmd["cmd"] = "configure";
+            configure_cmd["params"]["target_info"] = target_info;
+            if (!cc.send_command(initiator_port, configure_cmd.dump())) {
+                error_resp(500, "Failed to send target_info to initiator node");
+                return;
+            }
+
+            std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+            bool initiator_running = false;
+            auto init_resp = cc.send_command_with_response(initiator_port, status_cmd.dump());
+            if (!init_resp.empty()) {
+                try {
+                    auto ir = nlohmann::json::parse(init_resp);
+                    if (ir.contains("data") && ir["data"].contains("running")) {
+                        initiator_running = ir["data"]["running"].get<bool>();
+                    }
+                } catch (...) {}
+            }
+
+            ok({
+                {"target_node_id", target_node_id},
+                {"initiator_node_id", initiator_node_id},
+                {"target_info_length", target_info.size()},
+                {"initiator_running", initiator_running}
+            });
         } else if (path.find("/api/graph/nodes/") == 0 && path.find("/command") != std::string::npos && method == "POST") {
             auto prefix = std::string("/api/graph/nodes/");
             auto suffix_start = path.find("/command");

libs/dmf-engine/src/node_control_client.cpp

@@ -85,4 +85,60 @@ bool NodeControlClient::send_command(uint16_t port, const std::string& json_cmd)
     return ok;
 }
 
+std::string NodeControlClient::send_command_with_response(uint16_t port, const std::string& json_cmd) {
+    int fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (fd < 0) {
+        spdlog::error("NodeControlClient: socket() failed: {}", strerror(errno));
+        return "";
+    }
+
+    struct timeval tv;
+    tv.tv_sec = 2;
+    tv.tv_usec = 0;
+    setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+    setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+
+    struct sockaddr_in addr;
+    std::memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(port);
+    inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);
+
+    if (connect(fd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) < 0) {
+        spdlog::error("NodeControlClient: connect to port {} failed: {}", port, strerror(errno));
+        close(fd);
+        return "";
+    }
+
+    std::ostringstream req;
+    req << "POST /cmd HTTP/1.1\r\n"
+        << "Host: 127.0.0.1:" << port << "\r\n"
+        << "Content-Type: application/json\r\n"
+        << "Content-Length: " << json_cmd.size() << "\r\n"
+        << "Connection: close\r\n"
+        << "\r\n"
+        << json_cmd;
+
+    auto request = req.str();
+    auto sent = write(fd, request.data(), request.size());
+    if (sent != static_cast<ssize_t>(request.size())) {
+        spdlog::error("NodeControlClient: write failed on port {}", port);
+        close(fd);
+        return "";
+    }
+
+    std::string full_resp;
+    char resp_buf[4096];
+    while (true) {
+        auto n = read(fd, resp_buf, sizeof(resp_buf));
+        if (n <= 0) break;
+        full_resp.append(resp_buf, n);
+    }
+    close(fd);
+
+    auto body_start = full_resp.find("\r\n\r\n");
+    if (body_start == std::string::npos) return "";
+    return full_resp.substr(body_start + 4);
+}
+
 } // namespace dmf_engine

libs/dmf-node/include/dmf-node/control_server.hpp

@@ -11,7 +11,7 @@ struct lws;
 
 namespace dmf_node {
 
-using CommandHandler = std::function<void(const nlohmann::json& payload)>;
+using CommandHandler = std::function<nlohmann::json(const nlohmann::json& payload)>;
 using StatusCallback = std::function<void(const nlohmann::json& event)>;
 
 class ControlServer {

libs/dmf-node/include/dmf-node/node.hpp

@@ -9,6 +9,7 @@
 #include <nlohmann/json.hpp>
 
 #include <memory>
+#include <unordered_map>
 #include <vector>
 
 namespace dmf_node {
@@ -21,12 +22,25 @@ public:
     virtual std::vector<PortDef> input_ports() const = 0;
     virtual std::vector<PortDef> output_ports() const = 0;
     virtual void configure(const nlohmann::json& params) = 0;
-    virtual void on_add_writer(const std::string& port_id, mxlFlowWriter writer) = 0;
+    virtual void on_add_writer(const std::string& port_id, mxlFlowWriter writer) {}
-    virtual void on_add_reader(const std::string& port_id, mxlFlowReader reader) = 0;
+    virtual void on_add_reader(const std::string& port_id, mxlFlowReader reader) {}
     virtual void on_remove_writer(const std::string& port_id) = 0;
     virtual void on_remove_reader(const std::string& port_id) = 0;
     virtual void process() = 0;
     virtual nlohmann::json status() const = 0;
+
+    void set_mxl_instance(mxlInstance instance) { mxl_instance_ = instance; }
+    mxlInstance get_mxl_instance() const { return mxl_instance_; }
+
+    void set_writer_config(const std::string& port_id, const mxlFlowConfigInfo& config) { writer_configs_[port_id] = config; }
+    void set_reader_config(const std::string& port_id, const mxlFlowConfigInfo& config) { reader_configs_[port_id] = config; }
+    const mxlFlowConfigInfo* get_writer_config(const std::string& port_id) const { auto it = writer_configs_.find(port_id); return it != writer_configs_.end() ? &it->second : nullptr; }
+    const mxlFlowConfigInfo* get_reader_config(const std::string& port_id) const { auto it = reader_configs_.find(port_id); return it != reader_configs_.end() ? &it->second : nullptr; }
+
+private:
+    mxlInstance mxl_instance_ = nullptr;
+    std::unordered_map<std::string, mxlFlowConfigInfo> writer_configs_;
+    std::unordered_map<std::string, mxlFlowConfigInfo> reader_configs_;
 };
 
 } // namespace dmf_node

libs/dmf-node/src/control_server.cpp

@@ -29,17 +29,18 @@ struct ControlServer::Impl {
     struct lws_context* context = nullptr;
 };
 
-static void dispatch_command(ControlServerData* data, const nlohmann::json& msg) {
+static nlohmann::json dispatch_command(ControlServerData* data, const nlohmann::json& msg) {
     if (!msg.contains("cmd")) {
         spdlog::warn("Control: message missing 'cmd' field");
-        return;
+        return {{"error", "missing 'cmd' field"}};
     }
     auto cmd = msg["cmd"].get<std::string>();
     auto it = data->commands.find(cmd);
     if (it != data->commands.end()) {
-        it->second(msg);
+        return it->second(msg);
     } else {
         spdlog::warn("Control: unknown command '{}'", cmd);
+        return {{"error", "unknown command: " + cmd}};
     }
 }
 
@@ -102,8 +103,8 @@ static int callback_all(struct lws* wsi, enum lws_callback_reasons reason,
         }
         try {
             auto msg = nlohmann::json::parse(ps->http_body);
-            dispatch_command(ps->data, msg);
+            auto result = dispatch_command(ps->data, msg);
-            return send_http_json(wsi, "200 OK", R"({"ok":true})");
+            return send_http_json(wsi, "200 OK", result.dump());
         } catch (const nlohmann::json::parse_error& e) {
             return send_http_json(wsi, "400 Bad Request",
                                  nlohmann::json({{"error", e.what()}}).dump());

libs/dmf-node/src/node_runner.cpp

@@ -57,6 +57,7 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
     std::signal(SIGINT, signal_handler);
     std::signal(SIGTERM, signal_handler);
 
+    spdlog::flush_on(spdlog::level::info);
     spdlog::info("Starting node '{}' type='{}'", node_id_, node->type());
 
     if (!std::filesystem::exists(mxl_domain_)) {
@@ -71,6 +72,8 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
     }
     spdlog::info("MXL instance created on domain: {}", mxl_domain_);
 
+    node->set_mxl_instance(mxl_instance_);
+
     mxlGarbageCollectFlows(mxl_instance_);
 
     if (!config_str_.empty()) {
@@ -101,7 +104,7 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
     };
     std::vector<FlowResource> flow_resources;
 
-    control_server->register_command("add_writer", [&](const nlohmann::json& msg) {
+    control_server->register_command("add_writer", [&](const nlohmann::json& msg) -> nlohmann::json {
         auto flow_id = msg["flow_id"].get<std::string>();
         auto port_id = msg["port_id"].get<std::string>();
         auto flow_def = msg["flow_def"].dump();
@@ -112,14 +115,16 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
         auto status = mxlCreateFlowWriter(mxl_instance_, flow_def.c_str(), nullptr, &writer, &config_info, &created);
         if (status != MXL_STATUS_OK || !writer) {
             spdlog::error("Failed to create flow writer for flow {}: status={}", flow_id, static_cast<int>(status));
-            return;
+            return {{"ok", false}, {"error", "failed to create flow writer"}};
         }
-        spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created);
+spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created);
         flow_resources.push_back({port_id, writer, nullptr});
+        node->set_writer_config(port_id, config_info);
         node->on_add_writer(port_id, writer);
+        return {{"ok", true}, {"created", created}};
     });
 
-    control_server->register_command("add_reader", [&](const nlohmann::json& msg) {
+    control_server->register_command("add_reader", [&](const nlohmann::json& msg) -> nlohmann::json {
         auto flow_id = msg["flow_id"].get<std::string>();
         auto port_id = msg["port_id"].get<std::string>();
 
@@ -127,14 +132,18 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
         auto status = mxlCreateFlowReader(mxl_instance_, flow_id.c_str(), nullptr, &reader);
         if (status != MXL_STATUS_OK || !reader) {
             spdlog::error("Failed to create flow reader for flow {}: status={}", flow_id, static_cast<int>(status));
-            return;
+            return {{"ok", false}, {"error", "failed to create flow reader"}};
         }
-        spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id);
+spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id);
         flow_resources.push_back({port_id, nullptr, reader});
+        mxlFlowConfigInfo config{};
+        mxlFlowReaderGetConfigInfo(reader, &config);
+        node->set_reader_config(port_id, config);
         node->on_add_reader(port_id, reader);
+        return {{"ok", true}};
     });
 
-    control_server->register_command("remove_writer", [&](const nlohmann::json& msg) {
+    control_server->register_command("remove_writer", [&](const nlohmann::json& msg) -> nlohmann::json {
         auto port_id = msg["port_id"].get<std::string>();
         node->on_remove_writer(port_id);
         for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) {
@@ -145,9 +154,10 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
             }
         }
         spdlog::info("Removed writer on port '{}'", port_id);
+        return {{"ok", true}};
     });
 
-    control_server->register_command("remove_reader", [&](const nlohmann::json& msg) {
+    control_server->register_command("remove_reader", [&](const nlohmann::json& msg) -> nlohmann::json {
         auto port_id = msg["port_id"].get<std::string>();
         node->on_remove_reader(port_id);
         for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) {
@@ -158,26 +168,30 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
             }
         }
         spdlog::info("Removed reader on port '{}'", port_id);
+        return {{"ok", true}};
     });
 
-    control_server->register_command("configure", [&](const nlohmann::json& msg) {
+    control_server->register_command("configure", [&](const nlohmann::json& msg) -> nlohmann::json {
         if (msg.contains("params")) {
             node->configure(msg["params"]);
             spdlog::info("Reconfigured node '{}'", node_id_);
         }
+        return {{"ok", true}};
     });
 
-    control_server->register_command("status", [&](const nlohmann::json& /*msg*/) {
+    control_server->register_command("status", [&](const nlohmann::json& /*msg*/) -> nlohmann::json {
         nlohmann::json resp;
         resp["event"] = "status";
         resp["node_id"] = node_id_;
         resp["data"] = node->status();
         control_server->send_event(resp);
+        return resp;
     });
 
-    control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) {
+    control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) -> nlohmann::json {
         spdlog::info("Shutdown command received");
         g_running = false;
+        return {{"ok", true}};
     });
 
     nlohmann::json ready_event;

nodes/decklink-in/CMakeLists.txt

@@ -0,0 +1,21 @@
+cmake_minimum_required(VERSION 3.24)
+
+project(dmf-node-decklink-in LANGUAGES CXX)
+
+set(DECKLINK_INCLUDE "${DECKLINK_SDK_DIR}/Linux/include")
+
+add_executable(dmf-node-decklink-in
+    src/main.cpp
+    src/decklink_in_node.cpp
+    "${DECKLINK_SDK_DIR}/Linux/include/DeckLinkAPIDispatch.cpp"
+)
+
+target_include_directories(dmf-node-decklink-in PRIVATE
+    "${DECKLINK_INCLUDE}"
+)
+
+target_link_libraries(dmf-node-decklink-in PRIVATE
+    dmf-node
+)
+
+target_compile_options(dmf-node-decklink-in PRIVATE -Wno-unused-parameter)

nodes/decklink-in/src/decklink_in_node.cpp

@@ -0,0 +1,292 @@
+#include "decklink_in_node.hpp"
+
+#include <mxl/flow.h>
+#include <mxl/mxl.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPIConfiguration.h>
+
+#include <spdlog/spdlog.h>
+
+#include <cstring>
+
+namespace dmf_node {
+
+DeckLinkInNode::~DeckLinkInNode() {
+    close_device();
+}
+
+void DeckLinkInNode::configure(const nlohmann::json& params) {
+    if (params.contains("device_index")) {
+        device_index_ = params["device_index"].get<int>();
+    }
+    if (params.contains("mode")) {
+        auto mode_str = params["mode"].get<std::string>();
+        if (mode_str == "1080i50") display_mode_ = bmdModeHD1080i50;
+        else if (mode_str == "1080p50") display_mode_ = bmdModeHD1080p50;
+        else if (mode_str == "1080p25") display_mode_ = bmdModeHD1080p25;
+        else if (mode_str == "1080i5994") display_mode_ = bmdModeHD1080i5994;
+        else if (mode_str == "1080p5994") display_mode_ = bmdModeHD1080p5994;
+        else if (mode_str == "1080p2997") display_mode_ = bmdModeHD1080p2997;
+        else if (mode_str == "720p50") display_mode_ = bmdModeHD720p50;
+        else if (mode_str == "720p5994") display_mode_ = bmdModeHD720p5994;
+        else {
+            spdlog::warn("DeckLink-in: unknown mode '{}', defaulting to 1080i50", mode_str);
+        }
+    }
+    if (params.contains("input_connection")) {
+        auto conn_str = params["input_connection"].get<std::string>();
+        if (conn_str == "sdi") input_connection_ = bmdVideoConnectionSDI;
+        else if (conn_str == "hdmi") input_connection_ = bmdVideoConnectionHDMI;
+        else if (conn_str == "optical_sdi") input_connection_ = bmdVideoConnectionOpticalSDI;
+        else if (conn_str == "component") input_connection_ = bmdVideoConnectionComponent;
+        else if (conn_str == "composite") input_connection_ = bmdVideoConnectionComposite;
+        else if (conn_str == "svideo") input_connection_ = bmdVideoConnectionSVideo;
+        else {
+            spdlog::warn("DeckLink-in: unknown input_connection '{}', defaulting to auto", conn_str);
+        }
+    }
+
+    if (!open_device()) {
+        spdlog::error("DeckLink-in: failed to open device during configure");
+    }
+}
+
+void DeckLinkInNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
+    if (port_id == "video_out") {
+        writer_ = writer;
+
+        auto now = mxlGetTime();
+        write_index_ = mxlTimestampToIndex(&grain_rate_, now);
+
+        spdlog::info("DeckLink-in: writer added, grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
+
+        if (input_ && !capturing_) {
+            if (input_->StartStreams() != S_OK) {
+                spdlog::error("DeckLink-in: failed to start streams");
+                return;
+            }
+            capturing_ = true;
+            spdlog::info("DeckLink-in: capture started");
+        }
+    }
+}
+
+void DeckLinkInNode::on_remove_writer(const std::string& port_id) {
+    if (port_id == "video_out") {
+        close_device();
+        writer_.reset();
+        spdlog::info("DeckLink-in: writer removed");
+    }
+}
+
+bool DeckLinkInNode::open_device() {
+    auto* iter = CreateDeckLinkIteratorInstance();
+    if (!iter) {
+        spdlog::error("DeckLink-in: DeckLink drivers not found");
+        return false;
+    }
+
+    IDeckLink* device = nullptr;
+    for (int i = 0; i <= device_index_; ++i) {
+        if (iter->Next(&device) != S_OK) {
+            spdlog::error("DeckLink-in: device index {} not found", device_index_);
+            iter->Release();
+            return false;
+        }
+        if (i < device_index_) {
+            device->Release();
+        }
+    }
+    iter->Release();
+
+    const char* model_name = nullptr;
+    device->GetModelName(&model_name);
+    spdlog::info("DeckLink-in: opened device '{}'", model_name ? model_name : "unknown");
+
+    if (input_connection_ != bmdVideoConnectionUnspecified) {
+        IDeckLinkConfiguration* config = nullptr;
+        if (device->QueryInterface(IID_IDeckLinkConfiguration, (void**)&config) == S_OK && config) {
+            if (config->SetInt(bmdDeckLinkConfigVideoInputConnection, input_connection_) != S_OK) {
+                spdlog::warn("DeckLink-in: failed to set input connection");
+            } else {
+                spdlog::info("DeckLink-in: input connection configured");
+            }
+            config->Release();
+        }
+    }
+
+    if (device->QueryInterface(IID_IDeckLinkInput, (void**)&input_) != S_OK) {
+        spdlog::error("DeckLink-in: device has no input interface");
+        device->Release();
+        return false;
+    }
+
+    decklink_ = device;
+
+    callback_ = std::make_unique<CaptureCallback>(*this);
+    input_->SetCallback(callback_.get());
+
+    auto flags = bmdVideoInputEnableFormatDetection;
+    if (input_->EnableVideoInput(display_mode_, bmdFormat10BitYUV, flags) != S_OK) {
+        spdlog::error("DeckLink-in: failed to enable video input");
+        return false;
+    }
+
+    IDeckLinkDisplayMode* mode = nullptr;
+    if (input_->GetDisplayMode(display_mode_, &mode) == S_OK) {
+        frame_width_ = mode->GetWidth();
+        frame_height_ = mode->GetHeight();
+        BMDTimeValue duration = 0;
+        BMDTimeScale scale = 0;
+        mode->GetFrameRate(&duration, &scale);
+        mode->Release();
+
+        is_interlaced_ = (display_mode_ == bmdModeHD1080i50 ||
+                          display_mode_ == bmdModeHD1080i5994);
+
+        grain_rate_.numerator = static_cast<int32_t>(scale);
+        grain_rate_.denominator = static_cast<int32_t>(duration);
+
+        auto g = std::__gcd(grain_rate_.numerator, grain_rate_.denominator);
+        grain_rate_.numerator /= g;
+        grain_rate_.denominator /= g;
+
+        spdlog::info("DeckLink-in: {}x{} @ {}/{} fps, interlaced={}",
+                     frame_width_, frame_height_,
+                     grain_rate_.numerator, grain_rate_.denominator,
+                     is_interlaced_);
+    }
+
+    spdlog::info("DeckLink-in: device opened, waiting for writer to start capture");
+    return true;
+}
+
+void DeckLinkInNode::close_device() {
+    if (input_) {
+        if (capturing_) {
+            input_->StopStreams();
+        }
+        input_->DisableVideoInput();
+        input_->Release();
+        input_ = nullptr;
+    }
+    if (decklink_) {
+        decklink_->Release();
+        decklink_ = nullptr;
+    }
+    capturing_ = false;
+    callback_.reset();
+    spdlog::info("DeckLink-in: device closed");
+}
+
+void DeckLinkInNode::process() {
+    if (!writer_ || !capturing_) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        return;
+    }
+
+    auto deadline = mxlIndexToTimestamp(&grain_rate_, write_index_ + 1);
+    mxlSleepUntil(deadline);
+
+    void* src_data = nullptr;
+    long src_row_bytes = 0;
+    long src_width = 0;
+    long src_height = 0;
+
+    {
+        std::lock_guard<std::mutex> lock(frame_mutex_);
+        if (frame_data_) {
+            src_data = frame_data_;
+            src_row_bytes = frame_row_bytes_;
+            src_width = frame_width_;
+            src_height = frame_height_;
+        }
+        frame_ready_ = false;
+    }
+
+    if (!src_data) {
+        write_index_++;
+        return;
+    }
+
+    mxlGrainInfo out_grain{};
+    uint8_t* out_payload = nullptr;
+    auto status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload);
+    if (status != MXL_STATUS_OK) {
+        auto now = mxlGetTime();
+        auto current = mxlTimestampToIndex(&grain_rate_, now);
+        write_index_ = current + 1;
+        return;
+    }
+
+    auto dst_row_bytes = out_grain.grainSize / src_height;
+    auto copy_row_bytes = std::min(static_cast<long>(dst_row_bytes), src_row_bytes);
+
+    for (long y = 0; y < src_height && y < static_cast<long>(out_grain.grainSize / dst_row_bytes); ++y) {
+        std::memcpy(out_payload + y * dst_row_bytes,
+                    static_cast<uint8_t*>(src_data) + y * src_row_bytes,
+                    copy_row_bytes);
+    }
+
+    out_grain.validSlices = out_grain.totalSlices;
+    mxlFlowWriterCommitGrain(*writer_, &out_grain);
+
+    write_index_++;
+    grains_written_++;
+
+    if (grains_written_ == 1) {
+        spdlog::info("DeckLink-in: first grain written, index={}", write_index_ - 1);
+    }
+}
+
+nlohmann::json DeckLinkInNode::status() const {
+    return {
+        {"type", "decklink-in"},
+        {"grains_written", grains_written_},
+        {"write_index", write_index_},
+        {"capturing", capturing_.load()},
+        {"has_writer", writer_.has_value()},
+        {"grain_rate", {{"numerator", grain_rate_.numerator}, {"denominator", grain_rate_.denominator}}},
+        {"width", frame_width_},
+        {"height", frame_height_},
+        {"interlaced", is_interlaced_},
+    };
+}
+
+HRESULT DeckLinkInNode::CaptureCallback::VideoInputFormatChanged(
+    BMDVideoInputFormatChangedEvents, IDeckLinkDisplayMode* newMode, BMDDetectedVideoInputFormatFlags) {
+    spdlog::info("DeckLink-in: input format changed");
+    return S_OK;
+}
+
+HRESULT DeckLinkInNode::CaptureCallback::VideoInputFrameArrived(
+    IDeckLinkVideoInputFrame* videoFrame, IDeckLinkAudioInputPacket*) {
+    if (!videoFrame || (videoFrame->GetFlags() & bmdFrameHasNoInputSource)) {
+        return S_OK;
+    }
+
+    void* bytes = nullptr;
+    IDeckLinkVideoBuffer* buf = nullptr;
+    if (videoFrame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) == S_OK && buf) {
+        buf->StartAccess(bmdBufferAccessRead);
+        buf->GetBytes(&bytes);
+        buf->EndAccess(bmdBufferAccessRead);
+        buf->Release();
+    }
+
+    if (!bytes) {
+        return S_OK;
+    }
+
+    std::lock_guard<std::mutex> lock(owner_.frame_mutex_);
+    owner_.frame_data_ = bytes;
+    owner_.frame_row_bytes_ = videoFrame->GetRowBytes();
+    owner_.frame_width_ = videoFrame->GetWidth();
+    owner_.frame_height_ = videoFrame->GetHeight();
+    owner_.frame_ready_ = true;
+
+    return S_OK;
+}
+
+} // namespace dmf_node

nodes/decklink-in/src/decklink_in_node.hpp

@@ -0,0 +1,84 @@
+#pragma once
+
+#include <dmf-node/node.hpp>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPI.h>
+
+#include <atomic>
+#include <mutex>
+#include <optional>
+#include <thread>
+
+namespace dmf_node {
+
+class DeckLinkInNode : public Node {
+public:
+    DeckLinkInNode() = default;
+    ~DeckLinkInNode();
+
+    std::string type() const override { return "decklink-in"; }
+
+    std::vector<PortDef> input_ports() const override { return {}; }
+
+    std::vector<PortDef> output_ports() const override {
+        return {{"video_out", PortDirection::Output, MediaType::VideoV210}};
+    }
+
+    void configure(const nlohmann::json& params) override;
+    void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override;
+    void on_add_reader(const std::string& port_id, mxlFlowReader reader) override {}
+    void on_remove_writer(const std::string& port_id) override;
+    void on_remove_reader(const std::string& port_id) override {}
+
+    void process() override;
+    nlohmann::json status() const override;
+
+private:
+    bool open_device();
+    void close_device();
+
+    std::optional<mxlFlowWriter> writer_;
+    mxlRational grain_rate_{25, 1};
+    uint64_t write_index_ = 0;
+    uint64_t grains_written_ = 0;
+
+    int device_index_ = 0;
+    BMDDisplayMode display_mode_ = bmdModeHD1080i50;
+    BMDVideoConnection input_connection_ = bmdVideoConnectionUnspecified;
+
+    IDeckLink* decklink_ = nullptr;
+    IDeckLinkInput* input_ = nullptr;
+
+    std::atomic<bool> capturing_{false};
+    bool is_interlaced_ = false;
+    std::mutex frame_mutex_;
+    void* frame_data_ = nullptr;
+    long frame_row_bytes_ = 0;
+    long frame_width_ = 1920;
+    long frame_height_ = 1080;
+    bool frame_ready_ = false;
+
+    class CaptureCallback : public IDeckLinkInputCallback {
+    public:
+        CaptureCallback(DeckLinkInNode& owner) : owner_(owner) {}
+
+        HRESULT STDMETHODCALLTYPE QueryInterface(REFIID, void**) override { return E_NOINTERFACE; }
+        ULONG STDMETHODCALLTYPE AddRef() override { return 1; }
+        ULONG STDMETHODCALLTYPE Release() override { return 1; }
+
+        HRESULT STDMETHODCALLTYPE VideoInputFormatChanged(
+            BMDVideoInputFormatChangedEvents, IDeckLinkDisplayMode*, BMDDetectedVideoInputFormatFlags) override;
+
+        HRESULT STDMETHODCALLTYPE VideoInputFrameArrived(
+            IDeckLinkVideoInputFrame* videoFrame, IDeckLinkAudioInputPacket*) override;
+
+    private:
+        DeckLinkInNode& owner_;
+    };
+
+    std::unique_ptr<CaptureCallback> callback_;
+};
+
+} // namespace dmf_node

nodes/decklink-in/src/main.cpp

@@ -0,0 +1,6 @@
+#include <dmf-node/node_runner.hpp>
+#include "decklink_in_node.hpp"
+
+int main(int argc, char* argv[]) {
+    return dmf_node::NodeRunner::run<dmf_node::DeckLinkInNode>(argc, argv);
+}

nodes/decklink-out/CMakeLists.txt

@@ -0,0 +1,21 @@
+cmake_minimum_required(VERSION 3.24)
+
+project(dmf-node-decklink-out LANGUAGES CXX)
+
+set(DECKLINK_INCLUDE "${DECKLINK_SDK_DIR}/Linux/include")
+
+add_executable(dmf-node-decklink-out
+    src/main.cpp
+    src/decklink_out_node.cpp
+    "${DECKLINK_SDK_DIR}/Linux/include/DeckLinkAPIDispatch.cpp"
+)
+
+target_include_directories(dmf-node-decklink-out PRIVATE
+    "${DECKLINK_INCLUDE}"
+)
+
+target_link_libraries(dmf-node-decklink-out PRIVATE
+    dmf-node
+)
+
+target_compile_options(dmf-node-decklink-out PRIVATE -Wno-unused-parameter)

nodes/decklink-out/src/decklink_out_node.cpp

@@ -0,0 +1,365 @@
+#include "decklink_out_node.hpp"
+
+#include <mxl/flow.h>
+#include <mxl/mxl.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPIConfiguration.h>
+
+#include <spdlog/spdlog.h>
+
+#include <cstring>
+
+namespace dmf_node {
+
+DeckLinkOutNode::~DeckLinkOutNode() {
+    close_device();
+}
+
+void DeckLinkOutNode::configure(const nlohmann::json& params) {
+    if (params.contains("device_index")) {
+        device_index_ = params["device_index"].get<int>();
+    }
+    if (params.contains("mode")) {
+        auto mode_str = params["mode"].get<std::string>();
+        if (mode_str == "1080i50") display_mode_ = bmdModeHD1080i50;
+        else if (mode_str == "1080p50") display_mode_ = bmdModeHD1080p50;
+        else if (mode_str == "1080p25") display_mode_ = bmdModeHD1080p25;
+        else if (mode_str == "1080i5994") display_mode_ = bmdModeHD1080i5994;
+        else if (mode_str == "1080p5994") display_mode_ = bmdModeHD1080p5994;
+        else if (mode_str == "1080p2997") display_mode_ = bmdModeHD1080p2997;
+        else if (mode_str == "720p50") display_mode_ = bmdModeHD720p50;
+        else if (mode_str == "720p5994") display_mode_ = bmdModeHD720p5994;
+        else {
+            spdlog::warn("DeckLink-out: unknown mode '{}', defaulting to 1080i50", mode_str);
+        }
+    }
+    if (params.contains("output_connection")) {
+        auto conn_str = params["output_connection"].get<std::string>();
+        if (conn_str == "sdi") output_connection_ = bmdVideoConnectionSDI;
+        else if (conn_str == "hdmi") output_connection_ = bmdVideoConnectionHDMI;
+        else if (conn_str == "optical_sdi") output_connection_ = bmdVideoConnectionOpticalSDI;
+        else if (conn_str == "component") output_connection_ = bmdVideoConnectionComponent;
+        else if (conn_str == "composite") output_connection_ = bmdVideoConnectionComposite;
+        else if (conn_str == "svideo") output_connection_ = bmdVideoConnectionSVideo;
+        else {
+            spdlog::warn("DeckLink-out: unknown output_connection '{}', defaulting to auto", conn_str);
+        }
+    }
+}
+
+void DeckLinkOutNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
+    if (port_id == "video_in") {
+        reader_ = reader;
+
+        mxlFlowConfigInfo config{};
+        mxlFlowReaderGetConfigInfo(*reader_, &config);
+        flow_rate_ = config.common.grainRate;
+
+        auto now = mxlGetTime();
+        auto current_index = mxlTimestampToIndex(&flow_rate_, now);
+        read_index_ = current_index - 2;
+
+        spdlog::info("DeckLink-out: reader added, flow_rate={}/{}", flow_rate_.numerator, flow_rate_.denominator);
+
+        if (!open_device()) {
+            spdlog::error("DeckLink-out: failed to open device");
+            return;
+        }
+    }
+}
+
+void DeckLinkOutNode::on_remove_reader(const std::string& port_id) {
+    if (port_id == "video_in") {
+        close_device();
+        reader_.reset();
+        spdlog::info("DeckLink-out: reader removed");
+    }
+}
+
+bool DeckLinkOutNode::open_device() {
+    auto* iter = CreateDeckLinkIteratorInstance();
+    if (!iter) {
+        spdlog::error("DeckLink-out: DeckLink drivers not found");
+        return false;
+    }
+
+    IDeckLink* device = nullptr;
+    for (int i = 0; i <= device_index_; ++i) {
+        if (iter->Next(&device) != S_OK) {
+            spdlog::error("DeckLink-out: device index {} not found", device_index_);
+            iter->Release();
+            return false;
+        }
+        if (i < device_index_) {
+            device->Release();
+        }
+    }
+    iter->Release();
+
+    const char* model_name = nullptr;
+    device->GetModelName(&model_name);
+    spdlog::info("DeckLink-out: opened device '{}'", model_name ? model_name : "unknown");
+
+    if (device->QueryInterface(IID_IDeckLinkOutput, (void**)&output_) != S_OK) {
+        spdlog::error("DeckLink-out: device has no output interface");
+        device->Release();
+        return false;
+    }
+
+    decklink_ = device;
+
+    if (output_connection_ != bmdVideoConnectionUnspecified) {
+        IDeckLinkConfiguration* config = nullptr;
+        if (decklink_->QueryInterface(IID_IDeckLinkConfiguration, (void**)&config) == S_OK && config) {
+            if (config->SetInt(bmdDeckLinkConfigVideoOutputConnection, output_connection_) != S_OK) {
+                spdlog::warn("DeckLink-out: failed to set output connection");
+            } else {
+                spdlog::info("DeckLink-out: set output connection configured");
+            }
+            config->Release();
+        } else {
+            spdlog::warn("DeckLink-out: IDeckLinkConfiguration not available");
+        }
+    }
+
+    callback_ = std::make_unique<OutputCallback>(*this);
+    output_->SetScheduledFrameCompletionCallback(callback_.get());
+
+    if (output_->EnableVideoOutput(display_mode_, bmdVideoOutputFlagDefault) != S_OK) {
+        spdlog::error("DeckLink-out: failed to enable video output");
+        return false;
+    }
+
+    IDeckLinkDisplayMode* mode = nullptr;
+    if (output_->GetDisplayMode(display_mode_, &mode) == S_OK) {
+        out_width_ = mode->GetWidth();
+        out_height_ = mode->GetHeight();
+        BMDTimeValue duration = 0;
+        BMDTimeScale scale = 0;
+        mode->GetFrameRate(&duration, &scale);
+        frame_duration_ = duration;
+        time_scale_ = scale;
+        mode->Release();
+
+        output_rate_.numerator = static_cast<int32_t>(scale);
+        output_rate_.denominator = static_cast<int32_t>(duration);
+        auto g = std::__gcd(output_rate_.numerator, output_rate_.denominator);
+        output_rate_.numerator /= g;
+        output_rate_.denominator /= g;
+
+        is_interlaced_ = (display_mode_ == bmdModeHD1080i50 ||
+                          display_mode_ == bmdModeHD1080i5994);
+
+        spdlog::info("DeckLink-out: {}x{} @ {}/{} fps, interlaced={}",
+                     out_width_, out_height_,
+                     output_rate_.numerator, output_rate_.denominator,
+                     is_interlaced_);
+    }
+
+    preroll_frames_ = 5;
+    for (int i = 0; i < preroll_frames_; ++i) {
+        IDeckLinkMutableVideoFrame* frame = nullptr;
+        int32_t out_row_bytes = out_width_ * 16 / 6;
+        if (output_->CreateVideoFrame(out_width_, out_height_, out_row_bytes,
+                                       bmdFormat10BitYUV, bmdFrameFlagDefault, &frame) != S_OK || !frame) {
+            out_row_bytes = out_width_ * 2;
+            if (output_->CreateVideoFrame(out_width_, out_height_, out_row_bytes,
+                                           bmdFormat8BitYUV, bmdFrameFlagDefault, &frame) != S_OK || !frame) {
+                spdlog::error("DeckLink-out: failed to create preroll frame");
+                return false;
+            }
+            spdlog::info("DeckLink-out: using 8BitYUV output format");
+            output_format_ = bmdFormat8BitYUV;
+        } else {
+            output_format_ = bmdFormat10BitYUV;
+        }
+        IDeckLinkVideoBuffer* buf = nullptr;
+        if (frame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) == S_OK && buf) {
+            buf->StartAccess(bmdBufferAccessWrite);
+            void* dst = nullptr;
+            buf->GetBytes(&dst);
+            if (dst) {
+                std::memset(dst, 0, out_row_bytes * out_height_);
+            }
+            buf->EndAccess(bmdBufferAccessWrite);
+            buf->Release();
+        }
+        auto stream_time = i * frame_duration_;
+        output_->ScheduleVideoFrame(frame, stream_time, frame_duration_, time_scale_);
+        frame->Release();
+    }
+
+    stream_frame_ = preroll_frames_;
+    if (output_->StartScheduledPlayback(0, time_scale_, 1.0) != S_OK) {
+        spdlog::error("DeckLink-out: failed to start scheduled playback");
+        return false;
+    }
+
+    playing_ = true;
+    spdlog::info("DeckLink-out: playback started");
+    return true;
+}
+
+void DeckLinkOutNode::close_device() {
+    if (output_) {
+        if (playing_) {
+            output_->StopScheduledPlayback(0, nullptr, time_scale_);
+        }
+        output_->DisableVideoOutput();
+        output_->Release();
+        output_ = nullptr;
+    }
+    if (decklink_) {
+        decklink_->Release();
+        decklink_ = nullptr;
+    }
+    playing_ = false;
+    callback_.reset();
+    {
+        std::lock_guard<std::mutex> lock(frame_pool_mutex_);
+        for (auto* f : frame_pool_) {
+            f->Release();
+        }
+        frame_pool_.clear();
+    }
+    spdlog::info("DeckLink-out: device closed");
+}
+
+void DeckLinkOutNode::schedule_frame(void* mxl_payload, long width, long height, long row_bytes) {
+    if (!output_) return;
+
+    int32_t out_row_bytes = 0;
+    if (output_format_ == bmdFormat10BitYUV) {
+        if (output_->RowBytesForPixelFormat(bmdFormat10BitYUV, width, &out_row_bytes) != S_OK || out_row_bytes <= 0) {
+            out_row_bytes = ((width + 5) / 6) * 16;
+        }
+    } else {
+        out_row_bytes = width * 2;
+    }
+
+    IDeckLinkMutableVideoFrame* frame = nullptr;
+    {
+        std::lock_guard<std::mutex> lock(frame_pool_mutex_);
+        if (!frame_pool_.empty()) {
+            frame = frame_pool_.back();
+            frame_pool_.pop_back();
+        }
+    }
+
+    if (!frame) {
+        return;
+    }
+
+    IDeckLinkVideoBuffer* buf = nullptr;
+    if (frame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) == S_OK && buf) {
+        buf->StartAccess(bmdBufferAccessWrite);
+        void* dst = nullptr;
+        buf->GetBytes(&dst);
+        if (dst) {
+            auto copy_row_bytes = std::min(static_cast<long>(out_row_bytes), row_bytes);
+            for (long y = 0; y < height; ++y) {
+                std::memcpy(static_cast<uint8_t*>(dst) + y * out_row_bytes,
+                            static_cast<uint8_t*>(mxl_payload) + y * row_bytes,
+                            copy_row_bytes);
+            }
+        }
+        buf->EndAccess(bmdBufferAccessWrite);
+        buf->Release();
+    }
+
+    auto stream_time = stream_frame_ * frame_duration_;
+    output_->ScheduleVideoFrame(frame, stream_time, frame_duration_, time_scale_);
+    frame->Release();
+    stream_frame_++;
+}
+
+void DeckLinkOutNode::process() {
+    if (!reader_ || !playing_) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        return;
+    }
+
+    if (first_frame_) {
+        auto now = mxlGetTime();
+        grains_read_ = mxlTimestampToIndex(&output_rate_, now);
+        first_frame_ = false;
+        spdlog::info("DeckLink-out: starting at output index {}", grains_read_);
+    }
+
+    auto deadline = mxlIndexToTimestamp(&output_rate_, grains_read_ + 1);
+    mxlSleepUntil(deadline);
+
+    auto out_timestamp = mxlIndexToTimestamp(&output_rate_, grains_read_);
+    auto source_index = mxlTimestampToIndex(&flow_rate_, out_timestamp);
+
+    mxlGrainInfo grain_info{};
+    uint8_t* payload = nullptr;
+    auto status = mxlFlowReaderGetGrain(*reader_, source_index, 5000000ULL, &grain_info, &payload);
+    if (status != MXL_STATUS_OK) {
+        if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
+            auto now = mxlGetTime();
+            auto current_index = mxlTimestampToIndex(&flow_rate_, now);
+            spdlog::warn("DeckLink-out: grain too late, skipping to index {}", current_index - 2);
+            source_index = current_index - 2;
+            status = mxlFlowReaderGetGrain(*reader_, source_index, 5000000ULL, &grain_info, &payload);
+            if (status != MXL_STATUS_OK) {
+                grains_read_++;
+                return;
+            }
+        } else if (status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
+            return;
+        } else {
+            return;
+        }
+    }
+
+    mxlFlowConfigInfo config{};
+    mxlFlowReaderGetConfigInfo(*reader_, &config);
+    auto grain_size = grain_info.grainSize;
+
+    long src_row_bytes = (config.discrete.sliceSizes[0] > 0) ? static_cast<long>(config.discrete.sliceSizes[0]) : (grain_size / out_height_);
+    long src_height = (src_row_bytes > 0) ? (grain_size / src_row_bytes) : out_height_;
+    long src_width = (src_row_bytes * 3) / 8;
+
+    schedule_frame(payload, src_width, src_height, src_row_bytes);
+
+    grains_read_++;
+
+    if (grains_read_ == 1) {
+        spdlog::info("DeckLink-out: first grain output, source_index={}", source_index);
+    }
+}
+
+nlohmann::json DeckLinkOutNode::status() const {
+    return {
+        {"type", "decklink-out"},
+        {"grains_read", grains_read_},
+        {"read_index", read_index_},
+        {"playing", playing_.load()},
+        {"has_reader", reader_.has_value()},
+        {"flow_rate", {{"numerator", flow_rate_.numerator}, {"denominator", flow_rate_.denominator}}},
+        {"output_rate", {{"numerator", output_rate_.numerator}, {"denominator", output_rate_.denominator}}},
+        {"width", out_width_},
+        {"height", out_height_},
+    };
+}
+
+HRESULT DeckLinkOutNode::OutputCallback::ScheduledFrameCompleted(
+    IDeckLinkVideoFrame* completedFrame, BMDOutputFrameCompletionResult result) {
+    if (completedFrame) {
+        completedFrame->AddRef();
+        std::lock_guard<std::mutex> lock(owner_.frame_pool_mutex_);
+        owner_.frame_pool_.push_back(static_cast<IDeckLinkMutableVideoFrame*>(completedFrame));
+        owner_.frame_pool_cv_.notify_one();
+    }
+    return S_OK;
+}
+
+HRESULT DeckLinkOutNode::OutputCallback::ScheduledPlaybackHasStopped() {
+    owner_.playing_ = false;
+    spdlog::info("DeckLink-out: scheduled playback stopped");
+    return S_OK;
+}
+
+} // namespace dmf_node

nodes/decklink-out/src/decklink_out_node.hpp

@@ -0,0 +1,93 @@
+#pragma once
+
+#include <dmf-node/node.hpp>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPI.h>
+
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+#include <optional>
+#include <thread>
+#include <vector>
+
+namespace dmf_node {
+
+class DeckLinkOutNode : public Node {
+public:
+    DeckLinkOutNode() = default;
+    ~DeckLinkOutNode();
+
+    std::string type() const override { return "decklink-out"; }
+
+    std::vector<PortDef> input_ports() const override {
+        return {{"video_in", PortDirection::Input, MediaType::VideoV210}};
+    }
+
+    std::vector<PortDef> output_ports() const override { return {}; }
+
+    void configure(const nlohmann::json& params) override;
+    void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override {}
+    void on_add_reader(const std::string& port_id, mxlFlowReader reader) override;
+    void on_remove_writer(const std::string& port_id) override {}
+    void on_remove_reader(const std::string& port_id) override;
+
+    void process() override;
+    nlohmann::json status() const override;
+
+private:
+    bool open_device();
+    void close_device();
+    void schedule_frame(void* mxl_payload, long width, long height, long row_bytes);
+
+    std::optional<mxlFlowReader> reader_;
+    mxlRational flow_rate_{50, 1};
+    mxlRational output_rate_{50, 1};
+    uint64_t read_index_ = 0;
+    uint64_t grains_read_ = 0;
+    bool first_frame_ = true;
+
+    int device_index_ = 0;
+    BMDDisplayMode display_mode_ = bmdModeHD1080i50;
+    BMDVideoConnection output_connection_ = bmdVideoConnectionUnspecified;
+
+    IDeckLink* decklink_ = nullptr;
+    IDeckLinkOutput* output_ = nullptr;
+    BMDTimeValue frame_duration_ = 1000;
+    BMDTimeScale time_scale_ = 50000;
+    BMDPixelFormat output_format_ = bmdFormat10BitYUV;
+    int preroll_frames_ = 3;
+
+    std::atomic<bool> playing_{false};
+    int create_fail_count_ = 0;
+    uint64_t stream_frame_ = 0;
+    bool is_interlaced_ = false;
+    long out_width_ = 1920;
+    long out_height_ = 1080;
+
+    class OutputCallback : public IDeckLinkVideoOutputCallback {
+    public:
+        OutputCallback(DeckLinkOutNode& owner) : owner_(owner) {}
+
+        HRESULT STDMETHODCALLTYPE QueryInterface(REFIID, void**) override { return E_NOINTERFACE; }
+        ULONG STDMETHODCALLTYPE AddRef() override { return 1; }
+        ULONG STDMETHODCALLTYPE Release() override { return 1; }
+
+        HRESULT STDMETHODCALLTYPE ScheduledFrameCompleted(
+            IDeckLinkVideoFrame* completedFrame, BMDOutputFrameCompletionResult result) override;
+        HRESULT STDMETHODCALLTYPE ScheduledPlaybackHasStopped() override;
+
+    private:
+        DeckLinkOutNode& owner_;
+    };
+
+    std::unique_ptr<OutputCallback> callback_;
+    std::vector<IDeckLinkMutableVideoFrame*> frame_pool_;
+    std::mutex frame_pool_mutex_;
+    std::condition_variable frame_pool_cv_;
+    IDeckLinkMutableVideoFrame* scheduled_frame_ = nullptr;
+};
+
+} // namespace dmf_node

nodes/decklink-out/src/main.cpp

@@ -0,0 +1,6 @@
+#include <dmf-node/node_runner.hpp>
+#include "decklink_out_node.hpp"
+
+int main(int argc, char* argv[]) {
+    return dmf_node::NodeRunner::run<dmf_node::DeckLinkOutNode>(argc, argv);
+}

nodes/fabric-bridge/CMakeLists.txt

@@ -0,0 +1,28 @@
+cmake_minimum_required(VERSION 3.24 FATAL_ERROR)
+
+project(dmf-node-fabric-bridge VERSION 0.1.0 LANGUAGES CXX)
+
+find_package(fmt CONFIG REQUIRED)
+find_package(spdlog CONFIG REQUIRED)
+find_package(nlohmann_json CONFIG REQUIRED)
+find_package(PkgConfig REQUIRED)
+pkg_check_modules(libfabric REQUIRED IMPORTED_TARGET libfabric)
+
+add_executable(dmf-node-fabric-bridge
+    src/fabric_bridge_node.cpp
+    src/main.cpp
+)
+
+target_link_libraries(dmf-node-fabric-bridge PRIVATE
+    dmf-node
+    mxl-fabrics-objects
+    mxl-fabrics-headers
+    PkgConfig::libfabric
+    spdlog::spdlog
+    fmt::fmt
+    nlohmann_json::nlohmann_json
+)
+
+target_include_directories(dmf-node-fabric-bridge PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/src
+)

nodes/fabric-bridge/src/fabric_bridge_node.cpp

@@ -0,0 +1,470 @@
+#include "fabric_bridge_node.hpp"
+
+#include <mxl/mxl.h>
+
+#include <spdlog/spdlog.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <unistd.h>
+#include <netdb.h>
+#include <cerrno>
+
+namespace dmf_node {
+
+FabricBridgeNode::~FabricBridgeNode() {
+    running_ = false;
+    if (worker_thread_.has_value() && worker_thread_->joinable()) {
+        worker_thread_->join();
+    }
+}
+
+std::vector<PortDef> FabricBridgeNode::input_ports() const {
+    if (mode_ == FabricMode::Target) {
+        return {};
+    }
+    return {{"video_in", PortDirection::Input, MediaType::VideoV210}};
+}
+
+std::vector<PortDef> FabricBridgeNode::output_ports() const {
+    if (mode_ == FabricMode::Initiator) {
+        return {};
+    }
+    return {{"video_out", PortDirection::Output, MediaType::VideoV210}};
+}
+
+void FabricBridgeNode::configure(const nlohmann::json& params) {
+    if (params.contains("mode")) {
+        auto mode_str = params["mode"].get<std::string>();
+        if (mode_str == "initiator") mode_ = FabricMode::Initiator;
+        else if (mode_str == "target") mode_ = FabricMode::Target;
+        else spdlog::warn("Fabric-bridge: unknown mode '{}', defaulting to initiator", mode_str);
+    }
+
+    if (params.contains("bind_node")) bind_node_ = params["bind_node"].get<std::string>();
+    if (params.contains("bind_service")) bind_service_ = params["bind_service"].get<std::string>();
+    if (params.contains("provider")) {
+        auto prov_str = params["provider"].get<std::string>();
+        mxlFabricsProviderFromString(prov_str.c_str(), &provider_);
+    }
+
+    if (params.contains("target_info")) {
+        target_info_str_ = params["target_info"].get<std::string>();
+        if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_info_str_.empty()) {
+            start_requested_.store(true);
+            spdlog::info("Fabric-bridge: target_info set, signaling startup");
+        }
+    }
+
+    if (params.contains("target_host")) {
+        target_host_ = params["target_host"].get<std::string>();
+        target_port_ = params.value("target_port", static_cast<uint16_t>(9100));
+        spdlog::info("Fabric-bridge: will fetch target_info from {}:{}", target_host_, target_port_);
+    }
+}
+
+void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
+    if (port_id == "video_out") {
+        writer_ = writer;
+        start_requested_.store(true);
+        spdlog::info("Fabric-bridge: writer added (target mode), signaling startup");
+    }
+}
+
+void FabricBridgeNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
+    if (port_id == "video_in") {
+        reader_ = reader;
+        if (mode_ == FabricMode::Initiator) {
+            if (!target_info_str_.empty() || !target_host_.empty()) {
+                start_requested_.store(true);
+                spdlog::info("Fabric-bridge: reader added (initiator mode), signaling startup");
+            } else {
+                spdlog::info("Fabric-bridge: reader added (initiator mode), waiting for target_info");
+            }
+        }
+    }
+}
+
+void FabricBridgeNode::on_remove_writer(const std::string& port_id) {
+    if (port_id == "video_out") {
+        writer_.reset();
+        spdlog::info("Fabric-bridge: writer removed");
+    }
+}
+
+void FabricBridgeNode::on_remove_reader(const std::string& port_id) {
+    if (port_id == "video_in") {
+        reader_.reset();
+        spdlog::info("Fabric-bridge: reader removed");
+    }
+}
+
+void FabricBridgeNode::fetch_target_info() {
+    if (target_host_.empty()) return;
+
+    spdlog::info("Fabric-bridge: fetching target_info from {}:{}", target_host_, target_port_);
+
+    std::string body = R"({"cmd":"status"})";
+    std::string request = "POST /cmd HTTP/1.1\r\n"
+                           "Host: " + target_host_ + "\r\n"
+                           "Content-Type: application/json\r\n"
+                           "Connection: close\r\n"
+                           "Content-Length: " + std::to_string(body.size()) + "\r\n"
+                           "\r\n" +
+                           body;
+
+    for (int attempt = 0; attempt < 30 && target_info_str_.empty(); ++attempt) {
+        int sock = socket(AF_INET, SOCK_STREAM, 0);
+        if (sock < 0) {
+            spdlog::warn("Fabric-bridge: socket() failed, retry {}...", attempt + 1);
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        struct timeval tv{};
+        tv.tv_sec = 2;
+        tv.tv_usec = 0;
+        setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+        setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+
+        struct hostent* he = gethostbyname(target_host_.c_str());
+        if (!he) {
+            spdlog::warn("Fabric-bridge: cannot resolve {}, retry {}...", target_host_, attempt + 1);
+            close(sock);
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        struct sockaddr_in addr{};
+        addr.sin_family = AF_INET;
+        addr.sin_port = htons(target_port_);
+        std::memcpy(&addr.sin_addr, he->h_addr_list[0], he->h_length);
+
+        if (connect(sock, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) < 0) {
+            spdlog::warn("Fabric-bridge: connect to {}:{} failed (errno={}), retry {}...", target_host_, target_port_, errno, attempt + 1);
+            close(sock);
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        if (send(sock, request.c_str(), request.size(), 0) < 0) {
+            spdlog::warn("Fabric-bridge: send failed, retry {}...", attempt + 1);
+            close(sock);
+            std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            continue;
+        }
+
+        std::string response;
+        char buf[4096];
+        ssize_t n;
+        while ((n = recv(sock, buf, sizeof(buf), 0)) > 0) {
+            response.append(buf, n);
+        }
+        close(sock);
+
+        spdlog::info("Fabric-bridge: received {} bytes from target", response.size());
+
+        auto body_pos = response.find("\r\n\r\n");
+        if (body_pos == std::string::npos) {
+            spdlog::warn("Fabric-bridge: no HTTP body in response, retry {}...", attempt + 1);
+            std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            continue;
+        }
+
+        std::string response_body = response.substr(body_pos + 4);
+
+        try {
+            auto json = nlohmann::json::parse(response_body);
+            if (json.contains("data") && json["data"].contains("target_info")) {
+                auto ti = json["data"]["target_info"].get<std::string>();
+                if (!ti.empty()) {
+                    target_info_str_ = ti;
+                    spdlog::info("Fabric-bridge: fetched target_info ({} bytes)", target_info_str_.size());
+                    return;
+                }
+            }
+            spdlog::debug("Fabric-bridge: response data: {}", json.contains("data") ? json["data"].dump() : "(no data)");
+        } catch (const nlohmann::json::exception& e) {
+            spdlog::warn("Fabric-bridge: JSON parse error: {}", e.what());
+        }
+
+        spdlog::info("Fabric-bridge: target_info not ready yet, retry {}...", attempt + 1);
+        std::this_thread::sleep_for(std::chrono::milliseconds(500));
+    }
+
+    if (target_info_str_.empty()) {
+        spdlog::error("Fabric-bridge: failed to fetch target_info after 30 attempts");
+    }
+}
+
+void FabricBridgeNode::process() {
+    if (!worker_thread_.has_value() && start_requested_.load()) {
+        if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_host_.empty() && target_info_str_.empty()) {
+            fetch_target_info();
+        }
+        if (start_requested_.load()) {
+            running_ = true;
+            spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target");
+            if (mode_ == FabricMode::Initiator) {
+                worker_thread_ = std::thread(&FabricBridgeNode::run_initiator, this);
+            } else {
+                worker_thread_ = std::thread(&FabricBridgeNode::run_target, this);
+            }
+        }
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+}
+
+void FabricBridgeNode::run_initiator() {
+    spdlog::info("Fabric-bridge: starting initiator mode");
+
+    auto instance = get_mxl_instance();
+    if (!instance) {
+        spdlog::error("Fabric-bridge: no MXL instance");
+        running_ = false;
+        return;
+    }
+
+    spdlog::info("Fabric-bridge: creating fabrics instance...");
+    mxlFabricsInstance fab_inst = nullptr;
+    auto status = mxlFabricsCreateInstance(instance, nullptr, &fab_inst);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateInstance returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !fab_inst) {
+        spdlog::error("Fabric-bridge: failed to create fabrics instance: {}", static_cast<int>(status));
+        running_ = false;
+        return;
+    }
+    fabrics_instance_ = fab_inst;
+
+    spdlog::info("Fabric-bridge: creating initiator...");
+    mxlFabricsInitiator initiator = nullptr;
+    status = mxlFabricsCreateInitiator(fab_inst, &initiator);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateInitiator returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !initiator) {
+        spdlog::error("Fabric-bridge: failed to create initiator: {}", static_cast<int>(status));
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+    initiator_ = initiator;
+
+    mxlFlowConfigInfo config{};
+    mxlFlowReaderGetConfigInfo(*reader_, &config);
+    grain_rate_ = config.common.grainRate;
+
+    mxlFabricsInitiatorConfig init_cfg{};
+    init_cfg.version = MXL_FABRICS_API_VERSION;
+    init_cfg.interface.version = MXL_FABRICS_API_VERSION;
+    init_cfg.interface.provider = provider_;
+    init_cfg.interface.address.node = bind_node_.c_str();
+    init_cfg.interface.address.service = bind_service_.c_str();
+    init_cfg.reader = *reader_;
+
+    spdlog::info("Fabric-bridge: calling mxlFabricsInitiatorSetup on {}:{}", bind_node_, bind_service_);
+    status = mxlFabricsInitiatorSetup(initiator, &init_cfg, nullptr);
+    if (status != MXL_STATUS_OK) {
+        spdlog::error("Fabric-bridge: initiator setup failed: {}", static_cast<int>(status));
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    if (target_info_str_.empty()) {
+        spdlog::error("Fabric-bridge: no target_info available, cannot add target");
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    mxlFabricsTargetInfo target_info = nullptr;
+    spdlog::info("Fabric-bridge: parsing target_info ({} bytes)", target_info_str_.size());
+    status = mxlFabricsTargetInfoFromString(target_info_str_.c_str(), &target_info);
+    if (status != MXL_STATUS_OK || !target_info) {
+        spdlog::error("Fabric-bridge: failed to parse target_info: {}", static_cast<int>(status));
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    status = mxlFabricsInitiatorAddTarget(initiator, target_info);
+    if (status != MXL_STATUS_OK) {
+        spdlog::error("Fabric-bridge: failed to add target: {}", static_cast<int>(status));
+        mxlFabricsFreeTargetInfo(target_info);
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    spdlog::info("Fabric-bridge: waiting for connection...");
+    do {
+        status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 250);
+    } while (status == MXL_ERR_NOT_READY && running_);
+
+    auto now = mxlGetTime();
+    read_index_ = mxlTimestampToIndex(&grain_rate_, now) - 2;
+
+    spdlog::info("Fabric-bridge: initiator started, grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
+
+    while (running_) {
+        auto deadline = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
+        mxlSleepUntil(deadline);
+
+        mxlGrainInfo grain_info{};
+        uint8_t* payload = nullptr;
+        auto grain_status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &grain_info, &payload);
+        if (grain_status != MXL_STATUS_OK) {
+            if (grain_status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || grain_status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
+                auto ts = mxlGetTime();
+                read_index_ = mxlTimestampToIndex(&grain_rate_, ts) - 2;
+                continue;
+            }
+            read_index_++;
+            continue;
+        }
+
+        auto transfer_status = mxlFabricsInitiatorTransferGrain(initiator, read_index_, 0, grain_info.validSlices);
+        if (transfer_status != MXL_STATUS_OK) {
+            spdlog::warn("Fabric-bridge: transfer grain {} failed: {}", read_index_, static_cast<int>(transfer_status));
+            read_index_++;
+            continue;
+        }
+
+        do {
+            status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 10);
+        } while (status == MXL_ERR_NOT_READY && running_);
+
+        read_index_++;
+        grains_transferred_++;
+
+        if (grains_transferred_ == 1) {
+            spdlog::info("Fabric-bridge: first grain transferred, index={}", read_index_ - 1);
+        }
+    }
+
+    mxlFabricsInitiatorRemoveTarget(initiator, target_info);
+    mxlFabricsFreeTargetInfo(target_info);
+    mxlFabricsDestroyInitiator(fab_inst, initiator);
+    mxlFabricsDestroyInstance(fab_inst);
+    spdlog::info("Fabric-bridge: initiator stopped");
+}
+
+void FabricBridgeNode::run_target() {
+    spdlog::info("Fabric-bridge: starting target mode");
+
+    auto instance = get_mxl_instance();
+    if (!instance) {
+        spdlog::error("Fabric-bridge: no MXL instance");
+        running_ = false;
+        return;
+    }
+
+    spdlog::info("Fabric-bridge: creating fabrics instance...");
+    mxlFabricsInstance fab_inst = nullptr;
+    auto status = mxlFabricsCreateInstance(instance, nullptr, &fab_inst);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateInstance returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !fab_inst) {
+        spdlog::error("Fabric-bridge: failed to create fabrics instance: {}", static_cast<int>(status));
+        running_ = false;
+        return;
+    }
+    fabrics_instance_ = fab_inst;
+
+    spdlog::info("Fabric-bridge: creating target...");
+    mxlFabricsTarget tgt = nullptr;
+    status = mxlFabricsCreateTarget(fab_inst, &tgt);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateTarget returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !tgt) {
+        spdlog::error("Fabric-bridge: failed to create target: {}", static_cast<int>(status));
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+    target_ = tgt;
+
+    auto* config = get_writer_config("video_out");
+    if (config) {
+        grain_rate_ = config->common.grainRate;
+        spdlog::info("Fabric-bridge: target grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
+    } else {
+        spdlog::warn("Fabric-bridge: no writer config, using default grain_rate");
+    }
+
+    mxlFabricsTargetConfig target_cfg{};
+    target_cfg.version = MXL_FABRICS_API_VERSION;
+    target_cfg.interface.version = MXL_FABRICS_API_VERSION;
+    target_cfg.interface.provider = provider_;
+    target_cfg.interface.address.node = bind_node_.c_str();
+    target_cfg.interface.address.service = bind_service_.c_str();
+    target_cfg.writer = *writer_;
+
+    spdlog::info("Fabric-bridge: calling mxlFabricsTargetSetup on {}:{}", bind_node_, bind_service_);
+    mxlFabricsTargetInfo target_info = nullptr;
+    status = mxlFabricsTargetSetup(tgt, &target_cfg, nullptr, &target_info);
+    if (status != MXL_STATUS_OK || !target_info) {
+        spdlog::error("Fabric-bridge: target setup failed: {}", static_cast<int>(status));
+        mxlFabricsDestroyTarget(fab_inst, tgt);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    size_t info_size = 0;
+    mxlFabricsTargetInfoToString(target_info, nullptr, &info_size);
+    std::vector<char> info_buf(info_size);
+    mxlFabricsTargetInfoToString(target_info, info_buf.data(), &info_size);
+    target_info_str_ = std::string(info_buf.data(), info_size);
+
+    spdlog::info("Fabric-bridge: target ready, target_info length={}", target_info_str_.size());
+    spdlog::info("Fabric-bridge: TARGET_INFO={}", target_info_str_);
+
+    while (running_) {
+        uint64_t grain_index = 0;
+        status = mxlFabricsTargetReadGrain(tgt, 200, &grain_index);
+        if (status == MXL_ERR_TIMEOUT || status == MXL_ERR_NOT_READY) {
+            continue;
+        }
+        if (status != MXL_STATUS_OK) {
+            spdlog::warn("Fabric-bridge: target read grain failed: {}", static_cast<int>(status));
+            continue;
+        }
+
+        mxlGrainInfo grain_info{};
+        uint8_t* dummy_payload = nullptr;
+        auto open_status = mxlFlowWriterOpenGrain(*writer_, grain_index, &grain_info, &dummy_payload);
+        if (open_status == MXL_STATUS_OK) {
+            grain_info.validSlices = grain_info.totalSlices;
+            mxlFlowWriterCommitGrain(*writer_, &grain_info);
+            grains_transferred_++;
+            if (grains_transferred_ == 1) {
+                spdlog::info("Fabric-bridge: first grain committed, index={}", grain_index);
+            }
+        } else {
+            spdlog::warn("Fabric-bridge: open grain {} failed: {}", grain_index, static_cast<int>(open_status));
+        }
+    }
+
+    mxlFabricsFreeTargetInfo(target_info);
+    mxlFabricsDestroyTarget(fab_inst, tgt);
+    mxlFabricsDestroyInstance(fab_inst);
+    spdlog::info("Fabric-bridge: target stopped");
+}
+
+nlohmann::json FabricBridgeNode::status() const {
+    return {
+        {"type", "fabric-bridge"},
+        {"mode", mode_ == FabricMode::Initiator ? "initiator" : "target"},
+        {"grains_transferred", grains_transferred_},
+        {"running", running_.load()},
+        {"has_reader", reader_.has_value()},
+        {"has_writer", writer_.has_value()},
+        {"target_info_length", target_info_str_.size()},
+        {"target_info", target_info_str_},
+    };
+}
+
+} // namespace dmf_node

nodes/fabric-bridge/src/fabric_bridge_node.hpp

@@ -0,0 +1,68 @@
+#pragma once
+
+#include <dmf-node/node.hpp>
+#include <mxl/flow.h>
+#include <mxl/fabrics.h>
+#include <mxl/time.h>
+
+#include <atomic>
+#include <optional>
+#include <string>
+#include <thread>
+
+namespace dmf_node {
+
+enum class FabricMode { Initiator, Target };
+
+class FabricBridgeNode : public Node {
+public:
+    FabricBridgeNode() = default;
+    ~FabricBridgeNode();
+
+    std::string type() const override { return "fabric-bridge"; }
+
+    std::vector<PortDef> input_ports() const override;
+    std::vector<PortDef> output_ports() const override;
+
+    void configure(const nlohmann::json& params) override;
+    void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override;
+    void on_add_reader(const std::string& port_id, mxlFlowReader reader) override;
+    void on_remove_writer(const std::string& port_id) override;
+    void on_remove_reader(const std::string& port_id) override;
+
+    void process() override;
+    nlohmann::json status() const override;
+
+private:
+    void run_initiator();
+    void run_target();
+
+    FabricMode mode_ = FabricMode::Initiator;
+
+    std::string bind_node_ = "0.0.0.0";
+    std::string bind_service_ = "0";
+    mxlFabricsProvider provider_ = MXL_FABRICS_PROVIDER_ANY;
+
+    std::optional<mxlFlowReader> reader_;
+    std::optional<mxlFlowWriter> writer_;
+    mxlRational grain_rate_{50, 1};
+    uint64_t read_index_ = 0;
+    uint64_t grains_transferred_ = 0;
+
+    std::string target_info_str_;
+
+    std::string target_host_;
+    uint16_t target_port_ = 9100;
+
+    void fetch_target_info();
+
+    mxlFabricsInstance fabrics_instance_ = nullptr;
+    mxlFabricsInitiator initiator_ = nullptr;
+    mxlFabricsTarget target_ = nullptr;
+
+    std::optional<std::thread> worker_thread_;
+    std::atomic<bool> running_{false};
+    std::atomic<bool> start_requested_{false};
+};
+
+} // namespace dmf_node

nodes/fabric-bridge/src/main.cpp

@@ -0,0 +1,6 @@
+#include "fabric_bridge_node.hpp"
+#include <dmf-node/node_runner.hpp>
+
+int main(int argc, char* argv[]) {
+    return dmf_node::NodeRunner::run<dmf_node::FabricBridgeNode>(argc, argv);
+}