subscriber.py

#!/usr/bin/env python3
"""Conductor subscriber — listens to Redis and spawns agents.

Subscribes to Redis channels, and when an issue with a conductor label
is opened (or reopened, or triggered via /run comment), spawns `claude -p`
in the target repo's directory.

Usage:
    python3 subscriber.py                             # listen and spawn
    python3 subscriber.py --dry-run                   # log only, don't spawn
    python3 subscriber.py --channel conductor:issues   # issues only

Redis channels:
    conductor:issues    — issue opened/reopened/labeled/closed events
    conductor:comments  — issue comment events (created/edited)
    conductor:events    — all events (firehose)

Comment triggers:
    /run      — spawn an agent for this issue (re-trigger missed or stalled work)
    /run with additional context after the command is passed to the agent
"""

import argparse
import json
import logging
import os
import shutil
import subprocess
import sys
import threading
import time
from datetime import datetime, timezone
from pathlib import Path

import re

import redis

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("subscriber")

# Track active agents to prevent duplicates
active_agents: dict[str, dict] = {}  # key: "repo#number" → agent info
agent_lock = threading.Lock()

# Where agent logs go
LOG_DIR = Path("/home/tim/Projects/loqu8/conductor/logs")

# Where agent worktrees are created (isolated git checkouts)
WORKTREE_BASE = Path("/tmp/conductor-worktrees")

# Backend definitions — what can actually run
BACKENDS = {
    "claude": {"command": "claude", "description": "Claude Code (default)"},
    "claude-opus": {"command": "claude", "description": "Claude Code (Opus)"},
    "cursor": {"command": None, "description": "Cursor (manual)"},
    "marketing": {"command": "claude", "description": "Claude Code (marketing)"},
}

# Label/task-type → fallback chain mapping
# Each chain is tried in order; if a backend fails, the next is attempted.
# "task:*" keys are internal routing hints (never posted as GitHub labels).
TASK_ROUTING = {
    "agent:opus":      {"chain": ["claude-opus", "claude"], "description": "Complex tasks"},
    "agent:cursor":    {"chain": ["cursor", "claude"], "description": "Cursor-preferred"},
    "agent:marketing": {"chain": ["marketing", "claude"], "description": "Marketing copy"},
    "conductor":       {"chain": ["claude"], "description": "Default"},
    "task:feature":    {"chain": ["claude"], "description": "Well-specified features"},
    "task:test":       {"chain": ["claude"], "description": "Test writing"},
    "task:review":     {"chain": ["claude"], "description": "Code review"},
}

MAX_RETRIES = 2
AGENT_TIMEOUT_SECONDS = 30 * 60  # 30 minutes — kill agents that exceed this


def agent_key(repo: str, number: int) -> str:
    return f"{repo}#{number}"


def _try_claim(repo: str, number: int) -> bool:
    """Atomically check for duplicates and claim the slot.

    Returns True if this caller claimed the slot (proceed with spawn).
    Returns False if another agent already holds it (skip).
    """
    key = agent_key(repo, number)
    with agent_lock:
        if key in active_agents:
            info = active_agents[key]
            proc = info.get("process")
            # Still running or still spawning → duplicate
            if info.get("status") == "spawning" or (proc and proc.poll() is None):
                log.info("  Agent already running for %s (pid %s)", key, info.get("pid", "spawning"))
                return False
            # Process finished, clean up stale entry
            del active_agents[key]
        # Claim the slot before releasing the lock
        active_agents[key] = {
            "status": "spawning",
            "repo": repo,
            "issue_number": number,
            "started": datetime.now(timezone.utc).isoformat(),
        }
    return True


def _release_claim(repo: str, number: int):
    """Release a claimed slot if spawn fails."""
    key = agent_key(repo, number)
    with agent_lock:
        info = active_agents.get(key)
        if info and info.get("status") == "spawning":
            del active_agents[key]


def _slugify(title: str, max_len: int = 40) -> str:
    """Convert an issue title to a branch-safe slug."""
    slug = title.lower()
    slug = re.sub(r"[^a-z0-9]+", "-", slug)
    slug = slug.strip("-")
    if len(slug) > max_len:
        slug = slug[:max_len].rstrip("-")
    return slug or "task"


