claude-skills-reference/engineering/agenthub/scripts/hub_init.py

#!/usr/bin/env python3
"""Initialize an AgentHub collaboration session.

Creates the .agenthub/ directory structure, generates a session ID,
and writes config.yaml and state.json for the session.

Usage:
    python hub_init.py --task "Optimize API response time" --agents 3 \\
        --eval "pytest bench.py --json" --metric p50_ms --direction lower

    python hub_init.py --task "Refactor auth module" --agents 2

    python hub_init.py --demo
"""

import argparse
import json
import os
import sys
from datetime import datetime, timezone


def generate_session_id():
    """Generate a timestamp-based session ID."""
    return datetime.now().strftime("%Y%m%d-%H%M%S")


def create_directory_structure(base_path):
    """Create the .agenthub/ directory tree."""
    dirs = [
        os.path.join(base_path, "sessions"),
        os.path.join(base_path, "board", "dispatch"),
        os.path.join(base_path, "board", "progress"),
        os.path.join(base_path, "board", "results"),
    ]
    for d in dirs:
        os.makedirs(d, exist_ok=True)


def write_gitignore(base_path):
    """Write .agenthub/.gitignore to exclude worktree artifacts."""
    gitignore_path = os.path.join(base_path, ".gitignore")
    if not os.path.exists(gitignore_path):
        with open(gitignore_path, "w") as f:
            f.write("# AgentHub gitignore\n")
            f.write("# Keep board and sessions, ignore worktree artifacts\n")
            f.write("*.tmp\n")
            f.write("*.lock\n")


def write_board_index(base_path):
    """Initialize the board index file."""
    index_path = os.path.join(base_path, "board", "_index.json")
    if not os.path.exists(index_path):
        index = {
            "channels": ["dispatch", "progress", "results"],
            "counters": {"dispatch": 0, "progress": 0, "results": 0},
        }
        with open(index_path, "w") as f:
            json.dump(index, f, indent=2)
            f.write("\n")


def create_session(base_path, session_id, task, agents, eval_cmd, metric,
                   direction, base_branch):
    """Create a new session with config and state files."""
    session_dir = os.path.join(base_path, "sessions", session_id)
    os.makedirs(session_dir, exist_ok=True)

    # Write config.yaml (manual YAML to avoid dependency)
    config_path = os.path.join(session_dir, "config.yaml")
    config_lines = [
        f"session_id: {session_id}",
        f"task: \"{task}\"",
        f"agent_count: {agents}",
        f"base_branch: {base_branch}",
        f"created: {datetime.now(timezone.utc).isoformat()}",
    ]
    if eval_cmd:
        config_lines.append(f"eval_cmd: \"{eval_cmd}\"")
    if metric:
        config_lines.append(f"metric: {metric}")
    if direction:
        config_lines.append(f"direction: {direction}")

    with open(config_path, "w") as f:
        f.write("\n".join(config_lines))
        f.write("\n")

    # Write state.json
    state_path = os.path.join(session_dir, "state.json")
    state = {
        "session_id": session_id,
        "state": "init",
        "created": datetime.now(timezone.utc).isoformat(),
        "updated": datetime.now(timezone.utc).isoformat(),
        "agents": {},
    }
    with open(state_path, "w") as f:
        json.dump(state, f, indent=2)
        f.write("\n")

    return session_dir


def validate_git_repo():
    """Check if current directory is a git repository."""
    if not os.path.isdir(".git"):
        # Check parent dirs
        path = os.path.abspath(".")
        while path != "/":
            if os.path.isdir(os.path.join(path, ".git")):
                return True
            path = os.path.dirname(path)
        return False
    return True


def get_current_branch():
    """Get the current git branch name."""
    head_file = os.path.join(".git", "HEAD")
    if os.path.exists(head_file):
        with open(head_file) as f:
            ref = f.read().strip()
            if ref.startswith("ref: refs/heads/"):
                return ref[len("ref: refs/heads/"):]
    return "main"


