Rafal/node backend execution 2 execution scheduler pr to main

frontend/src/components/graph/AddNodeModal.tsx

@@ -1,11 +1,12 @@
-import { useState, useMemo, useRef, useEffect } from "react";
+import { useState, useMemo, useRef, useEffect, useCallback } from "react";
 import {
   Dialog,
   DialogContent,
   DialogHeader,
   DialogTitle,
   DialogDescription,
 } from "../ui/dialog";
+import type { FlowNodeData } from "../../lib/graphUtils";
 
 interface AddNodeModalProps {
   open: boolean;
@@ -36,8 +37,10 @@ interface AddNodeModalProps {
       | "tempo"
       | "prompt_list"
       | "prompt_blend"
-      | "scheduler",
-    subType?: string
+      | "scheduler"
+      | "custom_node",
+    subType?: string,
+    extraData?: Partial<FlowNodeData>
   ) => void;
 }
 
@@ -67,12 +70,15 @@ interface NodeCatalogItem {
     | "tempo"
     | "prompt_list"
     | "prompt_blend"
-    | "scheduler";
+    | "scheduler"
+    | "custom_node";
   subType?: string;
   name: string;
   description: string;
   color: string;
   category: string;
+  /** Full definition for custom nodes (inputs/outputs/params). */
+  customNodeDef?: Record<string, unknown>;
 }
 
 const NODE_CATALOG: NodeCatalogItem[] = [
@@ -85,8 +91,8 @@ const NODE_CATALOG: NodeCatalogItem[] = [
   },
   {
     type: "pipeline",
-    name: "Pipeline",
-    description: "Processing pipeline node",
+    name: "Node",
+    description: "Video processing node (pick a model after dropping it)",
     color: "#60a5fa",
     category: "I/O",
   },
@@ -294,7 +300,15 @@ const NODE_CATALOG: NodeCatalogItem[] = [
   },
 ];
 
