nick0lay · April 18, 2026 19:47
diff --git a/paw.py b/paw.py
 #!/usr/bin/env python3
 """parallel-agents helper CLI.

 Manages config/state for the parallel-agents plugin and performs the
 mechanical parts of worktree/task orchestration. Intended to be called
 from slash commands; emits JSON on stdout for easy consumption.
 """
 from __future__ import annotations

 import argparse
 import copy
 import datetime as _dt
 import json
 import os
 import re
 import signal
 import subprocess
 import sys
 import time
 import uuid
 from pathlib import Path
 from typing import Any

 DEFAULT_CONFIG: dict[str, Any] = {
    "version": 1,
    "worktree_root": ".claude/worktrees",
    "max_worktrees_per_repo": 4,
    "monitoring": {
        "_comment_mode": (
            "live = after 'task start', the skill also runs 'task wait' in a "
            "Bash background and attaches the Monitor tool, so the assistant "
            "is notified when events arrive. wait = no live watcher; the "
            "assistant polls via ScheduleWakeup or when the user asks."
        ),
        "mode": "live",
        "_comment_event_granularity": (
            "Only applies when mode=live. Controls what 'task wait' emits on "
            "stdout (each line wakes the assistant up via Monitor). "
            "exit = one WATCHING line + one EXITED line per task — quiet, "
            "recommended default, good for 'notify me when it's done'. "
            "tool = also emit one TOOL line per tool-use event inside the "
            "headless session (Edit/Write/Bash/…). Useful for narrating "
            "progress; higher wake-up cost. "
            "all = stream every --include-hook-events JSON line. Very noisy, "
            "debug only."
        ),
        "event_granularity": "exit",
    },
    "repositories": [],
 }

 DEFAULT_STATE: dict[str, Any] = {
    "version": 1,
    "gen_seq": 0,
    "tasks": [],
 }

 CONFIG_FILE = "config.json"
 STATE_FILE = "state.json"
 EXCLUDE_MARKER = "# parallel-agents: managed worktree path"
 EXCLUDE_PATTERN = ".claude/worktrees/"


 def config_path(root: Path) -> Path:
    return root / CONFIG_FILE


 def state_path(root: Path) -> Path:
    return root / STATE_FILE


 def _load_json(path: Path, default: dict[str, Any]) -> dict[str, Any]:
    if not path.exists():
        return copy.deepcopy(default)
    with path.open("r", encoding="utf-8") as f:
        return json.load(f)


 def _save_json(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)
        f.write("\n")
    tmp.replace(path)


 def _migrate_config(cfg: dict[str, Any]) -> bool:
    """Add missing top-level sections so consumers can rely on them being present.

    Returns True if `cfg` was mutated. Call before reading any optional section.
    """
    changed = False
    if "monitoring" not in cfg:
        cfg["monitoring"] = copy.deepcopy(DEFAULT_CONFIG["monitoring"])
        changed = True
    else:
        monitoring = cfg["monitoring"]
        if not isinstance(monitoring, dict):
            cfg["monitoring"] = copy.deepcopy(DEFAULT_CONFIG["monitoring"])
            changed = True
        else:
            if "mode" not in monitoring:
                monitoring["mode"] = DEFAULT_CONFIG["monitoring"]["mode"]
                changed = True
            if "event_granularity" not in monitoring:
                monitoring["event_granularity"] = DEFAULT_CONFIG["monitoring"]["event_granularity"]
                changed = True
    return changed


 def load_config(root: Path) -> dict[str, Any]:
    cfg = _load_json(config_path(root), DEFAULT_CONFIG)
    _migrate_config(cfg)  # in-memory only — ensures defaults without touching the file
    return cfg


 def save_config(root: Path, cfg: dict[str, Any]) -> None:
    _save_json(config_path(root), cfg)


 def load_state(root: Path) -> dict[str, Any]:
    return _load_json(state_path(root), DEFAULT_STATE)


 def save_state(root: Path, st: dict[str, Any]) -> None:
    _save_json(state_path(root), st)


 def discover_sibling_git_repos(start: Path) -> list[Path]:
    """Return sibling directories of `start` that contain a `.git` entry."""
    start = start.resolve()
    parent = start.parent
    if not parent.is_dir():
        return []
    results: list[Path] = []
    for child in sorted(parent.iterdir()):
        if not child.is_dir():
            continue
        if child.resolve() == start:
            continue
        if (child / ".git").exists():
            results.append(child)
    return results


 def _resolve_git_dir(repo_path: Path) -> Path:
    """Resolve the real `.git` directory for a repo or worktree checkout."""
    git_entry = repo_path / ".git"
    if git_entry.is_dir():
        return git_entry
    if git_entry.is_file():
        content = git_entry.read_text(encoding="utf-8")
        for line in content.splitlines():
            if line.startswith("gitdir:"):
                ref = line.split(":", 1)[1].strip()
                return (repo_path / ref).resolve()
    raise FileNotFoundError(f"No .git entry at {repo_path}")


 def add_to_git_exclude(repo_path: Path, pattern: str = EXCLUDE_PATTERN) -> bool:
    """Append `pattern` to `<repo>/.git/info/exclude` if not already present.

    Returns True when the exclude file was modified.
    """
    git_dir = _resolve_git_dir(repo_path)
    info_dir = git_dir / "info"
    info_dir.mkdir(parents=True, exist_ok=True)
    exclude_file = info_dir / "exclude"
    existing = exclude_file.read_text(encoding="utf-8") if exclude_file.exists() else ""
    if pattern in existing.splitlines():
        return False
    new_content = existing
    if new_content and not new_content.endswith("\n"):
        new_content += "\n"
    new_content += f"{EXCLUDE_MARKER}\n{pattern}\n"
    exclude_file.write_text(new_content, encoding="utf-8")
    return True


 _SLUG_RE = re.compile(r"[^a-z0-9]+")
 _PRD_ID_RE = re.compile(r"^PRD-\d+$")
 _GEN_ID_RE = re.compile(r"^GEN-\d+$")

 ACTIVE_STATUSES = {"created", "running", "exited", "failed", "awaiting-merge"}
 DEFAULT_CLAUDE_BIN = os.environ.get("PARALLEL_AGENTS_CLAUDE_BIN", "claude")


 def slugify(text: str, *, max_len: int = 40) -> str:
    s = _SLUG_RE.sub("-", text.strip().lower()).strip("-")
    if len(s) > max_len:
        s = s[:max_len].rstrip("-")
    return s or "task"


 def _now_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")


 def allocate_generic_id(state: dict[str, Any]) -> str:
    """Mutate `state` to bump `gen_seq`, return the next `GEN-NNN` id."""
    state["gen_seq"] = int(state.get("gen_seq", 0)) + 1
    return f"GEN-{state['gen_seq']:03d}"


 def _find_repo(cfg: dict[str, Any], name: str) -> dict[str, Any]:
    for r in cfg["repositories"]:
        if r["name"] == name:
            return r
    raise SystemExit(f"Unknown repository: {name}")


 def _active_tasks_for_repo(state: dict[str, Any], repo_name: str) -> list[dict[str, Any]]:
    return [
        t for t in state["tasks"]
        if t["repo"] == repo_name and t["status"] in ACTIVE_STATUSES
    ]


 def _find_active_task(state: dict[str, Any], task_id: str) -> dict[str, Any] | None:
    for t in state["tasks"]:
        if t["id"] == task_id and t["status"] in ACTIVE_STATUSES:
            return t
    return None


 def _run_git(args: list[str], *, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
    cmd = ["git"]
    if cwd is not None:
        cmd += ["-C", str(cwd)]
    cmd += args
    return subprocess.run(cmd, capture_output=True, text=True)


 def _git_worktree_add(repo_path: Path, worktree_path: Path, branch: str, base_branch: str) -> None:
    worktree_path.parent.mkdir(parents=True, exist_ok=True)
    result = _run_git(
        ["worktree", "add", "-b", branch, str(worktree_path), base_branch],
        cwd=repo_path,
    )
    if result.returncode != 0:
        raise SystemExit(f"git worktree add failed: {result.stderr.strip() or result.stdout.strip()}")


 def _git_worktree_remove(repo_path: Path, worktree_path: Path, *, force: bool = False) -> None:
    args = ["worktree", "remove"]
    if force:
        args.append("--force")
    args.append(str(worktree_path))
    result = _run_git(args, cwd=repo_path)
    if result.returncode != 0:
        raise SystemExit(f"git worktree remove failed: {result.stderr.strip() or result.stdout.strip()}")


 def _git_delete_branch(repo_path: Path, branch: str, *, force: bool = True) -> None:
    flag = "-D" if force else "-d"
    result = _run_git(["branch", flag, branch], cwd=repo_path)
    if result.returncode != 0:
        raise SystemExit(f"git branch delete failed: {result.stderr.strip() or result.stdout.strip()}")


 def detect_default_branch(repo_path: Path) -> str | None:
    """Best-effort detection of the default branch for a registered repo."""
    try:
        result = subprocess.run(
            ["git", "-C", str(repo_path), "symbolic-ref", "--short", "refs/remotes/origin/HEAD"],
            capture_output=True,
            text=True,
            check=False,
        )
        if result.returncode == 0:
            ref = result.stdout.strip()
            if "/" in ref:
                return ref.split("/", 1)[1]
    except FileNotFoundError:
        return None
    return None


 def cmd_init(root: Path, *, discover_from: Path | None = None) -> dict[str, Any]:
    cfg_existed = config_path(root).exists()
    st_existed = state_path(root).exists()
    # Load raw config so we know whether migration mutated it; load_config() would
    # return a migrated copy regardless of whether the on-disk file had the fields.
    cfg = _load_json(config_path(root), DEFAULT_CONFIG)
    mutated = _migrate_config(cfg)
    st = load_state(root)
    if not cfg_existed or mutated:
        save_config(root, cfg)
    if not st_existed:
        save_state(root, st)
    siblings: list[str] = []
    if discover_from is not None:
        siblings = [str(p) for p in discover_sibling_git_repos(discover_from)]
    return {
        "root": str(root),
        "config_created": not cfg_existed,
        "state_created": not st_existed,
        "config": cfg,
        "state": st,
        "siblings": siblings,
    }


 def cmd_repo_add(
    root: Path,
    *,
    name: str,
    path: str,
    default_branch: str = "main",
    branch_prefix_prd: str = "prd/",
    branch_prefix_generic: str = "task/",
 ) -> dict[str, Any]:
    cfg = load_config(root)
    repo_path = Path(path).expanduser().resolve()
    if not repo_path.is_dir():
        raise SystemExit(f"Path does not exist: {repo_path}")
    if not (repo_path / ".git").exists():
        raise SystemExit(f"Not a git repo: {repo_path}")
    for r in cfg["repositories"]:
        if r["name"] == name:
            raise SystemExit(f"Repository '{name}' already registered")
        if Path(r["path"]).resolve() == repo_path:
            raise SystemExit(f"Path already registered under name '{r['name']}'")
    entry = {
        "name": name,
        "path": str(repo_path),
        "default_branch": default_branch,
        "branch_prefix_prd": branch_prefix_prd,
        "branch_prefix_generic": branch_prefix_generic,
    }
    cfg["repositories"].append(entry)
    save_config(root, cfg)
    excluded = add_to_git_exclude(repo_path)
    return {"added": entry, "exclude_updated": excluded}


 def cmd_repo_list(root: Path) -> dict[str, Any]:
    return {"repositories": load_config(root)["repositories"]}


 def cmd_discover(start: Path) -> dict[str, Any]:
    return {"siblings": [str(p) for p in discover_sibling_git_repos(start)]}


 def cmd_task_create(
    root: Path,
    *,
    task_type: str,
    description: str,
    repo_name: str,
    task_id: str | None = None,
    slug: str | None = None,
    blocked_by: list[str] | None = None,
 ) -> dict[str, Any]:
    if task_type not in ("prd", "generic"):
        raise SystemExit(f"Invalid task type: {task_type}")

    cfg = load_config(root)
    state = load_state(root)
    repo = _find_repo(cfg, repo_name)

    cap = int(cfg.get("max_worktrees_per_repo", 4))
    active = _active_tasks_for_repo(state, repo_name)
    if len(active) >= cap:
        raise SystemExit(
            f"Worktree cap reached for repo '{repo_name}' ({len(active)}/{cap}). "
            f"Destroy a task first."
        )

    if task_type == "prd":
        if not task_id:
            raise SystemExit("PRD task requires --task-id")
        if not _PRD_ID_RE.fullmatch(task_id):
            raise SystemExit(f"Invalid PRD id: {task_id} (expected PRD-<digits>)")
    else:
        if task_id is None:
            task_id = allocate_generic_id(state)
        elif not _GEN_ID_RE.fullmatch(task_id):
            raise SystemExit(f"Invalid generic id: {task_id} (expected GEN-<digits>)")

    if _find_active_task(state, task_id) is not None:
        raise SystemExit(f"Active task already exists: {task_id}")

    effective_slug = slug or slugify(description)
    prefix = repo["branch_prefix_prd"] if task_type == "prd" else repo["branch_prefix_generic"]
    branch = f"{prefix}{task_id}-{effective_slug}"

    repo_path = Path(repo["path"])
    worktree_path = repo_path / cfg["worktree_root"] / task_id

    _git_worktree_add(repo_path, worktree_path, branch, repo["default_branch"])

    now = _now_iso()
    task = {
        "id": task_id,
        "type": task_type,
        "description": description,
        "repo": repo_name,
        "branch": branch,
        "worktree_path": str(worktree_path.resolve()),
        "status": "created",
        "session_id": None,
        "session_name": None,
        "log_path": None,
        "initial_prompt": None,
        "queued_prompts": [],
        "blocked_by": list(blocked_by or []),
        "process": None,
        "pr": None,
        "created_at": now,
        "updated_at": now,
    }
    state["tasks"].append(task)
    save_state(root, state)
    return {"task": task}


 def cmd_task_list(root: Path, *, include_history: bool = False) -> dict[str, Any]:
    tasks = load_state(root)["tasks"]
    if not include_history:
        tasks = [t for t in tasks if t["status"] in ACTIVE_STATUSES]
    return {"tasks": tasks}


 def _process_alive(pid: int) -> bool:
    """Return True if a process with `pid` currently exists and is not a zombie.

    Zombies matter because when paw.py spawned the process itself, its exit turns
    it into a zombie in our process table — `kill(pid, 0)` still says "alive"
    until someone waits on it. We waitpid non-blockingly first to reap any such
    zombie we own.
    """
    if pid <= 0:
        return False
    try:
        wpid, _status = os.waitpid(pid, os.WNOHANG)
        if wpid == pid:
            return False
        if wpid == 0:
            return True
    except ChildProcessError:
        pass  # Not our child (or already reaped) — fall through to kill probe.
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


 def _kill_process_tree(pid: int, *, timeout: float = 2.0) -> bool:
    """SIGTERM a pid, wait up to `timeout`, SIGKILL if still alive. Returns True if it exited."""
    if not _process_alive(pid):
        return True
    try:
        os.kill(pid, signal.SIGTERM)
    except ProcessLookupError:
        return True
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if not _process_alive(pid):
            return True
        time.sleep(0.05)
    try:
        os.kill(pid, signal.SIGKILL)
    except ProcessLookupError:
        return True
    time.sleep(0.05)
    return not _process_alive(pid)


 def _tail_file(path: Path, n: int = 20) -> list[str]:
    if not path.exists():
        return []
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return []
    return text.splitlines()[-n:]


 def build_claude_command(
    *,
    prompt: str,
    session_id: str,
    session_name: str | None = None,
    resume: bool = False,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
    dangerously_skip_permissions: bool = True,
    include_hook_events: bool = True,
 ) -> list[str]:
    """Assemble argv for a headless `claude` invocation.

    First-launch shape: `claude -p <prompt> --session-id <id> --name <name> …`
    Resume shape:        `claude --resume <id> -p <prompt> …`
    """
    args: list[str] = [claude_bin]
    if resume:
        args += ["--resume", session_id, "-p", prompt]
    else:
        args += ["-p", prompt, "--session-id", session_id]
        if session_name:
            args += ["--name", session_name]
    if dangerously_skip_permissions:
        args.append("--dangerously-skip-permissions")
    if include_hook_events:
        args.append("--include-hook-events")
    return args


 BROADCAST_TARGET_STATUSES = {"created", "running", "exited"}


 def _spawn_claude_for_task(
    root: Path,
    task: dict[str, Any],
    prompt: str,
    *,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
 ) -> None:
    """Start `claude` in the task's worktree, mutate task state to reflect it."""
    worktree_path = Path(task["worktree_path"])
    if not worktree_path.is_dir():
        raise SystemExit(f"Worktree missing for {task['id']}: {worktree_path}")

    is_resume = bool(task.get("session_id"))
    session_id = task.get("session_id") or str(uuid.uuid4())
    session_name = task.get("session_name") or f"{task['id']}: {task['description']}"
    log_path = _log_path_for(root, task["id"])
    log_path.parent.mkdir(parents=True, exist_ok=True)

    cmd = build_claude_command(
        prompt=prompt,
        session_id=session_id,
        session_name=session_name,
        resume=is_resume,
        claude_bin=claude_bin,
    )

    log_fh = log_path.open("ab")
    try:
        proc = subprocess.Popen(
            cmd,
            cwd=str(worktree_path),
            stdout=log_fh,
            stderr=subprocess.STDOUT,
            stdin=subprocess.DEVNULL,
            start_new_session=True,
        )
    finally:
        log_fh.close()

    now = _now_iso()
    task["session_id"] = session_id
    task["session_name"] = session_name
    task["log_path"] = str(log_path)
    if not task.get("initial_prompt"):
        task["initial_prompt"] = prompt
    task["process"] = {
        "pid": proc.pid,
        "started_at": now,
        "exited_at": None,
        "exit_code": None,
    }
    task["status"] = "running"
    task["updated_at"] = now


 def _drain_queue_if_idle(
    root: Path,
    task: dict[str, Any],
    *,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
 ) -> bool:
    """If the task is idle and has queued prompts, spawn the next one. Returns True if spawned."""
    if task["status"] not in ("created", "exited"):
        return False
    queue = task.get("queued_prompts") or []
    if not queue:
        return False
    next_prompt = queue.pop(0)
    task["queued_prompts"] = queue
    _spawn_claude_for_task(root, task, next_prompt, claude_bin=claude_bin)
    return True


 def _log_path_for(root: Path, task_id: str) -> Path:
    return root / "logs" / f"{task_id}.jsonl"


 def _reap_if_exited(task: dict[str, Any]) -> bool:
    """If the task is marked running but the process is gone, update it. Return True if changed."""
    if task.get("status") != "running":
        return False
    proc = task.get("process") or {}
    pid = proc.get("pid")
    if not isinstance(pid, int):
        return False
    if _process_alive(pid):
        return False
    proc["exited_at"] = _now_iso()
    task["process"] = proc
    task["status"] = "exited"
    task["updated_at"] = _now_iso()
    return True


 def _find_task_any(state: dict[str, Any], task_id: str) -> dict[str, Any] | None:
    for t in state["tasks"]:
        if t["id"] == task_id and t["status"] != "destroyed":
            return t
    return None


 def cmd_task_start(
    root: Path,
    *,
    task_id: str,
    prompt: str,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
    dry_run: bool = False,
 ) -> dict[str, Any]:
    state = load_state(root)
    task = _find_task_any(state, task_id)
    if task is None:
        raise SystemExit(f"No active task: {task_id}")
    # If task was previously running but the process is gone, reconcile first.
    _reap_if_exited(task)
    if task["status"] == "running":
        pid = (task.get("process") or {}).get("pid")
        raise SystemExit(f"Task {task_id} is already running (pid={pid})")

    worktree_path = Path(task["worktree_path"])
    if not worktree_path.is_dir():
        raise SystemExit(f"Worktree missing for {task_id}: {worktree_path}")

    if dry_run:
        is_resume = bool(task.get("session_id"))
        session_id = task.get("session_id") or str(uuid.uuid4())
        session_name = task.get("session_name") or f"{task_id}: {task['description']}"
        cmd = build_claude_command(
            prompt=prompt,
            session_id=session_id,
            session_name=session_name,
            resume=is_resume,
            claude_bin=claude_bin,
        )
        return {
            "dry_run": True,
            "task_id": task_id,
            "session_id": session_id,
            "session_name": session_name,
            "log_path": str(_log_path_for(root, task_id)),
            "cwd": str(worktree_path),
            "command": cmd,
        }

    _spawn_claude_for_task(root, task, prompt, claude_bin=claude_bin)
    save_state(root, state)
    return {"task": task}


 def cmd_task_send(
    root: Path,
    *,
    task_id: str,
    prompt: str,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
 ) -> dict[str, Any]:
    """Enqueue a prompt for a task and drain the queue if the task is idle.

    Semantics:
    - If task is currently `running`: prompt is appended to `queued_prompts`.
    - If task is `created` or `exited`: the new prompt goes to the back of the
      queue and the head of the queue is immediately dispatched via spawn
      (using `--resume` when a session_id already exists).
    """
    state = load_state(root)
    task = _find_task_any(state, task_id)
    if task is None:
        raise SystemExit(f"No active task: {task_id}")
    _reap_if_exited(task)
    task.setdefault("queued_prompts", []).append(prompt)
    dispatched = _drain_queue_if_idle(root, task, claude_bin=claude_bin)
    task["updated_at"] = _now_iso()
    save_state(root, state)
    return {
        "task_id": task_id,
        "dispatched": dispatched,
        "queued_count": len(task.get("queued_prompts") or []),
        "task": task,
    }


 def cmd_task_broadcast(
    root: Path,
    *,
    prompt: str,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
 ) -> dict[str, Any]:
    """Append a prompt to every broadcast-eligible task. Drain idle ones."""
    state = load_state(root)
    results: list[dict[str, Any]] = []
    for task in state["tasks"]:
        if task["status"] not in BROADCAST_TARGET_STATUSES:
            continue
        _reap_if_exited(task)
        task.setdefault("queued_prompts", []).append(prompt)
        dispatched = _drain_queue_if_idle(root, task, claude_bin=claude_bin)
        task["updated_at"] = _now_iso()
        results.append({
            "task_id": task["id"],
            "dispatched": dispatched,
            "queued_count": len(task.get("queued_prompts") or []),
        })
    save_state(root, state)
    return {"total": len(results), "results": results}


 def cmd_task_status(
    root: Path,
    *,
    task_id: str,
    tail_lines: int = 20,
    claude_bin: str = DEFAULT_CLAUDE_BIN,
 ) -> dict[str, Any]:
    state = load_state(root)
    task = _find_task_any(state, task_id)
    if task is None:
        raise SystemExit(f"No task: {task_id}")
    reaped = _reap_if_exited(task)
    drained = _drain_queue_if_idle(root, task, claude_bin=claude_bin)
    if reaped or drained:
        task["updated_at"] = _now_iso()
        save_state(root, state)
    log_path = task.get("log_path")
    tail: list[str] = []
    if log_path:
        tail = _tail_file(Path(log_path), n=tail_lines)
    return {"task": task, "log_tail": tail, "reaped": reaped, "drained": drained}


 def _gh_pr_info(
    repo_path: Path,
    branch: str,
    *,
    gh_bin: str = "gh",
 ) -> dict[str, Any] | None:
    """Look up the most recent PR for `branch`. Returns raw dict or None."""
    try:
        result = subprocess.run(
            [
                gh_bin, "pr", "list",
                "--head", branch,
                "--state", "all",
                "--json", "number,state,mergedAt,url",
                "--limit", "1",
            ],
            cwd=str(repo_path),
            capture_output=True,
            text=True,
        )
    except FileNotFoundError:
        return None
    if result.returncode != 0:
        return None
    try:
        data = json.loads(result.stdout or "[]")
    except json.JSONDecodeError:
        return None
    if not isinstance(data, list) or not data:
        return None
    return data[0]


 def cmd_task_pr_check(
    root: Path,
    *,
    task_id: str,
    gh_bin: str = "gh",
 ) -> dict[str, Any]:
    """Query gh for PR state on this task's branch. Persist into task.pr."""
    cfg = load_config(root)
    state = load_state(root)
    task = _find_task_any(state, task_id)
    if task is None:
        raise SystemExit(f"No task: {task_id}")
    repo = _find_repo(cfg, task["repo"])
    info = _gh_pr_info(Path(repo["path"]), task["branch"], gh_bin=gh_bin)
    if info is not None:
        task["pr"] = {
            "number": info.get("number"),
            "state": info.get("state"),
            "merged_at": info.get("mergedAt"),
            "url": info.get("url"),
        }
        task["updated_at"] = _now_iso()
        save_state(root, state)
    return {
        "task_id": task_id,
        "branch": task["branch"],
        "found": info is not None,
        "pr": task.get("pr"),
    }


 def cmd_task_attach(root: Path, *, task_id: str) -> dict[str, Any]:
    state = load_state(root)
    task = _find_task_any(state, task_id)
    if task is None:
        raise SystemExit(f"No task: {task_id}")
    session_id = task.get("session_id")
    if not session_id:
        raise SystemExit(f"Task {task_id} has no session yet")
    worktree = task["worktree_path"]
    resume_cmd = f'cd "{worktree}" && claude --resume {session_id}'
    return {
        "task_id": task_id,
        "session_id": session_id,
        "session_name": task.get("session_name"),
        "worktree_path": worktree,
        "resume_command": resume_cmd,
    }


 def _extract_tool_use(line: str) -> str | None:
    """Best-effort parser for `--include-hook-events` JSON lines.

    Returns the tool name if the line looks like a tool-use hook event.
    Returns None for log noise, non-JSON, or non-tool events.
    Tolerant of schema variations — we don't know the exact shape at
    write-time, so we check several common field names.
    """
    line = line.strip()
    if not line or not line.startswith("{"):
        return None
    try:
        data = json.loads(line)
    except (json.JSONDecodeError, ValueError):
        return None
    if not isinstance(data, dict):
        return None
    event = str(
        data.get("hook_event_name")
        or data.get("event")
        or data.get("type")
        or ""
    )
    tool = data.get("tool_name") or data.get("tool")
    if not tool:
        tool_input = data.get("tool_input")
        if isinstance(tool_input, dict):
            tool = tool_input.get("name")
    if not tool:
        payload = data.get("payload") or data.get("data")
        if isinstance(payload, dict):
            tool = payload.get("tool_name") or payload.get("tool")
    if not tool:
        return None
    event_lower = event.lower()
    if "tooluse" in event_lower or "tool_use" in event_lower or event == "":
        return str(tool)
    return None


 def _emit(line: str) -> None:
    """Line-buffered stdout write, flushed immediately so Monitor sees it."""
    sys.stdout.write(line + "\n")
    sys.stdout.flush()


 def cmd_task_wait(
    root: Path,
    *,
    task_id: str,
    granularity: str = "exit",
    timeout: float | None = None,
    poll_interval: float = 0.5,
 ) -> int:
    """Block until the task's headless process exits, emitting events to stdout.

    Designed to be launched via `Bash(run_in_background=true)` and attached to
    the `Monitor` tool — each emitted line becomes a notification.

    Granularities:
    - `exit`: one WATCHING line up front, one EXITED line when the process dies.
      Quietest; one wake-up per task.
    - `tool`: additionally emit a TOOL line for each parsed tool-use event in
      the log stream. Higher wake-up volume; narrates progress.
    - `all`: emit every line from the log as a LOG event. Very noisy; debug only.
    """
    if granularity not in ("exit", "tool", "all"):
        _emit(f"ERROR task={task_id} reason=invalid_granularity={granularity}")
        return 2

    state = load_state(root)
    task = _find_task_any(state, task_id)
    if task is None:
        _emit(f"ERROR task={task_id} reason=not_found")
        return 3
    proc = task.get("process") or {}
    pid = proc.get("pid")
    if not isinstance(pid, int):
        _emit(f"ERROR task={task_id} reason=no_process")
        return 4

    log_path_str = task.get("log_path")
    log_fh = None
    if granularity in ("tool", "all") and log_path_str:
        log_path = Path(log_path_str)
        if log_path.exists():
            try:
                log_fh = log_path.open("r", encoding="utf-8", errors="replace")
                # Seek to start so we replay everything the child has written
                # so far, then follow forward.
                log_fh.seek(0, 0)
            except OSError:
                log_fh = None

    _emit(f"WATCHING task={task_id} pid={pid} granularity={granularity}")

    deadline = (time.monotonic() + timeout) if timeout else None
    try:
        while True:
            if log_fh is not None:
                while True:
                    line = log_fh.readline()
                    if not line:
                        break
                    stripped = line.rstrip("\n")
                    if granularity == "all":
                        _emit(f"LOG task={task_id}: {stripped}")
                    elif granularity == "tool":
                        tool_name = _extract_tool_use(stripped)
                        if tool_name:
                            _emit(f"TOOL task={task_id} tool={tool_name}")

            if not _process_alive(pid):
                # Drain any last-written bytes now that the process is gone.
                if log_fh is not None:
                    while True:
                        line = log_fh.readline()
                        if not line:
                            break
                        stripped = line.rstrip("\n")
                        if granularity == "all":
                            _emit(f"LOG task={task_id}: {stripped}")
                        elif granularity == "tool":
                            tool_name = _extract_tool_use(stripped)
                            if tool_name:
                                _emit(f"TOOL task={task_id} tool={tool_name}")
                _emit(f"EXITED task={task_id}")
                return 0

            if deadline is not None and time.monotonic() > deadline:
                _emit(f"TIMEOUT task={task_id}")
                return 5

            time.sleep(poll_interval)
    finally:
        if log_fh is not None:
            log_fh.close()


 def cmd_task_cleanup_merged(
    root: Path,
    *,
    gh_bin: str = "gh",
    dry_run: bool = False,
 ) -> dict[str, Any]:
    """Refresh PR state for every active PRD task and destroy those whose PR is MERGED.

    Generic tasks are never touched (they have no PR concept).
    Tasks where gh returns no data or a non-merged state are left alone.
    """
    state = load_state(root)
    # Snapshot active PRD task ids so destroy operations don't invalidate iteration.
    candidate_ids = [
        t["id"]
        for t in state["tasks"]
        if t["type"] == "prd" and t["status"] in ACTIVE_STATUSES
    ]
    results: list[dict[str, Any]] = []

    for tid in candidate_ids:
        try:
            pr_out = cmd_task_pr_check(root, task_id=tid, gh_bin=gh_bin)
        except SystemExit as e:
            results.append({"task_id": tid, "action": "error", "error": str(e)})
            continue
        pr = pr_out.get("pr")
        pr_state = str((pr or {}).get("state", "")).upper()
        if pr_state != "MERGED":
            results.append({
                "task_id": tid,
                "action": "skipped",
                "reason": f"pr_state={pr_state or 'none'}",
                "pr": pr,
            })
            continue
        if dry_run:
            results.append({"task_id": tid, "action": "would_destroy", "pr": pr})
            continue
        try:
            destroy_out = cmd_task_destroy(
                root, task_id=tid, delete_branch=True, force=False,
            )
            results.append({
                "task_id": tid,
                "action": "destroyed",
                "pr": pr,
                "detail": destroy_out,
            })
        except SystemExit as e:
            results.append({
                "task_id": tid,
                "action": "error",
                "error": str(e),
                "pr": pr,
            })

    destroyed = sum(1 for r in results if r["action"] == "destroyed")
    return {"results": results, "scanned": len(candidate_ids), "destroyed": destroyed}


 def cmd_task_destroy(
    root: Path,
    *,
    task_id: str,
    delete_branch: bool = False,
    force: bool = False,
 ) -> dict[str, Any]:
    cfg = load_config(root)
    state = load_state(root)
    task = _find_active_task(state, task_id)
    if task is None:
        raise SystemExit(f"No active task: {task_id}")

    repo = _find_repo(cfg, task["repo"])
    repo_path = Path(repo["path"])
    worktree_path = Path(task["worktree_path"])
    branch = task["branch"]

    process_killed = False
    proc = task.get("process") or {}
    pid = proc.get("pid") if isinstance(proc, dict) else None
    if isinstance(pid, int) and _process_alive(pid):
        process_killed = _kill_process_tree(pid)

    worktree_removed = False
    if worktree_path.exists():
        _git_worktree_remove(repo_path, worktree_path, force=force)
        worktree_removed = True

    branch_deleted = False
    branch_delete_error: str | None = None
    if delete_branch:
        try:
            _git_delete_branch(repo_path, branch, force=True)
            branch_deleted = True
        except SystemExit as e:
            branch_delete_error = str(e)

    task["status"] = "destroyed"
    task["updated_at"] = _now_iso()
    save_state(root, state)
    return {
        "destroyed": task_id,
        "worktree_removed": worktree_removed,
        "branch_deleted": branch_deleted,
        "branch_delete_error": branch_delete_error,
        "process_killed": process_killed,
    }


 def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="paw", description="parallel-agents helper")
    p.add_argument(
        "--root",
        type=Path,
        default=Path.cwd() / ".claude" / "parallel-agents",
        help="Plugin data root (holds config.json and state.json)",
    )
    sub = p.add_subparsers(dest="command", required=True)

    sp_init = sub.add_parser("init", help="Create config/state if missing")
    sp_init.add_argument("--discover-from", type=Path, default=None)

    sp_discover = sub.add_parser("discover", help="List sibling git repos of a path")
    sp_discover.add_argument("--from", dest="from_", type=Path, default=Path.cwd())

    sp_repo = sub.add_parser("repo", help="Repository management")
    sp_repo_sub = sp_repo.add_subparsers(dest="repo_command", required=True)

    sp_repo_add = sp_repo_sub.add_parser("add", help="Register a repository")
    sp_repo_add.add_argument("--name", required=True)
    sp_repo_add.add_argument("--path", required=True)
    sp_repo_add.add_argument("--default-branch", default="main")
    sp_repo_add.add_argument("--branch-prefix-prd", default="prd/")
    sp_repo_add.add_argument("--branch-prefix-generic", default="task/")

    sp_repo_sub.add_parser("list", help="List registered repositories")

    sp_task = sub.add_parser("task", help="Task/worktree management")
    sp_task_sub = sp_task.add_subparsers(dest="task_command", required=True)

    sp_task_create = sp_task_sub.add_parser("create", help="Create a worktree + task entry")
    sp_task_create.add_argument("--type", dest="task_type", choices=["prd", "generic"], required=True)
    sp_task_create.add_argument("--repo", dest="repo_name", required=True)
    sp_task_create.add_argument("--description", required=True)
    sp_task_create.add_argument("--task-id", default=None, help="Required for PRD; auto-allocated for generic")
    sp_task_create.add_argument("--slug", default=None, help="Override the derived branch slug")
    sp_task_create.add_argument(
        "--blocked-by",
        default="",
        help="Comma-separated task ids that block this one (informational, no enforcement)",
    )

    sp_task_list = sp_task_sub.add_parser("list", help="List tasks")
    sp_task_list.add_argument("--include-history", action="store_true")

    sp_task_destroy = sp_task_sub.add_parser("destroy", help="Remove worktree, mark task destroyed")
    sp_task_destroy.add_argument("--task-id", required=True)
    sp_task_destroy.add_argument("--delete-branch", action="store_true")
    sp_task_destroy.add_argument("--force", action="store_true", help="Pass --force to git worktree remove")

    sp_task_start = sp_task_sub.add_parser("start", help="Launch a headless claude session in the worktree")
    sp_task_start.add_argument("--task-id", required=True)
    sp_task_start.add_argument("--prompt", required=True)
    sp_task_start.add_argument("--claude-bin", default=DEFAULT_CLAUDE_BIN)
    sp_task_start.add_argument("--dry-run", action="store_true")

    sp_task_status = sp_task_sub.add_parser("status", help="Inspect one task (reconciles running→exited)")
    sp_task_status.add_argument("--task-id", required=True)
    sp_task_status.add_argument("--tail", type=int, default=20, dest="tail_lines")

    sp_task_attach = sp_task_sub.add_parser("attach", help="Print the resume command for a task")
    sp_task_attach.add_argument("--task-id", required=True)

    sp_task_send = sp_task_sub.add_parser("send", help="Queue a prompt for a task (drains if idle)")
    sp_task_send.add_argument("--task-id", required=True)
    sp_task_send.add_argument("--prompt", required=True)
    sp_task_send.add_argument("--claude-bin", default=DEFAULT_CLAUDE_BIN)

    sp_task_broadcast = sp_task_sub.add_parser("broadcast", help="Append a prompt to every active task")
    sp_task_broadcast.add_argument("--prompt", required=True)
    sp_task_broadcast.add_argument("--claude-bin", default=DEFAULT_CLAUDE_BIN)

    sp_task_pr = sp_task_sub.add_parser("pr-check", help="Look up PR state via gh")
    sp_task_pr.add_argument("--task-id", required=True)
    sp_task_pr.add_argument("--gh-bin", default="gh")

    sp_task_cleanup = sp_task_sub.add_parser(
        "cleanup",
        help="Destroy all active PRD tasks whose PR is MERGED (refreshes PR state first).",
    )
    sp_task_cleanup.add_argument("--gh-bin", default="gh")
    sp_task_cleanup.add_argument("--dry-run", action="store_true")

    sp_task_wait = sp_task_sub.add_parser(
        "wait",
        help="Block until the task exits; streams events for live monitoring.",
    )
    sp_task_wait.add_argument("--task-id", required=True)
    sp_task_wait.add_argument(
        "--granularity",
        choices=["exit", "tool", "all"],
        default="exit",
        help="exit = one EXITED line only; tool = also one TOOL line per tool use; all = every log line",
    )
    sp_task_wait.add_argument("--timeout", type=float, default=None)
    sp_task_wait.add_argument("--poll-interval", type=float, default=0.5)

    return p


 def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    root: Path = args.root
    if args.command == "init":
        out = cmd_init(root, discover_from=args.discover_from)
    elif args.command == "discover":
        out = cmd_discover(args.from_)
    elif args.command == "repo" and args.repo_command == "add":
        out = cmd_repo_add(
            root,
            name=args.name,
            path=args.path,
            default_branch=args.default_branch,
            branch_prefix_prd=args.branch_prefix_prd,
            branch_prefix_generic=args.branch_prefix_generic,
        )
    elif args.command == "repo" and args.repo_command == "list":
        out = cmd_repo_list(root)
    elif args.command == "task" and args.task_command == "create":
        blocked_by_list = [s.strip() for s in (args.blocked_by or "").split(",") if s.strip()]
        out = cmd_task_create(
            root,
            task_type=args.task_type,
            repo_name=args.repo_name,
            description=args.description,
            task_id=args.task_id,
            slug=args.slug,
            blocked_by=blocked_by_list,
        )
    elif args.command == "task" and args.task_command == "list":
        out = cmd_task_list(root, include_history=args.include_history)
    elif args.command == "task" and args.task_command == "destroy":
        out = cmd_task_destroy(
            root,
            task_id=args.task_id,
            delete_branch=args.delete_branch,
            force=args.force,
        )
    elif args.command == "task" and args.task_command == "start":
        out = cmd_task_start(
            root,
            task_id=args.task_id,
            prompt=args.prompt,
            claude_bin=args.claude_bin,
            dry_run=args.dry_run,
        )
    elif args.command == "task" and args.task_command == "status":
        out = cmd_task_status(
            root,
            task_id=args.task_id,
            tail_lines=args.tail_lines,
        )
    elif args.command == "task" and args.task_command == "attach":
        out = cmd_task_attach(root, task_id=args.task_id)
    elif args.command == "task" and args.task_command == "send":
        out = cmd_task_send(
            root,
            task_id=args.task_id,
            prompt=args.prompt,
            claude_bin=args.claude_bin,
        )
    elif args.command == "task" and args.task_command == "broadcast":
        out = cmd_task_broadcast(
            root,
            prompt=args.prompt,
            claude_bin=args.claude_bin,
        )
    elif args.command == "task" and args.task_command == "pr-check":
        out = cmd_task_pr_check(
            root,
            task_id=args.task_id,
            gh_bin=args.gh_bin,
        )
    elif args.command == "task" and args.task_command == "cleanup":
        out = cmd_task_cleanup_merged(
            root,
            gh_bin=args.gh_bin,
            dry_run=args.dry_run,
        )
    elif args.command == "task" and args.task_command == "wait":
        # Streams plain-text event lines to stdout instead of a JSON blob,
        # so the Monitor tool can read line-by-line notifications.
        return cmd_task_wait(
            root,
            task_id=args.task_id,
            granularity=args.granularity,
            timeout=args.timeout,
            poll_interval=args.poll_interval,
        )
    else:
        print(f"Unknown command: {args.command}", file=sys.stderr)
        return 2
    json.dump(out, sys.stdout, indent=2)
    sys.stdout.write("\n")
    return 0


 if __name__ == "__main__":
    raise SystemExit(main())