def run_demo():
    """Show a demo of what hub_init creates."""
    print("=" * 60)
    print("AgentHub Init — Demo Mode")
    print("=" * 60)
    print()
    print("Session ID: 20260317-143022")
    print("Task: Optimize API response time below 100ms")
    print("Agents: 3")
    print("Eval: pytest bench.py --json")
    print("Metric: p50_ms (lower is better)")
    print("Base branch: dev")
    print()
    print("Directory structure created:")
    print("  .agenthub/")
    print("  ├── .gitignore")
    print("  ├── sessions/")
    print("  │   └── 20260317-143022/")
    print("  │       ├── config.yaml")
    print("  │       └── state.json")
    print("  └── board/")
    print("      ├── _index.json")
    print("      ├── dispatch/")
    print("      ├── progress/")
    print("      └── results/")
    print()
    print("config.yaml:")
    print('  session_id: 20260317-143022')
    print('  task: "Optimize API response time below 100ms"')
    print("  agent_count: 3")
    print("  base_branch: dev")
    print('  eval_cmd: "pytest bench.py --json"')
    print("  metric: p50_ms")
    print("  direction: lower")
    print()
    print("state.json:")
    print('  { "state": "init", "agents": {} }')
    print()
    print("Next step: Run /hub:spawn to launch agents")


def main():
    parser = argparse.ArgumentParser(
        description="Initialize an AgentHub collaboration session"
    )
    parser.add_argument("--task", type=str, help="Task description for agents")
    parser.add_argument("--agents", type=int, default=3,
                        help="Number of parallel agents (default: 3)")
    parser.add_argument("--eval", type=str, dest="eval_cmd",
                        help="Evaluation command to run in each worktree")
    parser.add_argument("--metric", type=str,
                        help="Metric name to extract from eval output")
    parser.add_argument("--direction", choices=["lower", "higher"],
                        help="Whether lower or higher metric is better")
    parser.add_argument("--base-branch", type=str,
                        help="Base branch (default: current branch)")
    parser.add_argument("--format", choices=["text", "json"], default="text",
                        help="Output format (default: text)")
    parser.add_argument("--demo", action="store_true",
                        help="Show demo output without creating files")
    args = parser.parse_args()

    if args.demo:
        run_demo()
        return

    if not args.task:
        print("Error: --task is required", file=sys.stderr)
        print("Usage: hub_init.py --task 'description' [--agents N] "
              "[--eval 'cmd'] [--metric name] [--direction lower|higher]",
              file=sys.stderr)
        sys.exit(1)

    if not validate_git_repo():
        print("Error: Not a git repository. AgentHub requires git.",
              file=sys.stderr)
        sys.exit(1)

    base_branch = args.base_branch or get_current_branch()
    base_path = ".agenthub"
    session_id = generate_session_id()

    # Create structure
    create_directory_structure(base_path)
    write_gitignore(base_path)
    write_board_index(base_path)

    # Create session
    session_dir = create_session(
        base_path, session_id, args.task, args.agents,
        args.eval_cmd, args.metric, args.direction, base_branch
    )

    if args.format == "json":
        output = {
            "session_id": session_id,
            "session_dir": session_dir,
            "task": args.task,
            "agent_count": args.agents,
            "eval_cmd": args.eval_cmd,
            "metric": args.metric,
            "direction": args.direction,
            "base_branch": base_branch,
            "state": "init",
        }
        print(json.dumps(output, indent=2))
    else:
        print(f"AgentHub session initialized")
        print(f"  Session ID: {session_id}")
        print(f"  Task: {args.task}")
        print(f"  Agents: {args.agents}")
        if args.eval_cmd:
            print(f"  Eval: {args.eval_cmd}")
        if args.metric:
            direction_str = "lower is better" if args.direction == "lower" else "higher is better"
            print(f"  Metric: {args.metric} ({direction_str})")
        print(f"  Base branch: {base_branch}")
        print(f"  State: init")
        print()
        print(f"Next step: Run /hub:spawn to launch {args.agents} agents")


if __name__ == "__main__":
    main()