def _setup_agent_workspace(
    repo_dir: str, number: int, title: str, base_branch: str, branch_strategy: str
) -> tuple[str | None, str | None, str | None]:
    """Create an isolated workspace for the agent using git worktrees.

    For feature-branch strategy: creates a git worktree with a feature branch.
    For direct-commit: works in the main checkout (no isolation).

    Returns (work_dir, branch_name, worktree_path) or (None, None, None) on failure.
    worktree_path is set when a worktree was created (for cleanup later).
    """
    if branch_strategy != "feature-branch":
        return repo_dir, None, None

    slug = _slugify(title)
    branch = f"feature/issue-{number}-{slug}"
    safe_key = f"{repo_dir.replace('/', '-').lstrip('-')}-{number}"
    worktree_path = WORKTREE_BASE / safe_key

    try:
        # Clean up any stale worktree at this path (e.g., from crashed agent)
        if worktree_path.exists():
            subprocess.run(
                ["git", "worktree", "remove", "--force", str(worktree_path)],
                cwd=repo_dir, capture_output=True, timeout=15,
            )
            # If git worktree remove failed, force-remove the directory
            if worktree_path.exists():
                shutil.rmtree(worktree_path, ignore_errors=True)

        # Fetch latest from remote
        subprocess.run(
            ["git", "fetch", "origin"],
            cwd=repo_dir, capture_output=True, timeout=30,
        )

        # Determine base ref
        base_ref = f"origin/{base_branch}"
        result = subprocess.run(
            ["git", "rev-parse", "--verify", base_ref],
            cwd=repo_dir, capture_output=True, text=True, timeout=10,
        )
        if result.returncode != 0:
            base_ref = base_branch

        # Ensure worktree base directory exists
        WORKTREE_BASE.mkdir(parents=True, exist_ok=True)

        # Check if branch already exists (e.g., /run re-trigger)
        result = subprocess.run(
            ["git", "rev-parse", "--verify", branch],
            cwd=repo_dir, capture_output=True, text=True, timeout=10,
        )
        if result.returncode == 0:
            # Branch exists — create worktree on existing branch
            result = subprocess.run(
                ["git", "worktree", "add", str(worktree_path), branch],
                cwd=repo_dir, capture_output=True, text=True, timeout=15,
            )
            log.info("  Worktree (existing branch): %s → %s", branch, worktree_path)
        else:
            # Create new branch in worktree
            result = subprocess.run(
                ["git", "worktree", "add", "-b", branch, str(worktree_path), base_ref],
                cwd=repo_dir, capture_output=True, text=True, timeout=15,
            )
            log.info("  Worktree (new branch): %s from %s → %s", branch, base_ref, worktree_path)

        if result.returncode != 0:
            log.error("  Failed to create worktree: %s", result.stderr.strip())
            return None, None, None

        return str(worktree_path), branch, str(worktree_path)

    except Exception as e:
        log.error("  Failed to create agent workspace: %s", e)
        return None, None, None


def _cleanup_worktree(repo_dir: str, worktree_path: str | None):
    """Remove a git worktree. Best-effort."""
    if not worktree_path:
        return
    try:
        # Prune any stale worktree references first
        subprocess.run(
            ["git", "worktree", "prune"],
            cwd=repo_dir, capture_output=True, timeout=15,
        )
        # Remove the worktree
        result = subprocess.run(
            ["git", "worktree", "remove", "--force", str(worktree_path)],
            cwd=repo_dir, capture_output=True, text=True, timeout=15,
        )
        if result.returncode == 0:
            log.info("  Cleaned up worktree: %s", worktree_path)
        else:
            # Fallback: force-remove directory
            if os.path.isdir(worktree_path):
                shutil.rmtree(worktree_path, ignore_errors=True)
                log.info("  Force-removed worktree dir: %s", worktree_path)
    except Exception as e:
        log.warning("  Failed to clean up worktree %s: %s", worktree_path, e)


WORKTREE_STALE_HOURS = 2  # worktrees older than this are considered stale


def _cleanup_stale_worktrees():
    """Remove worktrees older than WORKTREE_STALE_HOURS at startup.

    Selective cleanup: only removes worktrees that are likely from crashed agents,
    not ones that might belong to currently running agents on another subscriber instance.
    """
    if not WORKTREE_BASE.exists():
        return
    cutoff = time.time() - (WORKTREE_STALE_HOURS * 3600)
    removed = 0
    for entry in WORKTREE_BASE.iterdir():
        if not entry.is_dir():
            continue
        try:
            mtime = entry.stat().st_mtime
            if mtime < cutoff:
                shutil.rmtree(entry, ignore_errors=True)
                removed += 1
                log.info("  Removed stale worktree: %s (age: %.1fh)",
                         entry.name, (time.time() - mtime) / 3600)
        except OSError:
            pass
    if removed:
        log.info("Cleaned up %d stale worktree(s)", removed)


def _resolve_route(labels: list[str]) -> tuple[str, dict]:
    """Resolve labels to a route key and TASK_ROUTING entry.

    Returns (route_key, route_entry). Scans labels for the first match
    in TASK_ROUTING; falls back to 'conductor'.
    """
    for label in labels:
        if label in TASK_ROUTING:
            return label, TASK_ROUTING[label]
    return "conductor", TASK_ROUTING["conductor"]


