#!/usr/bin/env python3 """ Ralph Workflow Helper - pentru Echo Gestionează workflow-ul complet de creare și execuție proiecte cu Ralph + Claude Code """ import os import sys import json import subprocess from pathlib import Path WORKSPACE = Path.home() / "workspace" CLAUDE_BIN = Path.home() / ".local" / "bin" / "claude" def create_prd(project_name: str, description: str, workspace_dir: Path = WORKSPACE): """ Pas 1: Creează PRD folosind Claude Code cu skill /ralph:prd Args: project_name: Nume proiect (kebab-case) description: Descriere feature/proiect workspace_dir: Director workspace (default: ~/workspace) Returns: Path către PRD generat """ project_dir = workspace_dir / project_name project_dir.mkdir(parents=True, exist_ok=True) # Verifică dacă există package.json (feature mode) sau nu (new project mode) config_exists = (project_dir / "package.json").exists() mode = "FEATURE" if config_exists else "NEW_PROJECT" print(f"🔄 Creez PRD pentru {project_name} (mode: {mode})") print(f"📁 Workspace: {project_dir}") # Construiește prompt pentru Claude Code prompt = f"""/ralph:prd {description} """ # Rulează Claude Code în director proiect try: result = subprocess.run( [str(CLAUDE_BIN), "exec", prompt], cwd=project_dir, capture_output=True, text=True, timeout=300 # 5 min timeout ) if result.returncode != 0: print(f"❌ Eroare la generare PRD: {result.stderr}") return None # Caută PRD generat tasks_dir = project_dir / "tasks" if tasks_dir.exists(): prd_files = list(tasks_dir.glob("prd-*.md")) if prd_files: prd_file = prd_files[0] print(f"✅ PRD generat: {prd_file}") return prd_file print("⚠️ PRD generat dar nu găsit în tasks/") return None except subprocess.TimeoutExpired: print("❌ Timeout la generare PRD (>5 min)") return None except Exception as e: print(f"❌ Eroare: {e}") return None def convert_prd(prd_file: Path): """ Pas 2: Convertește PRD în prd.json folosind /ralph:convert Args: prd_file: Path către PRD markdown Returns: Path către prd.json generat """ project_dir = prd_file.parent.parent print(f"🔄 Convertesc PRD în prd.json") print(f"📄 PRD: {prd_file}") # Construiește prompt pentru conversie prompt = f"""/ralph:convert Convertește PRD-ul din {prd_file.relative_to(project_dir)} """ try: result = subprocess.run( [str(CLAUDE_BIN), "exec", prompt], cwd=project_dir, capture_output=True, text=True, timeout=180 # 3 min timeout ) if result.returncode != 0: print(f"❌ Eroare la conversie: {result.stderr}") return None # Verifică prd.json prd_json = project_dir / "scripts" / "ralph" / "prd.json" if prd_json.exists(): print(f"✅ prd.json generat: {prd_json}") # Afișează stories with open(prd_json) as f: data = json.load(f) print(f"\n📋 Stories: {len(data.get('userStories', []))}") for story in data.get('userStories', [])[:3]: print(f" - {story['id']}: {story['title']}") if len(data.get('userStories', [])) > 3: print(f" ... și {len(data['userStories']) - 3} mai multe") return prd_json print("⚠️ Conversie completă dar prd.json nu găsit") return None except subprocess.TimeoutExpired: print("❌ Timeout la conversie (>3 min)") return None except Exception as e: print(f"❌ Eroare: {e}") return None def run_ralph(prd_json: Path, max_iterations: int = 20, background: bool = False): """ Pas 3: Rulează ralph.sh pentru execuție autonomă Args: prd_json: Path către prd.json max_iterations: Max iterații Ralph background: Dacă True, rulează în background (nohup) Returns: subprocess.Popen object dacă background, altfel None """ project_dir = prd_json.parent.parent.parent ralph_script = prd_json.parent / "ralph.sh" if not ralph_script.exists(): print(f"❌ ralph.sh nu există în {ralph_script.parent}") return None print(f"🚀 Lansez Ralph loop") print(f"📁 Project: {project_dir}") print(f"🔁 Max iterations: {max_iterations}") print(f"🌙 Background: {background}") cmd = [str(ralph_script), str(max_iterations)] if background: # Rulează în background cu nohup log_file = ralph_script.parent / "logs" / "ralph.log" log_file.parent.mkdir(exist_ok=True) with open(log_file, 'w') as f: process = subprocess.Popen( cmd, cwd=project_dir, stdout=f, stderr=subprocess.STDOUT, start_new_session=True # Detach de terminal ) print(f"✅ Ralph pornit în background (PID: {process.pid})") print(f"📋 Log: {log_file}") # Salvează PID pentru tracking pid_file = ralph_script.parent / ".ralph.pid" with open(pid_file, 'w') as f: f.write(str(process.pid)) return process else: # Rulează în foreground (pentru debug) print("⚠️ Rulare în foreground (pentru debug)") result = subprocess.run(cmd, cwd=project_dir) return None def check_status(project_dir: Path): """ Verifică status Ralph pentru un proiect Args: project_dir: Director proiect Returns: Dict cu status (stories complete/incomplete, learnings) """ prd_json = project_dir / "scripts" / "ralph" / "prd.json" progress_file = project_dir / "scripts" / "ralph" / "progress.txt" pid_file = project_dir / "scripts" / "ralph" / ".ralph.pid" status = { "project": project_dir.name, "running": False, "complete": [], "incomplete": [], "learnings": [] } # Verifică dacă rulează if pid_file.exists(): with open(pid_file) as f: pid = int(f.read().strip()) try: os.kill(pid, 0) # Verifică dacă procesul există status["running"] = True status["pid"] = pid except OSError: status["running"] = False # Citește stories if prd_json.exists(): with open(prd_json) as f: data = json.load(f) for story in data.get('userStories', []): if story.get('passes'): status["complete"].append({ "id": story['id'], "title": story['title'] }) else: status["incomplete"].append({ "id": story['id'], "title": story['title'], "priority": story.get('priority', 999) }) # Citește learnings (ultimele 10 linii) if progress_file.exists(): with open(progress_file) as f: lines = f.readlines() status["learnings"] = [l.strip() for l in lines[-10:] if l.strip()] return status def main(): """CLI pentru testing""" if len(sys.argv) < 2: print("Usage:") print(" python ralph_workflow.py create PROJECT_NAME 'description'") print(" python ralph_workflow.py status PROJECT_NAME") sys.exit(1) command = sys.argv[1] if command == "create": if len(sys.argv) < 4: print("Usage: python ralph_workflow.py create PROJECT_NAME 'description'") sys.exit(1) project_name = sys.argv[2] description = sys.argv[3] # Pas 1: Creează PRD prd_file = create_prd(project_name, description) if not prd_file: print("❌ Eroare la creare PRD") sys.exit(1) # Pas 2: Convertește în prd.json prd_json = convert_prd(prd_file) if not prd_json: print("❌ Eroare la conversie") sys.exit(1) # Pas 3: Lansează Ralph în background run_ralph(prd_json, max_iterations=20, background=True) print("\n✅ Workflow complet!") print(f"📁 Project: {prd_json.parent.parent.parent}") elif command == "status": if len(sys.argv) < 3: print("Usage: python ralph_workflow.py status PROJECT_NAME") sys.exit(1) project_name = sys.argv[2] project_dir = WORKSPACE / project_name if not project_dir.exists(): print(f"❌ Proiect nu există: {project_dir}") sys.exit(1) status = check_status(project_dir) print(f"\n📊 Status: {status['project']}") print(f"🔄 Running: {'DA (PID: ' + str(status.get('pid', '')) + ')' if status['running'] else 'NU'}") print(f"✅ Complete: {len(status['complete'])}") print(f"🔄 Incomplete: {len(status['incomplete'])}") if status['complete']: print("\n✅ Stories complete:") for s in status['complete'][:5]: print(f" - {s['id']}: {s['title']}") if status['incomplete']: print("\n🔄 Stories incomplete:") for s in sorted(status['incomplete'], key=lambda x: x['priority'])[:5]: print(f" - {s['id']} (P{s['priority']}): {s['title']}") if status['learnings']: print("\n📚 Recent learnings:") for l in status['learnings'][-5:]: print(f" {l}") else: print(f"❌ Comandă necunoscută: {command}") sys.exit(1) if __name__ == "__main__": main()