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

Store plans in content-addressed cache (IPFS)

Browse Source 
- Plans now go through cache_manager.put() for IPFS pinning
- Returns plan_cache_id and plan_ipfs_cid in result
- Plan S-expression is content-addressed like everything else

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-12 00:30:59 +00:00
parent 10fe05a049
commit 3dbbb52d23
1 changed files with 20 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
tasks/orchestrate.py
View File

@@ -296,22 +296,26 @@ def run_recipe(
logger.info(f"Generated plan with {len(plan.steps)} steps")
# Save plan as S-expression (content-addressed by plan_id hash)
PLAN_CACHE_DIR.mkdir(parents=True, exist_ok=True)
plan_sexp_path = PLAN_CACHE_DIR / f"{plan.plan_id}.sexp"
if hasattr(plan, 'to_sexp_string'):
with open(plan_sexp_path, "w") as f:
f.write(plan.to_sexp_string())
else:
# Fallback to JSON for legacy plans
plan_json_path = PLAN_CACHE_DIR / f"{plan.plan_id}.json"
with open(plan_json_path, "w") as f:
f.write(plan.to_json())
# Save plan as S-expression through cache manager (goes to IPFS)
import tempfile
from cache_manager import get_cache_manager
cache_mgr = get_cache_manager()
# Also save a reference by run_id for easy lookup
plan_sexp = plan.to_sexp_string() if hasattr(plan, 'to_sexp_string') else plan.to_json()
plan_suffix = ".sexp" if hasattr(plan, 'to_sexp_string') else ".json"
with tempfile.NamedTemporaryFile(delete=False, suffix=plan_suffix, mode="w") as tmp:
tmp.write(plan_sexp)
tmp_path = Path(tmp.name)
# Store in cache (content-addressed, auto-pins to IPFS)
cached, plan_ipfs_cid = cache_mgr.put(tmp_path, node_type="plan", move=True)
logger.info(f"Plan cached: hash={cached.content_hash}, ipfs={plan_ipfs_cid}")
# Also save to plans dir for legacy lookup by run_id
PLAN_CACHE_DIR.mkdir(parents=True, exist_ok=True)
run_plan_path = PLAN_CACHE_DIR / f"{run_id}.sexp"
if plan_sexp_path.exists():
run_plan_path.write_text(plan_sexp_path.read_text())
run_plan_path.write_text(plan_sexp)
# Phase 3: Execute
logger.info("Phase 3: Executing plan...")
@@ -323,6 +327,8 @@ def run_recipe(
"run_id": run_id,
"recipe": recipe.name,
"plan_id": plan.plan_id,
"plan_cache_id": cached.content_hash,
"plan_ipfs_cid": plan_ipfs_cid,
"output_path": result.get("output_path"),
"output_cache_id": result.get("output_cache_id"),
"output_ipfs_cid": result.get("output_ipfs_cid"),