def spawn_agent(event: dict, dry_run: bool = False, chain_index: int = 0):
    """Spawn a claude agent for a conductor issue.

    Args:
        event: Issue event dict with repo, issue_number, labels, etc.
        dry_run: Log only, don't actually spawn.
        chain_index: Position in the fallback chain to start from (0 = first backend).
    """
    repo = event["repo"]
    number = event["issue_number"]
    title = event["issue_title"]
    repo_dir = event["repo_dir"]
    labels = event.get("labels", [])
    issue_url = event.get("issue_url", "")

    if not repo_dir:
        log.error("  No directory mapping for %s — cannot spawn", repo)
        return

    if not os.path.isdir(repo_dir):
        log.error("  Directory %s does not exist — cannot spawn", repo_dir)
        return

    if not _try_claim(repo, number):
        return

    # Resolve route from labels → fallback chain
    route_key, route_entry = _resolve_route(labels)
    chain = route_entry["chain"]

    # Skip backends that can't spawn (command=None), advancing through chain
    backend_key = None
    backend = None
    while chain_index < len(chain):
        candidate = chain[chain_index]
        if BACKENDS[candidate]["command"] is not None:
            backend_key = candidate
            backend = BACKENDS[candidate]
            break
        log.info("  Skipping %s (no command) — trying next in chain", BACKENDS[candidate]["description"])
        chain_index += 1

    if backend is None:
        log.info("  No spawnable backend in chain for %s — logging only", route_entry["description"])
        return

    # Workspace setup — isolated worktree for feature branches
    repo_config = event.get("repo_config", {})
    branch_strategy = repo_config.get("branch_strategy", "feature-branch")
    base_branch = repo_config.get("base_branch", "develop")

    # Allow `direct-commit` label to override branch strategy for simple tasks
    if "direct-commit" in labels:
        branch_strategy = "direct-commit"
        log.info("  direct-commit label — working on current branch (no worktree)")

    work_dir, branch_name, worktree_path = _setup_agent_workspace(
        repo_dir, number, title, base_branch, branch_strategy,
    )
    if work_dir is None:
        log.warning("  Failed to create workspace — agent will work in main checkout")
        work_dir = repo_dir

    branch_instructions = ""
    if branch_name:
        branch_instructions = (
            f"\n\n## Git Workflow\n"
            f"You are on feature branch `{branch_name}` (based on `{base_branch}`).\n"
            f"Do NOT push directly to `{base_branch}`, `master`, or `main`.\n"
            f"When your work is complete:\n"
            f"1. Commit and push your branch: `git push -u origin {branch_name}`\n"
            f"2. Open a PR: `gh pr create --base {base_branch} --head {branch_name} "
            f"--title \"<title>\" --body \"Closes #{number}\\n\\n<summary>\"`\n"
            f"3. Call complete() with a link to the PR"
        )

    # Build the lean prompt — protocol details come from conductor MCP instructions
    extra_context = event.get("extra_context", "")
    prompt = (
        f"You are a conductor agent assigned to {repo}#{number}: {title}\n"
        f"URL: {issue_url}\n\n"
        f"You have the conductor MCP connected. Call my_assignment() to get full context,\n"
        f"then follow the protocol in the conductor instructions.\n"
        f"Start by calling acknowledge().\n\n"
        f"If conductor MCP tools fail (stale session), fall back to gh CLI:\n"
        f"  Comment: gh issue comment {number} --repo {repo} --body \"<message>\"\n"
        f"  Close:   gh issue close {number} --repo {repo}\n"
        f"  Re-trigger: comment `/run` on the issue to re-spawn an agent"
        f"{branch_instructions}"
    )
    if extra_context:
        prompt += f"\n\nAdditional context:\n{extra_context}"

    # Build command — use claudea for full permissions via Anthropic API
    cmd = [
        "/home/tim/bin/claudea",
        "-p", prompt,
        "--output-format", "json",
    ]

    key = agent_key(repo, number)
    log.info("  SPAWNING: %s in %s (chain: %s, index: %d)",
             backend["description"], work_dir, route_key, chain_index)
    if worktree_path:
        log.info("  Worktree: %s", worktree_path)
    log.info("  Issue: %s — %s", key, title)

    if dry_run:
        log.info("  [DRY RUN] Would run: claude -p '...' in %s", work_dir)
        return

    # Ensure log directory exists
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    safe_key = key.replace("/", "-").replace("#", "-")
    log_file = LOG_DIR / f"{timestamp}-{safe_key}.log"

    try:
        with open(log_file, "w") as f:
            proc = subprocess.Popen(
                cmd,
                cwd=work_dir,
                stdout=f,
                stderr=subprocess.STDOUT,
                env={
                    **os.environ,
                    "CLAUDE_CODE_ENTRYPOINT": "conductor",
                    "CONDUCTOR_AGENT_REPO": repo,
                    "CONDUCTOR_AGENT_ISSUE": str(number),
                },
            )

        # Upgrade the "spawning" placeholder to full agent info
        with agent_lock:
            active_agents[key] = {
                "process": proc,
                "pid": proc.pid,
                "repo": repo,
                "issue_number": number,
                "title": title,
                "started": datetime.now(timezone.utc).isoformat(),
                "log_file": str(log_file),
                "route_key": route_key,
                "chain_index": chain_index,
                "backend_key": backend_key,
                "event": event,
                "repo_dir": repo_dir,
                "worktree_path": worktree_path,
            }

        log.info("  Started agent pid %d — log: %s", proc.pid, log_file)

        # Track agent spawn
        _track_event("agent_started", repo=repo, issue_number=number,
                      title=title, pid=proc.pid, backend=backend_key,
                      route=route_key)

        # Comment on the issue that an agent has been dispatched
        branch_note = f" on `{branch_name}`" if branch_name else ""
        worktree_note = " (worktree)" if worktree_path else ""
        try:
            subprocess.run(
                ["gh", "issue", "comment", str(number),
                 "--repo", repo,
                 "--body", f"📡 Conductor dispatched agent (pid {proc.pid}, {backend['description']}) in `{repo_dir}`{branch_note}{worktree_note}"],
                capture_output=True, text=True, timeout=15,
            )
        except Exception:
            pass  # Best effort

        # Publish agent started event
        try:
            r = redis.Redis()
            r.publish("conductor:agent_started", json.dumps({
                "key": key,
                "repo": repo,
                "issue_number": number,
                "title": title,
                "pid": proc.pid,
                "repo_dir": repo_dir,
                "log_file": str(log_file),
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }))
        except Exception:
            pass

        # Start a thread to monitor completion
        t = threading.Thread(
            target=_monitor_agent, args=(key, proc, log_file, repo, number),
            daemon=True,
        )
        t.start()

    except Exception as e:
        _release_claim(repo, number)
        log.error("  Failed to spawn agent: %s", e)


