Files
celery/app/routers/recipes.py
2026-01-12 16:57:15 +00:00

686 lines
22 KiB
Python

"""
Recipe management routes for L1 server.
Handles recipe upload, listing, viewing, and execution.
"""
import json
import logging
from typing import Any, Dict, List, Optional, Tuple
from fastapi import APIRouter, Request, Depends, HTTPException, UploadFile, File
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from artdag_common import render
from artdag_common.middleware import wants_html, wants_json
from artdag_common.middleware.auth import UserContext
from ..dependencies import require_auth, get_current_user, get_templates, get_redis_client, get_cache_manager
from ..services.auth_service import AuthService
from ..services.recipe_service import RecipeService
from ..types import (
CompiledNode, TransformedNode, Registry, Recipe,
is_variable_input, get_effect_cid,
)
router = APIRouter()
logger = logging.getLogger(__name__)
class RecipeUploadRequest(BaseModel):
content: str # S-expression or YAML
name: Optional[str] = None
description: Optional[str] = None
class RecipeRunRequest(BaseModel):
"""Request to run a recipe with variable inputs."""
inputs: Dict[str, str] = {} # Map input names to CIDs
def get_recipe_service() -> RecipeService:
"""Get recipe service instance."""
return RecipeService(get_redis_client(), get_cache_manager())
def transform_node(
node: CompiledNode,
assets: Dict[str, Dict[str, Any]],
effects: Dict[str, Dict[str, Any]],
) -> TransformedNode:
"""
Transform a compiled node to artdag execution format.
- Resolves asset references to CIDs for SOURCE nodes
- Resolves effect references to CIDs for EFFECT nodes
- Renames 'type' to 'node_type', 'id' to 'node_id'
"""
node_id = node.get("id", "")
config = dict(node.get("config", {})) # Copy to avoid mutation
# Resolve asset references for SOURCE nodes
if node.get("type") == "SOURCE" and "asset" in config:
asset_name = config["asset"]
if asset_name in assets:
config["cid"] = assets[asset_name].get("cid")
# Resolve effect references for EFFECT nodes
if node.get("type") == "EFFECT" and "effect" in config:
effect_name = config["effect"]
if effect_name in effects:
config["cid"] = effects[effect_name].get("cid")
return {
"node_id": node_id,
"node_type": node.get("type", "EFFECT"),
"config": config,
"inputs": node.get("inputs", []),
"name": node.get("name"),
}
def build_input_name_mapping(
nodes: Dict[str, TransformedNode],
) -> Dict[str, str]:
"""
Build a mapping from input names to node IDs for variable inputs.
Variable inputs can be referenced by:
- node_id directly
- config.name (e.g., "Second Video")
- snake_case version (e.g., "second_video")
- kebab-case version (e.g., "second-video")
- node.name (def binding name)
"""
input_name_to_node: Dict[str, str] = {}
for node_id, node in nodes.items():
if node.get("node_type") != "SOURCE":
continue
config = node.get("config", {})
if not is_variable_input(config):
continue
# Map by node_id
input_name_to_node[node_id] = node_id
# Map by config.name
name = config.get("name")
if name:
input_name_to_node[name] = node_id
input_name_to_node[name.lower().replace(" ", "_")] = node_id
input_name_to_node[name.lower().replace(" ", "-")] = node_id
# Map by node.name (def binding)
node_name = node.get("name")
if node_name:
input_name_to_node[node_name] = node_id
input_name_to_node[node_name.replace("-", "_")] = node_id
return input_name_to_node
def bind_inputs(
nodes: Dict[str, TransformedNode],
input_name_to_node: Dict[str, str],
user_inputs: Dict[str, str],
) -> List[str]:
"""
Bind user-provided input CIDs to source nodes.
Returns list of warnings for inputs that couldn't be bound.
"""
warnings: List[str] = []
for input_name, cid in user_inputs.items():
# Try direct node ID match first
if input_name in nodes:
node = nodes[input_name]
if node.get("node_type") == "SOURCE":
node["config"]["cid"] = cid
logger.info(f"Bound input {input_name} directly to node, cid={cid[:16]}...")
continue
# Try input name lookup
if input_name in input_name_to_node:
node_id = input_name_to_node[input_name]
node = nodes[node_id]
node["config"]["cid"] = cid
logger.info(f"Bound input {input_name} via lookup to node {node_id}, cid={cid[:16]}...")
continue
# Input not found
warnings.append(f"Input '{input_name}' not found in recipe")
logger.warning(f"Input {input_name} not found in nodes or input_name_to_node")
return warnings
async def resolve_friendly_names_in_registry(
registry: dict,
actor_id: str,
) -> dict:
"""
Resolve friendly names to CIDs in the registry.
Friendly names are identified by containing a space (e.g., "brightness 01hw3x9k")
or by not being a valid CID format.
"""
from ..services.naming_service import get_naming_service
import re
naming = get_naming_service()
resolved = {"assets": {}, "effects": {}}
# CID patterns: IPFS CID (Qm..., bafy...) or SHA256 hash (64 hex chars)
cid_pattern = re.compile(r'^(Qm[a-zA-Z0-9]{44}|bafy[a-zA-Z0-9]+|[a-f0-9]{64})$')
for asset_name, asset_info in registry.get("assets", {}).items():
cid = asset_info.get("cid", "")
if cid and not cid_pattern.match(cid):
# Looks like a friendly name, resolve it
resolved_cid = await naming.resolve(actor_id, cid, item_type="media")
if resolved_cid:
asset_info = dict(asset_info)
asset_info["cid"] = resolved_cid
asset_info["_resolved_from"] = cid
resolved["assets"][asset_name] = asset_info
for effect_name, effect_info in registry.get("effects", {}).items():
cid = effect_info.get("cid", "")
if cid and not cid_pattern.match(cid):
# Looks like a friendly name, resolve it
resolved_cid = await naming.resolve(actor_id, cid, item_type="effect")
if resolved_cid:
effect_info = dict(effect_info)
effect_info["cid"] = resolved_cid
effect_info["_resolved_from"] = cid
resolved["effects"][effect_name] = effect_info
return resolved
async def prepare_dag_for_execution(
recipe: Recipe,
user_inputs: Dict[str, str],
actor_id: str = None,
) -> Tuple[str, List[str]]:
"""
Prepare a recipe DAG for execution by transforming nodes and binding inputs.
Resolves friendly names to CIDs if actor_id is provided.
Returns (dag_json, warnings).
"""
recipe_dag = recipe.get("dag")
if not recipe_dag or not isinstance(recipe_dag, dict):
raise ValueError("Recipe has no DAG definition")
# Deep copy to avoid mutating original
dag_copy = json.loads(json.dumps(recipe_dag))
nodes = dag_copy.get("nodes", {})
# Get registry for resolving references
registry = recipe.get("registry", {})
# Resolve friendly names to CIDs
if actor_id and registry:
registry = await resolve_friendly_names_in_registry(registry, actor_id)
assets = registry.get("assets", {}) if registry else {}
effects = registry.get("effects", {}) if registry else {}
# Transform nodes from list to dict if needed
if isinstance(nodes, list):
nodes_dict: Dict[str, TransformedNode] = {}
for node in nodes:
node_id = node.get("id")
if node_id:
nodes_dict[node_id] = transform_node(node, assets, effects)
nodes = nodes_dict
dag_copy["nodes"] = nodes
# Build input name mapping and bind user inputs
input_name_to_node = build_input_name_mapping(nodes)
logger.info(f"Input name to node mapping: {input_name_to_node}")
logger.info(f"User-provided inputs: {user_inputs}")
warnings = bind_inputs(nodes, input_name_to_node, user_inputs)
# Log final SOURCE node configs for debugging
for nid, n in nodes.items():
if n.get("node_type") == "SOURCE":
logger.info(f"Final SOURCE node {nid}: config={n.get('config')}")
# 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"] = {}
return json.dumps(dag_copy), warnings
@router.post("/upload")
async def upload_recipe(
file: UploadFile = File(...),
ctx: UserContext = Depends(require_auth),
recipe_service: RecipeService = Depends(get_recipe_service),
):
"""Upload a new recipe from S-expression or YAML file."""
import yaml
# Read content from the uploaded file
content = (await file.read()).decode("utf-8")
# Detect format (skip comments starting with ;)
def is_sexp_format(text):
for line in text.split('\n'):
stripped = line.strip()
if not stripped or stripped.startswith(';'):
continue
return stripped.startswith('(')
return False
is_sexp = is_sexp_format(content)
try:
from artdag.sexp import compile_string, ParseError, CompileError
SEXP_AVAILABLE = True
except ImportError:
SEXP_AVAILABLE = False
recipe_name = None
recipe_version = "1.0"
recipe_description = None
variable_inputs = []
fixed_inputs = []
if is_sexp:
if not SEXP_AVAILABLE:
raise HTTPException(500, "S-expression recipes require artdag.sexp module (not installed on server)")
# Parse S-expression
try:
compiled = compile_string(content)
recipe_name = compiled.name
recipe_version = compiled.version
recipe_description = compiled.description
for node in compiled.nodes:
if node.get("type") == "SOURCE":
config = node.get("config", {})
if config.get("input"):
variable_inputs.append(config.get("name", node.get("id")))
elif config.get("asset"):
fixed_inputs.append(config.get("asset"))
except Exception as e:
raise HTTPException(400, f"Parse error: {e}")
else:
# Parse YAML
try:
recipe_data = yaml.safe_load(content)
recipe_name = recipe_data.get("name")
recipe_version = recipe_data.get("version", "1.0")
recipe_description = recipe_data.get("description")
inputs = recipe_data.get("inputs", {})
for input_name, input_def in inputs.items():
if isinstance(input_def, dict) and input_def.get("fixed"):
fixed_inputs.append(input_name)
else:
variable_inputs.append(input_name)
except yaml.YAMLError as e:
raise HTTPException(400, f"Invalid YAML: {e}")
# Use filename as recipe name if not specified
if not recipe_name and file.filename:
recipe_name = file.filename.rsplit(".", 1)[0]
recipe_id, error = await recipe_service.upload_recipe(
content=content,
uploader=ctx.actor_id,
name=recipe_name,
description=recipe_description,
)
if error:
raise HTTPException(400, error)
return {
"recipe_id": recipe_id,
"name": recipe_name or "unnamed",
"version": recipe_version,
"variable_inputs": variable_inputs,
"fixed_inputs": fixed_inputs,
"message": "Recipe uploaded successfully",
}
@router.get("")
async def list_recipes(
request: Request,
offset: int = 0,
limit: int = 20,
recipe_service: RecipeService = Depends(get_recipe_service),
ctx: UserContext = Depends(require_auth),
):
"""List available recipes."""
recipes = await recipe_service.list_recipes(ctx.actor_id, offset=offset, limit=limit)
has_more = len(recipes) >= limit
if wants_json(request):
return {"recipes": recipes, "offset": offset, "limit": limit, "has_more": has_more}
from ..dependencies import get_nav_counts
nav_counts = await get_nav_counts(ctx.actor_id)
templates = get_templates(request)
return render(templates, "recipes/list.html", request,
recipes=recipes,
user=ctx,
nav_counts=nav_counts,
active_tab="recipes",
offset=offset,
limit=limit,
has_more=has_more,
)
@router.get("/{recipe_id}")
async def get_recipe(
recipe_id: str,
request: Request,
recipe_service: RecipeService = Depends(get_recipe_service),
ctx: UserContext = Depends(require_auth),
):
"""Get recipe details."""
recipe = await recipe_service.get_recipe(recipe_id)
if not recipe:
raise HTTPException(404, "Recipe not found")
# Add friendly name if available
from ..services.naming_service import get_naming_service
naming = get_naming_service()
friendly = await naming.get_by_cid(ctx.actor_id, recipe_id)
if friendly:
recipe["friendly_name"] = friendly["friendly_name"]
recipe["base_name"] = friendly["base_name"]
recipe["version_id"] = friendly["version_id"]
if wants_json(request):
return recipe
# Build DAG elements for visualization and convert nodes to steps format
dag_elements = []
steps = []
node_colors = {
"SOURCE": "#3b82f6",
"EFFECT": "#8b5cf6",
"SEQUENCE": "#ec4899",
"transform": "#10b981",
"output": "#f59e0b",
}
# Debug: log recipe structure
logger.info(f"Recipe keys: {list(recipe.keys())}")
# Get nodes from dag - can be list or dict, can be under "dag" or directly on recipe
dag = recipe.get("dag", {})
logger.info(f"DAG type: {type(dag)}, keys: {list(dag.keys()) if isinstance(dag, dict) else 'not dict'}")
nodes = dag.get("nodes", []) if isinstance(dag, dict) else []
logger.info(f"Nodes from dag.nodes: {type(nodes)}, len: {len(nodes) if hasattr(nodes, '__len__') else 'N/A'}")
# Also check for nodes directly on recipe (alternative formats)
if not nodes:
nodes = recipe.get("nodes", [])
logger.info(f"Nodes from recipe.nodes: {type(nodes)}, len: {len(nodes) if hasattr(nodes, '__len__') else 'N/A'}")
if not nodes:
nodes = recipe.get("pipeline", [])
logger.info(f"Nodes from recipe.pipeline: {type(nodes)}, len: {len(nodes) if hasattr(nodes, '__len__') else 'N/A'}")
if not nodes:
nodes = recipe.get("steps", [])
logger.info(f"Nodes from recipe.steps: {type(nodes)}, len: {len(nodes) if hasattr(nodes, '__len__') else 'N/A'}")
logger.info(f"Final nodes count: {len(nodes) if hasattr(nodes, '__len__') else 'N/A'}")
# Convert list of nodes to steps format
if isinstance(nodes, list):
for node in nodes:
node_id = node.get("id", "")
node_type = node.get("type", "EFFECT")
inputs = node.get("inputs", [])
config = node.get("config", {})
steps.append({
"id": node_id,
"name": node_id,
"type": node_type,
"inputs": inputs,
"params": config,
})
dag_elements.append({
"data": {
"id": node_id,
"label": node_id,
"color": node_colors.get(node_type, "#6b7280"),
}
})
for inp in inputs:
if isinstance(inp, str):
dag_elements.append({
"data": {"source": inp, "target": node_id}
})
elif isinstance(nodes, dict):
for node_id, node in nodes.items():
node_type = node.get("type", "EFFECT")
inputs = node.get("inputs", [])
config = node.get("config", {})
steps.append({
"id": node_id,
"name": node_id,
"type": node_type,
"inputs": inputs,
"params": config,
})
dag_elements.append({
"data": {
"id": node_id,
"label": node_id,
"color": node_colors.get(node_type, "#6b7280"),
}
})
for inp in inputs:
if isinstance(inp, str):
dag_elements.append({
"data": {"source": inp, "target": node_id}
})
# Add steps to recipe for template
recipe["steps"] = steps
# Use S-expression source if available
if "sexp" not in recipe:
recipe["sexp"] = "; No S-expression source available"
from ..dependencies import get_nav_counts
nav_counts = await get_nav_counts(ctx.actor_id)
templates = get_templates(request)
return render(templates, "recipes/detail.html", request,
recipe=recipe,
dag_elements=dag_elements,
user=ctx,
nav_counts=nav_counts,
active_tab="recipes",
)
@router.delete("/{recipe_id}")
async def delete_recipe(
recipe_id: str,
ctx: UserContext = Depends(require_auth),
recipe_service: RecipeService = Depends(get_recipe_service),
):
"""Delete a recipe."""
success, error = await recipe_service.delete_recipe(recipe_id, ctx.actor_id)
if error:
raise HTTPException(400 if "Cannot" in error else 404, error)
return {"deleted": True, "recipe_id": recipe_id}
@router.post("/{recipe_id}/run")
async def run_recipe(
recipe_id: str,
req: RecipeRunRequest,
ctx: UserContext = Depends(require_auth),
recipe_service: RecipeService = Depends(get_recipe_service),
):
"""Run a recipe with given inputs."""
from ..services.run_service import RunService
from ..dependencies import get_cache_manager
import database
recipe = await recipe_service.get_recipe(recipe_id)
if not recipe:
raise HTTPException(404, "Recipe not found")
try:
# Create run using run service
run_service = RunService(database, get_redis_client(), get_cache_manager())
# Prepare DAG for execution (transform nodes, bind inputs, resolve friendly names)
dag_json = None
if recipe.get("dag"):
dag_json, warnings = await prepare_dag_for_execution(recipe, req.inputs, actor_id=ctx.actor_id)
for warning in warnings:
logger.warning(warning)
run, error = await run_service.create_run(
recipe=recipe_id, # Use recipe hash as primary identifier
inputs=req.inputs,
use_dag=True,
dag_json=dag_json,
actor_id=ctx.actor_id,
l2_server=ctx.l2_server,
recipe_name=recipe.get("name"), # Store name for display
)
if error:
raise HTTPException(400, error)
if not run:
raise HTTPException(500, "Run creation returned no result")
return {
"run_id": run["run_id"] if isinstance(run, dict) else run.run_id,
"status": run.get("status", "pending") if isinstance(run, dict) else run.status,
"message": "Recipe execution started",
}
except HTTPException:
raise
except Exception as e:
logger.exception(f"Error running recipe {recipe_id}")
raise HTTPException(500, f"Run failed: {e}")
@router.get("/{recipe_id}/dag")
async def recipe_dag(
recipe_id: str,
request: Request,
recipe_service: RecipeService = Depends(get_recipe_service),
):
"""Get recipe DAG visualization data."""
recipe = await recipe_service.get_recipe(recipe_id)
if not recipe:
raise HTTPException(404, "Recipe not found")
dag_elements = []
node_colors = {
"input": "#3b82f6",
"effect": "#8b5cf6",
"analyze": "#ec4899",
"transform": "#10b981",
"output": "#f59e0b",
}
for i, step in enumerate(recipe.get("steps", [])):
step_id = step.get("id", f"step-{i}")
dag_elements.append({
"data": {
"id": step_id,
"label": step.get("name", f"Step {i+1}"),
"color": node_colors.get(step.get("type", "effect"), "#6b7280"),
}
})
for inp in step.get("inputs", []):
dag_elements.append({
"data": {"source": inp, "target": step_id}
})
return {"elements": dag_elements}
@router.delete("/{recipe_id}/ui", response_class=HTMLResponse)
async def ui_discard_recipe(
recipe_id: str,
request: Request,
recipe_service: RecipeService = Depends(get_recipe_service),
):
"""HTMX handler: discard a recipe."""
ctx = await get_current_user(request)
if not ctx:
return HTMLResponse('<div class="text-red-400">Login required</div>', status_code=401)
success, error = await recipe_service.delete_recipe(recipe_id, ctx.actor_id)
if error:
return HTMLResponse(f'<div class="text-red-400">{error}</div>')
return HTMLResponse(
'<div class="text-green-400">Recipe deleted</div>'
'<script>setTimeout(() => window.location.href = "/recipes", 1500);</script>'
)
@router.post("/{recipe_id}/publish")
async def publish_recipe(
recipe_id: str,
request: Request,
ctx: UserContext = Depends(require_auth),
recipe_service: RecipeService = Depends(get_recipe_service),
):
"""Publish recipe to L2 and IPFS."""
from ..services.cache_service import CacheService
from ..dependencies import get_cache_manager
import database
# Verify recipe exists
recipe = await recipe_service.get_recipe(recipe_id)
if not recipe:
raise HTTPException(404, "Recipe not found")
# Use cache service to publish (recipes are stored in cache)
cache_service = CacheService(database, get_cache_manager())
ipfs_cid, error = await cache_service.publish_to_l2(
cid=recipe_id,
actor_id=ctx.actor_id,
l2_server=ctx.l2_server,
auth_token=request.cookies.get("auth_token"),
)
if error:
if wants_html(request):
return HTMLResponse(f'<span class="text-red-400">{error}</span>')
raise HTTPException(400, error)
if wants_html(request):
return HTMLResponse(f'<span class="text-green-400">Shared: {ipfs_cid[:16]}...</span>')
return {"ipfs_cid": ipfs_cid, "published": True}