diff --git a/frontend/src/components/graph/AddNodeModal.tsx b/frontend/src/components/graph/AddNodeModal.tsx
index 59e91ed89..65588a9d2 100644
--- a/frontend/src/components/graph/AddNodeModal.tsx
+++ b/frontend/src/components/graph/AddNodeModal.tsx
@@ -1,4 +1,4 @@
-import { useState, useMemo, useRef, useEffect } from "react";
+import { useState, useMemo, useRef, useEffect, useCallback } from "react";
 import {
   Dialog,
   DialogContent,
@@ -6,6 +6,7 @@ import {
   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,45 @@ 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), so we filter
+        // them out of the plugin listing here to avoid duplication.
+        const items: NodeCatalogItem[] = (data.nodes ?? [])
+          // eslint-disable-next-line @typescript-eslint/no-explicit-any
+          .filter((n: any) => n.pipeline_meta == null)
+          // 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 +449,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();
diff --git a/frontend/src/components/graph/GraphEditor.tsx b/frontend/src/components/graph/GraphEditor.tsx
index 15446cc80..18b71338a 100644
--- a/frontend/src/components/graph/GraphEditor.tsx
+++ b/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 = {
diff --git a/frontend/src/components/graph/contextMenuItems.tsx b/frontend/src/components/graph/contextMenuItems.tsx
index 0d4903f9e..caeadd698 100644
--- a/frontend/src/components/graph/contextMenuItems.tsx
+++ b/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",
diff --git a/frontend/src/components/graph/hooks/graph/useGraphPersistence.ts b/frontend/src/components/graph/hooks/graph/useGraphPersistence.ts
index 8a2248d31..f31df93f8 100644
--- a/frontend/src/components/graph/hooks/graph/useGraphPersistence.ts
+++ b/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,
     ]
   );
 
diff --git a/frontend/src/components/graph/hooks/node/useNodeFactories.ts b/frontend/src/components/graph/hooks/node/useNodeFactories.ts
index 36982fa76..60803a265 100644
--- a/frontend/src/components/graph/hooks/node/useNodeFactories.ts
+++ b/frontend/src/components/graph/hooks/node/useNodeFactories.ts
@@ -44,7 +44,8 @@ type NodeTypeKey =
   | "tempo"
   | "prompt_list"
   | "prompt_blend"
-  | "scheduler";
+  | "scheduler"
+  | "custom_node";
 
 interface NodeDefaults {
   /** The React Flow node `type` */
@@ -470,6 +471,15 @@ const NODE_DEFAULTS: Record<NodeTypeKey, NodeDefaults> = {
       ],
     },
   },