def _record_attempt(repo: str, number: int, backend_key: str, success: bool, exit_code: int, elapsed: str):
    """Record attempt in Redis: per-issue history + per-backend stats."""
    try:
        r = redis.Redis()
        # Per-issue retry history (7-day TTL)
        history_key = f"conductor:retries:{repo}#{number}"
        entry = json.dumps({
            "backend": backend_key,
            "success": success,
            "exit_code": exit_code,
            "elapsed": elapsed,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        })
        r.rpush(history_key, entry)
        r.expire(history_key, 7 * 86400)

        # Per-backend aggregate stats
        stat_field = f"{backend_key}:{'successes' if success else 'failures'}"
        r.hincrby("conductor:backend_stats", stat_field, 1)
    except Exception as e:
        log.warning("Failed to record attempt stats: %s", e)


def _post_retry_comment(repo: str, number: int, tried_desc: str, exit_code: int, next_desc: str, attempt: int):
    """Post a retry comment on the GitHub issue."""
    body = f"🔄 Retry {attempt}/{MAX_RETRIES} — {tried_desc} failed (exit {exit_code}), trying {next_desc}"
    try:
        subprocess.run(
            ["gh", "issue", "comment", str(number), "--repo", repo, "--body", body],
            capture_output=True, text=True, timeout=15,
        )
    except Exception:
        pass


def _post_exhausted_comment(repo: str, number: int):
    """Post a comment that all backends in the chain have been exhausted."""
    body = "❌ All backends exhausted — no more retries available. Comment `/run` to try again manually."
    try:
        subprocess.run(
            ["gh", "issue", "comment", str(number), "--repo", repo, "--body", body],
            capture_output=True, text=True, timeout=15,
        )
    except Exception:
        pass


def _extract_cost_info(log_file: Path) -> dict:
    """Try to extract cost/token info from the agent's JSON output log.

    Claude Code with --output-format json writes a final JSON summary.
    We look for it in the last few lines of the log file.
    """
    try:
        with open(log_file) as f:
            # Read last 50 lines to find the JSON summary
            lines = f.readlines()
            for line in reversed(lines[-50:]):
                line = line.strip()
                if not line.startswith("{"):
                    continue
                try:
                    data = json.loads(line)
                    result = {}
                    # Claude Code JSON output includes cost_usd and token usage
                    if "cost_usd" in data:
                        result["total_cost"] = data["cost_usd"]
                    if "total_tokens" in data:
                        result["total_tokens"] = data["total_tokens"]
                    if "input_tokens" in data:
                        result["input_tokens"] = data["input_tokens"]
                    if "output_tokens" in data:
                        result["output_tokens"] = data["output_tokens"]
                    # Also check nested stats
                    stats = data.get("stats", {})
                    if stats:
                        if "cost_usd" in stats:
                            result["total_cost"] = stats["cost_usd"]
                        if "total_tokens" in stats:
                            result["total_tokens"] = stats["total_tokens"]
                    if result:
                        return result
                except (json.JSONDecodeError, KeyError):
                    continue
    except Exception:
        pass
    return {}


