conductor.py

#!/usr/bin/env python3
"""Loqu8 Conductor — webhook listener + MCP server + Redis event bridge.

Receives GitHub webhook payloads from `gh webhook forward` on port 9999,
publishes structured events to Redis pub/sub, and exposes MCP tools for
AI agents to dispatch work, check status, and review results.

Usage:
    python3 conductor.py                  # start on :9999
    python3 conductor.py --port 9999      # explicit port
    python3 conductor.py --dry-run        # log without Redis

MCP tools (protocol):
    acknowledge         — Claim an issue and start working
    update_progress     — Post a milestone update
    report_blocked      — Flag issue as blocked + notify dispatcher
    complete            — Post summary and close issue
    my_assignment       — Get full context for your assigned issue
    get_issue_context   — Read any issue across repos

MCP tools (orchestration):
    dispatch            — Create a GitHub issue to trigger agent work
    dashboard           — All-in-one overview: health, agents, events
    list_repos          — Show monitored repos and directories
    agent_status        — Detailed status of a specific agent
    tail_log            — Read last N lines of an agent's log file

Redis channels:
    conductor:issues    — issue opened/labeled/closed events
    conductor:comments  — issue comment events (created/edited)
    conductor:pull_requests — PR opened/closed events
    conductor:check_runs    — CI check run completed events
    conductor:events    — all events (firehose)
"""

import argparse
import json
import logging
import os
import subprocess
import sys
import threading
import time
from collections import deque
from datetime import datetime, timedelta, timezone

import redis
import uvicorn
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route

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

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

_REPOS_FILE = os.path.join(os.path.dirname(__file__), "repos.json")
_repos_raw_cache: dict[str, dict | str] = {}
_repos_mtime: float = 0

# Default repo config values
_REPO_DEFAULTS = {
    "branch_strategy": "feature-branch",
    "base_branch": "develop",
    "auto_merge": False,
}


def _load_repos() -> dict[str, dict | str]:
    """Load raw repos.json, hot-reloading on change."""
    global _repos_raw_cache, _repos_mtime
    try:
        mtime = os.path.getmtime(_REPOS_FILE)
        if mtime != _repos_mtime:
            with open(_REPOS_FILE) as f:
                _repos_raw_cache = json.load(f)
            _repos_mtime = mtime
            log.info("Loaded %d repos from %s", len(_repos_raw_cache), _REPOS_FILE)
    except (FileNotFoundError, json.JSONDecodeError, OSError) as e:
        if not _repos_raw_cache:
            log.error("Failed to load %s: %s", _REPOS_FILE, e)
    return _repos_raw_cache


def _get_repo_dirs() -> dict[str, str]:
    """Load repo → directory mapping from repos.json, hot-reloading on change.

    Supports both formats:
      - Legacy: {"repo": "/path/to/dir"}
      - New:    {"repo": {"directory": "/path/to/dir", "branch_strategy": "...", ...}}
    """
    raw = _load_repos()
    result = {}
    for repo, value in raw.items():
        if isinstance(value, str):
            result[repo] = value
        elif isinstance(value, dict):
            result[repo] = value.get("directory", "")
        else:
            log.warning("Invalid config for %s in repos.json", repo)
    return result


def _get_repo_config(repo: str) -> dict:
    """Get full config for a repo, with defaults applied.

    Returns dict with keys: directory, branch_strategy, base_branch, auto_merge.
    """
    raw = _load_repos()
    value = raw.get(repo)
    if value is None:
        return {}
    if isinstance(value, str):
        return {"directory": value, **_REPO_DEFAULTS}
    config = {**_REPO_DEFAULTS}
    config.update(value)
    return config


def _guess_repo_directory(repo: str) -> str:
    """Guess the local directory for a repo based on existing repo paths.

    Looks at existing repos to find a common base directory for the org,
    then appends the repo name. Falls back to ~/Projects/{org}/{name}.
    """
    org, name = repo.split("/", 1) if "/" in repo else ("", repo)
    # Find existing repos from the same org to infer the base path
    for existing_repo, directory in _get_repo_dirs().items():
        existing_org = existing_repo.split("/", 1)[0] if "/" in existing_repo else ""
        if existing_org == org and directory:
            # Use the parent of this repo's directory as the org base
            org_base = os.path.dirname(directory)
            return os.path.join(org_base, name)
    return os.path.expanduser(f"~/Projects/{org}/{name}")


# Eagerly load once at startup, then hot-reload on change
# Eagerly load at startup for import-time references (parse_issue_event etc.)
# All tool functions call _get_repo_dirs() directly for hot-reload.

# Webhook delivery deduplication (GitHub can re-deliver on timeout)
_seen_deliveries: deque = deque(maxlen=200)

CONDUCTOR_LABELS = {"conductor", "agent:opus", "agent:cursor", "agent:marketing"}

# Maximum dispatch depth to prevent infinite recursion (A dispatches to B dispatches to A...)
MAX_DISPATCH_DEPTH = 3

# Keyword → label mapping for auto-labeling in dispatch()
# Keys are matched case-insensitively against issue title and body.
KEYWORD_LABELS = {
    # Bug indicators
    "fix": "bug",
    "bug": "bug",
    "broken": "bug",
    "crash": "bug",
    "error": "bug",
    "fail": "bug",
    "wrong": "bug",
    "regression": "bug",
    # Feature indicators
    "feat": "enhancement",
    "add": "enhancement",
    "implement": "enhancement",
    "new": "enhancement",
    "support": "enhancement",
    # Documentation
    "doc": "documentation",
    "readme": "documentation",
    "typo": "documentation",
    # Refactoring / chores
    "refactor": "refactor",
    "clean": "refactor",
    "rename": "refactor",
    "move": "refactor",
    # Testing
    "test": "testing",
    "spec": "testing",
    "coverage": "testing",
}

# Labels that signal complexity → auto-add agent:opus
COMPLEX_KEYWORDS = {"refactor", "migrate", "redesign", "architect", "security", "performance"}

# Task-type keywords → internal routing hints (not posted as GitHub labels)
# These are included in the event's labels list so subscriber can match them
# in TASK_ROUTING, but filtered out before creating GitHub issues.
TASK_TYPE_KEYWORDS = {
    "test": "task:test", "spec": "task:test", "coverage": "task:test",
    "review": "task:review", "audit": "task:review",
    "feat": "task:feature", "feature": "task:feature", "implement": "task:feature",
}


def infer_labels(title: str, body: str) -> list[str]:
    """Infer issue labels from title and body keywords.

    Scans title first (higher signal), then body. Returns deduplicated labels.
    Title words are matched as whole words; body uses substring matching
    for the first 500 chars to avoid false positives in long descriptions.
    """
    inferred: dict[str, None] = {}  # ordered set
    title_lower = title.lower()
    body_lower = (body or "")[:500].lower()

    # Check title words (higher confidence — match whole words)
    title_words = set(title_lower.split())
    for keyword, label in KEYWORD_LABELS.items():
        if keyword in title_words:
            inferred[label] = None

    # Check body (substring match, but only first 500 chars)
    for keyword, label in KEYWORD_LABELS.items():
        if keyword in body_lower:
            inferred[label] = None

    # Auto-add agent:opus for complex tasks
    all_text = title_lower + " " + body_lower
    if any(kw in all_text for kw in COMPLEX_KEYWORDS):
        inferred["agent:opus"] = None

    # Infer task-type routing hints (internal, filtered before GitHub label creation)
    for keyword, task_label in TASK_TYPE_KEYWORDS.items():
        if keyword in title_words or keyword in body_lower:
            inferred[task_label] = None

    return list(inferred)


