art-dag/celery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix effect loading from IPFS and improve COMPOUND handling

Browse Source 
- Remove effect:identity shortcut executor so effects load from IPFS by CID
- COMPOUND nodes now fall back to generic EFFECT executor for dynamic effects
- EFFECT nodes also fall back to generic executor when specific not found
- Update test assertions to match current implementation
- Raise error instead of silently skipping when effect executor not found

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-13 03:04:52 +00:00
parent ad15ef1ce7
commit c3d6427883
2 changed files with 32 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
legacy_tasks.py
View File

@@ -94,15 +94,8 @@ class DogExecutor(Executor):
return dog_process(inputs, output_path, config, None)
@register_executor("effect:identity")
class IdentityExecutor(Executor):
"""Executor for the identity effect (passthrough)."""
def execute(self, config: Dict, inputs: List[Path], output_path: Path) -> Path:
from artdag.nodes.effect import effect_identity
if len(inputs) != 1:
raise ValueError(f"Identity effect expects 1 input, got {len(inputs)}")
return effect_identity(inputs[0], output_path, config)
# Note: No specific executor for identity - let the generic EFFECT executor
# handle it so effects can be loaded from IPFS by CID when specified.
@register_executor(NodeType.SOURCE)
@@ -796,20 +789,23 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
effect_name = filter_config.get("effect")
effect_cid = filter_config.get("cid")
if effect_name and effect_cid:
# Get effect executor
if effect_name:
# Try specific executor first, fall back to generic EFFECT executor
effect_executor = get_executor(f"effect:{effect_name}")
if not effect_executor:
effect_executor = get_executor("EFFECT")
if effect_executor:
temp_dir = Path(tempfile.mkdtemp())
temp_output = temp_dir / f"compound_{i}_{effect_name}.mkv"
logger.info(f"COMPOUND: Running effect {effect_name} (step {i+1}/{len(filter_chain)})")
logger.info(f"COMPOUND: Running effect {effect_name} (cid={effect_cid[:16] if effect_cid else 'built-in'}...) step {i+1}/{len(filter_chain)}")
result_path = effect_executor.execute(filter_config, [current_input], temp_output)
current_input = result_path
temp_files.append(temp_dir)
else:
logger.warning(f"COMPOUND: No executor for effect {effect_name}, skipping")
raise ValueError(f"COMPOUND: No executor for effect {effect_name}")
# Store final result
output_dir = CACHE_DIR / "nodes" / step.cache_id
@@ -918,7 +914,11 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
if not effect_name:
raise ValueError(f"EFFECT node missing 'effect' in config: {step.config}")
# Try specific executor first (e.g., effect:dog, effect:identity)
executor = get_executor(f"effect:{effect_name}")
if not executor:
# Fall back to generic EFFECT executor (handles IPFS effects)
executor = get_executor("EFFECT")
if not executor:
raise ValueError(f"No executor for effect: {effect_name}")
@@ -929,7 +929,8 @@ def execute_recipe(self, recipe_sexp: str, input_hashes: Dict[str, str], run_id:
output_dir.mkdir(parents=True, exist_ok=True)
output_path = output_dir / "output.mkv"
logger.info(f"EFFECT: Running {effect_name} on input")
effect_cid = step.config.get("cid")
logger.info(f"EFFECT: Running {effect_name} (cid={effect_cid[:16] if effect_cid else 'built-in'}...)")
result_path = executor.execute(step.config, input_paths, output_path)
cached, content_cid = cache_manager.put(