def _monitor_agent(key: str, proc, log_file: Path, repo: str, number: int):
    """Monitor an agent process and retry on failure using the fallback chain."""
    try:
        proc.wait(timeout=AGENT_TIMEOUT_SECONDS)
    except subprocess.TimeoutExpired:
        log.warning("Agent %s exceeded timeout (%ds) — killing", key, AGENT_TIMEOUT_SECONDS)
        proc.kill()
        proc.wait(timeout=10)  # reap the zombie
    elapsed = "unknown"

    with agent_lock:
        info = active_agents.pop(key, {})
        if info.get("started"):
            start = datetime.fromisoformat(info["started"])
            elapsed = str(datetime.now(timezone.utc) - start).split(".")[0]

    exit_code = proc.returncode
    route_key = info.get("route_key", "conductor")
    chain_index = info.get("chain_index", 0)
    backend_key = info.get("backend_key", "claude")
    event = info.get("event", {})
    agent_repo_dir = info.get("repo_dir", "")
    worktree_path = info.get("worktree_path")

    status = "completed" if exit_code == 0 else f"failed (exit {exit_code})"
    log.info("Agent %s %s after %s — log: %s", key, status, elapsed, log_file)

    # Try to extract cost/token info from agent JSON output
    cost_info = _extract_cost_info(log_file)
    if cost_info:
        log.info("  Cost: $%.4f, tokens: %s", cost_info.get("total_cost", 0), cost_info.get("total_tokens", "?"))

    # Record attempt stats + tracking
    _record_attempt(repo, number, backend_key, exit_code == 0, exit_code, elapsed)
    _track_event("agent_completed", repo=repo, issue_number=number,
                  exit_code=exit_code, elapsed=elapsed, backend=backend_key,
                  status="success" if exit_code == 0 else "failure",
                  **cost_info)

    # Clean up worktree before retry or exit (frees the branch for reuse)
    _cleanup_worktree(agent_repo_dir, worktree_path)

    # Retry logic on failure
    if exit_code != 0 and event:
        chain = TASK_ROUTING.get(route_key, {}).get("chain", [])
        next_index = chain_index + 1
        if next_index < len(chain) and next_index <= MAX_RETRIES:
            next_backend_key = chain[next_index]
            tried_desc = BACKENDS.get(backend_key, {}).get("description", backend_key)
            next_desc = BACKENDS.get(next_backend_key, {}).get("description", next_backend_key)
            log.info("  Retrying %s: %s → %s (chain index %d)", key, tried_desc, next_desc, next_index)
            _post_retry_comment(repo, number, tried_desc, exit_code, next_desc, next_index)
            spawn_agent(event, chain_index=next_index)
            return
        else:
            log.info("  All backends exhausted for %s", key)
            _post_exhausted_comment(repo, number)

    # Notify dispatching agent on completion (best-effort)
    if exit_code == 0 and event:
        prompt = _build_completed_prompt(repo, number, event, elapsed)
        notify_callback(repo, number, event, reason="completed", prompt=prompt)

    # Publish completion event to Redis
    try:
        r = redis.Redis()
        r.publish("conductor:agent_complete", json.dumps({
            "key": key,
            "repo": repo,
            "issue_number": number,
            "status": "success" if exit_code == 0 else "failure",
            "exit_code": exit_code,
            "elapsed": elapsed,
            "log_file": str(log_file),
            "backend": backend_key,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }))
    except Exception:
        pass  # Best effort


def notify_callback(repo: str, number: int, event: dict, reason: str, prompt: str, dry_run: bool = False):
    """Resume the dispatching agent's session with a notification.

    Looks up the callback session ID stored by dispatch(), then spawns
    `claudea --resume <session-id>` with the provided prompt.

    Args:
        repo: Target repo
        number: Issue number
        event: Original event dict (for repo_dir, issue_url, etc.)
        reason: Short label for logging (e.g., "blocked", "completed")
        prompt: Full prompt to send to the resumed session
        dry_run: Log only, don't spawn
    """
    try:
        r = redis.Redis()
        callback_key = f"conductor:callback:{repo}#{number}"
        callback = r.hgetall(callback_key)
        if not callback:
            log.info("  No callback session for %s#%s — no one to notify", repo, number)
            return

        session_id = callback.get(b"session_id", b"").decode()
        if not session_id:
            log.info("  Callback has no session_id — skipping")
            return

        log.info("  Callback session found for %s: %s", reason, session_id[:8])

        if dry_run:
            log.info("  [DRY RUN] Would resume session %s with %s notification", session_id[:8], reason)
            return

        cmd = [
            "/home/tim/bin/claudea",
            "--resume", session_id,
            "-p", prompt,
            "--output-format", "json",
        ]

        LOG_DIR.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        safe_key = f"{repo.replace('/', '-')}-{number}"
        log_file = LOG_DIR / f"{timestamp}-callback-{reason}-{safe_key}.log"

        with open(log_file, "w") as f:
            proc = subprocess.Popen(
                cmd,
                cwd=event.get("repo_dir") or "/home/tim",
                stdout=f,
                stderr=subprocess.STDOUT,
            )

        log.info("  Callback agent spawned (pid %d, session %s, %s) — log: %s",
                 proc.pid, session_id[:8], reason, log_file)

    except Exception as e:
        log.error("  Failed to notify callback (%s): %s", reason, e)


def _build_blocked_prompt(repo: str, number: int, event: dict) -> str:
    """Build prompt for blocked callback notification."""
    try:
        result = subprocess.run(
            ["gh", "issue", "view", str(number), "--repo", repo,
             "--json", "body,comments", "--jq",
             '.comments[-1].body // "No reason given"'],
            capture_output=True, text=True, timeout=15,
        )
        blocked_reason = result.stdout.strip() if result.returncode == 0 else "Could not fetch reason"
    except Exception:
        blocked_reason = "Could not fetch reason"

    return (
        f"⚠️ BLOCKED AGENT NOTIFICATION\n\n"
        f"An agent you dispatched to {repo}#{number} is blocked and needs help.\n\n"
        f"Issue: {event.get('issue_url', f'https://github.com/{repo}/issues/{number}')}\n"
        f"Title: {event.get('issue_title', 'unknown')}\n\n"
        f"Blocked reason (from latest comment):\n{blocked_reason}\n\n"
        f"Please review the issue thread with "
        f"`gh issue view {number} --repo {repo} --comments` "
        f"and either:\n"
        f"1. Comment on the issue with guidance for the agent, then "
        f"comment `/run` to re-trigger\n"
        f"2. Resolve the blocker yourself and comment `/run`\n"
        f"3. Close the issue if the task should be abandoned"
    )


