Files
celery/app/services/recipe_service.py
gilesb 9df78f771d Fix run detail: add username, total_steps, recipe_name
- Extract username from actor_id format (@user@server)
- Set total_steps and executed from recipe nodes
- Use recipe name for display instead of hash

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-11 22:23:08 +00:00

245 lines
8.5 KiB
Python

"""
Recipe Service - business logic for recipe management.
Recipes are content-addressed S-expression files stored in the cache (and IPFS).
The recipe ID is the content hash of the S-expression file.
"""
import tempfile
from pathlib import Path
from typing import Optional, List, Dict, Any, Tuple
from artdag.sexp import compile_string, parse, serialize, CompileError, ParseError
class RecipeService:
"""
Service for managing recipes.
Recipes are stored in the content-addressed cache, not Redis.
"""
def __init__(self, redis, cache):
# Redis kept for compatibility but not used for recipe storage
self.redis = redis
self.cache = cache
async def get_recipe(self, recipe_id: str) -> Optional[Dict[str, Any]]:
"""Get a recipe by ID (content hash)."""
# Get from cache (content-addressed storage)
path = self.cache.get_by_content_hash(recipe_id)
if not path or not path.exists():
return None
with open(path) as f:
sexp_content = f.read()
# Compile S-expression recipe
try:
compiled = compile_string(sexp_content)
recipe_data = compiled.to_dict()
except (ParseError, CompileError) as e:
# Return basic error info
return {"error": str(e), "recipe_id": recipe_id}
# Add the recipe_id to the data for convenience
recipe_data["recipe_id"] = recipe_id
recipe_data["sexp"] = sexp_content # Keep original S-expression
# Get IPFS CID if available
ipfs_cid = self.cache.get_ipfs_cid(recipe_id)
if ipfs_cid:
recipe_data["ipfs_cid"] = ipfs_cid
# Compute step_count from nodes
nodes = recipe_data.get("dag", {}).get("nodes", [])
recipe_data["step_count"] = len(nodes) if isinstance(nodes, list) else 0
return recipe_data
async def list_recipes(self, actor_id: str = None, offset: int = 0, limit: int = 20) -> list:
"""
List available recipes for a user.
L1 data is isolated per-user - only shows recipes owned by actor_id.
Note: This scans the cache for recipe files. For production,
you might want a database index of recipes by owner.
"""
import logging
logger = logging.getLogger(__name__)
# Get all cached items and filter for recipes
recipes = []
# Check if cache has a list method for recipes
if hasattr(self.cache, 'list_by_type'):
items = self.cache.list_by_type('recipe')
logger.info(f"Found {len(items)} recipe items in cache")
for content_hash in items:
recipe = await self.get_recipe(content_hash)
if recipe:
uploader = recipe.get("uploader")
logger.info(f"Recipe {content_hash[:12]}: uploader={uploader}, actor_id={actor_id}")
# Filter by actor - L1 is per-user
if actor_id is None or uploader == actor_id:
recipes.append(recipe)
# Sort by name
recipes.sort(key=lambda r: r.get("name", ""))
# Paginate
return recipes[offset:offset + limit]
async def upload_recipe(
self,
sexp_content: str,
uploader: str,
name: str = None,
description: str = None,
) -> Tuple[Optional[str], Optional[str]]:
"""
Upload a recipe from S-expression content.
The recipe is stored in the cache and optionally pinned to IPFS.
Returns (recipe_id, error_message).
"""
# Validate and compile S-expression
try:
compiled = compile_string(sexp_content)
except ParseError as e:
return None, f"Parse error: {e}"
except CompileError as e:
return None, f"Compile error: {e}"
# For now, store the original S-expression content
# The uploader info is not embedded in the S-expression (kept in metadata)
# In a full implementation, we might add a :uploader keyword
# Write to temp file for caching
try:
with tempfile.NamedTemporaryFile(delete=False, suffix=".sexp", mode="w") as tmp:
tmp.write(sexp_content)
tmp_path = Path(tmp.name)
# Store in cache (content-addressed, auto-pins to IPFS)
cached, ipfs_cid = self.cache.put(tmp_path, node_type="recipe", move=True)
recipe_id = cached.content_hash
return recipe_id, None
except Exception as e:
return None, f"Failed to cache recipe: {e}"
async def delete_recipe(self, recipe_id: str, actor_id: str = None) -> Tuple[bool, Optional[str]]:
"""
Delete a recipe.
Note: This only removes from local cache. IPFS copies persist.
Returns (success, error_message).
"""
# Get recipe to check ownership
recipe = await self.get_recipe(recipe_id)
if not recipe:
return False, "Recipe not found"
# Check ownership if actor_id provided
if actor_id:
recipe_owner = recipe.get("uploader")
if recipe_owner and recipe_owner != actor_id:
return False, "Cannot delete: you don't own this recipe"
# Delete from cache
try:
if hasattr(self.cache, 'delete_by_content_hash'):
success, msg = self.cache.delete_by_content_hash(recipe_id)
if not success:
return False, msg
else:
# Fallback: get path and delete directly
path = self.cache.get_by_content_hash(recipe_id)
if path and path.exists():
path.unlink()
return True, None
except Exception as e:
return False, f"Failed to delete: {e}"
def parse_recipe(self, sexp_content: str) -> Dict[str, Any]:
"""Parse and compile recipe S-expression content."""
compiled = compile_string(sexp_content)
return compiled.to_dict()
def build_dag(self, recipe: Dict[str, Any]) -> Dict[str, Any]:
"""
Build DAG visualization data from recipe.
Returns nodes and edges for Cytoscape.js.
"""
vis_nodes = []
edges = []
dag = recipe.get("dag", {})
dag_nodes = dag.get("nodes", [])
output_node = dag.get("output")
# Handle list format from compiled S-expression recipes
if isinstance(dag_nodes, list):
for node_def in dag_nodes:
node_id = node_def.get("id")
node_type = node_def.get("type", "EFFECT")
vis_nodes.append({
"data": {
"id": node_id,
"label": node_id,
"nodeType": node_type,
"isOutput": node_id == output_node,
}
})
# Build edges from inputs
for input_ref in node_def.get("inputs", []):
if isinstance(input_ref, dict):
source = input_ref.get("node") or input_ref.get("input")
else:
source = input_ref
if source:
edges.append({
"data": {
"source": source,
"target": node_id,
}
})
# Handle dict format (legacy)
elif isinstance(dag_nodes, dict):
for node_id, node_def in dag_nodes.items():
node_type = node_def.get("type", "EFFECT")
vis_nodes.append({
"data": {
"id": node_id,
"label": node_id,
"nodeType": node_type,
"isOutput": node_id == output_node,
}
})
for input_ref in node_def.get("inputs", []):
if isinstance(input_ref, dict):
source = input_ref.get("node") or input_ref.get("input")
else:
source = input_ref
if source:
edges.append({
"data": {
"source": source,
"target": node_id,
}
})
return {"nodes": vis_nodes, "edges": edges}