diff --git a/scripts/recipe_engine.py b/scripts/recipe_engine.py new file mode 100644 index 00000000..4b8c7913 --- /dev/null +++ b/scripts/recipe_engine.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +""" +recipe_engine.py — Deterministic Workflow Orchestrator for the Timmy Foundation. + +Executes "Recipes" — sequences of deterministic steps (shell commands, scripts) +to reduce reliance on LLM reasoning for common, repetitive tasks. + +Usage: + python3 scripts/recipe_engine.py recipes/deploy_beacon.json --vars REPO_PATH=/opt/the-beacon +""" + +import os +import sys +import json +import subprocess +import argparse +import time +from datetime import datetime +from typing import List, Dict, Any + +class RecipeEngine: + def __init__(self, vars: Dict[str, str] = None): + self.vars = vars or {} + self.vars["TIMESTAMP"] = datetime.now().strftime("%Y%m%d-%H%M%S") + + def log(self, message: str): + ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"[{ts}] {message}") + + def substitute_vars(self, text: str) -> str: + for k, v in self.vars.items(): + text = text.replace(f"{{{{{k}}}}}", v) + return text + + def execute_step(self, step: Dict[str, Any]) -> bool: + name = step.get("name", "Unnamed Step") + command = step.get("command") + if not command: + self.log(f" [SKIP] {name}: No command provided.") + return True + + command = self.substitute_vars(command) + self.log(f" [EXEC] {name}: {command}") + + try: + start_time = time.time() + # Use shell=True to allow pipes and redirections in recipes + process = subprocess.run(command, shell=True, capture_output=True, text=True) + duration = time.time() - start_time + + if process.returncode == 0: + self.log(f" [OK] {name} completed in {duration:.2f}s") + if process.stdout.strip(): + # Print first few lines of output + lines = process.stdout.strip().split("\n") + for line in lines[:3]: + print(f" > {line}") + if len(lines) > 3: + print(f" > ... ({len(lines)-3} more lines)") + return True + else: + self.log(f" [FAIL] {name} failed with exit code {process.returncode}") + print(f" [ERR] {process.stderr.strip()}") + return False + except Exception as e: + self.log(f" [ERROR] {name} encountered an exception: {e}") + return False + + def run_recipe(self, recipe_path: str) -> bool: + try: + with open(recipe_path, "r") as f: + recipe = json.load(f) + except Exception as e: + self.log(f"Failed to load recipe {recipe_path}: {e}") + return False + + name = recipe.get("name", "Unknown Recipe") + description = recipe.get("description", "") + steps = recipe.get("steps", []) + + self.log(f"--- Starting Recipe: {name} ---") + if description: + self.log(f"Description: {description}") + + self.log(f"Steps to execute: {len(steps)}") + + for i, step in enumerate(steps, 1): + self.log(f"Step {i}/{len(steps)}: {step.get('name')}") + success = self.execute_step(step) + if not success and not step.get("continue_on_failure", False): + self.log(f"Recipe {name} aborted due to failure in step {i}.") + return False + + self.log(f"--- Recipe {name} completed successfully ---") + return True + +def main(): + parser = argparse.ArgumentParser(description="Deterministic Recipe Engine") + parser.add_argument("recipe", help="Path to the JSON recipe file") + parser.add_argument("--vars", nargs="*", help="Variables in KEY=VALUE format") + + args = parser.parse_args() + + variables = {} + if args.vars: + for v in args.vars: + if "=" in v: + k, val = v.split("=", 1) + variables[k] = val + + engine = RecipeEngine(vars=variables) + success = engine.run_recipe(args.recipe) + if not success: + sys.exit(1) + +if __name__ == "__main__": + main()