# Recent events buffer (in-memory, last 100)
recent_events: deque = deque(maxlen=100)

# Active agents tracker (populated from Redis events, persisted to Redis hash)
active_agents: dict[str, dict] = {}  # key: "repo#number" → agent info

ACTIVE_AGENTS_REDIS_KEY = "conductor:active_agents"

# Redis client (set in main)
redis_client: redis.Redis | None = None


def _persist_active_agents():
    """Write active_agents dict to Redis so it survives restarts."""
    if redis_client is None:
        return
    try:
        if active_agents:
            # Store each agent as a hash field (key → JSON)
            mapping = {k: json.dumps(v, default=str) for k, v in active_agents.items()}
            redis_client.delete(ACTIVE_AGENTS_REDIS_KEY)
            redis_client.hset(ACTIVE_AGENTS_REDIS_KEY, mapping=mapping)
        else:
            redis_client.delete(ACTIVE_AGENTS_REDIS_KEY)
        redis_client.expire(ACTIVE_AGENTS_REDIS_KEY, 86400)  # 24h TTL
    except Exception:
        pass


def _restore_active_agents():
    """Restore active_agents from Redis on startup."""
    if redis_client is None:
        return
    try:
        data = redis_client.hgetall(ACTIVE_AGENTS_REDIS_KEY)
        for key, value in data.items():
            k = key.decode() if isinstance(key, bytes) else key
            v = json.loads(value)
            active_agents[k] = v
        if active_agents:
            log.info("Restored %d active agent(s) from Redis", len(active_agents))
    except Exception as e:
        log.warning("Failed to restore active agents: %s", e)

# ---------------------------------------------------------------------------
# Redis helpers
# ---------------------------------------------------------------------------

def publish(channel: str, event: dict):
    if redis_client is None:
        return 0
    try:
        msg = json.dumps(event, default=str)
        count = redis_client.publish(channel, msg)
        log.info("  → Published to %s (%d subscribers)", channel, count)
        return count
    except redis.ConnectionError:
        log.error("  → Redis connection failed")
        return 0