def _build_completed_prompt(repo: str, number: int, event: dict, elapsed: str) -> str:
    """Build prompt for completion callback notification."""
    try:
        result = subprocess.run(
            ["gh", "issue", "view", str(number), "--repo", repo,
             "--json", "comments", "--jq",
             '.comments[-1].body // "No summary available"'],
            capture_output=True, text=True, timeout=15,
        )
        summary = result.stdout.strip() if result.returncode == 0 else "No summary available"
    except Exception:
        summary = "No summary available"

    return (
        f"✅ DISPATCHED TASK COMPLETED\n\n"
        f"The agent you dispatched to {repo}#{number} has finished ({elapsed}).\n\n"
        f"Issue: {event.get('issue_url', f'https://github.com/{repo}/issues/{number}')}\n"
        f"Title: {event.get('issue_title', 'unknown')}\n\n"
        f"Agent's last comment:\n{summary}\n\n"
        f"Review with `gh issue view {number} --repo {repo} --comments` if needed."
    )


def handle_issue_event(event: dict, dry_run: bool = False):
    """Process an issue event from Redis."""
    action = event.get("action")
    repo = event.get("repo")
    number = event.get("issue_number")
    title = event.get("issue_title")
    labels = event.get("labels", [])
    repo_dir = event.get("repo_dir")
    has_conductor = event.get("has_conductor_label", False)

    log.info("Issue %s: %s#%s — %s", action, repo, number, title)
    log.info("  Labels: %s", ", ".join(labels) if labels else "none")
    log.info("  Directory: %s", repo_dir or "UNKNOWN")

    if not has_conductor:
        log.info("  No conductor label — skipping")
        return

    # Detect blocked label being added
    if action == "labeled" and "blocked" in labels:
        log.info("  🚨 BLOCKED label detected — notifying callback")
        prompt = _build_blocked_prompt(repo, number, event)
        notify_callback(repo, number, event, reason="blocked", prompt=prompt, dry_run=dry_run)
        return

    if action not in ("opened", "reopened"):
        # Only spawn on 'opened' or 'reopened', not 'labeled' — avoids double-spawn
        # (GitHub sends both 'opened' and 'labeled' when issue is created with labels)
        log.info("  Action '%s' — skipping (only 'opened'/'reopened' triggers spawn)", action)
        return

    spawn_agent(event, dry_run=dry_run)


def handle_comment_event(event: dict, dry_run: bool = False):
    """Process an issue comment event from Redis.

    Supports /run command to trigger agent spawn, with optional extra context.
    Any text after /run on the same or subsequent lines is passed as additional
    context to the spawned agent.
    """
    action = event.get("action")
    repo = event.get("repo")
    number = event.get("issue_number")
    title = event.get("issue_title")
    labels = event.get("labels", [])
    has_conductor = event.get("has_conductor_label", False)
    comment_body = event.get("comment_body", "")
    comment_author = event.get("comment_author", "")

    log.info("Comment %s: %s#%s by %s", action, repo, number, comment_author)

    if action != "created":
        log.info("  Comment action '%s' — skipping (only 'created' triggers)", action)
        return

    if not has_conductor:
        log.info("  No conductor label on issue — skipping comment")
        return

    # Check for /run command
    stripped = comment_body.strip()
    if not stripped.startswith("/run"):
        log.info("  Comment does not start with /run — skipping")
        return

    # Extract extra context: everything after "/run" (same line + subsequent lines)
    extra = stripped[4:].strip()

    log.info("  /run command detected from %s", comment_author)
    if extra:
        log.info("  Extra context: %s", extra[:100])

    # Build a synthetic issue event for spawn_agent
    spawn_event = {
        "repo": repo,
        "issue_number": number,
        "issue_title": title,
        "repo_dir": event.get("repo_dir"),
        "repo_config": event.get("repo_config", {}),
        "labels": labels,
        "issue_url": event.get("issue_url", ""),
        "extra_context": extra,
    }

    spawn_agent(spawn_event, dry_run=dry_run)


# ---------------------------------------------------------------------------
# PR Reviewer Agent
# ---------------------------------------------------------------------------

def handle_pr_event(event: dict, dry_run: bool = False):
    """Process a pull_request event from Redis."""
    action = event.get("action")
    repo = event.get("repo")
    pr_number = event.get("pr_number")
    pr_title = event.get("pr_title", "")
    pr_base = event.get("pr_base", "")
    pr_draft = event.get("pr_draft", False)
    repo_config = event.get("repo_config", {})
    repo_dir = event.get("repo_dir")

    log.info("PR %s: %s#%s — %s", action, repo, pr_number, pr_title)

    # Only review on opened (not synchronize, edited, etc.)
    if action != "opened":
        log.info("  PR action '%s' — skipping (only 'opened' triggers review)", action)
        return

    # Skip drafts
    if pr_draft:
        log.info("  Draft PR — skipping review")
        return

    # Only review PRs targeting the repo's base_branch
    base_branch = repo_config.get("base_branch", "develop")
    if pr_base != base_branch:
        log.info("  PR targets %s, not %s — skipping review", pr_base, base_branch)
        return

    # Only review if repo is monitored
    if not repo_dir:
        log.info("  Repo not monitored — skipping review")
        return

    spawn_reviewer(event, dry_run=dry_run)