+  custom_node: {
+    type: "custom_node",
+    idPrefix: "custom",
+    defaultX: 300,
+    data: {
+      label: "Custom Node",
+      nodeType: "custom_node" as const,
+    },
+  },
 };
 
 interface UseNodeFactoriesArgs {
@@ -565,8 +575,10 @@ export function useNodeFactories({
         | "tempo"
         | "prompt_list"
         | "prompt_blend"
-        | "scheduler",
-      subType?: string
+        | "scheduler"
+        | "custom_node",
+      subType?: string,
+      extraData?: Partial<FlowNodeData>
     ) => {
       if (!pendingNodePosition) return;
 
@@ -597,6 +609,12 @@ export function useNodeFactories({
           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);
       }
diff --git a/frontend/src/components/graph/nodes/CustomNode.tsx b/frontend/src/components/graph/nodes/CustomNode.tsx
new file mode 100644
index 000000000..a051674af
--- /dev/null
+++ b/frontend/src/components/graph/nodes/CustomNode.tsx
@@ -0,0 +1,223 @@
+import { Handle, Position } from "@xyflow/react";
+import type { NodeProps, Node } from "@xyflow/react";
+import type { FlowNodeData } from "../../../lib/graphUtils";
+import { buildHandleId } from "../../../lib/graphUtils";
+import { useNodeData } from "../hooks/node/useNodeData";
+import { useNodeCollapse } from "../hooks/node/useNodeCollapse";
+import { NodeCard, NodeHeader, NodeBody, collapsedHandleStyle } from "../ui";
+
+type CustomNodeType = Node<FlowNodeData, "custom_node">;
+
+/* Port type -> color mapping for custom types */
+const PORT_COLORS: Record<string, string> = {
+  audio: "#22c55e",
+  video: "#eeeeee",
+  number: "#38bdf8",
+  string: "#fbbf24",
+  boolean: "#34d399",
+  trigger: "#f97316",
+  latent: "#a855f7",
+  model: "#f59e0b",
+  vae: "#f59e0b",
+  clip: "#f59e0b",
+  conditioning: "#3b82f6",
+  semantic_hints: "#06b6d4",
+  config: "#6b7280",
+  curve: "#ec4899",
+  mask: "#ef4444",
+  lora: "#f472b6",
+};
+
+function portColor(portType: string): string {
+  return PORT_COLORS[portType] ?? "#9ca3af";
+}
+
+export function CustomNode({ id, data, selected }: NodeProps<CustomNodeType>) {
+  const { updateData } = useNodeData(id);
+  const { collapsed, toggleCollapse } = useNodeCollapse();
+
+  const inputs = data.customNodeInputs ?? [];
+  const outputs = data.customNodeOutputs ?? [];
+  const params = data.customNodeParamDefs ?? [];
+  const displayName =
+    data.customTitle ||
+    data.customNodeDisplayName ||
+    data.customNodeTypeId ||
+    "Custom Node";
+  const category = data.customNodeCategory ?? "";
+  const onParameterChange = data.onParameterChange as
+    | ((nodeId: string, key: string, value: unknown) => void)
+    | undefined;
+
+  const setParam = (name: string, value: unknown) => {
+    updateData({
+      customNodeParams: {
+        ...data.customNodeParams,
+        [name]: value,
+      },
+    });
+    onParameterChange?.(id, name, value);
+  };
+
+  return (
+    <NodeCard
+      selected={selected}
+      collapsed={collapsed}
+      autoMinHeight={!collapsed}
+    >
+      <NodeHeader
+        title={displayName}
+        onTitleChange={t => updateData({ customTitle: t })}
+        collapsed={collapsed}
+        onCollapseToggle={toggleCollapse}
+      />
+      {!collapsed && (
+        <NodeBody>
+          {/* Show category badge */}
+          {category && (
+            <div className="px-2 pb-1">
+              <span className="text-[10px] text-zinc-500 uppercase tracking-wider">
+                {category}
+              </span>
+            </div>
+          )}
+          {/* Show input ports */}
+          {inputs.length > 0 && (
+            <div className="flex flex-col gap-0.5 px-2 py-1">
+              {inputs.map(p => (
+                <div
+                  key={p.name}
+                  className="text-[11px] text-zinc-400 flex items-center gap-1"
+                >
+                  <span
+                    className="w-2 h-2 rounded-full inline-block"
+                    style={{ backgroundColor: portColor(p.port_type) }}
+                  />
+                  {p.name}
+                  <span className="text-zinc-600 text-[9px]">
+                    {p.port_type}
+                  </span>
+                </div>
+              ))}
+            </div>
+          )}
+          {/* Show output ports */}
+          {outputs.length > 0 && (
+            <div className="flex flex-col gap-0.5 px-2 py-1">
+              {outputs.map(p => (
+                <div
+                  key={p.name}
+                  className="text-[11px] text-zinc-400 flex items-center gap-1 justify-end"
+                >
+                  <span className="text-zinc-600 text-[9px]">
+                    {p.port_type}
+                  </span>
+                  {p.name}
+                  <span
+                    className="w-2 h-2 rounded-full inline-block"
+                    style={{ backgroundColor: portColor(p.port_type) }}
+                  />
+                </div>
+              ))}
+            </div>
+          )}
+          {/* Parameter widgets (ComfyUI-style editable params) */}
+          {params.length > 0 && (
+            <div className="flex flex-col gap-1 px-2 py-1 border-t border-zinc-800">
+              {params.map(p => {
+                const val = data.customNodeParams?.[p.name] ?? p.default ?? "";
+                return (
+                  <div
+                    key={p.name}
+                    className="flex items-center justify-between gap-2 text-[11px]"
+                  >
+                    <span
+                      className="text-zinc-500 shrink-0"
+                      title={p.description || p.name}
+                    >
+                      {p.description || p.name}
+                    </span>
+                    {p.param_type === "select" &&
+                    Array.isArray(p.ui?.options) ? (
+                      <select
+                        className="bg-zinc-900 text-zinc-200 rounded px-1 py-0.5 text-[11px] max-w-[130px]"
+                        value={String(val)}
+                        onChange={e => setParam(p.name, e.target.value)}
+                      >
+                        {(p.ui?.options as string[]).map(o => (
+                          <option key={o} value={o}>
+                            {o}
+                          </option>
+                        ))}
+                      </select>
+                    ) : p.param_type === "boolean" ? (
+                      <input
+                        type="checkbox"
+                        checked={Boolean(val)}
+                        onChange={e => setParam(p.name, e.target.checked)}
+                        className="accent-blue-500"
+                      />
+                    ) : p.param_type === "number" ? (
+                      <input
+                        type="number"
+                        className="bg-zinc-900 text-zinc-200 rounded px-1 py-0.5 text-[11px] w-[80px]"
+                        value={Number(val)}
+                        min={(p.ui?.min as number | undefined) ?? undefined}
+                        max={(p.ui?.max as number | undefined) ?? undefined}
+                        step={(p.ui?.step as number | undefined) ?? undefined}
+                        onChange={e => setParam(p.name, Number(e.target.value))}
+                      />
+                    ) : (
+                      <input
+                        type="text"
+                        className="bg-zinc-900 text-zinc-200 rounded px-1 py-0.5 text-[11px] max-w-[130px]"
+                        value={String(val)}
+                        onChange={e => setParam(p.name, e.target.value)}
+                      />
+                    )}
+                  </div>
+                );
+              })}
+            </div>
+          )}
+        </NodeBody>
+      )}
+
+      {/* Input handles (left side) */}
+      {inputs.map((p, i) => (
+        <Handle
+          key={`in-${p.name}`}
+          type="target"
+          position={Position.Left}
+          id={buildHandleId("stream", p.name)}
+          style={{
+            background: portColor(p.port_type),
+            top: collapsed ? "50%" : `${44 + (category ? 18 : 0) + i * 18}px`,
+            width: 8,
+            height: 8,
+            ...(collapsed ? collapsedHandleStyle : {}),
+          }}
+        />
+      ))}
+
+      {/* Output handles (right side) */}
+      {outputs.map((p, i) => (
+        <Handle
+          key={`out-${p.name}`}
+          type="source"
+          position={Position.Right}
+          id={buildHandleId("stream", p.name)}
+          style={{
+            background: portColor(p.port_type),
+            top: collapsed
+              ? "50%"
+              : `${44 + (category ? 18 : 0) + (inputs.length > 0 ? inputs.length * 18 + 4 : 0) + i * 18}px`,
+            width: 8,
+            height: 8,
+            ...(collapsed ? collapsedHandleStyle : {}),
+          }}
+        />
+      ))}
+    </NodeCard>
+  );
+}
diff --git a/frontend/src/components/graph/nodes/PipelineNode.tsx b/frontend/src/components/graph/nodes/PipelineNode.tsx
index 7b8408448..4cd461886 100644
--- a/frontend/src/components/graph/nodes/PipelineNode.tsx
+++ b/frontend/src/components/graph/nodes/PipelineNode.tsx
@@ -84,7 +84,7 @@ export function PipelineNode({
   const supportsLoRA = data.supportsLoRA ?? false;
   const isStreaming = data.isStreaming ?? false;
 
-  const pipelineName = data.pipelineId || "Pipeline";
+  const pipelineName = data.pipelineId || "Node";
 
   // Inject unavailable pipelineId into options
   const isUnavailable =
@@ -187,8 +187,8 @@ export function PipelineNode({
       />
       {!collapsed && (
         <NodeBody withGap>
-          {/* Pipeline selector */}
-          <NodeParamRow label="Pipeline">
+          {/* Node type selector (video pipeline model) */}
+          <NodeParamRow label="Model">
             <NodePillSearchableSelect
               value={data.pipelineId || ""}
               onChange={newValue => {
diff --git a/frontend/src/components/graph/nodes/SinkNode.tsx b/frontend/src/components/graph/nodes/SinkNode.tsx
index 198695265..92dd2b166 100644
--- a/frontend/src/components/graph/nodes/SinkNode.tsx
+++ b/frontend/src/components/graph/nodes/SinkNode.tsx
@@ -23,6 +23,7 @@ export function SinkNode({ id, data, selected }: NodeProps<SinkNodeType>) {
   const isPlaying = (data.isPlaying as boolean | undefined) ?? true;
   const onPlayPauseToggle = data.onPlayPauseToggle as (() => void) | undefined;
   const videoRef = useRef<HTMLVideoElement>(null);
+  const audioRef = useRef<HTMLAudioElement>(null);
   const [videoSize, setVideoSize] = useState<{
     width: number;
     height: number;
@@ -40,14 +41,26 @@ export function SinkNode({ id, data, selected }: NodeProps<SinkNodeType>) {
   }, []);
 
   useEffect(() => {
-    if (videoRef.current && remoteStream instanceof MediaStream) {
-      videoRef.current.srcObject = remoteStream;
+    if (remoteStream instanceof MediaStream) {
+      if (videoRef.current) {
+        videoRef.current.srcObject = remoteStream;
+      }
       setHasAudioTrack(remoteStream.getAudioTracks().length > 0);
       setHasVideoTrack(remoteStream.getVideoTracks().length > 0);
 
+      // Also attach to audio element for reliable audio-only playback
+      if (audioRef.current) {
+        audioRef.current.srcObject = remoteStream;
+        audioRef.current.play().catch(() => {});
+      }
+
       const handleTrackAdded = () => {
         setHasAudioTrack(remoteStream.getAudioTracks().length > 0);
         setHasVideoTrack(remoteStream.getVideoTracks().length > 0);
+        if (audioRef.current && audioRef.current.srcObject !== remoteStream) {
+          audioRef.current.srcObject = remoteStream;
+          audioRef.current.play().catch(() => {});
+        }
       };
       remoteStream.addEventListener("addtrack", handleTrackAdded);
       return () => {
@@ -60,8 +73,18 @@ export function SinkNode({ id, data, selected }: NodeProps<SinkNodeType>) {
     if (videoRef.current) {
       videoRef.current.muted = isMuted;
     }
+    if (audioRef.current) {
+      audioRef.current.muted = isMuted;
+    }
   }, [isMuted]);
 
+  // Auto-unmute when audio arrives without video (audio-only workflow)
+  useEffect(() => {
+    if (hasAudioTrack && !hasVideoTrack) {
+      setIsMuted(false);
+    }
+  }, [hasAudioTrack, hasVideoTrack]);
+
   const toggleMute = useCallback((e: React.MouseEvent) => {
     e.stopPropagation();
     setIsMuted(prev => !prev);
@@ -116,6 +139,13 @@ export function SinkNode({ id, data, selected }: NodeProps<SinkNodeType>) {
                   playsInline
                   onResize={handleResize}
                 />
+                {/* Always-present audio element for reliable audio playback */}
+                <audio
+                  ref={audioRef}
+                  autoPlay
+                  muted={isMuted}
+                  style={{ display: "none" }}
+                />
                 {!hasVideoTrack && (
                   <div className="flex flex-col items-center justify-center h-full gap-1 text-[#8c8c8d]">
                     <Volume2 className="h-5 w-5" />
@@ -237,6 +267,17 @@ export function SinkNode({ id, data, selected }: NodeProps<SinkNodeType>) {
             : { top: handleY, left: 0, backgroundColor: "#ffffff" }
         }
       />
+      <Handle
+        type="target"
+        position={Position.Left}
+        id="stream:audio"
+        className="!w-2.5 !h-2.5 !border-0"
+        style={
+          collapsed
+            ? collapsedHandleStyle("left")
+            : { top: (handleY ?? 0) + 16, left: 0, backgroundColor: "#22c55e" }
+        }
+      />
       <Handle
         type="source"
         position={Position.Right}
diff --git a/frontend/src/components/graph/utils/connectionValidation.ts b/frontend/src/components/graph/utils/connectionValidation.ts
index 1a3bf183c..475bb3a1e 100644
--- a/frontend/src/components/graph/utils/connectionValidation.ts
+++ b/frontend/src/components/graph/utils/connectionValidation.ts
@@ -302,9 +302,32 @@ export function validateConnection(
     return true;
   }
 
-  // Stream ↔ stream always ok
-  if (sourceParsed.kind === "stream" && targetParsed.kind === "stream")
-    return true;
+  // Stream ↔ stream: for custom_node edges, enforce port-type matching
+  // against the node's declared inputs/outputs. For built-in source /
+  // pipeline / sink nodes, streams are untyped (video) and always ok.
+  if (sourceParsed.kind === "stream" && targetParsed.kind === "stream") {
+    const sourceNode = nodes.find(n => n.id === connection.source);
+    const targetNode = nodes.find(n => n.id === connection.target);
+    if (!sourceNode || !targetNode) return true;
+    const srcIsCustom = sourceNode.data.nodeType === "custom_node";
+    const tgtIsCustom = targetNode.data.nodeType === "custom_node";
+    if (!srcIsCustom && !tgtIsCustom) return true;
+
+    const srcType = srcIsCustom
+      ? sourceNode.data.customNodeOutputs?.find(
+          p => p.name === sourceParsed.name
+        )?.port_type
+      : undefined;
+    const tgtType = tgtIsCustom
+      ? targetNode.data.customNodeInputs?.find(
+          p => p.name === targetParsed.name
+        )?.port_type
+      : undefined;
+
+    // If we can't look up both types, allow (assume compatible).
+    if (!srcType || !tgtType) return true;
+    return srcType === tgtType;
+  }
 
   // Param ↔ param
   if (sourceParsed.kind === "param" && targetParsed.kind === "param") {
diff --git a/frontend/src/components/graph/utils/nodeEnrichment.ts b/frontend/src/components/graph/utils/nodeEnrichment.ts
index 00b264ac6..16476021b 100644
--- a/frontend/src/components/graph/utils/nodeEnrichment.ts
+++ b/frontend/src/components/graph/utils/nodeEnrichment.ts
@@ -155,6 +155,16 @@ export function enrichNodes(
         },
       };
     }
+    if (n.data.nodeType === "custom_node") {
+      return {
+        ...n,
+        data: {
+          ...n.data,
+          onParameterChange: deps.handleNodeParameterChange,
+          isStreaming: deps.isStreaming,
+        },
+      };
+    }
     if (n.data.nodeType === "source") {
       // Per-node stream if available (multi-source), else fall back to global
       const nodeStream = deps.localStreams?.[n.id] ?? deps.localStream;
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts
index fc8ce79c5..66a6227d9 100644
--- a/frontend/src/lib/api.ts
+++ b/frontend/src/lib/api.ts
@@ -907,8 +907,12 @@ export const deleteApiKey = async (
 
 export interface GraphNode {
   id: string;
-  type: "source" | "pipeline" | "sink" | "record";
+  type: "source" | "pipeline" | "sink" | "record" | "node";
   pipeline_id?: string | null;
+  /** Node type ID (NodeRegistry key) when type is "node" */
+  node_type_id?: string | null;
+  /** Per-node parameter values for custom nodes */
+  params?: Record<string, unknown> | null;
   x?: number | null;
   y?: number | null;
   w?: number | null;
@@ -920,6 +924,63 @@ export interface GraphNode {
   sink_name?: string | null;
 }
 
+export interface NodePortDef {
+  name: string;
+  port_type: string;
+  required?: boolean;
+  description?: string;
+  default_value?: unknown;
+}
+
+export interface NodeParamDef {
+  name: string;
+  param_type: "number" | "string" | "boolean" | "select";
+  default?: unknown;
+  description?: string;
+  /**
+   * Free-form widget hints (``min``/``max``/``step`` for number,
+   * ``options`` for select, …). Keeps widget-specific fields off the
+   * base schema; the frontend renderer dispatches on ``param_type``
+   * and reads whichever ``ui`` keys apply.
+   */
+  ui?: Record<string, unknown> | null;
+  convertible_to_input?: boolean;
+}
+
+export interface NodeDefinitionDto {
+  node_type_id: string;
+  display_name: string;
+  category: string;
+  description: string;
+  inputs: NodePortDef[];
+  outputs: NodePortDef[];
+  params: NodeParamDef[];
+  continuous: boolean;
+  /**
+   * Rich pipeline-only metadata (config_schema, mode_defaults,
+   * supports_lora, supports_vace, etc.) populated for entries whose
+   * underlying class is a Pipeline subclass. ``null`` for plain nodes.
+   */
+  pipeline_meta?: Record<string, unknown> | null;
+}
+
+export interface NodeDefinitionsResponse {
+  nodes: NodeDefinitionDto[];
+}
+
+export const fetchNodeDefinitions = async (
+  options: { signal?: AbortSignal } = {}
+): Promise<NodeDefinitionsResponse> => {
+  const response = await fetch("/api/v1/nodes/definitions", {
+    method: "GET",
+    signal: options.signal,
+  });
+  if (!response.ok) {
+    throw new Error(`Failed to fetch node definitions: ${response.statusText}`);
+  }
+  return response.json();
+};
+
 export interface GraphEdge {
   from: string;
   from_port: string;
diff --git a/frontend/src/lib/graphUtils.ts b/frontend/src/lib/graphUtils.ts
index 861df74da..189ab5708 100644
--- a/frontend/src/lib/graphUtils.ts
+++ b/frontend/src/lib/graphUtils.ts
@@ -5,6 +5,8 @@ import type {
   GraphEdge,
   PipelineSchemaInfo,
   LoRAFileInfo,
+  NodePortDef,
+  NodeParamDef,
 } from "./api";
 import { inferPrimitiveFieldType } from "./schemaSettings";
 import { resolveLoRAPath } from "./workflowSettings";
@@ -104,7 +106,8 @@ export interface FlowNodeData {
     | "prompt_list"
     | "prompt_blend"
     | "scheduler"
-    | "audio";
+    | "audio"
+    | "custom_node";
   availablePipelineIds?: string[];
   /** Declared input ports for the selected pipeline */
   streamInputs?: string[];
@@ -384,6 +387,22 @@ export interface FlowNodeData {
   /* ── Tempo beat count offset ── */
   tempoBeatCountOffset?: number;
 
+  /* ── Custom node fields ── */
+  /** For custom_node: the node_type_id from the backend registry */
+  customNodeTypeId?: string;
+  /** For custom_node: display name from node definition */
+  customNodeDisplayName?: string;
+  /** For custom_node: category from node definition */
+  customNodeCategory?: string;
+  /** For custom_node: input port definitions */
+  customNodeInputs?: NodePortDef[];
+  /** For custom_node: output port definitions */
+  customNodeOutputs?: NodePortDef[];
+  /** For custom_node: current parameter values (user-editable) */
+  customNodeParams?: Record<string, unknown>;
+  /** For custom_node: parameter definitions from API (widget metadata) */
+  customNodeParamDefs?: NodeParamDef[];
+
   /* ── Node lock / pin / collapse ── */
   /** When true, parameter inputs on this node are disabled (read-only). */
   locked?: boolean;
@@ -779,6 +798,40 @@ export function graphConfigToFlow(
     });
   });
 
+  // Backend custom nodes (type="node"). Port metadata is hydrated later
+  // from GET /api/v1/nodes/definitions in useGraphPersistence.
+  const customNodes = graph.nodes.filter(
+    n => n.type === "node" && !isSubgraphInnerNode(n.id)
+  );
+  customNodes.forEach((n, i) => {
+    const savedX = n.x ?? undefined;
+    const savedY = n.y ?? undefined;
+    const sizeProps =
+      n.w != null || n.h != null
+        ? {
+            width: n.w ?? undefined,
+            height: n.h ?? undefined,
+            style: { width: n.w ?? undefined, height: n.h ?? undefined },
+          }
+        : {};
+    nodes.push({
+      id: n.id,
+      type: "custom_node",
+      position: {
+        x: savedX !== undefined ? savedX : START_X + COLUMN_GAP * 1.5,
+        y:
+          savedY !== undefined ? savedY : START_Y + i * (NODE_HEIGHT + ROW_GAP),
+      },
+      ...sizeProps,
+      data: {
+        label: n.node_type_id || n.id,
+        nodeType: "custom_node",
+        customNodeTypeId: n.node_type_id ?? undefined,
+        customNodeParams: (n.params as Record<string, unknown>) ?? undefined,
+      },
+    });
+  });
+
   // Convert edges - add stream: prefix to handle IDs
   // Skip edges that reference flattened inner subgraph nodes
   const edges: Edge[] = graph.edges
@@ -1208,11 +1261,21 @@ export function flowToGraphConfig(
             ? "sink"
             : n.data.nodeType === "record"
               ? "record"
-              : "pipeline",
+              : n.data.nodeType === "custom_node"
+                ? "node"
+                : "pipeline",
       pipeline_id:
         n.data.nodeType === "pipeline"
           ? (n.data.pipelineId ?? null)
           : undefined,
+      node_type_id:
+        n.data.nodeType === "custom_node"
+          ? (n.data.customNodeTypeId ?? null)
+          : undefined,
+      params:
+        n.data.nodeType === "custom_node" && n.data.customNodeParams
+          ? n.data.customNodeParams
+          : undefined,
       x: n.position.x,
       y: n.position.y,
       w: w && !Number.isNaN(w) ? w : undefined,
diff --git a/frontend/src/pages/StreamPage.tsx b/frontend/src/pages/StreamPage.tsx
index c9db65363..9aad87463 100644
--- a/frontend/src/pages/StreamPage.tsx
+++ b/frontend/src/pages/StreamPage.tsx
@@ -2707,10 +2707,13 @@ export function StreamPage() {
         );
       }
 
-      const loadSuccess = await loadPipeline(loadItems);
-      if (!loadSuccess) {
-        console.error("Failed to load pipeline, cannot start stream");
-        return false;
+      // Node-only graphs (no pipeline nodes) skip the pipeline load step.
+      if (loadItems.length > 0) {
+        const loadSuccess = await loadPipeline(loadItems);
+        if (!loadSuccess) {
+          console.error("Failed to load pipeline, cannot start stream");
+          return false;
+        }
       }
 
       // Check video requirements based on input mode
@@ -2806,7 +2809,9 @@ export function StreamPage() {
       }
 
       // Pipeline chain: preprocessors + main pipeline (already built above)
-      initialParameters.pipeline_ids = pipelineIds;
+      if (pipelineIds.length > 0) {
+        initialParameters.pipeline_ids = pipelineIds;
+      }
 
       // Media modalities from pipeline status — used by the backend to decide
       // which tracks to create (avoids unnecessary audio processing for
diff --git a/src/scope/core/nodes/__init__.py b/src/scope/core/nodes/__init__.py
new file mode 100644
index 000000000..60ce40162
--- /dev/null
+++ b/src/scope/core/nodes/__init__.py
@@ -0,0 +1,29 @@
+"""Backend node system for Scope.
+
+Provides a base class and registry for defining fine-grained processing
+nodes that can be wired into pipeline graphs. Nodes are simpler than full
+pipelines — they declare typed input/output ports, editable parameters,
+and a small execution contract. Built-in nodes and plugin-provided nodes
+are discovered here and rendered generically by the frontend via
+``GET /api/v1/nodes/definitions``.
+"""
+
+from .base import BaseNode, NodeDefinition, NodeParam, NodePort
+from .builtins import AudioSourceNode
+from .registry import NodeRegistry
+
+
+def register_builtin_nodes() -> None:
+    """Register all built-in node types shipped with the foundation."""
+    NodeRegistry.register(AudioSourceNode)
+
+
+__all__ = [
+    "AudioSourceNode",
+    "BaseNode",
+    "NodeDefinition",
+    "NodeParam",
+    "NodePort",
+    "NodeRegistry",
+    "register_builtin_nodes",
+]
diff --git a/src/scope/core/nodes/base.py b/src/scope/core/nodes/base.py
new file mode 100644
index 000000000..96d70363b
--- /dev/null
+++ b/src/scope/core/nodes/base.py
@@ -0,0 +1,171 @@
+"""Base classes for the Scope node system.
+
+Nodes are lightweight, fine-grained processing units that can be wired into
+pipeline graphs alongside pipelines. Each node type declares typed input/
+output ports and editable parameters, and subclasses implement their own
+execution contract (which may differ between execution backends).
+
+This module intentionally keeps ``BaseNode`` minimal — only a class-level
+identifier and a ``get_definition()`` classmethod are required. Concrete
+execution backends (graph executor integration, event-driven runtime, etc.)
+layer their own abstract methods on top.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any, ClassVar, Literal
+
+from pydantic import BaseModel, Field
+
+
+class NodePort(BaseModel):
+    """Describes an input or output port on a node."""
+
+    name: str = Field(..., description="Port identifier (used in edge wiring)")
+    port_type: str = Field(
+        ...,
+        description=(
+            "Type of data carried by this port. Built-in types: "
+            "'audio', 'video', 'number', 'string', 'boolean', 'trigger'. "
+            "Plugins may define custom types (e.g. 'latent', 'model')."
+        ),
+    )
+    required: bool = Field(default=True, description="Whether this input is required")
+    description: str = Field(default="", description="Human-readable description")
+    default_value: Any = Field(default=None, description="Default value for inputs")
+
+
+class NodeParam(BaseModel):
+    """Describes an editable parameter (widget) on a node.
+
+    Parameters are user-configurable values that live on the node card.
+    Like ComfyUI widgets, a parameter may be overridden by connecting
+    an incoming wire to the corresponding input port — the widget then
+    becomes an input and the default value is ignored.
+
+    Widget-specific hints (number min/max/step, select options, etc.)
+    go into the free-form ``ui`` dict so the base schema doesn't grow
+    as new widget kinds are added. The frontend renderer dispatches on
+    ``param_type`` and reads whichever ``ui`` keys apply.
+    """
+
+    name: str = Field(..., description="Parameter identifier")
+    param_type: Literal["number", "string", "boolean", "select"] = Field(
+        ..., description="Widget type for the frontend"
+    )
+    default: Any = Field(default=None, description="Default value")
+    description: str = Field(default="", description="Human-readable label")
+    ui: dict[str, Any] | None = Field(
+        default=None,
+        description=(
+            "Widget-specific hints consumed by the frontend renderer. "
+            "Number widgets read ``min``/``max``/``step``; select "
+            "widgets read ``options``; plugin-defined widget kinds may "
+            "use any keys they like."
+        ),
+    )
+    convertible_to_input: bool = Field(
+        default=True,
+        description=(
+            "If True, this parameter can be overridden by connecting an "
+            "input wire (ComfyUI-style widget-to-input conversion)."
+        ),
+    )
+
+
+class NodeDefinition(BaseModel):
+    """Static metadata describing a node type."""
+
+    node_type_id: str = Field(..., description="Unique node type identifier")
+    display_name: str = Field(..., description="Human-readable name")
+    category: str = Field(default="general", description="Category for grouping")
+    description: str = Field(default="", description="What this node does")
+    inputs: list[NodePort] = Field(default_factory=list)
+    outputs: list[NodePort] = Field(default_factory=list)
+    params: list[NodeParam] = Field(
+        default_factory=list,
+        description="Editable parameters (widgets) displayed on the node card.",
+    )
+    continuous: bool = Field(
+        default=False,
+        description=(
+            "If True, source nodes (no inputs) re-execute continuously "
+            "instead of executing once. Useful for streaming generators."
+        ),
+    )
+    pipeline_meta: dict[str, Any] | None = Field(
+        default=None,
+        description=(
+            "Rich pipeline-only metadata (config_schema, mode_defaults, "
+            "supports_lora, supports_vace, etc.) for nodes that are "
+            ":class:`Pipeline` subclasses. ``None`` for plain nodes. "
+            "Populated by ``Pipeline.get_definition()`` from the config "
+            "class's ``get_schema_with_metadata()``."
+        ),
+    )
+
+
+class BaseNode(ABC):
+    """Abstract base class for all backend node types.
+
+    Subclasses must set ``node_type_id`` as a ``ClassVar`` and implement
+    ``get_definition()`` and ``execute()``. Nodes run inside a
+    :class:`NodeProcessor` in the pipeline graph: input port values arrive
+    in the ``inputs`` dict and the return dict fans out to downstream
+    nodes or pipelines via queue edges.
+
+    Example::
+
+        class MyNode(BaseNode):
+            node_type_id: ClassVar[str] = "my-plugin.my-node"
+
+            @classmethod
+            def get_definition(cls) -> NodeDefinition:
+                return NodeDefinition(
+                    node_type_id=cls.node_type_id,
+                    display_name="My Node",
+                    inputs=[NodePort(name="audio", port_type="audio")],
+                    outputs=[NodePort(name="audio", port_type="audio")],
+                )
+
+            def execute(self, inputs, **kwargs):
+                return {"audio": inputs["audio"]}
+    """
+
+    node_type_id: ClassVar[str]
+
+    def __init__(self, node_id: str = "", config: dict[str, Any] | None = None):
+        self.node_id = node_id
+        self.config = config or {}
+
+    @classmethod
+    @abstractmethod
+    def get_definition(cls) -> NodeDefinition:
+        """Return static metadata for this node type."""
+
+    def execute(
+        self,
+        inputs: dict[str, Any],
+        **kwargs,
+    ) -> dict[str, Any]:
+        """Execute the node with the given inputs.
+
+        Plain backend nodes (scheduled by ``NodeProcessor``) override
+        this. :class:`Pipeline` subclasses inherit the default because
+        they're driven by ``__call__`` from ``PipelineProcessor`` and
+        never reach this code path.
+
+        Args:
+            inputs: Dict mapping input port names to their values.
+                Audio ports receive ``(tensor, sample_rate)`` tuples.
+                Video ports receive frame tensors.
+            **kwargs: Additional context (pipeline parameters, etc.)
+
+        Returns:
+            Dict mapping output port names to their values.
+        """
+        raise NotImplementedError(
+            f"{type(self).__name__} must override execute() or be invoked "
+            "via PipelineProcessor (Pipeline subclass) instead of NodeProcessor."
+        )
diff --git a/src/scope/core/nodes/builtins/__init__.py b/src/scope/core/nodes/builtins/__init__.py
new file mode 100644
index 000000000..88e8c96f5
--- /dev/null
+++ b/src/scope/core/nodes/builtins/__init__.py
@@ -0,0 +1,5 @@
+"""Built-in nodes shipped with the foundation abstraction."""
+
+from .audio_io import AudioSourceNode
+
+__all__ = ["AudioSourceNode"]
diff --git a/src/scope/core/nodes/builtins/audio_io.py b/src/scope/core/nodes/builtins/audio_io.py
new file mode 100644
index 000000000..81f1a7ccf
--- /dev/null
+++ b/src/scope/core/nodes/builtins/audio_io.py
@@ -0,0 +1,331 @@
+"""Built-in audio I/O nodes: AudioSource (WAV file → audio stream).
+
+Terminal audio output is handled by the regular Sink node: audio edges
+into a Sink are routed straight to the WebRTC audio track via the
+session's audio_output_queue, with no intermediate node needed.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import struct
+import time
+from typing import Any, ClassVar
+
+import numpy as np
+import torch
+
+from ..base import BaseNode, NodeDefinition, NodeParam, NodePort
+
+logger = logging.getLogger(__name__)
+
+SAMPLE_RATE = 48000
+CHUNK_DURATION = 0.1  # 100ms chunks for streaming
+CHUNK_SAMPLES = int(SAMPLE_RATE * CHUNK_DURATION)
+
+
+def _read_wav_float32(path: str) -> tuple[np.ndarray, int]:
+    """Parse a WAV file into float32 samples without the stdlib ``wave``
+    module, which rejects IEEE-float (format 3) files.
+
+    Returns (data, sample_rate) where ``data`` has shape (samples, channels).
+    Supports formats 1 (PCM int) and 3 (IEEE float) — the two common cases.
+    WAVE_FORMAT_EXTENSIBLE (0xFFFE) is unwrapped to its underlying format.
+    """
+    with open(path, "rb") as f:
+        header = f.read(12)
+        if len(header) < 12 or header[:4] != b"RIFF" or header[8:12] != b"WAVE":
+            raise ValueError(f"Not a WAV file: {path}")
+
+        fmt_code: int | None = None
+        n_channels = 1
+        sample_rate = 0
+        bits_per_sample = 0
+        pcm_bytes = b""
+
+        while True:
+            chunk_header = f.read(8)
+            if len(chunk_header) < 8:
+                break
+            chunk_id, chunk_size = struct.unpack("<4sI", chunk_header)
+            chunk_data = f.read(chunk_size)
+            if chunk_size % 2 == 1:
+                f.read(1)  # RIFF pads odd-sized chunks
+
+            if chunk_id == b"fmt " and len(chunk_data) >= 16:
+                (
+                    fmt_code,
+                    n_channels,
+                    sample_rate,
+                    _byte_rate,
+                    _block_align,
+                    bits_per_sample,
+                ) = struct.unpack("<HHIIHH", chunk_data[:16])
+                # Unwrap WAVE_FORMAT_EXTENSIBLE: real format is first 2 bytes of the GUID.
+                if fmt_code == 0xFFFE and len(chunk_data) >= 26:
+                    fmt_code = struct.unpack("<H", chunk_data[24:26])[0]
+            elif chunk_id == b"data":
+                pcm_bytes = chunk_data
+
+        if fmt_code is None or not pcm_bytes or sample_rate <= 0:
+            raise ValueError(f"WAV missing fmt/data chunk: {path}")
+
+        if fmt_code == 1:  # PCM integer
+            if bits_per_sample == 16:
+                samples = (
+                    np.frombuffer(pcm_bytes, dtype="<i2").astype(np.float32) / 32768.0
+                )
+            elif bits_per_sample == 32:
+                samples = (
+                    np.frombuffer(pcm_bytes, dtype="<i4").astype(np.float32)
+                    / 2147483648.0
+                )
+            elif bits_per_sample == 8:
+                samples = (
+                    np.frombuffer(pcm_bytes, dtype=np.uint8).astype(np.float32) / 128.0
+                    - 1.0
+                )
+            elif bits_per_sample == 24:
+                raw = np.frombuffer(pcm_bytes, dtype=np.uint8).reshape(-1, 3)
+                as32 = (
+                    raw[:, 0].astype(np.int32)
+                    | (raw[:, 1].astype(np.int32) << 8)
+                    | (raw[:, 2].astype(np.int32) << 16)
+                )
+                # Sign-extend the 24-bit value.
+                as32 = np.where(as32 & 0x800000, as32 | ~0xFFFFFF, as32)
+                samples = as32.astype(np.float32) / 8388608.0
+            else:
+                raise ValueError(f"Unsupported PCM bit depth: {bits_per_sample}")
+        elif fmt_code == 3:  # IEEE float
+            if bits_per_sample == 32:
+                samples = np.frombuffer(pcm_bytes, dtype="<f4").astype(
+                    np.float32, copy=True
+                )
+            elif bits_per_sample == 64:
+                samples = np.frombuffer(pcm_bytes, dtype="<f8").astype(np.float32)
+            else:
+                raise ValueError(f"Unsupported float bit depth: {bits_per_sample}")
+        else:
+            raise ValueError(f"Unsupported WAV format code: {fmt_code}")
+
+        samples = samples.reshape(-1, n_channels)
+        return samples, sample_rate
+
+
+class AudioSourceNode(BaseNode):
+    """Load audio from a WAV file and stream it in 100ms chunks, looping."""
+
+    node_type_id: ClassVar[str] = "audio.AudioSource"
+
+    def __init__(self, node_id: str, config: dict[str, Any] | None = None):
+        super().__init__(node_id, config)
+        self._audio_data: np.ndarray | None = None
+        self._position = 0
+        self._loaded_file: str = ""
+        self._last_call_time: float | None = None
+        # In mode=full + loop=false we emit the entire clip once and then
+        # stay silent until the loaded file or mode changes. Otherwise
+        # every worker tick would re-push a 15s clip and flood the graph.
+        self._full_emitted_key: tuple[str, str] | None = None
+        # In mode=full + loop=true we re-emit the full clip paced to
+        # wall clock so downstream (e.g. ACEStep) re-processes at roughly
+        # real-time without flooding.
+        self._last_full_emit_time: float | None = None
+
+    @classmethod
+    def get_definition(cls) -> NodeDefinition:
+        return NodeDefinition(
+            node_type_id=cls.node_type_id,
+            display_name="Audio Source",
+            category="audio",
+            description="Load audio from a WAV file at 48kHz stereo.",
+            continuous=True,
+            inputs=[],
+            outputs=[
+                NodePort(name="audio", port_type="audio", description="Audio waveform"),
+            ],
+            params=[
+                NodeParam(
+                    name="file_id",
+                    param_type="string",
+                    default="",
+                    description="Audio file path",
+                ),
+                NodeParam(
+                    name="duration",
+                    param_type="number",
+                    default=15.0,
+                    description="Duration (s)",
+                    ui={"min": 1, "max": 600, "step": 1},
+                ),
+                NodeParam(
+                    name="mode",
+                    param_type="select",
+                    default="full",
+                    description="Output mode",
+                    ui={"options": ["full", "stream"]},
+                ),
+                NodeParam(
+                    name="loop",
+                    param_type="boolean",
+                    default=False,
+                    description="Repeat the audio continuously",
+                ),
+                NodeParam(
+                    name="pacing",
+                    param_type="select",
+                    default="realtime",
+                    description="Loop pacing",
+                    ui={"options": ["realtime", "downstream"]},
+                ),
+            ],
+        )
+
+    def _load_audio(self, file_path: str, duration: float) -> None:
+        """Load, decode, resample to 48kHz stereo, and clip to duration."""
+        data, sr = _read_wav_float32(file_path)  # (samples, channels)
+
+        if data.shape[1] == 1:
+            data = np.concatenate([data, data], axis=1)
+        elif data.shape[1] > 2:
+            data = data[:, :2]
+        data = data.T  # (channels, samples)
+
+        if sr != SAMPLE_RATE and sr > 0:
+            num_samples = data.shape[1]
+            new_len = int(num_samples * SAMPLE_RATE / sr)
+            old_indices = np.linspace(0, num_samples - 1, new_len)
+            resampled = np.zeros((data.shape[0], new_len), dtype=np.float32)
+            for ch in range(data.shape[0]):
+                resampled[ch] = np.interp(old_indices, np.arange(num_samples), data[ch])
+            data = resampled
+
+        max_samples = int(duration * SAMPLE_RATE)
+        if data.shape[1] > max_samples:
+            data = data[:, :max_samples]
+
+        self._audio_data = data
+        self._position = 0
+        self._loaded_file = file_path
+        logger.info(
+            "AudioSource loaded: %s (%.1fs)",
+            file_path,
+            data.shape[1] / SAMPLE_RATE,
+        )
+
+    def execute(self, inputs: dict[str, Any], **kwargs) -> dict[str, Any]:
+        file_id = kwargs.get("file_id", "")
+        duration = float(kwargs.get("duration", 15.0))
+        # "full" = emit entire clip once (for batch DAGs); "stream" = 100ms chunks
+        mode = kwargs.get("mode", "stream")
+        loop = bool(kwargs.get("loop", False))
+        pacing = kwargs.get("pacing", "realtime")
+
+        if not file_id:
+            return {}
+        file_id = self._resolve_path(file_id)
+        if not file_id:
+            return {}
+
+        if file_id != self._loaded_file:
+            try:
+                self._load_audio(file_id, duration)
+            except Exception as e:
+                logger.error("AudioSourceNode failed to load %s: %s", file_id, e)
+                return {}
+
+        if self._audio_data is None or self._audio_data.shape[1] == 0:
+            return {}
+
+        if mode == "full":
+            if loop:
+                if pacing == "downstream":
+                    # Skip the wall-clock gate: re-emit every tick and let
+                    # the downstream edge queue's maxsize=1 + blocking put
+                    # in _route_outputs backpressure us to consumer speed.
+                    self._last_full_emit_time = None
+                    return self._emit_full()
+                # realtime: re-emit the full clip paced by clip duration
+                # on the wall clock so downstream re-processes at roughly
+                # real-time without flooding the audio track.
+                clip_seconds = self._audio_data.shape[1] / SAMPLE_RATE
+                now = time.monotonic()
+                if (
+                    self._last_full_emit_time is not None
+                    and now - self._last_full_emit_time < clip_seconds
+                ):
+                    return {}
+                self._last_full_emit_time = now
+                return self._emit_full()
+            # Loop=false: emit the entire clip once per (file, mode) pair
+            # and then stay silent.
+            self._last_full_emit_time = None
+            key = (self._loaded_file, "full")
+            if self._full_emitted_key == key:
+                return {}
+            self._full_emitted_key = key
+            return self._emit_full()
+        # Stream mode: clear the full-mode flags so switching back to full
+        # later re-emits once.
+        self._full_emitted_key = None
+        self._last_full_emit_time = None
+        return self._emit_chunk(loop=loop, pacing=pacing)
+
+    @staticmethod
+    def _resolve_path(file_id: str) -> str | None:
+        """Resolve a file path. Absolute → cwd → ~/.daydream-scope/assets."""
+        if os.path.isabs(file_id) and os.path.exists(file_id):
+            return file_id
+        if os.path.exists(file_id):
+            return os.path.abspath(file_id)
+        from pathlib import Path
+
+        candidate = Path.home() / ".daydream-scope" / "assets" / file_id
+        if candidate.exists():
+            return str(candidate)
+        logger.warning("AudioSource: file not found: %s", file_id)
+        return None
+
+    def _emit_full(self) -> dict[str, Any]:
+        return {"audio": (torch.from_numpy(self._audio_data.copy()), SAMPLE_RATE)}
+
+    def _emit_chunk(
+        self, loop: bool = True, pacing: str = "realtime"
+    ) -> dict[str, Any]:
+        # Pace to real-time unless downstream-paced — in that mode the
+        # maxsize=1 edge queues handle rate limiting via backpressure.
+        if pacing != "downstream":
+            now = time.monotonic()
+            if self._last_call_time is not None:
+                elapsed = now - self._last_call_time
+                if elapsed < CHUNK_DURATION * 0.8:
+                    time.sleep(CHUNK_DURATION - elapsed)
+            self._last_call_time = time.monotonic()
+        else:
+            self._last_call_time = None
+
+        total = self._audio_data.shape[1]
+        if not loop and self._position >= total:
+            return {}
+
+        chunk = np.zeros((self._audio_data.shape[0], CHUNK_SAMPLES), dtype=np.float32)
+        remaining = CHUNK_SAMPLES
+        offset = 0
+        while remaining > 0:
+            avail = min(remaining, total - self._position)
+            if avail <= 0:
+                if not loop:
+                    # Stop at end of clip; emit a partial chunk (remaining
+                    # samples stay zero-padded).
+                    break
+                self._position = 0
+                avail = min(remaining, total - self._position)
+            chunk[:, offset : offset + avail] = self._audio_data[
+                :, self._position : self._position + avail
+            ]
+            self._position += avail
+            offset += avail
+            remaining -= avail
+        return {"audio": (torch.from_numpy(chunk), SAMPLE_RATE)}
diff --git a/src/scope/core/nodes/processor.py b/src/scope/core/nodes/processor.py
new file mode 100644
index 000000000..d8f776484
--- /dev/null
+++ b/src/scope/core/nodes/processor.py
@@ -0,0 +1,347 @@
+"""Node processor — wraps a BaseNode for execution in the pipeline graph.
+
+Adapts the node interface (typed I/O ports) to the pipeline processor
+interface (input/output queues, worker thread).
+"""
+
+import logging
+import os
+import queue
+import threading
+import time
+from typing import Any
+
+import torch
+
+from .base import BaseNode
+
+logger = logging.getLogger(__name__)
+
+SLEEP_TIME = 0.01
+
+# Set SCOPE_NODE_TRACE=1 for per-node diagnostics: param updates,
+# route-outputs backpressure events. Off by default.
+_TRACE = os.environ.get("SCOPE_NODE_TRACE", "").lower() in ("1", "true", "yes")
+
+
+class NodeProcessor:
+    """Runs a BaseNode in a dedicated thread. Input queues feed the node,
+    output queues fan out its results to downstream nodes.
+
+    Source nodes (no inputs) execute once by default; nodes marked
+    ``continuous=True`` in their definition re-execute on every tick, which
+    is how streaming sources (audio) and sinks (audio loop) stay alive.
+    """
+
+    def __init__(
+        self,
+        node: BaseNode,
+        node_id: str,
+        initial_parameters: dict | None = None,
+    ):
+        self.node = node
+        self.node_id = node_id
+        self.parameters = initial_parameters or {}
+
+        # Port-based queues wired by the graph executor
+        self.input_queues: dict[str, queue.Queue] = {}
+        self.output_queues: dict[str, list[queue.Queue]] = {}
+        self.input_queue_lock = threading.Lock()
+        self.external_queue_refs: list[tuple[dict, str]] = []
+
+        definition = node.get_definition()
+        self.audio_input_ports: set[str] = {
+            p.name for p in definition.inputs if p.port_type == "audio"
+        }
+        # Output ports wired to a sink node via graph_executor. Only these
+        # route through ``audio_output_queue`` → FrameProcessor.get_audio().
+        # A node whose audio output feeds another node (e.g. AudioSource →
+        # VAEEncodeAudio) must NOT push to audio_output_queue: with
+        # maxsize=1 + blocking put, nothing would ever drain it and the
+        # worker would deadlock after the second emission.
+        self.audio_sink_ports: set[str] = set()
+        # Names of parameters this node actually declares. Global param
+        # updates (no node_id) are broadcast to every processor; without
+        # this filter a graph-level tweak (quantize_mode, modulations…)
+        # would spuriously flag every custom node for re-execution.
+        self._declared_param_names: set[str] = {p.name for p in definition.params}
+
+        # Audio output queue consumed by FrameProcessor.get_audio() on the
+        # sink. Size 1 + blocking producer (see _route_audio) gives us
+        # backpressure: audio_decode stalls until AudioProcessingTrack has
+        # served enough of the current chunk to pull a new one, which
+        # cascades upstream through node-to-node edge queues and
+        # rate-limits batch generators to real-time playback.
+        self.audio_output_queue: queue.Queue[tuple[torch.Tensor, int]] = queue.Queue(
+            maxsize=1
+        )
+
+        self.worker_thread: threading.Thread | None = None
+        self.shutdown_event = threading.Event()
+        self.running = False
+
+        # Execution state
+        self._source_executed = False
+        self._has_executed = False
+        self._continuous = definition.continuous
+        # Cached inputs from the last successful run, replayed when
+        # parameters change so non-continuous nodes can refresh their
+        # output without requiring upstream to re-emit.
+        self._last_inputs: dict[str, Any] = {}
+        self._needs_rerun = False
+
+        # Input ports whose upstream is transitively reachable from a
+        # ``continuous=True`` node. Populated by graph_executor at build
+        # time. When non-empty, ``_process_once`` waits until every
+        # dynamic port has received fresh data since the last run before
+        # firing, so a single upstream cycle triggers exactly one run
+        # (not one run per fresh-port arrival). Static ports — typically
+        # one-shot handles like MODEL/VAE/CONFIG — always latch from
+        # ``_last_inputs``. Empty set falls back to the legacy "fire on
+        # any fresh input" behaviour.
+        self.dynamic_input_ports: set[str] = set()
+        # Fresh values drained from input queues while waiting for every
+        # dynamic port to catch up. Draining unblocks upstream producers
+        # without committing to a run yet.
+        self._pending_inputs: dict[str, Any] = {}
+
+        # PipelineProcessor interface compatibility: graph_executor populates
+        # this for every processor; kept as an empty dict so that write is safe.
+        self.output_consumers: dict[str, list] = {}
+        self.paused = False
+
+    @property
+    def output_queue(self) -> queue.Queue | None:
+        qs = self.output_queues.get("video")
+        return qs[0] if qs else None
+
+    def start(self) -> None:
+        if self.running:
+            return
+        self.running = True
+        self.shutdown_event.clear()
+        self.worker_thread = threading.Thread(
+            target=self._worker_loop, daemon=True, name=f"NodeProcessor[{self.node_id}]"
+        )
+        self.worker_thread.start()
+
+    def stop(self) -> None:
+        if not self.running:
+            return
+        self.running = False
+        self.shutdown_event.set()
+        if self.worker_thread is not None:
+            self.worker_thread.join(timeout=5.0)
+        logger.info("NodeProcessor stopped: %s", self.node_id)
+
+    def update_parameters(self, parameters: dict[str, Any]) -> None:
+        # Only mark the node dirty when the update actually touches a
+        # parameter this node declares AND the value differs from the
+        # current one. FrameProcessor.update_parameters broadcasts
+        # global updates (no node_id) to every processor, so without
+        # this guard a stream-level tweak like quantize_mode would fire
+        # _needs_rerun on every custom node and cascade through the DAG.
+        changed = False
+        changed_keys: list[str] = []
+        for key, value in parameters.items():
+            if key not in self._declared_param_names:
+                continue
+            if self.parameters.get(key) != value:
+                changed = True
+                changed_keys.append(key)
+        self.parameters.update(parameters)
+        if changed:
+            self._needs_rerun = True
+        if _TRACE:
+            accepted = {k: parameters[k] for k in changed_keys}
+            rejected = [
+                k for k in parameters.keys() if k not in self._declared_param_names
+            ]
+            if accepted or rejected:
+                logger.info(
+                    "[scope.update_parameters][%s] accepted=%s rejected_unknown=%s",
+                    self.node_id,
+                    accepted,
+                    rejected,
+                )
+
+    def set_beat_cache_reset_rate(self, rate):  # PipelineProcessor compat
+        pass
+
+    def get_fps(self) -> float:
+        return 30.0
+
+    def _worker_loop(self) -> None:
+        while not self.shutdown_event.is_set():
+            try:
+                self._process_once()
+            except Exception:
+                logger.exception("Error in node processor %s", self.node_id)
+                with self.input_queue_lock:
+                    is_source = not self.input_queues
+                if is_source:
+                    # Avoid infinite retry on failing source nodes
+                    self._source_executed = True
+                    self._continuous = False
+                self.shutdown_event.wait(SLEEP_TIME)
+
+    def _process_once(self) -> None:
+        if self.paused:
+            self.shutdown_event.wait(SLEEP_TIME)
+            return
+
+        with self.input_queue_lock:
+            all_queues = dict(self.input_queues)
+
+        is_source_node = not all_queues
+        inputs: dict[str, Any] = {}
+
+        if is_source_node:
+            # Source nodes run once, then re-run only when params change
+            # (or every tick when continuous).
+            if self._source_executed and not self._continuous and not self._needs_rerun:
+                self.shutdown_event.wait(1.0)
+                return
+        elif self._continuous:
+            # Continuous nodes: pick up whatever's currently in the queues
+            # and run every tick.
+            for port_name, q in all_queues.items():
+                try:
+                    inputs[port_name] = q.get_nowait()
+                except queue.Empty:
+                    pass
+        else:
+            # Non-continuous node with inputs:
+            # - First run waits until every port has received data at least
+            #   once so the latch cache (_last_inputs) is populated.
+            # - Subsequent runs drain fresh values into ``_pending_inputs``
+            #   (so upstream producers unblock immediately) and fire once
+            #   every dynamic input port has caught up. Static ports —
+            #   upstreams that never re-emit, e.g. MODEL/VAE handles —
+            #   are latched from ``_last_inputs``. Parameter changes on
+            #   this node bypass the gate via ``_needs_rerun``.
+            # - When ``dynamic_input_ports`` is empty (graph_executor did
+            #   not classify this node, e.g. a static-only subgraph), we
+            #   fall back to the legacy "fire on any fresh input" gate.
+            if not self._has_executed:
+                if any(q.empty() for q in all_queues.values()):
+                    self.shutdown_event.wait(SLEEP_TIME)
+                    return
+                inputs = {name: q.get_nowait() for name, q in all_queues.items()}
+            else:
+                for port_name, q in all_queues.items():
+                    try:
+                        self._pending_inputs[port_name] = q.get_nowait()
+                    except queue.Empty:
+                        pass
+
+                if self._needs_rerun:
+                    fire = True
+                elif self.dynamic_input_ports:
+                    fire = self.dynamic_input_ports <= self._pending_inputs.keys()
+                else:
+                    fire = bool(self._pending_inputs)
+
+                if not fire:
+                    self.shutdown_event.wait(SLEEP_TIME)
+                    return
+
+                inputs = {}
+                for port_name in all_queues:
+                    if port_name in self._pending_inputs:
+                        inputs[port_name] = self._pending_inputs[port_name]
+                    elif port_name in self._last_inputs:
+                        inputs[port_name] = self._last_inputs[port_name]
+                self._pending_inputs.clear()
+
+        outputs = self.node.execute(inputs, **self.parameters)
+
+        if is_source_node:
+            self._source_executed = True
+        self._needs_rerun = False
+
+        if not outputs:
+            self.shutdown_event.wait(SLEEP_TIME)
+            return
+
+        self._has_executed = True
+        self._last_inputs = inputs
+        self._route_outputs(outputs)
+
+    def _route_outputs(self, outputs: dict[str, Any]) -> None:
+        for port_name, value in outputs.items():
+            if value is None:
+                continue
+
+            # Audio outputs also feed the FrameProcessor's audio path
+            # — but only for ports that graph_executor wired to a sink,
+            # not every port whose type is "audio". Otherwise an
+            # intermediate audio-producing node (AudioSource → encoder)
+            # would also push to its own audio_output_queue, which
+            # nothing drains → worker deadlocks on the blocking put
+            # after the second emission.
+            if port_name in self.audio_sink_ports:
+                self._route_audio(value)
+
+            # Fan out to all downstream queues on this port. Block briefly
+            # when queues are full so producers throttle to consumer pace
+            # and GPU tensors don't pile up in memory.
+            out_queues = self.output_queues.get(port_name)
+            if out_queues:
+                for oq in out_queues:
+                    block_start = time.monotonic() if _TRACE else 0.0
+                    blocks = 0
+                    while not self.shutdown_event.is_set():
+                        try:
+                            oq.put(value, timeout=0.1)
+                            break
+                        except queue.Full:
+                            blocks += 1
+                            continue
+                    if _TRACE and blocks > 0:
+                        logger.info(
+                            "[scope._route_outputs][%s:%s] blocked %.2fs (%d retries) on full downstream queue",
+                            self.node_id,
+                            port_name,
+                            time.monotonic() - block_start,
+                            blocks,
+                        )
+
+    def _route_audio(self, value: Any) -> None:
+        """Extract audio tensor and push to audio_output_queue for WebRTC."""
+        if isinstance(value, tuple) and len(value) == 2:
+            audio_tensor, audio_sr = value
+        else:
+            audio_tensor = getattr(value, "waveform", None)
+            audio_sr = getattr(value, "sample_rate", 48000)
+        if audio_tensor is None:
+            return
+        if hasattr(audio_tensor, "is_cuda") and audio_tensor.is_cuda:
+            audio_tensor = audio_tensor.detach().cpu()
+        # VAE decoders (e.g. ACEStep) return (1, C, T); the audio track
+        # expects (C, T). Drop a leading singleton batch dim so the
+        # channel/interleave path in AudioProcessingTrack doesn't misread
+        # the layout and produce slowed-down / garbled playback.
+        if hasattr(audio_tensor, "dim") and audio_tensor.dim() == 3:
+            if audio_tensor.shape[0] == 1:
+                audio_tensor = audio_tensor.squeeze(0)
+        # Blocking-with-retry put. Stalls the worker thread when the audio
+        # track hasn't finished serving the previous chunk, which is the
+        # backpressure mechanism that rate-limits batch generators to
+        # real-time playback instead of silently dropping audio.
+        block_start = time.monotonic() if _TRACE else 0.0
+        blocks = 0
+        while not self.shutdown_event.is_set():
+            try:
+                self.audio_output_queue.put((audio_tensor, audio_sr), timeout=0.1)
+                break
+            except queue.Full:
+                blocks += 1
+                continue
+        if _TRACE and blocks > 0:
+            logger.info(
+                "[scope._route_audio][%s] blocked %.2fs (%d retries) on full audio_output_queue (WebRTC backpressure)",
+                self.node_id,
+                time.monotonic() - block_start,
+                blocks,
+            )
diff --git a/src/scope/core/nodes/registry.py b/src/scope/core/nodes/registry.py
new file mode 100644
index 000000000..59ac5cb4c
--- /dev/null
+++ b/src/scope/core/nodes/registry.py
@@ -0,0 +1,80 @@
+"""Unified registry for every node type on the graph (plain nodes + pipelines)."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .base import BaseNode, NodeDefinition
+
+logger = logging.getLogger(__name__)
+
+
+def _derive_node_type_id(node_class: type) -> str | None:
+    """Return the registry key for a node class, or None if not derivable.
+
+    Plain nodes carry the id as the ``node_type_id`` classvar; pipelines
+    keep it on their config class as ``pipeline_id``.
+    """
+    node_type_id = getattr(node_class, "node_type_id", None)
+    if node_type_id is not None:
+        return node_type_id
+    # Lazy import: nodes.registry is loaded before pipelines.interface.
+    # Narrow to ImportError so real bugs (AttributeError on a broken
+    # config class, typos in pipeline_id, etc.) surface instead of being
+    # silently swallowed.
+    try:
+        from scope.core.pipelines.interface import Pipeline
+    except ImportError:
+        return None
+
+    if issubclass(node_class, Pipeline):
+        return node_class.get_config_class().pipeline_id
+    return None
+
+
+class NodeRegistry:
+    """Central registry for all available node types."""
+
+    _nodes: dict[str, type[BaseNode]] = {}
+
+    @classmethod
+    def register(cls, node_class: type[BaseNode]) -> None:
+        """Register a :class:`BaseNode` subclass (plain node or pipeline)."""
+        node_type_id = _derive_node_type_id(node_class)
+        if node_type_id is None:
+            raise ValueError(
+                f"Cannot determine node_type_id for {node_class.__name__}; "
+                "set a ClassVar[str] `node_type_id` on plain nodes or a "
+                "`pipeline_id` on the pipeline config class."
+            )
+        cls._nodes[node_type_id] = node_class
+        logger.debug("Registered node type: %s", node_type_id)
+
+    @classmethod
+    def get(cls, node_type_id: str) -> type[BaseNode] | None:
+        return cls._nodes.get(node_type_id)
+
+    @classmethod
+    def is_registered(cls, node_type_id: str) -> bool:
+        return node_type_id in cls._nodes
+
+    @classmethod
+    def list_node_types(cls) -> list[str]:
+        return list(cls._nodes.keys())
+
+    @classmethod
+    def get_all_definitions(cls) -> list[NodeDefinition]:
+        return [nc.get_definition() for nc in cls._nodes.values()]
+
+    @classmethod
+    def unregister(cls, node_type_id: str) -> bool:
+        if node_type_id in cls._nodes:
+            del cls._nodes[node_type_id]
+            return True
+        return False
+
+    @classmethod
+    def clear(cls) -> None:
+        cls._nodes.clear()
diff --git a/src/scope/core/pipelines/interface.py b/src/scope/core/pipelines/interface.py
index 18312c647..acfbebe87 100644
--- a/src/scope/core/pipelines/interface.py
+++ b/src/scope/core/pipelines/interface.py
@@ -1,10 +1,20 @@
-"""Base interface for all pipelines."""
+"""Base interface for all pipelines.
+
+A :class:`Pipeline` is a :class:`scope.core.nodes.BaseNode` subclass —
+the "heavy" kind that batches video frames, loads GPU models, and
+carries a rich Pydantic config class. The graph editor and user-facing
+docs call them *Nodes*; the name ``Pipeline`` survives as the
+implementation base class so existing subclasses and plugins keep
+working unchanged.
+"""
 
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
 from pydantic import BaseModel
 
+from scope.core.nodes.base import BaseNode, NodeDefinition, NodePort
+
 if TYPE_CHECKING:
     from .schema import BasePipelineConfig
 
@@ -15,50 +25,57 @@ class Requirements(BaseModel):
     input_size: int
 
 
-class Pipeline(ABC):
-    """Abstract base class for all pipelines.
-
-    Pipelines must implement get_config_class() to return their Pydantic config model.
-    This enables:
-    - Validation via model_validate() / model_validate_json()
-    - JSON Schema generation via model_json_schema()
-    - Type-safe configuration access
-    - API introspection and automatic UI generation
+class Pipeline(BaseNode, ABC):
+    """Abstract base class for video-pipeline nodes.
 
-    See schema.py for the BasePipelineConfig model and pipeline-specific configs.
-    For multi-mode pipeline support (text/video), pipelines use helper functions
-    from defaults.py (resolve_input_mode, apply_mode_defaults_to_state, etc.).
+    Subclasses implement ``__call__`` (the per-chunk processing
+    function) and ``get_config_class`` (returning a Pydantic config
+    that drives validation, JSON-schema generation, the parameter
+    panel, and parameter defaults). Everything else — registry,
+    plugin hook, graph editor — is the same as for plain nodes.
     """
 
     @classmethod
     def get_config_class(cls) -> type["BasePipelineConfig"]:
         """Return the Pydantic config class for this pipeline.
 
-        The config class should inherit from BasePipelineConfig and define:
-        - pipeline_id: Unique identifier
-        - pipeline_name: Human-readable name
-        - pipeline_description: Capabilities description
-        - pipeline_version: Version string
-        - Default parameter values for the pipeline
-
-        Returns:
-            Pydantic config model class
-
-        Note:
-            Subclasses should override this method to return their config class.
-            The default implementation returns BasePipelineConfig.
-
-        Example:
-            from .schema import LongLiveConfig
-
-            @classmethod
-            def get_config_class(cls) -> type[BasePipelineConfig]:
-                return LongLiveConfig
+        Subclasses override to return their concrete config; the
+        default returns ``BasePipelineConfig``.
         """
         from .schema import BasePipelineConfig
 
         return BasePipelineConfig
 
+    @classmethod
+    def get_definition(cls) -> NodeDefinition:
+        """Project the pipeline's config class into a :class:`NodeDefinition`.
+
+        Populates the compact node-catalog fields (id, ports, etc.)
+        and stuffs the full ``get_schema_with_metadata()`` output into
+        ``pipeline_meta``, which is the rich data ``PipelineNode.tsx``
+        renders in the parameter panel. ``params`` is left empty
+        because the Pydantic schema is too structured to flatten into
+        ``NodeParam[]`` widgets.
+        """
+        config = cls.get_config_class()
+        return NodeDefinition(
+            node_type_id=config.pipeline_id,
+            display_name=getattr(config, "pipeline_name", config.pipeline_id),
+            category="pipeline",
+            description=getattr(config, "pipeline_description", "") or "",
+            inputs=[
+                NodePort(name=name, port_type="video")
+                for name in (getattr(config, "inputs", ["video"]) or ["video"])
+            ],
+            outputs=[
+                NodePort(name=name, port_type="video")
+                for name in (getattr(config, "outputs", ["video"]) or ["video"])
+            ],
+            params=[],
+            continuous=False,
+            pipeline_meta=config.get_schema_with_metadata(),
+        )
+
     @abstractmethod
     def __call__(self, **kwargs) -> dict:
         """
diff --git a/src/scope/core/pipelines/registry.py b/src/scope/core/pipelines/registry.py
index 030f7a3d1..d50e9d2fc 100644
--- a/src/scope/core/pipelines/registry.py
+++ b/src/scope/core/pipelines/registry.py
@@ -1,8 +1,10 @@
-"""Pipeline registry for centralized pipeline management.
+"""Pipeline registry — a filtering view over :class:`NodeRegistry`.
 
-This module provides a registry pattern to eliminate if/elif chains when
-accessing pipelines by ID. It enables dynamic pipeline discovery and
-metadata retrieval.
+Pipelines and plain custom nodes share the same ``NodeRegistry._nodes``
+storage. ``PipelineRegistry`` projects that storage down to entries
+whose class is a :class:`Pipeline` subclass and exposes the same API
+the rest of the codebase always used, so existing call sites and
+plugins keep working unchanged.
 """
 
 import importlib
@@ -11,6 +13,8 @@
 
 import torch
 
+from scope.core.nodes.registry import NodeRegistry
+
 if TYPE_CHECKING:
     from .interface import Pipeline
     from .schema import BasePipelineConfig
@@ -18,83 +22,67 @@
 logger = logging.getLogger(__name__)
 
 
-class PipelineRegistry:
-    """Registry for managing available pipelines."""
+def _is_pipeline(node_class: object) -> bool:
+    """Return True when ``node_class`` is a :class:`Pipeline` subclass.
+
+    Lazily imports :class:`Pipeline` to dodge the import cycle between
+    the pipelines and nodes packages at module load time.
+    """
+    from .interface import Pipeline
+
+    return isinstance(node_class, type) and issubclass(node_class, Pipeline)
+
 
-    _pipelines: dict[str, type["Pipeline"]] = {}
+class PipelineRegistry:
+    """Filtering view over :class:`NodeRegistry` for pipeline classes."""
 
     @classmethod
     def register(cls, pipeline_id: str, pipeline_class: type["Pipeline"]) -> None:
-        """Register a pipeline class with its ID.
+        """Plant a pipeline class into the unified :class:`NodeRegistry`.
 
-        Args:
-            pipeline_id: Unique identifier for the pipeline
-            pipeline_class: Pipeline class to register
+        Delegates to ``NodeRegistry.register`` so the same logging and
+        id-derivation path runs for built-in pipelines and plugin nodes
+        alike. The explicit ``pipeline_id`` argument is asserted against
+        the derived id to catch drift between the registry key and the
+        config class's ``pipeline_id``.
         """
-        cls._pipelines[pipeline_id] = pipeline_class
+        config_pipeline_id = pipeline_class.get_config_class().pipeline_id
+        if pipeline_id != config_pipeline_id:
+            class_name = getattr(pipeline_class, "__name__", repr(pipeline_class))
+            raise ValueError(
+                f"Pipeline id mismatch: registered as '{pipeline_id}' but "
+                f"{class_name}.get_config_class().pipeline_id is "
+                f"'{config_pipeline_id}'."
+            )
+        NodeRegistry.register(pipeline_class)
 
     @classmethod
     def get(cls, pipeline_id: str) -> type["Pipeline"] | None:
-        """Get a pipeline class by its ID.
-
-        Args:
-            pipeline_id: Pipeline identifier
-
-        Returns:
-            Pipeline class if found, None otherwise
-        """
-        return cls._pipelines.get(pipeline_id)
+        node_class = NodeRegistry.get(pipeline_id)
+        return node_class if _is_pipeline(node_class) else None
 
     @classmethod
     def unregister(cls, pipeline_id: str) -> bool:
-        """Remove a pipeline from the registry.
-
-        Args:
-            pipeline_id: Pipeline identifier to remove
-
-        Returns:
-            True if pipeline was removed, False if not found
-        """
-        if pipeline_id in cls._pipelines:
-            del cls._pipelines[pipeline_id]
-            return True
-        return False
+        if cls.get(pipeline_id) is None:
+            return False
+        return NodeRegistry.unregister(pipeline_id)
 
     @classmethod
     def is_registered(cls, pipeline_id: str) -> bool:
-        """Check if a pipeline is registered.
-
-        Args:
-            pipeline_id: Pipeline identifier
-
-        Returns:
-            True if pipeline is registered, False otherwise
-        """
-        return pipeline_id in cls._pipelines
+        return cls.get(pipeline_id) is not None
 
     @classmethod
     def get_config_class(cls, pipeline_id: str) -> type["BasePipelineConfig"] | None:
-        """Get config class for a specific pipeline.
-
-        Args:
-            pipeline_id: Pipeline identifier
-
-        Returns:
-            Pydantic config class if found, None otherwise
-        """
         pipeline_class = cls.get(pipeline_id)
-        if pipeline_class is None:
-            return None
-        return pipeline_class.get_config_class()
+        return pipeline_class.get_config_class() if pipeline_class else None
 
     @classmethod
     def list_pipelines(cls) -> list[str]:
-        """Get list of all registered pipeline IDs.
-
-        Returns:
-            List of pipeline IDs
-        """
-        return list(cls._pipelines.keys())
+        return [
+            pid
+            for pid in NodeRegistry.list_node_types()
+            if _is_pipeline(NodeRegistry.get(pid))
+        ]
 
     @classmethod
     def chain_produces_video(cls, pipeline_ids: list[str]) -> bool:
@@ -246,28 +234,41 @@ def _register_pipelines():
 
 
 def _initialize_registry():
-    """Initialize registry with built-in pipelines and plugins."""
+    """Initialize registry with built-in pipelines, nodes, and plugins."""
     # Register built-in pipelines first
     _register_pipelines()
 
-    # Load and register plugin pipelines
+    # Register built-in nodes (no-op on the base abstraction branch)
+    from scope.core.nodes import register_builtin_nodes
+
+    register_builtin_nodes()
+
+    # Load and register plugins. The unified register_plugin_nodes fires
+    # both register_pipelines and register_nodes hooks, so old and new
+    # plugins are picked up in one call.
     try:
         from scope.core.plugins import (
             ensure_plugins_installed,
             load_plugins,
-            register_plugin_pipelines,
+            register_plugin_nodes,
         )
 
         ensure_plugins_installed()
         load_plugins()
-        register_plugin_pipelines(PipelineRegistry)
+        register_plugin_nodes()
     except Exception as e:
         logger.error(
             f"Failed to load plugins: {e}. Built-in pipelines are still available."
         )
 
+    from scope.core.nodes.registry import NodeRegistry
+
     pipeline_count = len(PipelineRegistry.list_pipelines())
-    logger.info(f"Registry initialized with {pipeline_count} pipeline(s)")
+    node_count = len(NodeRegistry.list_node_types())
+    logger.info(
+        f"Registry initialized with {pipeline_count} pipeline(s) and "
+        f"{node_count} node(s)"
+    )
 
 
 # Auto-register pipelines on module import
diff --git a/src/scope/core/plugins/__init__.py b/src/scope/core/plugins/__init__.py
index a681707d7..652f85883 100644
--- a/src/scope/core/plugins/__init__.py
+++ b/src/scope/core/plugins/__init__.py
@@ -14,6 +14,7 @@
     get_plugin_manager,
     load_plugins,
     pm,
+    register_plugin_nodes,
     register_plugin_pipelines,
 )
 
@@ -22,6 +23,7 @@
     "ensure_plugins_installed",
     "load_plugins",
     "pm",
+    "register_plugin_nodes",
     "register_plugin_pipelines",
     "get_plugin_manager",
     "FailedPluginInfo",
diff --git a/src/scope/core/plugins/hookspecs.py b/src/scope/core/plugins/hookspecs.py
index 62ce4f672..2c0aa78b7 100644
--- a/src/scope/core/plugins/hookspecs.py
+++ b/src/scope/core/plugins/hookspecs.py
@@ -22,3 +22,17 @@ def register_pipelines(self, register):
             def register_pipelines(register):
                 register(MyPipeline)
         """
+
+    @hookspec
+    def register_nodes(self, register):
+        """Register custom node types.
+
+        Args:
+            register: Callback to register node classes.
+                     Usage: register(NodeClass)
+
+        Example:
+            @scope.core.hookimpl
+            def register_nodes(register):
+                register(MyCustomNode)
+        """
diff --git a/src/scope/core/plugins/manager.py b/src/scope/core/plugins/manager.py
index 87ee5a48a..d18d7118f 100644
--- a/src/scope/core/plugins/manager.py
+++ b/src/scope/core/plugins/manager.py
@@ -550,30 +550,36 @@ def get_failed_plugins(self) -> list[FailedPluginInfo]:
         with self._lock:
             return list(self._failed_plugins)
 
-    def register_plugin_pipelines(self, registry: "PipelineRegistry") -> None:
-        """Call register_pipelines hook for all plugins.
-
-        Args:
-            registry: PipelineRegistry to register pipelines with
+    def register_plugin_nodes(self, registry: Any = None) -> None:
+        """Fire ``register_nodes`` and ``register_pipelines`` hooks.
+
+        Both hooks plant into the unified :class:`NodeRegistry` storage,
+        so existing plugins using ``register_pipelines(register)`` keep
+        working unchanged alongside new ones using ``register_nodes``.
+        The ``registry`` argument is accepted for legacy callers but
+        ignored — the unified storage is always used.
         """
+        from scope.core.nodes.registry import NodeRegistry
+        from scope.core.pipelines.registry import PipelineRegistry
+
+        del registry  # legacy parameter, kept for callsite compat
+
         with self._lock:
-            # Clear previous mappings
             self._pipeline_to_plugin.clear()
 
-            def register_callback(pipeline_class: Any) -> None:
-                """Callback function passed to plugins."""
-                config_class = pipeline_class.get_config_class()
-                pipeline_id = config_class.pipeline_id
-                registry.register(pipeline_id, pipeline_class)
-
-                # Track which plugin owns this pipeline
-                # We'll update this mapping after the hook call
-                logger.info(f"Registered plugin pipeline: {pipeline_id}")
+            def register_callback(node_class: Any) -> None:
+                NodeRegistry.register(node_class)
+                node_id = getattr(node_class, "node_type_id", None) or (
+                    node_class.get_config_class().pipeline_id
+                )
+                logger.info(f"Registered plugin node: {node_id}")
 
+            self._pm.hook.register_nodes(register=register_callback)
             self._pm.hook.register_pipelines(register=register_callback)
+            self._update_pipeline_plugin_mapping(PipelineRegistry)
 
-            # Update pipeline-to-plugin mapping by checking which plugins provide which pipelines
-            self._update_pipeline_plugin_mapping(registry)
+    # Backwards-compat alias for internal callers using the legacy name.
+    register_plugin_pipelines = register_plugin_nodes
 
     def _update_pipeline_plugin_mapping(self, registry: "PipelineRegistry") -> None:
         """Update the mapping of pipeline IDs to plugin names."""
@@ -1658,10 +1664,15 @@ def load_plugins() -> None:
     get_plugin_manager().load_plugins()
 
 
-def register_plugin_pipelines(registry: "PipelineRegistry") -> None:
-    """Call register_pipelines hook for all plugins.
+def register_plugin_nodes(registry: Any = None) -> None:
+    """Fire ``register_nodes`` + ``register_pipelines`` hooks.
 
-    Args:
-        registry: PipelineRegistry to register pipelines with
+    Both hookspecs plant into the unified :class:`NodeRegistry` storage,
+    so old and new plugins coexist. The ``registry`` argument is kept
+    for legacy callers and ignored.
     """
-    get_plugin_manager().register_plugin_pipelines(registry)
+    get_plugin_manager().register_plugin_nodes(registry)
+
+
+# Backwards-compat alias for internal callers using the legacy name.
+register_plugin_pipelines = register_plugin_nodes
diff --git a/src/scope/server/app.py b/src/scope/server/app.py
index 470771551..e6feca3f9 100644
--- a/src/scope/server/app.py
+++ b/src/scope/server/app.py
@@ -823,47 +823,62 @@ async def get_pipeline_schemas(
     http_request: Request,
     cloud_manager: ScopeCloudBackend = Depends(get_scope_cloud),
 ):
-    """Get configuration schemas and defaults for all available pipelines.
+    """Compat alias for the pipeline-rich subset of the unified node catalog.
 
-    Returns the output of each pipeline's get_schema_with_metadata() method,
-    which includes:
-    - Pipeline metadata (id, name, description, version)
-    - supported_modes: List of supported input modes ("text", "video")
-    - default_mode: Default input mode for this pipeline
-    - mode_defaults: Mode-specific default overrides (if any)
-    - config_schema: Full JSON schema with defaults
+    Derives its response from the unified :class:`NodeRegistry` —
+    every entry whose :attr:`NodeDefinition.pipeline_meta` is set is a
+    pipeline, and ``pipeline_meta`` already holds the full output of
+    ``get_schema_with_metadata()`` (config_schema, mode_defaults,
+    supports_lora, supports_vace, etc.). Kept so existing frontend
+    callers in ``usePipelines.ts`` keep working without migration; new
+    code should read from ``GET /api/v1/nodes/definitions`` instead.
 
-    The frontend should use this as the source of truth for parameter defaults.
-
-    In cloud mode (when connected to cloud), this proxies the request to the
-    cloud-hosted scope backend to get the available pipelines there.
+    In cloud mode this proxies to the cloud-hosted scope backend.
     """
     global _pipeline_schemas_cache
     if _pipeline_schemas_cache is not None:
         return _pipeline_schemas_cache
 
-    from scope.core.pipelines.registry import PipelineRegistry
+    from scope.core.nodes.registry import NodeRegistry
     from scope.core.plugins import get_plugin_manager
 
     plugin_manager = get_plugin_manager()
     pipelines: dict = {}
 
-    for pipeline_id in PipelineRegistry.list_pipelines():
-        config_class = PipelineRegistry.get_config_class(pipeline_id)
-        if config_class:
-            # get_schema_with_metadata() includes supported_modes, default_mode,
-            # and mode_defaults directly from the config class
-            schema_data = config_class.get_schema_with_metadata()
-            schema_data["plugin_name"] = plugin_manager.get_plugin_for_pipeline(
-                pipeline_id
-            )
-            pipelines[pipeline_id] = schema_data
+    for definition in NodeRegistry.get_all_definitions():
+        if definition.pipeline_meta is None:
+            continue
+        schema_data = dict(definition.pipeline_meta)
+        schema_data["plugin_name"] = plugin_manager.get_plugin_for_pipeline(
+            definition.node_type_id
+        )
+        pipelines[definition.node_type_id] = schema_data
 
     response = PipelineSchemasResponse(pipelines=pipelines)
     _pipeline_schemas_cache = response
     return response
 
 
+# ---------------------------------------------------------------------------
+# Node definitions
+# ---------------------------------------------------------------------------
+
+
+@app.get("/api/v1/nodes/definitions")
+async def get_node_definitions():
+    """Return definitions for every registered node — pipelines included.
+
+    The unified discovery endpoint. Pipelines (Pipeline subclasses)
+    appear with ``pipeline_meta`` populated; plain custom nodes leave
+    ``pipeline_meta`` ``None``. Frontend consumers that only want plain
+    nodes filter on ``pipeline_meta == null`` client-side; consumers
+    that want rich pipeline data read from ``pipeline_meta`` directly.
+    """
+    from scope.core.nodes.registry import NodeRegistry
+
+    return {"nodes": [d.model_dump() for d in NodeRegistry.get_all_definitions()]}
+
+
 # ---------------------------------------------------------------------------
 # OSC endpoints
 # ---------------------------------------------------------------------------
@@ -1201,13 +1216,31 @@ async def handle_webrtc_offer(
             logger.info("Using relay mode - video will flow through backend to cloud")
             return await webrtc_manager.handle_offer_with_relay(request, cloud_manager)
 
-        # Local mode: ensure pipeline is loaded before proceeding
-        status_info = await pipeline_manager.get_status_info_async()
-        if status_info["status"] != "loaded":
-            raise HTTPException(
-                status_code=400,
-                detail="Pipeline not loaded. Please load pipeline first.",
+        # Local mode: ensure pipeline is loaded before proceeding.
+        # Node-only graphs (no pipeline nodes) skip this check — the graph
+        # executor handles custom nodes directly without loading pipelines.
+        graph_data = (
+            request.initialParameters.graph if request.initialParameters else None
+        )
+        has_graph_nodes = False
+        if graph_data is not None:
+            nodes = (
+                graph_data.get("nodes", [])
+                if isinstance(graph_data, dict)
+                else getattr(graph_data, "nodes", [])
+            )
+            has_graph_nodes = any(
+                (n.get("type") if isinstance(n, dict) else getattr(n, "type", None))
+                == "node"
+                for n in nodes
             )
+        if not has_graph_nodes:
+            status_info = await pipeline_manager.get_status_info_async()
+            if status_info["status"] != "loaded":
+                raise HTTPException(
+                    status_code=400,
+                    detail="Pipeline not loaded. Please load pipeline first.",
+                )
 
         return await webrtc_manager.handle_offer(
             request, pipeline_manager, tempo_sync=tempo_sync
diff --git a/src/scope/server/audio_track.py b/src/scope/server/audio_track.py
index 6d8b03433..0c9a28541 100644
--- a/src/scope/server/audio_track.py
+++ b/src/scope/server/audio_track.py
@@ -2,9 +2,12 @@
 import collections
 import fractions
 import logging
+import threading
 import time
+import weakref
 
 import numpy as np
+import torch
 from aiortc import MediaStreamTrack
 from aiortc.mediastreams import MediaStreamError
 from av import AudioFrame
@@ -25,6 +28,28 @@
 AUDIO_MAX_BUFFER_SAMPLES = AUDIO_CLOCK_RATE * 60
 
 
+# Registry of live audio tracks so graph nodes that need the playhead (e.g.
+# ACE-Step's StreamVAEDecode skip gate, which mirrors the realtime demo's
+# ``audio_eng.position / SAMPLE_RATE`` read in pipeline.py) can query it
+# without a hard dependency on FrameProcessor. Weak refs so closed tracks
+# drop out automatically.
+_PLAYHEAD_LOCK = threading.Lock()
+_PLAYHEAD_TRACKS: "weakref.WeakSet[AudioProcessingTrack]" = weakref.WeakSet()
+
+
+def get_current_playhead_seconds() -> float | None:
+    """Return the playhead position of the first live audio track, in seconds.
+
+    Returns None if no track is registered yet or none are live. Callers
+    should treat None as "skip gate disabled this tick".
+    """
+    with _PLAYHEAD_LOCK:
+        for track in _PLAYHEAD_TRACKS:
+            if track.readyState == "live":
+                return track.playhead_seconds
+    return None
+
+
 class AudioProcessingTrack(MediaStreamTrack):
     """WebRTC audio track that streams generated audio from the pipeline.
 
@@ -54,6 +79,19 @@ def __init__(
         self._start: float | None = None
         self._timestamp: int = 0
 
+        with _PLAYHEAD_LOCK:
+            _PLAYHEAD_TRACKS.add(self)
+
+    @property
+    def playhead_seconds(self) -> float:
+        """Current playback position in seconds (monotonic recv timestamp).
+
+        Mirrors the realtime demo's ``audio_eng.position / SAMPLE_RATE``
+        (see ``demos/realtime_motion_graph/pipeline.py``). Value is 0 before
+        the first ``recv`` call.
+        """
+        return self._timestamp / AUDIO_CLOCK_RATE
+
     @staticmethod
     def _resample_audio(
         audio: np.ndarray, source_rate: int, target_rate: int
@@ -97,9 +135,15 @@ async def recv(self) -> AudioFrame:
         if self.frame_processor.paused:
             return self._create_silence_frame()
 
-        # Drain all available audio from the queue to minimise latency
-        # for bursty or small-chunk pipelines.
-        while True:
+        samples_needed = self._samples_per_frame * self.channels
+
+        # Lazy drain: pull chunks from the queue only until the local buffer
+        # has enough samples to serve this frame. Leaving unconsumed chunks
+        # in ``audio_output_queue`` creates natural backpressure — producers
+        # block on the (small) queue's blocking put and cascade the stall
+        # upstream through node-to-node edge queues, matching production
+        # rate to real-time consumption without silent drops.
+        while self._buffered_samples < samples_needed:
             audio_tensor, sample_rate = self.frame_processor.get_audio()
             if audio_tensor is None and sample_rate is None:
                 break
@@ -119,6 +163,10 @@ async def recv(self) -> AudioFrame:
                 )
                 self._first_audio_logged = True
 
+            # numpy() rejects bfloat16/float16; upcast so half-precision tensors
+            # from GPU models (e.g. ACE-Step VAE decode) don't silently become silence.
+            if audio_tensor.dtype in (torch.bfloat16, torch.float16):
+                audio_tensor = audio_tensor.float()
             audio_np = audio_tensor.numpy()
             if audio_np.ndim == 1:
                 audio_np = audio_np.reshape(1, -1)
@@ -142,10 +190,9 @@ async def recv(self) -> AudioFrame:
             self._chunks.append(interleaved)
             self._buffered_samples += len(interleaved)
 
-        # Cap buffer to prevent unbounded growth.
-        # Trim-to-tail: concatenate and keep only the newest samples so a
-        # single large chunk (e.g. LTX2 delivering >1s at once) is never
-        # discarded entirely.
+        # Safety net: if something managed to push more than the cap anyway
+        # (shouldn't happen with lazy drain + maxsize=1 backpressure), keep
+        # the tail so we don't grow unbounded.
         max_interleaved = AUDIO_MAX_BUFFER_SAMPLES * self.channels
         if self._buffered_samples > max_interleaved:
             flat = np.concatenate(list(self._chunks))
@@ -156,7 +203,6 @@ async def recv(self) -> AudioFrame:
             logger.warning("Audio buffer overflow, dropped oldest chunks")
 
         # Serve a 20ms frame from the buffer
-        samples_needed = self._samples_per_frame * self.channels
         if self._buffered_samples >= samples_needed:
             flat = np.concatenate(list(self._chunks))
             self._chunks.clear()
diff --git a/src/scope/server/frame_processor.py b/src/scope/server/frame_processor.py
index 9f5724bda..571479432 100644
--- a/src/scope/server/frame_processor.py
+++ b/src/scope/server/frame_processor.py
@@ -212,8 +212,25 @@ def start(self):
             )
             return
 
-        # Local mode: setup pipeline graph
-        if not self.pipeline_ids:
+        # Local mode: setup pipeline graph.
+        # Node-only graphs (custom nodes, audio-only workflows) are allowed
+        # to start without any pipeline IDs — the graph executor still runs
+        # custom nodes via NodeProcessor and the audio path works through
+        # the standard audio_output_queue.
+        graph_param = (self.parameters or {}).get("graph")
+        _has_custom_nodes = False
+        if graph_param is not None:
+            _nodes = (
+                graph_param.get("nodes", [])
+                if isinstance(graph_param, dict)
+                else getattr(graph_param, "nodes", [])
+            )
+            _has_custom_nodes = any(
+                (n.get("type") if isinstance(n, dict) else getattr(n, "type", None))
+                == "node"
+                for n in _nodes
+            )
+        if not self.pipeline_ids and not _has_custom_nodes:
             error_msg = "No pipeline IDs provided, cannot start"
             logger.error(error_msg)
             self.running = False
@@ -745,6 +762,18 @@ def schedule_quantized_update(self, params: dict):
 
     def update_parameters(self, parameters: dict[str, Any]):
         """Update parameters that will be used in the next pipeline call."""
+        # Trace: surface every inbound parameter update at its entry point
+        # so we can distinguish "UI never sent anything" from "UI sent but
+        # was stripped upstream of the NodeProcessor dispatch".
+        import os as _os
+
+        if _os.environ.get("SCOPE_NODE_TRACE", "").lower() in ("1", "true", "yes"):
+            logger.info(
+                "[scope.FrameProcessor.update_parameters] inbound keys=%s node_id=%s",
+                sorted(parameters.keys()),
+                parameters.get("node_id"),
+            )
+
         # Always strip tempo-control keys so they never leak into pipelines,
         # even when the corresponding helper (scheduler/engine/tempo_sync) is absent.
 
diff --git a/src/scope/server/graph_executor.py b/src/scope/server/graph_executor.py
index 3939f510f..8924527b0 100644
--- a/src/scope/server/graph_executor.py
+++ b/src/scope/server/graph_executor.py
@@ -12,6 +12,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from scope.core.nodes.registry import NodeRegistry
+
 from .graph_schema import GraphConfig
 from .pipeline_processor import PipelineProcessor
 
@@ -92,7 +94,11 @@ def build_graph(
     # when multiple pipelines fan-in to the same sink.
     _sink_record_ids = {n.id for n in graph.nodes if n.type in ("sink", "record")}
 
-    # 1) Create one queue per edge (all edges are stream; frame-by-frame)
+    # 1) Create one queue per edge (all edges are stream; frame-by-frame).
+    # Node→node edges use maxsize=1 so the DAG executes one cycle at a
+    # time and large tensors don't pile up in memory; pipeline edges use
+    # the larger default so video chunks can accumulate.
+    _node_ids = {n.id for n in graph.nodes if n.type == "node"}
     stream_queues: dict[tuple[str, str], queue.Queue] = {}
     for e in graph.edges:
         if e.kind == "stream":
@@ -105,10 +111,12 @@ def build_graph(
                     f"node={e.to_node!r}, port={e.to_port!r}. "
                     f"Fan-in to a single port is not supported."
                 )
-            stream_queues[key] = queue.Queue(maxsize=DEFAULT_INPUT_QUEUE_MAXSIZE)
+            both_nodes = e.from_node in _node_ids and e.to_node in _node_ids
+            size = 1 if both_nodes else DEFAULT_INPUT_QUEUE_MAXSIZE
+            stream_queues[key] = queue.Queue(maxsize=size)
 
-    # 2) Create a processor per pipeline node and wire input_queues per port
-    node_processors: dict[str, PipelineProcessor] = {}
+    # 2) Create a processor per pipeline/custom node and wire input_queues
+    node_processors: dict[str, Any] = {}  # PipelineProcessor | NodeProcessor
     pipeline_ids: list[str] = []
 
     # Per-pipeline tempo: if any node explicitly opts in via tempo_sync=True,
@@ -118,24 +126,46 @@ def build_graph(
     any_node_has_tempo = any(n.tempo_sync for n in graph.nodes if n.type == "pipeline")
 
     for node in graph.nodes:
-        if node.type != "pipeline" or node.pipeline_id is None:
+        if node.type == "pipeline" and node.pipeline_id is not None:
+            node_gets_tempo = node.tempo_sync or not any_node_has_tempo
+            pipeline = pipeline_manager.get_pipeline_by_id(node.id)
+            processor = PipelineProcessor(
+                pipeline=pipeline,
+                pipeline_id=node.pipeline_id,
+                initial_parameters=initial_parameters.copy(),
+                session_id=session_id,
+                user_id=user_id,
+                connection_id=connection_id,
+                connection_info=connection_info,
+                tempo_sync=tempo_sync if node_gets_tempo else None,
+                modulation_engine=modulation_engine if node_gets_tempo else None,
+                node_id=node.id,
+            )
+            node_processors[node.id] = processor
+            pipeline_ids.append(node.pipeline_id)
+        elif node.type == "node" and node.node_type_id is not None:
+            from scope.core.nodes.processor import NodeProcessor
+
+            node_cls = NodeRegistry.get(node.node_type_id)
+            if node_cls is None:
+                raise ValueError(
+                    f"Unknown node type '{node.node_type_id}' for node '{node.id}'"
+                )
+            node_instance = node_cls(node_id=node.id)
+            # Merge per-node params (from workflow) with global initial params.
+            # Per-node values take precedence (e.g. "steps": 8 on DiffusionConfig).
+            node_params = {**initial_parameters}
+            if node.params:
+                node_params.update(node.params)
+            processor = NodeProcessor(
+                node=node_instance,
+                node_id=node.id,
+                initial_parameters=node_params,
+            )
+            node_processors[node.id] = processor
+            pipeline_ids.append(f"node:{node.node_type_id}")
+        else:
             continue
-        node_gets_tempo = node.tempo_sync or not any_node_has_tempo
-        pipeline = pipeline_manager.get_pipeline_by_id(node.id)
-        processor = PipelineProcessor(
-            pipeline=pipeline,
-            pipeline_id=node.pipeline_id,
-            initial_parameters=initial_parameters.copy(),
-            session_id=session_id,
-            user_id=user_id,
-            connection_id=connection_id,
-            connection_info=connection_info,
-            tempo_sync=tempo_sync if node_gets_tempo else None,
-            modulation_engine=modulation_engine if node_gets_tempo else None,
-            node_id=node.id,
-        )
-        node_processors[node.id] = processor
-        pipeline_ids.append(node.pipeline_id)
 
         for e in graph.edges_to(node.id):
             if e.kind != "stream":
@@ -143,10 +173,13 @@ def build_graph(
             q = stream_queues.get((node.id, e.to_port))
             if q is not None:
                 processor.input_queues[e.to_port] = q
+                # Mark audio ports so the processor consumes them differently
+                if e.to_port == "audio":
+                    processor.audio_input_ports.add(e.to_port)
 
     # 3) Set each producer's output_queues per port
     for node in graph.nodes:
-        if node.type != "pipeline" or node.id not in node_processors:
+        if node.type not in ("pipeline", "node") or node.id not in node_processors:
             continue
         proc = node_processors[node.id]
         out_by_port: dict[str, list[queue.Queue]] = {}
@@ -174,6 +207,13 @@ def build_graph(
     # 3d) Resize inter-pipeline queues based on downstream pipeline requirements
     _resize_graph_queues(graph, node_processors)
 
+    # 3d2) Propagate "dynamic" marking from continuous source nodes forward.
+    # A non-continuous NodeProcessor only re-fires once every dynamic input
+    # port has caught up for the current upstream cycle; static ports latch
+    # from cache. Without this, latch-fallback fires the node once per
+    # fresh-port arrival, which can double or triple GPU work per cycle.
+    _mark_dynamic_input_ports(graph, node_processors)
+
     # 3e) Derive source_queues from processor input_queues (post-resize).
     # Also build per-source-node queue mappings for multi-source support.
     source_queues: list[queue.Queue] = []
@@ -207,6 +247,18 @@ def build_graph(
                 feeder_proc = node_processors.get(e.from_node)
                 if feeder_proc is not None:
                     sink_processors_by_node[sink_id] = feeder_proc
+                    # Audio edges to sinks are served via audio_output_queue,
+                    # not dedicated sink queues — skip queue allocation so
+                    # the feeder isn't blocked on a queue nobody drains.
+                    if e.from_port == "audio" or e.to_port == "audio":
+                        # Mark the feeder's audio output port as sink-bound
+                        # so its _route_outputs pushes into audio_output_queue
+                        # (NodeProcessor only — PipelineProcessor has its own
+                        # always-on audio path via put_nowait).
+                        sink_ports = getattr(feeder_proc, "audio_sink_ports", None)
+                        if sink_ports is not None:
+                            sink_ports.add(e.from_port)
+                        break
                     sink_node = node_by_id[sink_id]
                     sink_mode = sink_node.sink_mode
                     # WebRTC preview reads sink_queues_by_node; NDI/Spout/Syphon threads
@@ -334,11 +386,21 @@ def _validate_edge_ports(
     # Build a map of node_id -> (declared_inputs, declared_outputs)
     port_map: dict[str, tuple[set[str], set[str]]] = {}
     for node in graph.nodes:
-        if node.type != "pipeline" or node.pipeline_id is None:
-            continue
-        pipeline = pipeline_manager.get_pipeline_by_id(node.id)
-        config_class = pipeline.get_config_class()
-        port_map[node.id] = (set(config_class.inputs), set(config_class.outputs))
+        if node.type == "pipeline" and node.pipeline_id is not None:
+            pipeline = pipeline_manager.get_pipeline_by_id(node.id)
+            config_class = pipeline.get_config_class()
+            port_map[node.id] = (
+                set(config_class.inputs),
+                set(config_class.outputs),
+            )
+        elif node.type == "node" and node.node_type_id is not None:
+            node_cls = NodeRegistry.get(node.node_type_id)
+            if node_cls is not None:
+                defn = node_cls.get_definition()
+                port_map[node.id] = (
+                    {p.name for p in defn.inputs},
+                    {p.name for p in defn.outputs},
+                )
 
     errors: list[str] = []
     for e in graph.edges:
@@ -364,6 +426,64 @@ def _validate_edge_ports(
         raise ValueError(f"Invalid graph ports: {'; '.join(errors)}")
 
 
+def _mark_dynamic_input_ports(
+    graph: GraphConfig,
+    node_processors: dict[str, Any],
+) -> None:
+    """Populate ``dynamic_input_ports`` on each NodeProcessor.
+
+    A node is "dynamic" if it is ``continuous=True`` (streaming source)
+    or reachable from one through stream edges. An input port on a
+    non-continuous node is "dynamic" when its upstream producer is
+    dynamic — meaning fresh values are expected once per upstream cycle.
+    Ports whose upstream is a one-shot static node (e.g. ``LoadModel``,
+    ``DiffusionConfig``) stay empty and fall back to latch-from-cache.
+
+    PipelineProcessor has its own tick loop and does not use this
+    attribute; we guard on ``hasattr`` rather than isinstance to avoid
+    dragging a type import into graph_executor.
+    """
+    dynamic_nodes: set[str] = set()
+    for node_id, proc in node_processors.items():
+        node = getattr(proc, "node", None)
+        if node is None:
+            continue
+        try:
+            if node.get_definition().continuous:
+                dynamic_nodes.add(node_id)
+        except Exception:  # pragma: no cover — defensive
+            continue
+
+    # Fixed-point forward propagation over stream edges.
+    changed = True
+    while changed:
+        changed = False
+        for e in graph.edges:
+            if e.kind != "stream":
+                continue
+            if e.from_node in dynamic_nodes and e.to_node not in dynamic_nodes:
+                if e.to_node in node_processors:
+                    dynamic_nodes.add(e.to_node)
+                    changed = True
+
+    for node_id, proc in node_processors.items():
+        if not hasattr(proc, "dynamic_input_ports"):
+            continue
+        dynamic_ports: set[str] = set()
+        for e in graph.edges:
+            if e.kind != "stream" or e.to_node != node_id:
+                continue
+            if e.from_node in dynamic_nodes:
+                dynamic_ports.add(e.to_port)
+        proc.dynamic_input_ports = dynamic_ports
+        if dynamic_ports:
+            logger.debug(
+                "Node %s: dynamic input ports = %s",
+                node_id,
+                sorted(dynamic_ports),
+            )
+
+
 def _resize_graph_queues(
     graph: GraphConfig,
     node_processors: dict[str, PipelineProcessor],
diff --git a/src/scope/server/graph_schema.py b/src/scope/server/graph_schema.py
index 26f95341f..105016336 100644
--- a/src/scope/server/graph_schema.py
+++ b/src/scope/server/graph_schema.py
@@ -29,7 +29,7 @@
 
 from __future__ import annotations
 
-from typing import Literal
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
@@ -41,14 +41,25 @@ class GraphNode(BaseModel):
         ...,
         description="Unique node id (e.g. 'input', 'yolo_plugin', 'longlive', 'output')",
     )
-    type: Literal["source", "pipeline", "sink", "record"] = Field(
+    type: Literal["source", "pipeline", "sink", "record", "node"] = Field(
         ...,
-        description="source = external input, pipeline = pipeline instance, sink = output, record = file recorder",
+        description=(
+            "source = external input, pipeline = pipeline instance, "
+            "sink = output, record = file recorder, node = custom backend node"
+        ),
     )
     pipeline_id: str | None = Field(
         default=None,
         description="Pipeline ID (registry key) when type is 'pipeline'",
     )
+    node_type_id: str | None = Field(
+        default=None,
+        description="Node type ID (NodeRegistry key) when type is 'node'",
+    )
+    params: dict[str, Any] | None = Field(
+        default=None,
+        description="Per-node parameter values for custom nodes",
+    )
     source_mode: str | None = Field(
         default=None,
         description="Video source mode for source nodes: 'video', 'camera', 'spout', 'ndi', 'syphon'",
@@ -110,6 +121,10 @@ def get_record_node_ids(self) -> list[str]:
         """Return node ids that are record nodes."""
         return [n.id for n in self.nodes if n.type == "record"]
 
+    def get_backend_node_ids(self) -> list[str]:
+        """Return node ids that are backend (custom) nodes."""
+        return [n.id for n in self.nodes if n.type == "node"]
+
     def edges_from(self, node_id: str) -> list[GraphEdge]:
         """Return edges whose source is the given node."""
         return [e for e in self.edges if e.from_node == node_id]
@@ -145,10 +160,12 @@ def validate_structure(self) -> list[str]:
         if not self.get_sink_node_ids():
             errors.append("Graph must have at least one sink node")
 
-        # Pipeline nodes must have pipeline_id
+        # Pipeline nodes must have pipeline_id; backend nodes need node_type_id
         for node in self.nodes:
             if node.type == "pipeline" and not node.pipeline_id:
                 errors.append(f"Pipeline node '{node.id}' is missing pipeline_id")
+            if node.type == "node" and not node.node_type_id:
+                errors.append(f"Node '{node.id}' is missing node_type_id")
 
         # Edge references must point to existing nodes
         node_id_set = set(node_ids)
diff --git a/src/scope/server/mcp_router.py b/src/scope/server/mcp_router.py
index 0d1426e7a..ad2b2119c 100644
--- a/src/scope/server/mcp_router.py
+++ b/src/scope/server/mcp_router.py
@@ -384,10 +384,11 @@ async def start_stream(
                 detail=f"Invalid graph: {'; '.join(errors)}",
             )
         pipeline_ids = graph_config.get_pipeline_node_ids()
-        if not pipeline_ids:
+        backend_node_ids = graph_config.get_backend_node_ids()
+        if not pipeline_ids and not backend_node_ids:
             raise HTTPException(
                 status_code=400,
-                detail="Graph must contain at least one pipeline node",
+                detail="Graph must contain at least one pipeline or custom node",
             )
 
         pipeline_tuples = [
@@ -397,8 +398,8 @@ async def start_stream(
         ]
         pipeline_id_list = [t[1] for t in pipeline_tuples]
 
-        if not use_cloud:
-            # Local mode: load pipelines locally
+        if not use_cloud and pipeline_tuples:
+            # Local mode: load pipelines locally (skip for node-only graphs)
             await pipeline_manager.load_pipelines(pipeline_tuples)
 
         initial_params: dict = {
@@ -449,9 +450,25 @@ async def start_stream(
                 detail="FrameProcessor failed to start (check logs for details)",
             )
 
+        # Custom (backend) nodes can produce audio too — if any audio sink /
+        # audio-emitting custom node is in the graph, expect audio output.
+        expect_audio = PipelineRegistry.chain_produces_audio(pipeline_id_list)
+        if request.graph is not None and not expect_audio:
+            from scope.core.nodes.registry import NodeRegistry as _NR
+
+            for gnode in graph_config.nodes:
+                if gnode.type == "node" and gnode.node_type_id:
+                    nc = _NR.get(gnode.node_type_id)
+                    if nc is None:
+                        continue
+                    defn = nc.get_definition()
+                    if any(p.port_type == "audio" for p in defn.outputs):
+                        expect_audio = True
+                        break
+
         session = HeadlessSession(
             frame_processor=frame_processor,
-            expect_audio=PipelineRegistry.chain_produces_audio(pipeline_id_list),
+            expect_audio=expect_audio,
         )
         session.start_frame_consumer()
         webrtc_manager.add_headless_session(session)
diff --git a/src/scope/server/pipeline_processor.py b/src/scope/server/pipeline_processor.py
index d7c6456c1..b9e1471e4 100644
--- a/src/scope/server/pipeline_processor.py
+++ b/src/scope/server/pipeline_processor.py
@@ -84,6 +84,10 @@ def __init__(
         # record_queues_by_node). Updated by _resize_output_queue so cached
         # references stay in sync when a queue object is replaced.
         self.external_queue_refs: list[tuple[dict, str]] = []
+        # Input ports that carry audio (tensor, sample_rate) tuples instead
+        # of video frame tensors. Set by graph_executor when wiring audio
+        # edges so audio inputs don't get accumulated into a video chunk.
+        self.audio_input_ports: set[str] = set()
 
         # Audio output queue: (audio_tensor, sample_rate) tuples.
         # Consumed by FrameProcessor.get_audio() on the sink processor.
@@ -531,6 +535,16 @@ def process_chunk(self):
                         "Audio output queue full for %s, dropping audio chunk",
                         self.pipeline_id,
                     )
+                # Also fan out audio to graph edge queues on the "audio" port
+                # so downstream custom nodes can consume it.
+                audio_queues = self.output_queues.get("audio")
+                if audio_queues:
+                    packed = (audio_cpu, audio_sample_rate)
+                    for q in audio_queues:
+                        try:
+                            q.put_nowait(packed)
+                        except queue.Full:
+                            pass
 
             # Extract video from the returned dictionary
             output = output_dict.get("video")
diff --git a/src/scope/server/webrtc.py b/src/scope/server/webrtc.py
index 18c8618a8..a9505bb1e 100644
--- a/src/scope/server/webrtc.py
+++ b/src/scope/server/webrtc.py
@@ -60,6 +60,39 @@
 vpx.MAX_BITRATE = 10000000
 
 
+def _graph_produces_audio(graph_data: dict) -> bool:
+    """Return True when any node in the graph exposes an audio output.
+
+    Inspects custom-node definitions from :class:`NodeRegistry` and
+    pipeline configs so WebRTC can decide whether to open an audio
+    track for graphs that produce audio through custom nodes rather
+    than through a registered pipeline.
+    """
+    from scope.core.nodes.registry import NodeRegistry
+
+    for node in graph_data.get("nodes", []):
+        node_type = node.get("type")
+        if node_type == "node":
+            node_type_id = node.get("node_type_id")
+            if node_type_id:
+                node_cls = NodeRegistry.get(node_type_id)
+                if node_cls is not None:
+                    defn = node_cls.get_definition()
+                    if any(p.port_type == "audio" for p in defn.outputs):
+                        return True
+        elif node_type == "pipeline":
+            pid = node.get("pipeline_id")
+            if pid:
+                cfg = PipelineRegistry.get_config_class(pid)
+                if cfg and getattr(cfg, "produces_audio", False):
+                    return True
+    # Also treat any audio-kind edge as an audio-producing graph.
+    for edge in graph_data.get("edges", []):
+        if edge.get("from_port") == "audio" or edge.get("to_port") == "audio":
+            return True
+    return False
+
+
 def _parse_graph_node_ids(
     initial_parameters: dict,
 ) -> tuple[list[str], list[str], list[str], list[str], list[str], bool]:
@@ -478,6 +511,11 @@ async def handle_offer(
                 )
 
             produces_audio = PipelineRegistry.chain_produces_audio(pipeline_ids)
+            # Graphs that emit audio through custom nodes (e.g. ACEStep)
+            # have no pipeline-produced audio, so also inspect the graph.
+            graph_data_for_audio = initial_parameters.get("graph")
+            if not produces_audio and graph_data_for_audio:
+                produces_audio = _graph_produces_audio(graph_data_for_audio)
             if produces_audio:
                 audio_track = AudioProcessingTrack(
                     frame_processor=frame_processor,
@@ -619,6 +657,18 @@ def on_data_channel_message(message):
                         # Parse the JSON message
                         data = json.loads(message)
                         logger.debug(f"Received parameter update: {data}")
+                        import os as _os
+
+                        if _os.environ.get("SCOPE_NODE_TRACE", "").lower() in (
+                            "1",
+                            "true",
+                            "yes",
+                        ):
+                            logger.info(
+                                "[scope.webrtc.datachannel] inbound keys=%s node_id=%s",
+                                sorted(data.keys()) if isinstance(data, dict) else None,
+                                data.get("node_id") if isinstance(data, dict) else None,
+                            )
 
                         # Always handle paused immediately (before quantized
                         # scheduling) so pause/unpause is never delayed.