#!/usr/bin/env python3 import os import sys import json import subprocess from datetime import datetime # Path Configurations AGENCY_DIR = ".ai_agency" STATE_FILE = os.path.join(AGENCY_DIR, "memory/state.json") BACKLOG_FILE = os.path.join(AGENCY_DIR, "memory/backlog.json") SCRATCHPAD_FILE = os.path.join(AGENCY_DIR, "memory/scratchpad.md") AGENTS_DIR = os.path.join(AGENCY_DIR, "agents") def load_json(filepath): if not os.path.exists(filepath): return {} with open(filepath, "r", encoding="utf-8") as f: return json.load(f) def save_json(filepath, data): os.makedirs(os.path.dirname(filepath), exist_ok=True) with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) def run_cmd(cmd, check=True): print(f"[EXEC] {cmd}") res = subprocess.run(cmd, shell=True, text=True, capture_output=True) if check and res.returncode != 0: print(f"[ERROR] Command failed: {res.stderr}") return False, res.stdout, res.stderr return True, res.stdout, res.stderr def is_brownfield(): markers = ["package.json", "requirements.txt", "pyproject.toml", "Cargo.toml", "go.mod", "composer.json"] return any(os.path.exists(marker) for marker in markers) def call_ai_agent(agent_name, prompt_context): agent_file = os.path.join(AGENTS_DIR, f"{agent_name}.md") if not os.path.exists(agent_file): raise FileNotFoundError(f"Agent system prompt file missing: {agent_file}") with open(agent_file, "r", encoding="utf-8") as f: system_prompt = f.read() full_prompt = f"{system_prompt}\n\n--- CURRENT EXECUTION CONTEXT ---\n{prompt_context}" print(f"[AI RUNNING] Invoking agent: {agent_name}...") with open(SCRATCHPAD_FILE, "a", encoding="utf-8") as f: f.write(f"\n\n### Agent Call: {agent_name} @ {datetime.now().isoformat()}\n") f.write(prompt_context) return True def handle_git_commit(ticket_id, message, is_wip=False): prefix = "wip:" if is_wip else "feat:" commit_msg = f"{prefix} {ticket_id} - {message}" run_cmd("git add .", check=False) run_cmd(f'git commit -m "{commit_msg}"', check=False) def get_next_pending_task(): backlog = load_json(BACKLOG_FILE) tasks = backlog.get("tasks", []) for task in tasks: if task.get("status") in ["PENDING", "pending"]: deps = task.get("dependency_task_ids", []) completed_ids = [t["id"] for t in tasks if t.get("status") in ["DONE", "completed"]] if all(dep in completed_ids for dep in deps): return task return None def is_backlog_sufficient(): backlog = load_json(BACKLOG_FILE) state = load_json(STATE_FILE) total_tasks = len(backlog.get("tasks", [])) minimum_expected = state.get("context_buffer", {}).get("minimum_expected_tasks", 3) decomposition_pass = state.get("checkpoint", {}).get("decomposition_pass", 0) max_passes = state.get("checkpoint", {}).get("max_decomposition_passes", 3) if decomposition_pass >= max_passes: return True return total_tasks >= minimum_expected def mark_task_complete(ticket_id): backlog = load_json(BACKLOG_FILE) for task in backlog.get("tasks", []): if task["id"] == ticket_id: task["status"] = "DONE" break backlog["completed_tasks"] = sum(1 for t in backlog.get("tasks", []) if t.get("status") == "DONE") save_json(BACKLOG_FILE, backlog) def main(): print("=== AI Software House Orchestrator Running ===") state = load_json(STATE_FILE) if not state or state.get("project_name") == "UNINITIALIZED": state = { "project_name": "VibeForge Project", "project_mode": "UNKNOWN", "status": "RUNNING", "checkpoint": { "stage": "INITIALIZATION", "active_agent": "00_auditor" if is_brownfield() else "01_ceo", "current_ticket_id": None, "sub_step_index": 0, "total_sub_steps": 0, "decomposition_pass": 0, "max_decomposition_passes": 3 }, "execution_guards": { "retry_count": 0, "max_retry_attempts": 3, "max_tokens_per_ticket": 50000, "token_usage_total": 0 }, "git_state": { "active_branch": "main", "last_healthy_commit": None }, "context_buffer": { "last_agent_summary": "System initialized.", "minimum_expected_tasks": 3 }, "last_updated": datetime.now().isoformat() } state["project_mode"] = "BROWNFIELD" if is_brownfield() else "GREENFIELD" save_json(STATE_FILE, state) if state.get("status") == "BLOCKED_NEEDS_HUMAN": print("[BLOCKED] Orchestrator halted. Retry limits reached or manual intervention required.") sys.exit(1) active_agent = state["checkpoint"]["active_agent"] current_ticket = state["checkpoint"]["current_ticket_id"] print(f"State Hydrated: Agent={active_agent} | Mode={state['project_mode']} | Ticket={current_ticket}") # Branch Management per Ticket if current_ticket: branch_name = f"feature/{current_ticket}" run_cmd(f"git checkout -b {branch_name}", check=False) run_cmd(f"git checkout {branch_name}", check=False) # Prepare Context Prompt for current Agent Execution context_data = { "state": state, "active_ticket_info": current_ticket, "scratchpad": SCRATCHPAD_FILE } # Call AI Agent Logic success = call_ai_agent(active_agent, json.dumps(context_data, indent=2)) if not success: state["execution_guards"]["retry_count"] += 1 if state["execution_guards"]["retry_count"] >= state["execution_guards"]["max_retry_attempts"]: state["status"] = "BLOCKED_NEEDS_HUMAN" print(f"[FATAL] Agent {active_agent} failed {state['execution_guards']['max_retry_attempts']} times. Halting.") save_json(STATE_FILE, state) sys.exit(1) # Dynamic Transition Logic next_agent = None if active_agent == "00_auditor": next_agent = "01_ceo" elif active_agent == "01_ceo": next_agent = "02_product_manager" elif active_agent == "02_product_manager": next_agent = "03_architect" elif active_agent == "03_architect": state["checkpoint"]["decomposition_pass"] += 1 if not is_backlog_sufficient(): print(f"[DECOMPOSITION] Pass {state['checkpoint']['decomposition_pass']}: backlog too sparse. Re-invoking 02_product_manager for deeper decomposition.") next_agent = "02_product_manager" else: next_task = get_next_pending_task() if next_task: state["checkpoint"]["current_ticket_id"] = next_task["id"] next_agent = next_task.get("assigned_role", "04_dev_backend") else: next_agent = "09_tech_writer" elif active_agent in ["04_dev_backend", "05_dev_frontend"]: handle_git_commit(current_ticket, f"Work in progress by {active_agent}", is_wip=True) next_agent = "06_qa_engineer" elif active_agent == "06_qa_engineer": next_agent = "07_visual_qa" elif active_agent == "07_visual_qa": next_agent = "08_devops_security" elif active_agent == "08_devops_security": next_agent = "09_tech_writer" elif active_agent == "09_tech_writer": handle_git_commit(current_ticket, "Completed task and updated documentation", is_wip=False) mark_task_complete(current_ticket) next_task = get_next_pending_task() if next_task: state["checkpoint"]["current_ticket_id"] = next_task["id"] next_agent = next_task.get("assigned_role", "04_dev_backend") print(f"[BACKLOG] Moving to next task: {next_task['id']}") else: state["checkpoint"]["current_ticket_id"] = None next_agent = "COMPLETE" # Update State & Save Checkpoint if next_agent == "COMPLETE": state["status"] = "SUCCESS" state["checkpoint"]["active_agent"] = None print("ALL BACKLOG TASKS COMPLETED SUCCESSFULLY!") else: state["checkpoint"]["active_agent"] = next_agent state["execution_guards"]["retry_count"] = 0 print(f"[TRANSITION] Next Active Agent: {next_agent}") state["last_updated"] = datetime.now().isoformat() save_json(STATE_FILE, state) print("Orchestrator loop completed safely. State persisted to disk.") if __name__ == "__main__": main()