def spawn_reviewer(event: dict, dry_run: bool = False):
    """Spawn a reviewer agent for a pull request."""
    repo = event["repo"]
    pr_number = event["pr_number"]
    pr_title = event.get("pr_title", "")
    pr_url = event.get("pr_url", "")
    pr_head = event.get("pr_head", "")
    pr_base = event.get("pr_base", "")
    repo_dir = event["repo_dir"]
    repo_config = event.get("repo_config", {})
    auto_merge = repo_config.get("auto_merge", False)

    key = f"review:{repo}#{pr_number}"

    # Dedup check
    with agent_lock:
        if key in active_agents:
            proc = active_agents[key].get("process")
            if proc and proc.poll() is None:
                log.info("  Reviewer already running for %s", key)
                return
            del active_agents[key]

    if auto_merge:
        merge_instruction = (
            "\n\n## Auto-merge\n"
            f"This repo has auto_merge enabled. If your review passes AND all CI checks are green:\n"
            f"  gh pr merge {pr_number} --repo {repo} --squash --auto\n"
            f"If review fails, post a requesting-changes review and do NOT merge."
        )
    else:
        merge_instruction = (
            "\n\n## Merge policy\n"
            f"This repo does NOT have auto_merge. After reviewing, leave your verdict as a PR review.\n"
            f"Do NOT merge. A human will merge after reviewing your assessment."
        )

    prompt = (
        f"You are a code reviewer for {repo} PR #{pr_number}: {pr_title}\n"
        f"URL: {pr_url}\n"
        f"Branch: {pr_head} → {pr_base}\n\n"
        f"## Review checklist\n"
        f"1. Read the PR diff: `gh pr diff {pr_number} --repo {repo}`\n"
        f"2. Read the PR description: `gh pr view {pr_number} --repo {repo}`\n"
        f"3. Check for: correctness, edge cases, style consistency, test coverage\n"
        f"4. If there are tests, run them (check the repo's CLAUDE.md for test commands)\n"
        f"5. Check CI status: `gh pr checks {pr_number} --repo {repo} --watch`\n"
        f"   - If CI is still running, wait up to 5 minutes\n"
        f"6. Submit your review:\n"
        f"   - APPROVE: `gh pr review {pr_number} --repo {repo} --approve --body \"<summary>\"`\n"
        f"   - REQUEST CHANGES: `gh pr review {pr_number} --repo {repo} --request-changes --body \"<issues found>\"`\n"
        f"{merge_instruction}\n\n"
        f"## Guidelines\n"
        f"- Be concise. Flag real issues, not style nits.\n"
        f"- If the PR looks good and tests pass, approve it.\n"
        f"- If you find issues, be specific about what to fix.\n"
    )

    cmd = [
        "/home/tim/bin/claudea",
        "-p", prompt,
        "--output-format", "json",
    ]

    log.info("  SPAWNING REVIEWER: %s in %s (auto_merge=%s)", key, repo_dir, auto_merge)

    if dry_run:
        log.info("  [DRY RUN] Would spawn reviewer for %s", key)
        return

    LOG_DIR.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    safe_key = key.replace("/", "-").replace("#", "-").replace(":", "-")
    log_file = LOG_DIR / f"{timestamp}-{safe_key}.log"

    try:
        with open(log_file, "w") as f:
            proc = subprocess.Popen(
                cmd,
                cwd=repo_dir,  # reviewer runs in main checkout (read-only review)
                stdout=f,
                stderr=subprocess.STDOUT,
                env={
                    **os.environ,
                    "CLAUDE_CODE_ENTRYPOINT": "conductor-reviewer",
                },
            )

        with agent_lock:
            active_agents[key] = {
                "process": proc,
                "pid": proc.pid,
                "repo": repo,
                "pr_number": pr_number,
                "title": pr_title,
                "started": datetime.now(timezone.utc).isoformat(),
                "log_file": str(log_file),
                "type": "reviewer",
            }

        log.info("  Started reviewer pid %d — log: %s", proc.pid, log_file)

        # Comment on the PR
        try:
            subprocess.run(
                ["gh", "pr", "comment", str(pr_number),
                 "--repo", repo,
                 "--body", f"🔍 Conductor reviewer dispatched (pid {proc.pid})"],
                capture_output=True, text=True, timeout=15,
            )
        except Exception:
            pass

        # Publish reviewer started event + track
        _track_event("reviewer_started", repo=repo, pr_number=pr_number,
                      title=pr_title, pid=proc.pid, auto_merge=auto_merge)
        try:
            r = redis.Redis()
            r.publish("conductor:agent_started", json.dumps({
                "key": key, "repo": repo, "pr_number": pr_number,
                "title": pr_title, "pid": proc.pid, "type": "reviewer",
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }))
        except Exception:
            pass

        # Monitor in background thread
        t = threading.Thread(
            target=_monitor_reviewer, args=(key, proc, log_file, repo, pr_number),
            daemon=True,
        )
        t.start()

    except Exception as e:
        log.error("  Failed to spawn reviewer: %s", e)


