diff --git a/app/routers/recipes.py b/app/routers/recipes.py
index 12646ab..f27fdb9 100644
--- a/app/routers/recipes.py
+++ b/app/routers/recipes.py
@@ -273,13 +273,21 @@ async def run_recipe(
             dag_copy = json.loads(json.dumps(recipe_dag))  # Deep copy
             nodes = dag_copy.get("nodes", {})
 
-            # Convert nodes from list to dict if needed
+            # Convert nodes from list to dict if needed, and transform to artdag format
             if isinstance(nodes, list):
                 nodes_dict = {}
                 for node in nodes:
                     node_id = node.get("id")
                     if node_id:
-                        nodes_dict[node_id] = node
+                        # Transform to artdag format: type->node_type, id->node_id
+                        transformed = {
+                            "node_id": node_id,
+                            "node_type": node.get("type", "EFFECT"),
+                            "config": node.get("config", {}),
+                            "inputs": node.get("inputs", []),
+                            "name": node.get("name"),
+                        }
+                        nodes_dict[node_id] = transformed
                 nodes = nodes_dict
                 dag_copy["nodes"] = nodes
 
@@ -287,11 +295,19 @@ async def run_recipe(
             for input_name, content_hash in req.inputs.items():
                 if input_name in nodes:
                     node = nodes[input_name]
-                    if node.get("type") == "SOURCE":
+                    if node.get("node_type") == "SOURCE":
                         if "config" not in node:
                             node["config"] = {}
                         node["config"]["content_hash"] = content_hash
 
+            # Transform output to output_id
+            if "output" in dag_copy:
+                dag_copy["output_id"] = dag_copy.pop("output")
+
+            # Add metadata if not present
+            if "metadata" not in dag_copy:
+                dag_copy["metadata"] = {}
+
             dag_json = json.dumps(dag_copy)
 
         run, error = await run_service.create_run(