chore: phase 1 cleanup — proper shutdown, flow resource cleanup, test fix

- node_runner: track all readers/writers in flow_resources vector,
  release them via mxlReleaseFlowWriter/mxlReleaseFlowReader before
  destroying the MXL instance (fixes 'leaked flow writer' warning)
- node_runner: remove_writer/remove_reader commands now also release
  the MXL flow resources, not just reset the node's optional<>
- engine: call process_manager.stop_all(graph) on shutdown to kill
  child node processes (prevents orphaned passthrough processes)
- graph: add get_node_mut() for process_manager to update node state
- process_manager: implement stop_all(Graph&) that SIGTERMs all
  running node processes
- passthrough: reduce logging to first-grain and realign-once only
- tests: update flow format assertion to match NMOS
  (urn:x-nmos:format:video instead of video/v210)

engine/src/main.cpp

@@ -52,5 +52,6 @@ int main(int argc, char* argv[]) {
     }
 
     spdlog::info("DMF Studio Engine shutting down");
+    process_manager.stop_all(graph);
     return 0;
 }

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

@@ -42,6 +42,7 @@ public:
     NodeId add_node(const std::string& type, const nlohmann::json& config = {});
     bool remove_node(const NodeId& node_id);
     const GraphNode* get_node(const NodeId& node_id) const;
+    GraphNode* get_node_mut(const NodeId& node_id);
     std::vector<GraphNode> get_nodes() const;
 
     EdgeId add_edge(const NodeId& from_node, const PortId& from_port,

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

@@ -12,6 +12,7 @@ public:
     bool start_node(GraphNode& node, const std::string& mxl_domain, uint16_t base_port);
     bool stop_node(GraphNode& node);
     void stop_all();
+    void stop_all(Graph& graph);
 
     bool is_running(const NodeId& node_id) const;
 

libs/dmf-engine/src/graph.cpp

@@ -51,6 +51,11 @@ const GraphNode* Graph::get_node(const NodeId& node_id) const {
     return it != nodes_.end() ? &it->second : nullptr;
 }
 
+GraphNode* Graph::get_node_mut(const NodeId& node_id) {
+    auto it = nodes_.find(node_id);
+    return it != nodes_.end() ? &it->second : nullptr;
+}
+
 std::vector<GraphNode> Graph::get_nodes() const {
     std::vector<GraphNode> result;
     for (const auto& [_, node] : nodes_) {

libs/dmf-engine/src/process_manager.cpp

@@ -90,6 +90,22 @@ bool ProcessManager::stop_node(GraphNode& node) {
 void ProcessManager::stop_all() {
 }
 
+void ProcessManager::stop_all(Graph& graph) {
+    for (auto& node : graph.get_nodes()) {
+        if (node.state == NodeState::Running && node.pid > 0) {
+            kill(node.pid, SIGTERM);
+            int status = 0;
+            waitpid(node.pid, &status, 0);
+            auto* mut_node = graph.get_node_mut(node.id);
+            if (mut_node) {
+                mut_node->pid = 0;
+                mut_node->state = NodeState::Stopped;
+            }
+            spdlog::info("ProcessManager: stopped node '{}' pid={}", node.id, node.pid);
+        }
+    }
+}
+
 bool ProcessManager::is_running(const NodeId& node_id) const {
     return false;
 }

libs/dmf-node/src/node_runner.cpp

@@ -94,7 +94,14 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
         return arr;
     };
 
-    control_server->register_command("add_writer", [&, w = std::unordered_map<std::string, mxlFlowWriter>{}](const nlohmann::json& msg) mutable {
+    struct FlowResource {
+        std::string port_id;
+        mxlFlowWriter writer = nullptr;
+        mxlFlowReader reader = nullptr;
+    };
+    std::vector<FlowResource> flow_resources;
+
+    control_server->register_command("add_writer", [&](const nlohmann::json& msg) {
         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();
@@ -108,6 +115,7 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
             return;
         }
         spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created);
+        flow_resources.push_back({port_id, writer, nullptr});
         node->on_add_writer(port_id, writer);
     });
 
@@ -122,18 +130,33 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
             return;
         }
         spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id);
+        flow_resources.push_back({port_id, nullptr, reader});
         node->on_add_reader(port_id, reader);
     });
 
     control_server->register_command("remove_writer", [&](const nlohmann::json& msg) {
         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) {
+            if (it->port_id == port_id && it->writer) {
+                mxlReleaseFlowWriter(mxl_instance_, it->writer);
+                flow_resources.erase(it);
+                break;
+            }
+        }
         spdlog::info("Removed writer on port '{}'", port_id);
     });
 
     control_server->register_command("remove_reader", [&](const nlohmann::json& msg) {
         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) {
+            if (it->port_id == port_id && it->reader) {
+                mxlReleaseFlowReader(mxl_instance_, it->reader);
+                flow_resources.erase(it);
+                break;
+            }
+        }
         spdlog::info("Removed reader on port '{}'", port_id);
     });
 
@@ -184,6 +207,17 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
     process_thread.join();
 
     spdlog::info("Node '{}' shutting down", node_id_);
+
+    for (auto& res : flow_resources) {
+        if (res.writer) {
+            mxlReleaseFlowWriter(mxl_instance_, res.writer);
+        }
+        if (res.reader) {
+            mxlReleaseFlowReader(mxl_instance_, res.reader);
+        }
+    }
+    flow_resources.clear();
+
     mxlDestroyInstance(mxl_instance_);
     return 0;
 }

nodes/passthrough/src/passthrough_node.cpp

@@ -64,12 +64,12 @@ void PassthroughNode::process() {
     auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &grain_info, &payload);
     if (status != MXL_STATUS_OK) {
         if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
-            auto old_index = read_index_;
             auto now = mxlGetTime();
             auto current_index = mxlTimestampToIndex(&grain_rate_, now);
             read_index_ = current_index - READ_DELAY_GRAINS;
-            spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}",
-                        old_index, static_cast<int>(status), read_index_);
+            if (grains_processed_ == 0) {
+                spdlog::warn("Passthrough: realigned to index {}", read_index_);
+            }
         }
         return;
     }
@@ -94,8 +94,8 @@ void PassthroughNode::process() {
     read_index_++;
     grains_processed_++;
 
-    if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) {
-        spdlog::info("Passthrough: grain #{}, index={}, size={}", grains_processed_, grain_info.index, grain_info.grainSize);
+    if (grains_processed_ == 1) {
+        spdlog::info("Passthrough: first grain processed, index={}", grain_info.index);
     }
 }
 

tests/test_graph.cpp

@@ -68,9 +68,10 @@ TEST_CASE("FlowManager creates V210 flow definition", "[flow_manager]") {
     auto def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1);
 
     REQUIRE(def["id"] == flow_id);
-    REQUIRE(def["format"] == "video/v210");
-    REQUIRE(def["width"] == 1920);
-    REQUIRE(def["height"] == 1080);
+    REQUIRE(def["format"] == "urn:x-nmos:format:video");
+    REQUIRE(def["media_type"] == "video/v210");
+    REQUIRE(def["frame_width"] == 1920);
+    REQUIRE(def["frame_height"] == 1080);
 }
 
 TEST_CASE("Graph serialize", "[graph]") {