246 lines
7.1 KiB
Python
Executable File
246 lines
7.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Spawn an isolated Gemini CLI subagent for focused task execution.
|
|
|
|
This enables parallel execution by launching independent gemini instances
|
|
with isolated context and specific skill instructions.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def find_repo_root(start: Path) -> Path:
|
|
"""Traverse upwards to find the repository root (containing .agent/)."""
|
|
curr = start.resolve()
|
|
for _ in range(10):
|
|
if (curr / ".agent").exists():
|
|
return curr
|
|
if curr.parent == curr:
|
|
break
|
|
curr = curr.parent
|
|
return Path.cwd()
|
|
|
|
|
|
def load_skill_instructions(skill_path: Path) -> str:
|
|
"""Load skill instructions from SKILL.md file."""
|
|
if not skill_path.exists():
|
|
return ""
|
|
return skill_path.read_text(encoding="utf-8")
|
|
|
|
|
|
def spawn_subagent(
|
|
skill: str,
|
|
task: str,
|
|
repo_root: Path,
|
|
yolo: bool = True,
|
|
output_format: str = "text",
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Spawn a subagent with isolated context.
|
|
|
|
Args:
|
|
skill: Skill name (e.g., 'tdd', 'debug', 'review')
|
|
task: Task description for the subagent
|
|
repo_root: Repository root path
|
|
yolo: Auto-approve all actions (default: True for parallel execution)
|
|
output_format: Output format ('text' or 'json')
|
|
|
|
Returns:
|
|
dict with keys: success, output, error, log_file, duration_s
|
|
"""
|
|
# Generate unique subagent ID
|
|
subagent_id = uuid.uuid4().hex[:8]
|
|
timestamp = time.strftime("%Y%m%d-%H%M%S")
|
|
|
|
# Setup logging directory
|
|
log_dir = repo_root / "artifacts" / "superpowers" / "subagents"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / f"{skill}-{timestamp}-{subagent_id}.log"
|
|
|
|
# Load skill instructions
|
|
skill_file = repo_root / f".agent/skills/superpowers-{skill}/SKILL.md"
|
|
skill_instructions = load_skill_instructions(skill_file)
|
|
|
|
if not skill_instructions:
|
|
return {
|
|
"success": False,
|
|
"output": "",
|
|
"error": f"Skill not found: {skill_file}",
|
|
"log_file": str(log_file),
|
|
"duration_s": 0,
|
|
}
|
|
|
|
# Construct focused prompt
|
|
prompt = f"""You are a specialized subagent focused on: {skill}
|
|
|
|
IMPORTANT: You have ISOLATED CONTEXT. Do not assume knowledge from other conversations.
|
|
|
|
Task:
|
|
{task}
|
|
|
|
Skill Instructions:
|
|
{skill_instructions}
|
|
|
|
Requirements:
|
|
1. Follow the skill instructions exactly
|
|
2. Complete the task fully
|
|
3. Output ONLY the final result at the end
|
|
4. Do not include meta-commentary or thinking process in final output
|
|
5. Write any artifacts to artifacts/superpowers/subagent-{subagent_id}/
|
|
|
|
When complete, output:
|
|
---SUBAGENT-RESULT-START---
|
|
[Your final result here]
|
|
---SUBAGENT-RESULT-END---
|
|
"""
|
|
|
|
# Build command
|
|
cmd = ["gemini"]
|
|
if yolo:
|
|
cmd.append("--yolo")
|
|
|
|
# Execute subagent
|
|
start_time = time.time()
|
|
|
|
try:
|
|
with open(log_file, "w", encoding="utf-8") as log:
|
|
log.write("=== SUBAGENT EXECUTION LOG ===\n")
|
|
log.write(f"Skill: {skill}\n")
|
|
log.write(f"ID: {subagent_id}\n")
|
|
log.write(f"Timestamp: {timestamp}\n")
|
|
log.write(f"Task: {task}\n\n")
|
|
log.write("=== PROMPT ===\n")
|
|
log.write(prompt)
|
|
log.write("\n\n=== EXECUTION ===\n")
|
|
log.flush()
|
|
|
|
result = subprocess.run(
|
|
cmd,
|
|
input=prompt,
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=repo_root,
|
|
timeout=600, # 10 minute timeout
|
|
shell=True, # Required on Windows for .ps1/.cmd scripts
|
|
)
|
|
|
|
duration_s = time.time() - start_time
|
|
|
|
log.write("\n=== STDOUT ===\n")
|
|
log.write(result.stdout)
|
|
log.write("\n=== STDERR ===\n")
|
|
log.write(result.stderr)
|
|
log.write(f"\n=== EXIT CODE: {result.returncode} ===\n")
|
|
log.write(f"=== DURATION: {duration_s:.2f}s ===\n")
|
|
|
|
# Extract final result from markers
|
|
output = result.stdout
|
|
if "---SUBAGENT-RESULT-START---" in output:
|
|
parts = output.split("---SUBAGENT-RESULT-START---", 1)
|
|
if len(parts) > 1:
|
|
result_part = parts[1].split("---SUBAGENT-RESULT-END---", 1)
|
|
output = result_part[0].strip()
|
|
|
|
return {
|
|
"success": result.returncode == 0,
|
|
"output": output,
|
|
"error": result.stderr if result.returncode != 0 else "",
|
|
"log_file": str(log_file),
|
|
"duration_s": duration_s,
|
|
"subagent_id": subagent_id,
|
|
}
|
|
|
|
except subprocess.TimeoutExpired:
|
|
duration_s = time.time() - start_time
|
|
return {
|
|
"success": False,
|
|
"output": "",
|
|
"error": f"Subagent timed out after {duration_s:.0f}s",
|
|
"log_file": str(log_file),
|
|
"duration_s": duration_s,
|
|
"subagent_id": subagent_id,
|
|
}
|
|
except Exception as e:
|
|
duration_s = time.time() - start_time
|
|
return {
|
|
"success": False,
|
|
"output": "",
|
|
"error": f"Subagent execution failed: {e}",
|
|
"log_file": str(log_file),
|
|
"duration_s": duration_s,
|
|
"subagent_id": subagent_id,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Spawn a Gemini CLI subagent for parallel execution"
|
|
)
|
|
parser.add_argument(
|
|
"--skill",
|
|
required=True,
|
|
help="Skill to use (tdd, debug, review, rest-automation, python-automation)",
|
|
)
|
|
parser.add_argument(
|
|
"--task",
|
|
required=True,
|
|
help="Task description for the subagent",
|
|
)
|
|
parser.add_argument(
|
|
"--no-yolo",
|
|
action="store_true",
|
|
help="Disable auto-approval (interactive mode)",
|
|
)
|
|
parser.add_argument(
|
|
"--output-format",
|
|
choices=["text", "json"],
|
|
default="text",
|
|
help="Output format",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
repo_root = find_repo_root(Path.cwd())
|
|
|
|
if args.output_format == "text":
|
|
print(f"🤖 Spawning subagent: {args.skill}")
|
|
print(f"📋 Task: {args.task[:80]}{'...' if len(args.task) > 80 else ''}")
|
|
|
|
result = spawn_subagent(
|
|
skill=args.skill,
|
|
task=args.task,
|
|
repo_root=repo_root,
|
|
yolo=not args.no_yolo,
|
|
output_format=args.output_format,
|
|
)
|
|
|
|
if args.output_format == "json":
|
|
print(json.dumps(result, indent=2))
|
|
return 0 if result["success"] else 1
|
|
|
|
# Text output
|
|
print(f"\n{'✅' if result['success'] else '❌'} Subagent completed in {result['duration_s']:.1f}s")
|
|
print(f"📝 Full log: {result['log_file']}")
|
|
|
|
if result["success"]:
|
|
print(f"\n{'='*60}")
|
|
print("RESULT:")
|
|
print(f"{'='*60}")
|
|
print(result["output"])
|
|
return 0
|
|
else:
|
|
print(f"\n{'='*60}")
|
|
print("ERROR:")
|
|
print(f"{'='*60}")
|
|
print(result["error"])
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|