def parse_issue_event(payload: dict) -> dict:
    action = payload.get("action", "unknown")
    issue = payload.get("issue", {})
    repo = payload.get("repository", {})
    repo_full = repo.get("full_name", "unknown")
    labels = [l.get("name", "") for l in issue.get("labels", [])]

    repo_config = _get_repo_config(repo_full)
    return {
        "type": "issue",
        "action": action,
        "repo": repo_full,
        "repo_dir": repo_config.get("directory"),
        "repo_config": repo_config,
        "issue_number": issue.get("number"),
        "issue_title": issue.get("title", ""),
        "issue_body": issue.get("body", ""),
        "issue_url": issue.get("html_url", ""),
        "labels": labels,
        "has_conductor_label": bool(CONDUCTOR_LABELS & set(labels)),
        "author": issue.get("user", {}).get("login", ""),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


def parse_comment_event(payload: dict) -> dict:
    action = payload.get("action", "unknown")
    comment = payload.get("comment", {})
    issue = payload.get("issue", {})
    repo = payload.get("repository", {})
    repo_full = repo.get("full_name", "unknown")
    labels = [l.get("name", "") for l in issue.get("labels", [])]

    repo_config = _get_repo_config(repo_full)
    return {
        "type": "comment",
        "action": action,
        "repo": repo_full,
        "repo_dir": repo_config.get("directory"),
        "repo_config": repo_config,
        "issue_number": issue.get("number"),
        "issue_title": issue.get("title", ""),
        "issue_body": issue.get("body", ""),
        "issue_url": issue.get("html_url", ""),
        "comment_id": comment.get("id"),
        "comment_body": comment.get("body", ""),
        "comment_url": comment.get("html_url", ""),
        "comment_author": comment.get("user", {}).get("login", ""),
        "labels": labels,
        "has_conductor_label": bool(CONDUCTOR_LABELS & set(labels)),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


def parse_pull_request_event(payload: dict) -> dict:
    action = payload.get("action", "unknown")
    pr = payload.get("pull_request", {})
    repo = payload.get("repository", {})
    repo_full = repo.get("full_name", "unknown")
    labels = [l.get("name", "") for l in pr.get("labels", [])]

    repo_config = _get_repo_config(repo_full)
    return {
        "type": "pull_request",
        "action": action,
        "repo": repo_full,
        "repo_dir": repo_config.get("directory"),
        "repo_config": repo_config,
        "pr_number": pr.get("number"),
        "pr_title": pr.get("title", ""),
        "pr_body": pr.get("body", ""),
        "pr_url": pr.get("html_url", ""),
        "pr_head": pr.get("head", {}).get("ref", ""),
        "pr_base": pr.get("base", {}).get("ref", ""),
        "pr_author": pr.get("user", {}).get("login", ""),
        "pr_draft": pr.get("draft", False),
        "labels": labels,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


def parse_check_run_event(payload: dict) -> dict:
    action = payload.get("action", "unknown")
    check_run = payload.get("check_run", {})
    repo = payload.get("repository", {})
    repo_full = repo.get("full_name", "unknown")

    # Extract PR numbers from check_run's pull_requests array
    prs = check_run.get("pull_requests", [])
    pr_numbers = [pr.get("number") for pr in prs if pr.get("number")]

    repo_config = _get_repo_config(repo_full)
    return {
        "type": "check_run",
        "action": action,
        "repo": repo_full,
        "repo_dir": repo_config.get("directory"),
        "repo_config": repo_config,
        "check_name": check_run.get("name", ""),
        "check_status": check_run.get("status", ""),
        "check_conclusion": check_run.get("conclusion", ""),
        "pr_numbers": pr_numbers,
        "head_sha": check_run.get("head_sha", ""),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


# ---------------------------------------------------------------------------
# Webhook HTTP routes (Starlette)
# ---------------------------------------------------------------------------

async def webhook_handler(request: Request):
    try:
        payload = await request.json()
    except Exception:
        return JSONResponse({"error": "invalid json"}, status_code=400)

    event_type = request.headers.get("X-GitHub-Event", "unknown")
    delivery_id = request.headers.get("X-GitHub-Delivery", "")

    # Silently ack pings — GitHub sends these frequently as keepalives
    if event_type == "ping":
        return JSONResponse({"ok": True})

    # Deduplicate webhook deliveries (GitHub retries on timeout)
    if delivery_id and delivery_id in _seen_deliveries:
        log.info("Duplicate delivery %s — skipping", delivery_id[:8])
        return JSONResponse({"ok": True, "duplicate": True})
    if delivery_id:
        _seen_deliveries.append(delivery_id)

    log.info("Received %s event (delivery: %s)", event_type, delivery_id[:8])

    if event_type == "issues":
        event = parse_issue_event(payload)
        log.info(
            "  %s#%s [%s] %s (labels: %s)",
            event["repo"], event["issue_number"], event["action"],
            event["issue_title"],
            ", ".join(event["labels"]) or "none",
        )
        if event["has_conductor_label"]:
            log.info("  → CONDUCTOR LABEL DETECTED")
        publish("conductor:issues", event)
        recent_events.appendleft(event)
    elif event_type == "issue_comment":
        event = parse_comment_event(payload)
        log.info(
            "  %s#%s comment [%s] by %s (labels: %s)",
            event["repo"], event["issue_number"], event["action"],
            event["comment_author"],
            ", ".join(event["labels"]) or "none",
        )
        if event["has_conductor_label"]:
            log.info("  → CONDUCTOR LABEL on commented issue")
        publish("conductor:comments", event)
        recent_events.appendleft(event)
    elif event_type == "pull_request":
        event = parse_pull_request_event(payload)
        log.info(
            "  %s PR #%s [%s] %s → %s by %s",
            event["repo"], event["pr_number"], event["action"],
            event["pr_head"], event["pr_base"], event["pr_author"],
        )
        publish("conductor:pull_requests", event)
        recent_events.appendleft(event)
    elif event_type == "check_run":
        event = parse_check_run_event(payload)
        if event["action"] == "completed":
            log.info(
                "  %s check_run [%s] %s — %s (PRs: %s)",
                event["repo"], event["action"], event["check_name"],
                event["check_conclusion"],
                ", ".join(str(n) for n in event["pr_numbers"]) or "none",
            )
            publish("conductor:check_runs", event)
            recent_events.appendleft(event)
        # Ignore non-completed check_runs (created, in_progress)
    else:
        log.info("  %s event — firehose only", event_type)

    firehose = {
        "type": event_type,
        "delivery_id": delivery_id,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    publish("conductor:events", firehose)
    recent_events.appendleft(firehose)

    return JSONResponse({"ok": True})


async def health_handler(request: Request):
    return JSONResponse({
        "service": "conductor",
        "status": "running",
        "redis": redis_client is not None,
        "mcp": True,
        "repos": list(_get_repo_dirs().keys()),
        "recent_events": len(recent_events),
    })


# ---------------------------------------------------------------------------
# MCP Server — Instructions Template + Dynamic State
# ---------------------------------------------------------------------------

_INSTRUCTIONS_TEMPLATE = """Loqu8 Conductor — orchestrate work across repos.

{dynamic_section}
## Agent Protocol

When you are assigned an issue by the conductor, follow this protocol:

1. **Acknowledge**: Call `acknowledge()` — posts a comment and marks you active
2. **Read context**: Call `my_assignment()` — returns the full issue body, comments, and protocol state
3. **Work**: Implement the task. Post milestone updates with `update_progress(milestone)`
   - Good milestones: approach decided, major step completed, commit pushed
   - 2-4 updates for a typical task is right. Don't over-update.
4. **If blocked**: Call `report_blocked(reason)` — adds label, notifies dispatcher, stops work
5. **When done**: Call `complete(summary)` — posts summary, closes issue

## Tool Selection Guide

| Need | Tool |
|------|------|
| Claim your assigned issue | `acknowledge()` |
| Get full issue context | `my_assignment()` |
| Post a milestone update | `update_progress(milestone)` |
| Flag yourself as blocked | `report_blocked(reason)` |
| Close issue with summary | `complete(summary)` |
| Read any issue across repos | `get_issue_context(repo, number)` |
| Dispatch work to another repo | `dispatch(repo, title, body)` |
| See all active agents + health | `dashboard()` |
| Find stale/abandoned issues | `check_stale(hours?)` |
| List monitored repos | `list_repos()` |
| Add a repo to monitoring | `add_repo(repo, directory)` |
| Agent/reviewer performance data | `tracking(days?, repo?)` |
| Detailed status of one agent | `agent_status(repo, issue_number)` |
| Peek at agent's live log output | `tail_log(repo, issue_number, lines?)` |

## Protocol Tool Defaults

If you were spawned by the conductor, `repo` and `issue_number` are set automatically
from environment variables (`CONDUCTOR_AGENT_REPO`, `CONDUCTOR_AGENT_ISSUE`).
MCP-connected agents must pass them explicitly.

## Re-triggering Agents

To re-spawn an agent on an existing issue (e.g., after a stale session or failed run),
comment `/run` on the issue. Optional context after `/run` is passed to the new agent.

## MCP Staleness

If conductor tools return errors or the MCP session is stale (conductor was restarted),
fall back to direct gh CLI commands:
- Comment: `gh issue comment <number> --repo <repo> --body "<message>"`
- Close: `gh issue close <number> --repo <repo>`
- Re-trigger: comment `/run` on the issue

## Monitored Repos

{repos_section}

## Conductor Labels

Issues with these labels trigger agent work: {labels_list}
"""


def _build_dynamic_instructions() -> str:
    """Build the dynamic section with live agent/event/health state."""
    lines = []

    # Active agents
    active = [info for info in active_agents.values()
              if info.get("status") not in ("success", "failure")]
    if active:
        lines.append(f"## Live State — {len(active)} Active Agent(s)\n")
        for info in active:
            started = info.get("started", "")
            duration = ""
            if started:
                try:
                    start_dt = datetime.fromisoformat(started)
                    elapsed = datetime.now(timezone.utc) - start_dt
                    minutes = int(elapsed.total_seconds() // 60)
                    duration = f" ({minutes}m)" if minutes > 0 else " (<1m)"
                except (ValueError, TypeError):
                    pass
            lines.append(f"- **{info.get('repo', '?')}#{info.get('issue_number', '?')}** "
                         f"— {info.get('title', '?')}{duration}")
        lines.append("")
    else:
        lines.append("## Live State — No Active Agents\n")

    # Recent events (last 3)
    issue_events = [e for e in recent_events if e.get("type") == "issue"][:3]
    if issue_events:
        lines.append("### Recent Events")
        for ev in issue_events:
            lines.append(f"- {ev.get('repo', '?')}#{ev.get('issue_number', '?')} "
                         f"[{ev.get('action', '?')}] {ev.get('issue_title', '')}")
        lines.append("")

    # System health
    redis_ok = redis_client is not None
    try:
        if redis_ok:
            redis_client.ping()
    except Exception:
        redis_ok = False
    lines.append(f"### Health\n- Redis: {'connected' if redis_ok else 'disconnected'}")
    lines.append(f"- Events buffered: {len(recent_events)}")
    lines.append("")

    return "\n".join(lines)


def _build_repos_section() -> str:
    """Build the repos list for instructions."""
    return "\n".join(f"- `{repo}` → `{directory}`"
                     for repo, directory in _get_repo_dirs().items())


def _format_instructions() -> str:
    """Format the full instructions string with current state."""
    return _INSTRUCTIONS_TEMPLATE.format(
        dynamic_section=_build_dynamic_instructions(),
        repos_section=_build_repos_section(),
        labels_list=", ".join(f"`{l}`" for l in sorted(CONDUCTOR_LABELS)),
    )


def _refresh_instructions() -> None:
    """Update the MCP server instructions with current state."""
    try:
        mcp._mcp_server.instructions = _format_instructions()
    except Exception:
        pass  # Keep existing instructions on error


mcp = FastMCP(
    "conductor",
    instructions=_INSTRUCTIONS_TEMPLATE.format(
        dynamic_section="## Live State — Starting up...\n\n",
        repos_section=_build_repos_section(),
        labels_list=", ".join(f"`{l}`" for l in sorted(CONDUCTOR_LABELS)),
    ),
)

# ---------------------------------------------------------------------------
# Helper: resolve repo/issue from args or env
# ---------------------------------------------------------------------------

def _resolve_repo_issue(
    repo: str | None, issue_number: int | None
) -> tuple[str, int] | dict:
    """Resolve repo and issue_number from explicit args or env vars.

    Returns (repo, issue_number) tuple or an error dict.
    """
    repo = repo or os.environ.get("CONDUCTOR_AGENT_REPO")
    issue_str = str(issue_number) if issue_number else os.environ.get("CONDUCTOR_AGENT_ISSUE")
    if not repo or not issue_str:
        return {"error": "repo and issue_number are required (not set in args or env vars "
                         "CONDUCTOR_AGENT_REPO / CONDUCTOR_AGENT_ISSUE)"}
    try:
        return (repo, int(issue_str))
    except (ValueError, TypeError):
        return {"error": f"Invalid issue_number: {issue_str}"}


def _ensure_labels(repo: str, labels: list[str]) -> None:
    """Create any missing labels in the target repo. Best effort."""
    for label in labels:
        try:
            subprocess.run(
                ["gh", "label", "create", label, "--repo", repo,
                 "--description", "Conductor-managed", "--color", "5319E7"],
                capture_output=True, text=True, timeout=15,
            )
        except Exception:
            pass  # Label may already exist — gh returns non-zero, that's fine


def _gh_comment(repo: str, number: int, body: str) -> dict | None:
    """Post a comment on a GitHub issue. Returns error dict or None on success."""
    try:
        result = subprocess.run(
            ["gh", "issue", "comment", str(number), "--repo", repo, "--body", body],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return {"error": f"gh comment failed: {result.stderr.strip()}"}
        return None
    except subprocess.TimeoutExpired:
        return {"error": "gh issue comment timed out"}
    except FileNotFoundError:
        return {"error": "gh CLI not found"}


# ---------------------------------------------------------------------------
# MCP Tools — Protocol (for spawned agents)
# ---------------------------------------------------------------------------

@mcp.tool()
def acknowledge(
    repo: str | None = None,
    issue_number: int | None = None,
) -> dict:
    """Acknowledge your assigned issue — call this first when you start working.

    Posts a comment on the issue and marks you as the active agent.
    Repo and issue_number default from env vars if you were spawned by conductor.

    Args:
        repo: Target repo (e.g., 'loqu8/ice'). Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    repo_dir = _get_repo_dirs().get(repo, "unknown")
    comment = f"🤖 Agent picking this up. Working in `{repo_dir}`."
    err = _gh_comment(repo, number, comment)
    if err:
        return err

    # Track as active
    key = f"{repo}#{number}"
    active_agents[key] = {
        "repo": repo,
        "issue_number": number,
        "started": datetime.now(timezone.utc).isoformat(),
        "status": "running",
        "milestones": 0,
    }
    log.info("Agent acknowledged: %s", key)

    _refresh_instructions()
    return {"acknowledged": True, "repo": repo, "issue_number": number}


@mcp.tool()
def update_progress(
    milestone: str,
    repo: str | None = None,
    issue_number: int | None = None,
) -> dict:
    """Post a milestone update on your assigned issue.

    Good milestones: approach decided, major step completed, commit pushed.
    2-4 updates per task is typical — don't over-update.

    Args:
        milestone: Brief description of what was accomplished
        repo: Target repo. Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    comment = f"📍 {milestone}"
    err = _gh_comment(repo, number, comment)
    if err:
        return err

    # Update milestone count
    key = f"{repo}#{number}"
    if key in active_agents:
        active_agents[key]["milestones"] = active_agents[key].get("milestones", 0) + 1

    log.info("Progress update on %s: %s", key, milestone[:80])

    _refresh_instructions()
    return {"updated": True, "repo": repo, "issue_number": number, "milestone": milestone}


@mcp.tool()
def report_blocked(
    reason: str,
    repo: str | None = None,
    issue_number: int | None = None,
) -> dict:
    """Report that you are blocked on your assigned issue.

    Posts the reason as a comment, adds the 'blocked' label, and triggers
    a callback notification to the agent that dispatched this work (if any).
    After calling this, stop working — a human or dispatcher will help.

    Args:
        reason: What you tried and what's blocking you
        repo: Target repo. Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    comment = f"🚧 **Blocked**: {reason}"
    err = _gh_comment(repo, number, comment)
    if err:
        return err

    # Add blocked label
    try:
        subprocess.run(
            ["gh", "issue", "edit", str(number), "--repo", repo, "--add-label", "blocked"],
            capture_output=True, text=True, timeout=15,
        )
    except Exception:
        pass  # Best effort

    # Update agent status
    key = f"{repo}#{number}"
    if key in active_agents:
        active_agents[key]["status"] = "blocked"

    log.info("Agent blocked on %s: %s", key, reason[:80])

    _refresh_instructions()
    return {"blocked": True, "repo": repo, "issue_number": number, "reason": reason}


@mcp.tool()
def complete(
    summary: str,
    repo: str | None = None,
    issue_number: int | None = None,
) -> dict:
    """Complete your assigned issue — post summary and close it.

    Posts a completion comment with your summary, then closes the issue.
    Call this when all work is done and verified.

    Args:
        summary: What you changed, which files, test results
        repo: Target repo. Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    comment = f"✅ **Done**\n\n{summary}"
    err = _gh_comment(repo, number, comment)
    if err:
        return err

    # Close the issue
    try:
        result = subprocess.run(
            ["gh", "issue", "close", str(number), "--repo", repo],
            capture_output=True, text=True, timeout=15,
        )
        if result.returncode != 0:
            return {"error": f"Failed to close issue: {result.stderr.strip()}"}
    except subprocess.TimeoutExpired:
        return {"error": "gh issue close timed out"}
    except FileNotFoundError:
        return {"error": "gh CLI not found"}

    # Update agent status
    key = f"{repo}#{number}"
    if key in active_agents:
        active_agents[key]["status"] = "success"
        active_agents[key]["completed"] = datetime.now(timezone.utc).isoformat()

    log.info("Agent completed %s: %s", key, summary[:80])

    _refresh_instructions()
    return {"completed": True, "repo": repo, "issue_number": number, "summary": summary}


@mcp.tool()
def my_assignment(
    repo: str | None = None,
    issue_number: int | None = None,
) -> dict:
    """Get full context for your assigned issue.

    Returns the issue body, comments, labels, and your protocol state
    (milestones posted, time active). Call this after acknowledge() to
    understand what you need to do.

    Args:
        repo: Target repo. Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    # Fetch issue details
    try:
        result = subprocess.run(
            ["gh", "issue", "view", str(number), "--repo", repo,
             "--json", "title,body,labels,state,comments,assignees"],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return {"error": f"Failed to fetch issue: {result.stderr.strip()}"}
        issue_data = json.loads(result.stdout)
    except subprocess.TimeoutExpired:
        return {"error": "gh issue view timed out"}
    except (FileNotFoundError, json.JSONDecodeError) as e:
        return {"error": str(e)}

    # Add protocol state
    key = f"{repo}#{number}"
    agent_state = active_agents.get(key, {})

    return {
        "repo": repo,
        "issue_number": number,
        "title": issue_data.get("title", ""),
        "body": issue_data.get("body", ""),
        "labels": [l.get("name", "") for l in issue_data.get("labels", [])],
        "state": issue_data.get("state", ""),
        "comments": [
            {"author": c.get("author", {}).get("login", ""),
             "body": c.get("body", ""),
             "created": c.get("createdAt", "")}
            for c in issue_data.get("comments", [])
        ],
        "protocol_state": {
            "status": agent_state.get("status", "not_tracked"),
            "milestones": agent_state.get("milestones", 0),
            "started": agent_state.get("started"),
        },
    }


@mcp.tool()
def get_issue_context(
    repo: str,
    number: int,
) -> dict:
    """Read any issue across monitored repos — not just your assignment.

    Use this to check related issues, understand dependencies, or review
    work done by other agents.

    Args:
        repo: Target repo (e.g., 'loqu8/ice')
        number: Issue number
    """
    try:
        result = subprocess.run(
            ["gh", "issue", "view", str(number), "--repo", repo,
             "--json", "title,body,labels,state,comments"],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return {"error": f"Failed to fetch issue: {result.stderr.strip()}"}
        issue_data = json.loads(result.stdout)
    except subprocess.TimeoutExpired:
        return {"error": "gh issue view timed out"}
    except (FileNotFoundError, json.JSONDecodeError) as e:
        return {"error": str(e)}

    return {
        "repo": repo,
        "number": number,
        "title": issue_data.get("title", ""),
        "body": issue_data.get("body", ""),
        "labels": [l.get("name", "") for l in issue_data.get("labels", [])],
        "state": issue_data.get("state", ""),
        "comments": [
            {"author": c.get("author", {}).get("login", ""),
             "body": c.get("body", ""),
             "created": c.get("createdAt", "")}
            for c in issue_data.get("comments", [])
        ],
    }


# ---------------------------------------------------------------------------
# MCP Tools — Orchestration
# ---------------------------------------------------------------------------

@mcp.tool()
def dispatch(
    repo: str,
    title: str,
    body: str,
    labels: list[str] | None = None,
    callback_session: str | None = None,
    dispatch_depth: int = 0,
) -> dict:
    """Create a GitHub issue in a target repo to dispatch agent work.

    The issue is labeled with 'conductor' automatically. After creation,
    a synthetic event is published directly to Redis so the subscriber
    spawns an agent immediately (no webhook round-trip required).

    Args:
        repo: Target repo (e.g., 'loqu8/ice', 'srclight/srclight')
        title: Issue title describing the task
        body: Detailed issue body with context, acceptance criteria, etc.
        labels: Additional labels (e.g., ['agent:opus', 'enhancement']). 'conductor' is always added.
              If omitted, labels are auto-inferred from title/body keywords (e.g., 'fix' → 'bug',
              'feat' → 'enhancement'). Pass an explicit list (even empty) to skip auto-inference.
        callback_session: Your Claude Code session ID (UUID). If the spawned agent gets
              blocked, conductor will resume your session to notify you. Find your session
              ID from your conversation's .jsonl filename in ~/.claude/projects/.
        dispatch_depth: Current depth in the dispatch chain (0 = top-level). Used to prevent
              infinite recursion when agents dispatch to other agents. Automatically incremented.
    """
    if dispatch_depth >= MAX_DISPATCH_DEPTH:
        return {
            "error": f"Dispatch depth limit reached ({MAX_DISPATCH_DEPTH}). "
                     f"This prevents infinite recursion in dispatch chains.",
            "depth": dispatch_depth,
            "max_depth": MAX_DISPATCH_DEPTH,
        }

    if repo not in _get_repo_dirs():
        suggested_dir = _guess_repo_directory(repo)
        hint = (
            f"Repo '{repo}' is not monitored. "
            f"Add it with: add_repo(repo='{repo}', directory='{suggested_dir}')"
        )
        return {
            "error": hint,
            "suggested_directory": suggested_dir,
            "available": list(_get_repo_dirs().keys()),
        }

    # Auto-infer labels from title/body, then merge with explicit labels
    auto_labels = infer_labels(title, body) if labels is None else []
    all_labels = ["conductor"] + (labels or []) + auto_labels
    # Deduplicate while preserving order
    seen: set[str] = set()
    all_labels = [l for l in all_labels if not (l in seen or seen.add(l))]
    # Separate internal routing hints from GitHub labels
    github_labels = [l for l in all_labels if not l.startswith("task:")]
    label_args = []
    for label in github_labels:
        label_args.extend(["--label", label])

    # Auto-create any missing labels in the target repo
    _ensure_labels(repo, github_labels)

    try:
        result = subprocess.run(
            ["gh", "issue", "create",
             "--repo", repo,
             "--title", title,
             "--body", body,
             *label_args],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return {"error": result.stderr.strip(), "returncode": result.returncode}

        issue_url = result.stdout.strip()
        # Extract issue number from URL (e.g., .../issues/13 → 13)
        issue_number = issue_url.rstrip("/").split("/")[-1]
        log.info("Dispatched: %s → %s (labels: %s)", repo, issue_url, ", ".join(all_labels))

        # Store callback session in Redis for blocked notification
        if callback_session and redis_client:
            callback_key = f"conductor:callback:{repo}#{issue_number}"
            redis_client.hset(callback_key, mapping={
                "session_id": callback_session,
                "dispatched_at": datetime.now(timezone.utc).isoformat(),
                "repo": repo,
                "issue_number": issue_number,
            })
            # Expire after 24h — if the issue isn't resolved by then, callback is stale
            redis_client.expire(callback_key, 86400)
            log.info("  Callback session stored: %s → %s", callback_key, callback_session[:8])

        # Publish directly to Redis so subscriber spawns immediately,
        # bypassing the webhook round-trip (which only covers loqu8/conductor).
        repo_config = _get_repo_config(repo)
        event = {
            "type": "issue",
            "action": "opened",
            "repo": repo,
            "repo_dir": repo_config.get("directory"),
            "repo_config": repo_config,
            "issue_number": int(issue_number),
            "issue_title": title,
            "issue_body": body,
            "issue_url": issue_url,
            "labels": all_labels,
            "has_conductor_label": bool(CONDUCTOR_LABELS & set(all_labels)),
            "author": "conductor-dispatch",
            "dispatch_depth": dispatch_depth + 1,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        publish("conductor:issues", event)
        recent_events.appendleft(event)

        _refresh_instructions()
        return {
            "dispatched": True,
            "repo": repo,
            "repo_dir": _get_repo_dirs()[repo],
            "issue_url": issue_url,
            "title": title,
            "labels": all_labels,
            "auto_labels": auto_labels,
            "callback_session": callback_session,
        }
    except subprocess.TimeoutExpired:
        return {"error": "gh issue create timed out"}
    except FileNotFoundError:
        return {"error": "gh CLI not found"}


@mcp.tool()
def dashboard() -> dict:
    """All-in-one overview: service health, active agents, recent events.

    Combines status, agent, and event information into a single view.
    """
    # Health
    redis_ok = redis_client is not None
    try:
        if redis_ok:
            redis_client.ping()
    except Exception:
        redis_ok = False

    # Agents
    active = []
    completed = []
    for key, info in active_agents.items():
        entry = {"key": key, **info}
        if info.get("status") in ("success", "failure"):
            completed.append(entry)
        else:
            active.append(entry)

    # Recent events
    issue_events = [e for e in recent_events if e.get("type") == "issue"]
    conductor_events = [e for e in issue_events if e.get("has_conductor_label")]

    _refresh_instructions()
    return {
        "health": {
            "redis": "connected" if redis_ok else "disconnected",
            "events_buffered": len(recent_events),
            "monitored_repos": len(_get_repo_dirs()),
        },
        "agents": {
            "active_count": len(active),
            "active": active,
            "recently_completed_count": len(completed),
            "recently_completed": completed[-10:],  # last 10
        },
        "recent_events": {
            "total": len(recent_events),
            "issue_events": len(issue_events),
            "conductor_events": len(conductor_events),
            "last_3": list(recent_events)[:3],
        },
    }


@mcp.tool()
def check_stale(
    hours: int = 6,
    repos: list[str] | None = None,
) -> dict:
    """Scan monitored repos for open conductor issues with no recent activity.

    Returns a list of potentially stale issues with last activity time,
    labels, and retry history. Use this to find abandoned or stuck work.

    Args:
        hours: Consider issues stale after this many hours of inactivity (default: 6)
        repos: Repos to scan (default: all monitored repos)
    """
    target_repos = repos or list(_get_repo_dirs().keys())
    stale = []
    active = []
    errors = []

    for repo in target_repos:
        if repo not in _get_repo_dirs():
            errors.append({"repo": repo, "error": "not monitored"})
            continue

        # Fetch open issues with conductor labels
        label_filter = ",".join(sorted(CONDUCTOR_LABELS))
        try:
            result = subprocess.run(
                ["gh", "issue", "list", "--repo", repo, "--state", "open",
                 "--json", "number,title,labels,updatedAt,comments,createdAt",
                 "--limit", "50"],
                capture_output=True, text=True, timeout=30,
            )
            if result.returncode != 0:
                errors.append({"repo": repo, "error": result.stderr.strip()})
                continue
            issues = json.loads(result.stdout)
        except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError) as e:
            errors.append({"repo": repo, "error": str(e)})
            continue

        for issue in issues:
            issue_labels = {l.get("name", "") for l in issue.get("labels", [])}
            if not (CONDUCTOR_LABELS & issue_labels):
                continue

            number = issue["number"]
            updated = issue.get("updatedAt", "")
            comments = issue.get("comments", [])

            # Determine last activity
            last_activity = updated
            last_actor = None
            if comments:
                last_comment = comments[-1]
                last_activity = last_comment.get("createdAt", updated)
                last_actor = last_comment.get("author", {}).get("login", "")

            # Calculate staleness
            try:
                last_dt = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
                age = datetime.now(timezone.utc) - last_dt
                age_hours = age.total_seconds() / 3600
            except (ValueError, TypeError):
                age_hours = 999
                age = None

            # Check for retry history in Redis
            retry_count = 0
            if redis_client:
                try:
                    retry_key = f"conductor:retries:{repo}#{number}"
                    retry_count = redis_client.llen(retry_key)
                except Exception:
                    pass

            entry = {
                "repo": repo,
                "number": number,
                "title": issue["title"],
                "labels": sorted(issue_labels),
                "hours_since_activity": round(age_hours, 1),
                "last_actor": last_actor,
                "retry_count": retry_count,
                "url": f"https://github.com/{repo}/issues/{number}",
            }

            if age_hours >= hours:
                # Classify the issue
                if "blocked" in issue_labels:
                    entry["status"] = "blocked"
                    entry["suggestion"] = "Review blocker reason, then /run or close"
                elif retry_count > 0:
                    entry["status"] = "retried_and_stale"
                    entry["suggestion"] = "Agent retried but still stale — check logs, then /run or close"
                else:
                    entry["status"] = "abandoned"
                    entry["suggestion"] = "Agent may have died silently — /run to retry or close"
                stale.append(entry)
            else:
                entry["status"] = "active"
                active.append(entry)

    return {
        "stale": stale,
        "stale_count": len(stale),
        "active": active,
        "active_count": len(active),
        "threshold_hours": hours,
        "repos_scanned": len(target_repos),
        "errors": errors,
    }


@mcp.tool()
def list_repos() -> dict:
    """List all monitored repos and their local directories.

    Shows which repos the conductor watches for issue events
    and where agents would be spawned on disk.
    """
    return {
        "repos": [
            {"repo": repo, "directory": directory}
            for repo, directory in _get_repo_dirs().items()
        ],
        "conductor_labels": sorted(CONDUCTOR_LABELS),
    }


@mcp.tool()
def add_repo(
    repo: str,
    directory: str,
    branch_strategy: str = "feature-branch",
    base_branch: str = "develop",
    auto_merge: bool = False,
) -> dict:
    """Add a repo to conductor monitoring at runtime.

    Writes the new entry to repos.json (hot-reloaded, no restart needed).
    The directory must exist on disk and contain a git repo.

    Args:
        repo: GitHub repo in 'owner/name' format (e.g., 'loqu8/stroke')
        directory: Absolute path to the local checkout
        branch_strategy: 'feature-branch' (default) or 'direct-commit'
        base_branch: Branch to create features from (default: 'develop')
        auto_merge: Whether reviewer can auto-merge approved PRs (default: false)
    """
    if "/" not in repo or len(repo.split("/")) != 2:
        return {"error": f"Invalid repo format: '{repo}'. Expected 'owner/name'."}

    if not os.path.isdir(directory):
        return {"error": f"Directory does not exist: {directory}"}

    if not os.path.isdir(os.path.join(directory, ".git")):
        return {"error": f"Not a git repository: {directory}"}

    if branch_strategy not in ("feature-branch", "direct-commit"):
        return {"error": f"Invalid branch_strategy: '{branch_strategy}'. Use 'feature-branch' or 'direct-commit'."}

    # Check if already monitored
    if repo in _get_repo_dirs():
        return {"error": f"Repo '{repo}' is already monitored.", "directory": _get_repo_dirs()[repo]}

    # Load current file, add entry, write back
    try:
        with open(_REPOS_FILE) as f:
            repos_data = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError) as e:
        return {"error": f"Failed to read {_REPOS_FILE}: {e}"}

    repos_data[repo] = {
        "directory": directory,
        "branch_strategy": branch_strategy,
        "base_branch": base_branch,
        "auto_merge": auto_merge,
    }

    try:
        with open(_REPOS_FILE, "w") as f:
            json.dump(repos_data, f, indent=2)
            f.write("\n")
    except OSError as e:
        return {"error": f"Failed to write {_REPOS_FILE}: {e}"}

    # Invalidate cache so next call picks up the change
    global _repos_mtime
    _repos_mtime = 0

    log.info("Added repo %s → %s", repo, directory)
    _refresh_instructions()

    return {
        "added": True,
        "repo": repo,
        "directory": directory,
        "branch_strategy": branch_strategy,
        "base_branch": base_branch,
        "auto_merge": auto_merge,
        "total_repos": len(_get_repo_dirs()),
    }


# ---------------------------------------------------------------------------
# Browser-based article reading
# ---------------------------------------------------------------------------

_browser_lock = threading.Lock()
_browser_executor = __import__("concurrent.futures").futures.ThreadPoolExecutor(max_workers=1)
BROWSER_PROFILE = os.path.expanduser("~/.conductor/browser-profile")


def _fetch_article(url: str, wait_seconds: int) -> dict:
    """Run Playwright in a dedicated thread (sync API can't run inside asyncio)."""
    from playwright.sync_api import sync_playwright

    os.makedirs(BROWSER_PROFILE, exist_ok=True)

    # Remove stale lock file from crashed sessions
    lock_file = os.path.join(BROWSER_PROFILE, "SingletonLock")
    if os.path.exists(lock_file):
        try:
            os.remove(lock_file)
        except OSError:
            pass

    with _browser_lock:
        context = None
        try:
            pw = sync_playwright().start()
            context = pw.chromium.launch_persistent_context(
                user_data_dir=BROWSER_PROFILE,
                headless=True,
                args=[
                    "--no-sandbox",
                    "--disable-dev-shm-usage",
                    "--disable-blink-features=AutomationControlled",
                ],
                user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
            )
            page = context.new_page()
            page.goto(url, wait_until="domcontentloaded", timeout=30_000)

            if wait_seconds > 0:
                page.wait_for_timeout(wait_seconds * 1000)

            final_url = page.url
            title = page.title()

            # Check for auth redirect
            auth_indicators = ["/login", "/signin", "/auth", "/account/begin", "/i/flow/login"]
            if any(ind in final_url.lower() for ind in auth_indicators):
                return {
                    "error": "Authentication required",
                    "auth_required": True,
                    "redirected_to": final_url,
                    "hint": "Run: python browser_login.py --url <site> to log in interactively",
                }

            # Grab raw body text as fallback before extracting
            fallback_text = ""
            try:
                fallback_text = page.inner_text("body")
            except Exception:
                pass

            html = page.content()
            context.close()
            pw.stop()
            context = None

            # Tier 1: readability extraction
            text = ""
            article_title = title
            try:
                from readability import Document
                import html2text

                doc = Document(html)
                article_html = doc.summary()
                article_title = doc.short_title() or title

                h = html2text.HTML2Text()
                h.ignore_links = False
                h.ignore_images = True
                h.body_width = 0
                text = h.handle(article_html).strip()
            except Exception:
                pass

            # Tier 2: fallback to raw body text if readability got junk
            noscript_markers = ["javascript is not available", "javascript is disabled", "enable javascript"]
            if len(text) < 100 or any(m in text.lower() for m in noscript_markers):
                text = fallback_text.strip()

            max_len = 50_000
            return {
                "url": url,
                "final_url": final_url,
                "title": article_title,
                "content": text[:max_len],
                "content_length": len(text),
                "truncated": len(text) > max_len,
            }

        except Exception as exc:
            return {"error": f"Browser error: {exc}", "url": url}
        finally:
            if context is not None:
                try:
                    context.close()
                except Exception:
                    pass


@mcp.tool()
def tracking(days: int = 7, repo: str = "") -> dict:
    """Get tracking data for evaluating agent and reviewer performance.

    Returns aggregate stats, daily breakdowns, and recent events.
    Use this to understand merge rates, review outcomes, CI pass rates,
    and agent success/failure patterns.

    Args:
        days: Number of days of daily stats to return (default 7).
        repo: Filter to a specific repo (optional, e.g. "loqu8/ice").
    """
    if redis_client is None:
        return {"error": "Redis not connected"}

    prefix = "conductor:tracking"

    # Aggregate stats
    stats = {}
    raw = redis_client.hgetall(f"{prefix}:stats")
    for k, v in raw.items():
        stats[k.decode()] = int(v)

    # Daily stats
    daily = {}
    today = datetime.now(timezone.utc)
    for i in range(days):
        d = (today - timedelta(days=i)).strftime("%Y-%m-%d")
        raw = redis_client.hgetall(f"{prefix}:daily:{d}")
        if raw:
            daily[d] = {k.decode(): int(v) for k, v in raw.items()}

    # Per-repo stats (all repos or filtered)
    repo_stats = {}
    if repo:
        raw = redis_client.hgetall(f"{prefix}:repo:{repo}")
        if raw:
            repo_stats[repo] = {k.decode(): int(v) for k, v in raw.items()}
    else:
        # Get all repo keys
        for key in redis_client.scan_iter(f"{prefix}:repo:*"):
            repo_name = key.decode().replace(f"{prefix}:repo:", "")
            raw = redis_client.hgetall(key)
            if raw:
                repo_stats[repo_name] = {k.decode(): int(v) for k, v in raw.items()}

    # Recent events (last 20)
    recent = []
    raw_events = redis_client.lrange(f"{prefix}:events", 0, 19)
    for entry in raw_events:
        try:
            recent.append(json.loads(entry))
        except Exception:
            pass

    # Compute derived metrics
    metrics = {}
    total_agents = stats.get("agent_started", 0)
    total_completed = stats.get("agent_completed", 0)
    total_reviewers = stats.get("reviewer_started", 0)
    total_reviews_done = stats.get("reviewer_completed", 0)
    total_ci = stats.get("ci_check_completed", 0)

    if total_agents > 0:
        # Count successes from recent events
        success_count = sum(1 for e in recent
                          if e.get("event") == "agent_completed" and e.get("status") == "success")
        fail_count = sum(1 for e in recent
                        if e.get("event") == "agent_completed" and e.get("status") == "failure")
        if success_count + fail_count > 0:
            metrics["recent_agent_success_rate"] = f"{success_count}/{success_count + fail_count}"

    metrics["total_agents_spawned"] = total_agents
    metrics["total_agents_completed"] = total_completed
    metrics["total_reviews"] = total_reviewers
    metrics["total_reviews_completed"] = total_reviews_done
    metrics["total_ci_checks_tracked"] = total_ci

    return {
        "metrics": metrics,
        "aggregate_stats": stats,
        "daily": daily,
        "repo_stats": repo_stats,
        "recent_events": recent,
    }


@mcp.tool()
def read_article(url: str, wait_seconds: int = 5) -> dict:
    """Fetch a URL using an authenticated browser and return the article text.

    Uses a persistent Chromium profile so sites where you've logged in
    (X.com, etc.) stay authenticated across sessions. Extracts clean article
    text using readability, falling back to raw page text.

    Run `python browser_login.py --url <site>` once to log in interactively.

    Args:
        url: The URL to fetch.
        wait_seconds: Seconds to wait for dynamic content (SPAs) to load. Default 5.
    """
    try:
        from playwright.sync_api import sync_playwright  # noqa: F401
    except ImportError:
        return {"error": "playwright not installed. Run: pip install playwright && python -m playwright install chromium"}

    future = _browser_executor.submit(_fetch_article, url, wait_seconds)
    return future.result(timeout=60)


@mcp.tool()
def agent_status(
    repo: str | None = None,
    issue_number: int | None = None,
) -> dict:
    """Get detailed status of a running (or recently completed) agent.

    Shows pid, log file path, milestones, elapsed time, and last activity
    from the log file. More detailed than dashboard() for a single agent.

    Args:
        repo: Target repo (e.g., 'loqu8/ice'). Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    key = f"{repo}#{number}"
    info = active_agents.get(key)
    if not info:
        return {"error": f"No agent found for {key}. Check dashboard() for active agents."}

    result = {"key": key, **info}

    # Calculate elapsed time
    started = info.get("started")
    if started:
        try:
            start_dt = datetime.fromisoformat(started)
            elapsed = datetime.now(timezone.utc) - start_dt
            result["elapsed_seconds"] = int(elapsed.total_seconds())
            result["elapsed_human"] = str(timedelta(seconds=int(elapsed.total_seconds())))
        except (ValueError, TypeError):
            pass

    # Peek at last few lines of log file for activity hint
    log_file = info.get("log_file")
    if log_file and os.path.isfile(log_file):
        result["log_file_size"] = os.path.getsize(log_file)
        try:
            result["last_modified"] = datetime.fromtimestamp(
                os.path.getmtime(log_file), tz=timezone.utc
            ).isoformat()
            # Time since last log write = proxy for "is it still active?"
            since_modified = datetime.now(timezone.utc) - datetime.fromtimestamp(
                os.path.getmtime(log_file), tz=timezone.utc
            )
            result["seconds_since_last_activity"] = int(since_modified.total_seconds())
        except OSError:
            pass
    elif log_file:
        result["log_file_missing"] = True

    return result


@mcp.tool()
def tail_log(
    repo: str | None = None,
    issue_number: int | None = None,
    lines: int = 30,
) -> dict:
    """Read the last N lines of a running agent's log file.

    Gives immediate insight into what the agent is doing without waiting
    for milestone comments. Useful for checking if an agent is stuck or
    making progress.

    Args:
        repo: Target repo (e.g., 'loqu8/ice'). Defaults from CONDUCTOR_AGENT_REPO env.
        issue_number: Issue number. Defaults from CONDUCTOR_AGENT_ISSUE env.
        lines: Number of lines to return from the end of the log (default 30, max 200).
    """
    resolved = _resolve_repo_issue(repo, issue_number)
    if isinstance(resolved, dict):
        return resolved
    repo, number = resolved

    key = f"{repo}#{number}"
    info = active_agents.get(key)
    if not info:
        return {"error": f"No agent found for {key}. Check dashboard() for active agents."}

    log_file = info.get("log_file")
    if not log_file:
        return {"error": f"No log file recorded for {key}. Agent may have been tracked via acknowledge() only."}

    if not os.path.isfile(log_file):
        return {"error": f"Log file not found: {log_file}"}

    lines = max(1, min(lines, 200))  # clamp to [1, 200]

    try:
        # Read last N lines efficiently (read from end)
        with open(log_file, "rb") as f:
            # Seek to end, read backwards to find N newlines
            f.seek(0, 2)  # seek to end
            file_size = f.tell()
            if file_size == 0:
                return {"key": key, "status": info.get("status"), "lines": [], "message": "Log file is empty"}

            # Read in chunks from the end
            chunk_size = 8192
            data = b""
            while f.tell() > 0 and data.count(b"\n") <= lines:
                to_read = min(chunk_size, f.tell())
                f.seek(f.tell() - to_read)
                data = f.read(to_read) + data
                f.seek(f.tell() - to_read)

        text = data.decode("utf-8", errors="replace")
        tail_lines = text.splitlines()[-lines:]

        return {
            "key": key,
            "status": info.get("status"),
            "log_file": log_file,
            "file_size": file_size,
            "lines_returned": len(tail_lines),
            "tail": tail_lines,
        }
    except OSError as e:
        return {"error": f"Failed to read log file: {e}"}


# ---------------------------------------------------------------------------
# Application assembly
# ---------------------------------------------------------------------------

HEALTH_CHECK_INTERVAL = 300  # 5 minutes


def _health_watchdog():
    """Background thread that periodically verifies Redis connectivity and cleans stale agents."""
    while True:
        time.sleep(HEALTH_CHECK_INTERVAL)
        try:
            if redis_client:
                redis_client.ping()
            else:
                log.warning("WATCHDOG: Redis client is None")
                continue

            # Prune agents that have been "running" for over 1 hour (likely stale from restart)
            now = datetime.now(timezone.utc)
            stale_keys = []
            for key, info in list(active_agents.items()):
                if info.get("status") != "running":
                    continue
                started = info.get("started")
                if not started:
                    continue
                try:
                    age = now - datetime.fromisoformat(started)
                    if age.total_seconds() > 3600:
                        stale_keys.append(key)
                except (ValueError, TypeError):
                    pass
            for key in stale_keys:
                info = active_agents.pop(key, {})
                log.warning("WATCHDOG: Pruned stale agent %s (started %s)", key, info.get("started"))
            if stale_keys:
                _persist_active_agents()
                _refresh_instructions()

        except redis.ConnectionError:
            log.error("WATCHDOG: Redis connection lost!")
        except Exception as e:
            log.warning("WATCHDOG: %s", e)


def _agent_tracker():
    """Background thread that subscribes to agent lifecycle events from Redis."""
    if redis_client is None:
        return
    try:
        pubsub = redis_client.pubsub()
        pubsub.psubscribe("conductor:agent_*")
        log.info("Agent tracker subscribed to conductor:agent_*")
        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
            key = event.get("key", "")
            if "started" in channel:
                active_agents[key] = {
                    "repo": event.get("repo"),
                    "issue_number": event.get("issue_number"),
                    "title": event.get("title"),
                    "pid": event.get("pid"),
                    "repo_dir": event.get("repo_dir"),
                    "log_file": event.get("log_file"),
                    "started": event.get("timestamp"),
                    "status": "running",
                }
                log.info("Agent started: %s (pid %s)", key, event.get("pid"))
                _persist_active_agents()
                _refresh_instructions()
            elif "complete" in channel:
                if key in active_agents:
                    active_agents[key].update({
                        "status": event.get("status", "unknown"),
                        "exit_code": event.get("exit_code"),
                        "elapsed": event.get("elapsed"),
                        "completed": event.get("timestamp"),
                    })
                else:
                    active_agents[key] = {
                        "status": event.get("status", "unknown"),
                        "exit_code": event.get("exit_code"),
                        "elapsed": event.get("elapsed"),
                        "completed": event.get("timestamp"),
                    }
                log.info("Agent completed: %s (%s)", key, event.get("status"))
                _persist_active_agents()
                _refresh_instructions()
    except Exception as e:
        log.error("Agent tracker error: %s", e)


def create_app() -> Starlette:
    """Create Starlette app with webhook routes + MCP (StreamableHTTP + SSE).

    Serves both transports so Claude Code (SSE) and other clients
    (StreamableHTTP) can connect. Follows srclight's dual-transport pattern.
    """
    from contextlib import asynccontextmanager

    # Initialize the session manager (lazy, created on first call)
    mcp_app = mcp.streamable_http_app()

    # SSE transport for Claude Code compatibility
    sse_app = mcp.sse_app()
    sse_routes = [r for r in sse_app.routes if getattr(r, "path", None) in ("/sse", "/messages")]

    @asynccontextmanager
    async def lifespan(app):
        async with mcp.session_manager.run():
            yield

    from mcp.server.fastmcp.server import StreamableHTTPASGIApp
    mcp_handler = StreamableHTTPASGIApp(mcp.session_manager)

    routes = [
        Route("/webhooks", webhook_handler, methods=["POST"]),
        Route("/health", health_handler),
        Route("/", health_handler),
        Route("/mcp", endpoint=mcp_handler),
        *sse_routes,
    ]

    return Starlette(routes=routes, lifespan=lifespan)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    global redis_client

    parser = argparse.ArgumentParser(description="Loqu8 Conductor")
    parser.add_argument("--port", type=int, default=9999)
    parser.add_argument("--redis-url", default="redis://localhost:6379")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    if args.dry_run:
        log.info("Dry-run mode — no Redis")
        redis_client = None
    else:
        try:
            redis_client = redis.Redis.from_url(args.redis_url)
            redis_client.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)

    _refresh_instructions()
    log.info("Conductor starting on http://127.0.0.1:%d", args.port)
    log.info("  Webhooks: /webhooks")
    log.info("  MCP:      /mcp (StreamableHTTP)")
    log.info("  Health:   /health")
    log.info("  Repos:    %d monitored", len(_get_repo_dirs()))

    # Restore active agents from Redis and start lifecycle tracker + watchdog
    if redis_client:
        _restore_active_agents()
        tracker = threading.Thread(target=_agent_tracker, daemon=True)
        tracker.start()
        watchdog = threading.Thread(target=_health_watchdog, daemon=True)
        watchdog.start()

    app = create_app()
    uvicorn.run(app, host="127.0.0.1", port=args.port, log_level="warning")


if __name__ == "__main__":
    main()