diff --git a/SKILL.md b/SKILL.md
name	parallel-agents
description	Orchestrate parallel headless Claude Code sessions in isolated git worktrees. Each task (a PRD implementation OR a free-form generic prompt) runs in its own worktree with its own session. Activate when the user wants to run one or more tasks in parallel ("implement PRD-010 and PRD-012 in parallel", "start PRD-010 in a worktree", "run this prompt in the background", "investigate the flaky calendar test in a worktree", "try this refactor in isolation"), check on parallel tasks they already started ("how's PRD-010 going", "any updates", "what's running"), send a follow-up prompt to a running worktree ("tell PRD-010 to also update the CHANGELOG", "broadcast — run the linter in every worktree"), jump into a running session ("take me into PRD-010", "open PRD-010 in VSCode"), or clean up worktrees after PRs have merged ("close all completed tasks", "destroy PRD-010"). Also activates on the first mention of parallel-agents in a project that hasn't been configured yet — runs the init sub-flow inline to register repositories before proceeding.
argument-hint	start <PRD-id[s]\|prompt> \| status [id] \| send <id\|all> <prompt> \| attach <id> \| destroy <id> \| cleanup
Event line	When	What to do
`WATCHING task=<id> pid=<N> granularity=<mode>`	Fires once on attach	Acknowledge in summary; no action needed
`TOOL task=<id> tool=<name>`	Only if granularity=`tool`, per parsed tool-use event in the child's log	Optionally narrate ("GEN-001 is editing now")
`LOG task=<id>: <raw line>`	Only if granularity=`all`, per raw log line	Debug-only; don't surface to user unless asked
`EXITED task=<id>`	Child process died	Run `task status --task-id <id>`; for PRD tasks run `task pr-check`; update the TaskCreate todo
`TIMEOUT task=<id>`	`--timeout` deadline hit (rare — we don't set one)	Treat as a tool-side error; run manual status
`ERROR task=<id> reason=<…>`	Watcher startup failure	Log it; fall back to manual `task status` checks for this task