-const CATEGORIES = ["All", "I/O", "Values", "Controls", "UI", "Utility"];
+const CATEGORIES = [
+  "All",
+  "I/O",
+  "Values",
+  "Controls",
+  "UI",
+  "Utility",
+  "Plugins",
+];
 
 interface TooltipState {
   text: string;
@@ -388,10 +402,50 @@ export function AddNodeModal({
 }: AddNodeModalProps) {
   const [searchText, setSearchText] = useState("");
   const [activeCategory, setActiveCategory] = useState("All");
+  const [customNodes, setCustomNodes] = useState<NodeCatalogItem[]>([]);
+
+  useEffect(() => {
+    if (!open) return;
+    fetch("/api/v1/nodes/definitions")
+      .then(r => r.json())
+      .then(data => {
+        // The unified endpoint returns both pipelines (pipeline_meta != null)
+        // and plain custom nodes. Pipelines are still added via the hardcoded
+        // "Pipeline" catalog entry (placeholder + dropdown); the scheduler
+        // has its own catalog entry with a bespoke widget. Filter both out
+        // of the plugin listing to avoid duplication.
+        const items: NodeCatalogItem[] = (data.nodes ?? [])
+
+          .filter(
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            (n: any) =>
+              n.pipeline_meta == null && n.node_type_id !== "scheduler"
+          )
+          // eslint-disable-next-line @typescript-eslint/no-explicit-any
+          .map((n: any) => ({
+            type: "custom_node" as const,
+            subType: n.node_type_id,
+            name: n.display_name || n.node_type_id,
+            description: n.description || "",
+            color: "#9ca3af",
+            category: "Plugins",
+            customNodeDef: n,
+          }));
+        setCustomNodes(items);
+      })
+      .catch(() => {
+        /* ignore — custom nodes just won't appear */
+      });
+  }, [open]);
+
+  const fullCatalog = useMemo(
+    () => [...NODE_CATALOG, ...customNodes],
+    [customNodes]
+  );
 
   const filteredItems = useMemo(() => {
     const lowerSearch = searchText.toLowerCase();
-    return NODE_CATALOG.filter(item => {
+    return fullCatalog.filter(item => {
       const matchesSearch =
         !lowerSearch ||
         item.name.toLowerCase().includes(lowerSearch) ||
@@ -400,14 +454,37 @@ export function AddNodeModal({
         activeCategory === "All" || item.category === activeCategory;
       return matchesSearch && matchesCategory;
     });
-  }, [searchText, activeCategory]);
+  }, [searchText, activeCategory, fullCatalog]);
 
-  const handleSelect = (item: NodeCatalogItem) => {
-    onSelectNodeType(item.type, item.subType);
-    onClose();
-    setSearchText("");
-    setActiveCategory("All");
-  };
+  const handleSelect = useCallback(
+    (item: NodeCatalogItem) => {
+      if (item.type === "custom_node" && item.customNodeDef) {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        const def = item.customNodeDef as any;
+        onSelectNodeType("custom_node", item.subType, {
+          customNodeTypeId: def.node_type_id,
+          customNodeDisplayName: def.display_name || def.node_type_id,
+          customNodeCategory: def.category || "",
+          customNodeInputs: def.inputs || [],
+          customNodeOutputs: def.outputs || [],
+          customNodeParamDefs: def.params || [],
+          customNodeParams: Object.fromEntries(
+            (def.params || [])
+              // eslint-disable-next-line @typescript-eslint/no-explicit-any
+              .filter((p: any) => p.default != null)
+              // eslint-disable-next-line @typescript-eslint/no-explicit-any
+              .map((p: any) => [p.name, p.default])
+          ),
+        });
+      } else {
+        onSelectNodeType(item.type, item.subType);
+      }
+      onClose();
+      setSearchText("");
+      setActiveCategory("All");
+    },
+    [onSelectNodeType, onClose]
+  );
 
   const handleClose = () => {
     onClose();

frontend/src/components/graph/GraphEditor.tsx

@@ -55,6 +55,7 @@ import { TempoNode } from "./nodes/TempoNode";
 import { PromptListNode } from "./nodes/PromptListNode";
 import { PromptBlendNode } from "./nodes/PromptBlendNode";
 import { SchedulerNode } from "./nodes/SchedulerNode";
+import { CustomNode } from "./nodes/CustomNode";
 import { CustomEdge } from "./CustomEdge";
 import { ContextMenu } from "./ContextMenu";
 import { AddNodeModal } from "./AddNodeModal";
@@ -130,6 +131,7 @@ const nodeTypes = {
   prompt_list: PromptListNode,
   prompt_blend: PromptBlendNode,
   scheduler: SchedulerNode,
+  custom_node: CustomNode,
 };
 
 const edgeTypes = {

frontend/src/components/graph/contextMenuItems.tsx

@@ -96,10 +96,10 @@ export function buildPaneMenuItems(deps: {
       keywords: ["input", "camera", "video"],
     },
     {
-      label: "Pipeline",
+      label: "Node",
       icon: <Workflow />,
       onClick: () => handleNodeTypeSelect("pipeline"),
-      keywords: ["process", "effect", "filter"],
+      keywords: ["process", "effect", "filter", "pipeline"],
     },
     {
       label: "Sink",

frontend/src/components/graph/hooks/graph/useGraphPersistence.ts

@@ -7,8 +7,8 @@ import {
   parseHandleId,
 } from "../../../../lib/graphUtils";
 import type { FlowNodeData } from "../../../../lib/graphUtils";
-import type { PluginInfo } from "../../../../lib/api";
-import { resolveWorkflow } from "../../../../lib/api";
+import type { PluginInfo, NodeDefinitionDto } from "../../../../lib/api";
+import { resolveWorkflow, fetchNodeDefinitions } from "../../../../lib/api";
 import type {
   ScopeWorkflow,
   WorkflowResolutionPlan,
@@ -65,6 +65,72 @@ function clearGraphFromLocalStorage(): void {
   }
 }
 
+/**
+ * After loading or importing a workflow, fetch `/api/v1/nodes/definitions`
+ * and hydrate each custom_node flow node with its inputs/outputs/params/
+ * display metadata. Saved workflows only persist `node_type_id` and
+ * `params`, so the port definitions have to be re-attached before render.
+ * User-supplied param values override the defaults from the definition.
+ *
+ * Accepts an AbortSignal so rapid reloads / unmounts can cancel an
+ * in-flight fetch before its setNodes callback stomps on newer state.
+ */
+function hydrateCustomNodeDefinitions(
+  nodes: Node<FlowNodeData>[],
+  setNodes: React.Dispatch<React.SetStateAction<Node<FlowNodeData>[]>>,
+  signal: AbortSignal
+): void {
+  const customFlowNodes = nodes.filter(
+    n => n.data.nodeType === "custom_node" && !n.data.customNodeInputs
+  );
+  if (customFlowNodes.length === 0) return;
+  fetchNodeDefinitions({ signal })
+    .then(data => {
+      if (signal.aborted) return;
+      const defMap = new Map<string, NodeDefinitionDto>(
+        data.nodes.map(d => [d.node_type_id, d])
+      );
+      setNodes(prev => {
+        if (signal.aborted) return prev;
+        return prev.map(n => {
+          if (
+            n.data.nodeType !== "custom_node" ||
+            !n.data.customNodeTypeId ||
+            n.data.customNodeInputs
+          ) {
+            return n;
+          }
+          const def = defMap.get(n.data.customNodeTypeId);
+          if (!def) return n;
+          return {
+            ...n,
+            data: {
+              ...n.data,
+              customNodeDisplayName: def.display_name,
+              customNodeCategory: def.category,
+              customNodeInputs: def.inputs ?? [],
+              customNodeOutputs: def.outputs ?? [],
+              customNodeParamDefs: def.params ?? [],
+              customNodeParams: {
+                ...Object.fromEntries(
+                  (def.params ?? [])
+                    .filter(p => p.default != null)
+                    .map(p => [p.name, p.default] as const)
+                ),
+                // User-edited values take precedence over definition defaults
+                ...(n.data.customNodeParams || {}),
+              },
+            },
+          };
+        });
+      });
+    })
+    .catch((err: unknown) => {
+      if (err instanceof DOMException && err.name === "AbortError") return;
+      // custom nodes just won't be hydrated; render falls back to placeholders
+    });
+}
+
 interface UseGraphPersistenceArgs {
   nodes: Node<FlowNodeData>[];
   edges: Edge[];
@@ -129,6 +195,25 @@ export function useGraphPersistence({
   // to localStorage so we skip the expensive save when nothing changed.
   const lastSavedJsonRef = useRef<string>("");
 
+  // AbortController for in-flight custom-node hydration fetches. Aborted
+  // before each new hydrate and on unmount so a stale /api/v1/nodes/definitions
+  // response can't overwrite newer nodes state.
+  const hydrateAbortRef = useRef<AbortController | null>(null);
+  const startHydrate = useCallback(
+    (initialNodes: Node<FlowNodeData>[]) => {
+      hydrateAbortRef.current?.abort();
+      const controller = new AbortController();
+      hydrateAbortRef.current = controller;
+      hydrateCustomNodeDefinitions(initialNodes, setNodes, controller.signal);
+    },
+    [setNodes]
+  );
+  useEffect(() => {
+    return () => {
+      hydrateAbortRef.current?.abort();
+    };
+  }, []);
+
   const loadGraph = useCallback(() => {
     if (Object.keys(portsMap).length === 0) return;
     resetNavigationRef.current?.();
@@ -182,6 +267,8 @@ export function useGraphPersistence({
           }, 0);
         }
 
+        startHydrate(enriched);
+
         // Allow async side-effects (e.g. source mode restore) to settle
         // before re-enabling change notifications.
         setTimeout(() => {
@@ -202,6 +289,7 @@ export function useGraphPersistence({
     setEdges,
     setNodeParams,
     setNodes,
+    startHydrate,
   ]);
 
   useEffect(() => {
@@ -374,6 +462,8 @@ export function useGraphPersistence({
           }
         }, 0);
       }
+
+      startHydrate(enriched);
     },
     [
       portsMap,
@@ -384,6 +474,7 @@ export function useGraphPersistence({
       setNodeParams,
       enrichDepsRef,
       resetNavigationRef,
+      startHydrate,
     ]
   );
 

frontend/src/components/graph/hooks/node/useNodeFactories.ts

@@ -45,7 +45,8 @@
   | "tempo"
   | "prompt_list"
   | "prompt_blend"
-  | "scheduler";
+  | "scheduler"
+  | "custom_node";
 
 interface NodeDefaults {
   /** The React Flow node `type` */
@@ -471,6 +472,15 @@
       ],
     },
   },
+  custom_node: {
+    type: "custom_node",
+    idPrefix: "custom",
+    defaultX: 300,
+    data: {
+      label: "Custom Node",
+      nodeType: "custom_node" as const,
+    },
+  },
 };
 
 interface UseNodeFactoriesArgs {
@@ -566,8 +576,10 @@
         | "tempo"
         | "prompt_list"
         | "prompt_blend"
-        | "scheduler",
-      subType?: string
+        | "scheduler"
+        | "custom_node",
+      subType?: string,
+      extraData?: Partial<FlowNodeData>
     ) => {
       if (!pendingNodePosition) return;
 
@@ -598,13 +610,19 @@
           outputSinkType: defaultType,
           outputSinkName: defaultNames[defaultType] || "Scope",
         });
+      } else if (type === "custom_node") {
+        addNode("custom_node", pendingNodePosition, {
+          customNodeTypeId: subType,
+          label: subType || "Custom Node",
+          ...extraData,
+        });
       } else {
         addNode(type as NodeTypeKey, pendingNodePosition);
       }
 
       setPendingNodePosition(null);
     },
     [
       nodes,
       pendingNodePosition,
       addNode,