def _monitor_reviewer(key: str, proc, log_file: Path, repo: str, pr_number: int):
    """Monitor a reviewer process."""
    proc.wait()
    elapsed = "unknown"
    with agent_lock:
        info = active_agents.pop(key, {})
        if info.get("started"):
            start = datetime.fromisoformat(info["started"])
            elapsed = str(datetime.now(timezone.utc) - start).split(".")[0]

    exit_code = proc.returncode
    status = "completed" if exit_code == 0 else f"failed (exit {exit_code})"
    log.info("Reviewer %s %s after %s — log: %s", key, status, elapsed, log_file)

    _track_event("reviewer_completed", repo=repo, pr_number=pr_number,
                  exit_code=exit_code, elapsed=elapsed, status=status)

    try:
        r = redis.Redis()
        r.publish("conductor:agent_complete", json.dumps({
            "key": key, "repo": repo, "pr_number": pr_number,
            "status": "success" if exit_code == 0 else "failure",
            "exit_code": exit_code, "elapsed": elapsed, "type": "reviewer",
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }))
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Check Run Handling (CI feedback)
# ---------------------------------------------------------------------------

def handle_check_run_event(event: dict, dry_run: bool = False):
    """Process a check_run completed event — track CI results."""
    repo = event.get("repo")
    conclusion = event.get("check_conclusion", "")
    check_name = event.get("check_name", "")
    pr_numbers = event.get("pr_numbers", [])

    if not pr_numbers:
        return  # No associated PRs, nothing to do

    log.info("CI check '%s' %s for %s PRs: %s",
             check_name, conclusion, repo,
             ", ".join(str(n) for n in pr_numbers))

    # Track all CI results
    for pr_num in pr_numbers:
        _track_event("ci_check_completed", repo=repo, pr_number=pr_num,
                      check_name=check_name, conclusion=conclusion)


# ---------------------------------------------------------------------------
# Tracking / Observability
# ---------------------------------------------------------------------------

TRACKING_REDIS_PREFIX = "conductor:tracking"


def _track_event(event_type: str, **kwargs):
    """Record a tracking event in Redis for observability.

    Events are stored as:
    - conductor:tracking:events  — time-sorted list of all events (capped at 1000)
    - conductor:tracking:stats   — aggregate counters (hash)
    - conductor:tracking:daily:{date} — daily counters (hash, 30-day TTL)
    - conductor:tracking:repo:{repo} — per-repo counters (hash)
    """
    try:
        r = redis.Redis()
        now = datetime.now(timezone.utc)
        today = now.strftime("%Y-%m-%d")

        entry = {
            "event": event_type,
            "timestamp": now.isoformat(),
            **{k: v for k, v in kwargs.items() if v is not None},
        }

        # Append to capped event log
        r.lpush(f"{TRACKING_REDIS_PREFIX}:events", json.dumps(entry))
        r.ltrim(f"{TRACKING_REDIS_PREFIX}:events", 0, 999)

        # Aggregate stats
        r.hincrby(f"{TRACKING_REDIS_PREFIX}:stats", event_type, 1)

        # Daily stats
        daily_key = f"{TRACKING_REDIS_PREFIX}:daily:{today}"
        r.hincrby(daily_key, event_type, 1)
        r.expire(daily_key, 30 * 86400)

        # Per-repo stats
        repo = kwargs.get("repo")
        if repo:
            repo_key = f"{TRACKING_REDIS_PREFIX}:repo:{repo}"
            r.hincrby(repo_key, event_type, 1)

    except Exception as e:
        log.warning("Tracking failed: %s", e)


def main():
    parser = argparse.ArgumentParser(description="Conductor Subscriber")
    parser.add_argument("--redis-url", default="redis://localhost:6379")
    parser.add_argument("--channel", default="conductor:*",
                        help="Channel pattern to subscribe to (default: conductor:*)")
    parser.add_argument("--dry-run", action="store_true",
                        help="Log agent spawn decisions without actually spawning")
    args = parser.parse_args()

    # Clean up stale worktrees from any previous crashed agents
    _cleanup_stale_worktrees()

    try:
        r = redis.Redis.from_url(args.redis_url)
        r.ping()
        log.info("Connected to Redis at %s", args.redis_url)
    except redis.ConnectionError:
        log.error("Cannot connect to Redis at %s", args.redis_url)
        sys.exit(1)

    pubsub = r.pubsub()
    pubsub.psubscribe(args.channel)
    log.info("Subscribed to: %s", args.channel)
    log.info("Dry run: %s", args.dry_run)
    log.info("Agent log dir: %s", LOG_DIR)
    log.info("Waiting for events...")

    try:
        for message in pubsub.listen():
            if message["type"] not in ("pmessage", "message"):
                continue

            channel = message.get("channel", b"").decode()
            try:
                event = json.loads(message["data"])
            except (json.JSONDecodeError, TypeError):
                continue

            if channel == "conductor:issues":
                handle_issue_event(event, dry_run=args.dry_run)
            elif channel == "conductor:comments":
                handle_comment_event(event, dry_run=args.dry_run)
            elif channel == "conductor:pull_requests":
                handle_pr_event(event, dry_run=args.dry_run)
            elif channel == "conductor:check_runs":
                handle_check_run_event(event, dry_run=args.dry_run)
            elif channel == "conductor:agent_complete":
                log.info("Agent completed: %s", event.get("key", "?"))
            else:
                log.info("Event on %s: %s", channel, event.get("type", "unknown"))
    except KeyboardInterrupt:
        log.info("Shutting down")
        pubsub.close()


if __name__ == "__main__":
    main()