worker.py

import os
import shutil
import subprocess
import time
import logging

from orchestrator import analyze_result, suggest_labels
from telegram_notifier import (
    notify_pipeline_started,
    notify_subtasks_created,
    notify_stage_started,
    notify_artifact_done,
    notify_pr_created,
    notify_testing_done,
    notify_all_done,
    notify_merged,
    notify_error,
)
from jira_client import JiraClient
from github_client import GitHubClient
from dependency_tracker import (
    collect_artifact_context,
    trigger_next_stages,
    all_stages_done,
)
from prompts import build_stage_prompt
from config import (
    GITHUB_TOKEN_TARGET,
    GITHUB_REPO,
    GITHUB_TOKEN_BRIDGE,
    GITHUB_REPO_BRIDGE,
    STAGE_BRANCH,
    JOB_TIMEOUT_MINUTES,
    MAX_RETRIES,
    RETRY_DELAY_MINUTES,
    ARTIFACT_STAGES,
    CODE_STAGES,
    STATUS_DONE,
    STATUS_IN_REVIEW,
    STATUS_IN_PROGRESS,
    STATUS_MERGE,
    JIRA_PROJECT_KEY,
    PIPELINE_LABEL_PREFIX,
    ALL_STAGES,
    STAGE_PREREQUISITES,
    AUTO_TRANSITION_ON_COMPLETE,
)

logger = logging.getLogger("pipeline.worker")
jira = JiraClient()
github = GitHubClient()


# ── Repo routing ──────────────────────────────────────────────────────────────

def _get_repo_config(job: dict) -> dict:
    """Return {"repo": ..., "token": ...} based on job labels.

    If the job has label ``repo:bridge`` → secondary repo/token,
    otherwise the default target repo/token.
    """
    labels = job.get("labels", [])
    if "repo:bridge" in labels:
        return {"repo": GITHUB_REPO_BRIDGE, "token": GITHUB_TOKEN_BRIDGE}
    return {"repo": GITHUB_REPO, "token": GITHUB_TOKEN_TARGET}


def _github_for_repo(repo_cfg: dict) -> GitHubClient:
    """Return a GitHubClient configured for the given repo/token."""
    client = GitHubClient()
    client.repo = repo_cfg["repo"]
    client.token = repo_cfg["token"]
    client.headers = {
        "Authorization": f"Bearer {repo_cfg['token']}",
        "Accept": "application/vnd.github.v3+json",
    }
    return client


# ── Helpers ───────────────────────────────────────────────────────────────────

def _clone_repo(work_dir: str, branch_name: str, repo_cfg: dict | None = None) -> None:
    """Clone repo and create a new local branch (no remote tracking)."""
    if repo_cfg is None:
        repo_cfg = {"repo": GITHUB_REPO, "token": GITHUB_TOKEN_TARGET}
    repo_url = f"https://x-access-token:{repo_cfg['token']}@github.com/{repo_cfg['repo']}.git"
    result = subprocess.run(
        ["git", "clone", "--depth=1", repo_url, work_dir],
        capture_output=True,
        timeout=120,
    )
    if result.returncode != 0:
        stderr = result.stderr.decode(errors="replace").strip()
        raise Exception(f"git clone failed (rc={result.returncode}): {stderr[:400]}")
    subprocess.run(
        ["git", "checkout", "-b", branch_name],
        cwd=work_dir,
        check=True,
        capture_output=True,
    )


def _clone_repo_with_branch(work_dir: str, branch_name: str, repo_cfg: dict | None = None) -> None:
    """Clone repo; if branch_name exists on remote, check it out.
    Otherwise create a new branch. This allows code stages to pick up
    artifacts committed by earlier artifact stages."""
    if repo_cfg is None:
        repo_cfg = {"repo": GITHUB_REPO, "token": GITHUB_TOKEN_TARGET}
    repo_url = (
        f"https://x-access-token:{repo_cfg['token']}"
        f"@github.com/{repo_cfg['repo']}.git"
    )
    # Try cloning the existing branch first
    result = subprocess.run(
        ["git", "clone", "--depth=50", "-b", branch_name, repo_url, work_dir],
        capture_output=True,
        timeout=120,
    )
    if result.returncode == 0:
        logger.info("Cloned existing branch %s", branch_name)
        return
    # Branch doesn't exist on remote — clone default and create it
    result = subprocess.run(
        ["git", "clone", "--depth=1", repo_url, work_dir],
        capture_output=True,
        timeout=120,
    )
    if result.returncode != 0:
        stderr = result.stderr.decode(errors="replace").strip()
        raise Exception(f"git clone failed (rc={result.returncode}): {stderr[:400]}")
    subprocess.run(
        ["git", "checkout", "-b", branch_name],
        cwd=work_dir,
        check=True,
        capture_output=True,
    )


# Errors that should trigger a retry (transient / server-side)
_RETRYABLE_MARKERS = (
    # Rate limits
    "rate limit", "429", "overloaded", "exceeded your current quota",
    # Server errors
    "500", "internal server error", "api_error",
    "502", "bad gateway",
    "503", "service unavailable",
    "529", "overloaded",
    # Auth (token expired — refresh_token.py should handle, but retry just in case)
    "401", "authentication_error", "token has expired",
    # Network / transient
    "connection error", "timeout", "econnreset", "econnrefused",
    "socket hang up", "fetch failed",
)

# Longer delay for rate limits, shorter for server errors
_RATE_LIMIT_MARKERS = ("rate limit", "429", "overloaded", "exceeded your current quota")


def _run_claude(prompt: str, work_dir: str, job: dict) -> subprocess.CompletedProcess:
    """Run Claude Code Opus via Popen; stores process in job["process"] for cancellation."""
    # Refresh OAuth token before each run (non-blocking)
    # 1. Try our own refresh_token.py
    try:
        from refresh_token import main as _refresh_token
        _refresh_token()
    except Exception:
        pass
    # 2. Warm up Claude CLI's built-in OAuth refresh
    #    (claude --version doesn't trigger it, but a real prompt does)
    try:
        subprocess.run(
            ["claude", "-p", "ok", "--max-turns", "1", "--output-format", "text"],
            cwd=work_dir, capture_output=True, text=True, timeout=30,
        )
    except Exception:
        pass

    proc = subprocess.Popen(
        [
            "claude", "-p", prompt,
            "--model", "claude-opus-4-6",
            "--output-format", "text",
            "--max-turns", "50",
        ],
        cwd=work_dir,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    job["process"] = proc
    try:
        stdout, stderr = proc.communicate(timeout=JOB_TIMEOUT_MINUTES * 60)
    except subprocess.TimeoutExpired:
        proc.kill()
        stdout, stderr = proc.communicate()
        raise Exception(f"Claude Code timed out after {JOB_TIMEOUT_MINUTES}m")
    finally:
        job.pop("process", None)
    return subprocess.CompletedProcess(proc.args, proc.returncode, stdout, stderr)


def _sleep_interruptible(seconds: int, job: dict) -> None:
    """Sleep in 5s chunks, aborting early if job is cancelled."""
    end = time.time() + seconds
    while time.time() < end:
        if job.get("cancelled"):
            raise Exception("Cancelled during retry wait")
        time.sleep(5)


def _run_claude_with_retry(prompt: str, work_dir: str, job: dict) -> subprocess.CompletedProcess:
    """Run Claude Code with automatic retry on transient errors.

    Retries on: rate limits (429), server errors (500/502/503),
    network issues (connection reset, timeout), API errors.
    Rate limits get a longer delay; server errors get a shorter one.
    """
    for attempt in range(1, MAX_RETRIES + 1):
        if job.get("cancelled"):
            raise Exception("Cancelled")
        result = _run_claude(prompt, work_dir, job)
        if result.returncode == 0:
            return result

        # Check both stdout and stderr for error markers
        combined = ((result.stdout or "") + (result.stderr or "")).lower()
        is_retryable = any(m in combined for m in _RETRYABLE_MARKERS)
        is_rate_limit = any(m in combined for m in _RATE_LIMIT_MARKERS)

        if is_retryable and attempt < MAX_RETRIES:
            # Rate limits → full delay; server errors → shorter delay (2 min)
            if is_rate_limit:
                delay_min = RETRY_DELAY_MINUTES
                reason = "Rate limit"
            else:
                delay_min = min(RETRY_DELAY_MINUTES, 2)
                reason = "Server error"

            logger.warning(
                "[%s] %s, attempt %d/%d, waiting %dm",
                job["issue_key"], reason, attempt, MAX_RETRIES, delay_min,
            )
            notify_error(
                job["issue_key"], job.get("stage", "?"),
                f"{reason} — retry {attempt}/{MAX_RETRIES} in {delay_min}m",
                job.get("jira_domain", ""),
            )
            _sleep_interruptible(delay_min * 60, job)
            continue

        # Non-retryable error or last attempt
        out = (result.stdout or "")[:300]
        err = (result.stderr or "")[:300]
        raise Exception(f"Claude Code rc={result.returncode}\nstdout: {out}\nstderr: {err}")
    raise Exception(f"Claude Code failed after {MAX_RETRIES} retries")


def _git_changed_files(work_dir: str) -> list[str]:
    diff = subprocess.run(
        ["git", "diff", "--name-only"],
        cwd=work_dir, capture_output=True, text=True,
    )
    untracked = subprocess.run(
        ["git", "ls-files", "--others", "--exclude-standard"],
        cwd=work_dir, capture_output=True, text=True,
    )
    return [f for f in (diff.stdout + untracked.stdout).strip().split("\n") if f]


def _relaunch_subtask(sub: dict, parent_key: str, stage: str) -> None:
    """Re-launch a dead subtask (In Progress but no active job).

    Creates a new job directly without waiting for webhook.
    """
    import uuid
    import time as _time

    sub_key = sub["key"]
    job_id = str(uuid.uuid4())[:8]

    # Fetch full subtask info for the job dict
    try:
        full = jira.get_issue(sub_key)
        full_fields = full.get("fields", {})
    except Exception:
        full_fields = {}

    parent_ref = full_fields.get("parent", {})
    job = {
        "job_id": job_id,
        "issue_key": sub_key,
        "key": sub_key,
        "parent_key": parent_ref.get("key", parent_key),
        "summary": full_fields.get("summary", sub.get("summary", "")),
        "description": full_fields.get("description", {}),
        "description_text": "",
        "issue_type": full_fields.get("issuetype", {}).get("name", "Sub-task"),
        "stage": stage,
        "trigger": "In Progress",
        "jira_domain": f"{os.environ.get('JIRA_DOMAIN', '')}.atlassian.net",
        "priority": full_fields.get("priority", {}).get("name", "Medium"),
        "labels": full_fields.get("labels", sub.get("labels", [])),
        "components": [
            c.get("name", "") if isinstance(c, dict) else c
            for c in full_fields.get("components", [])
        ],
        "status": "queued",
        "created": _time.time(),
    }

    from main import _launch_job, active_pipelines
    # Ensure parent is in active pipelines
    active_pipelines.add(parent_key)
    _launch_job(job)
    logger.info("[%s] re-launched dead stage %s as job %s", parent_key, stage, job_id)


# ── Planning job: break feature into epics and tasks ─────────────────────────

def run_plan_job(job: dict) -> None:
    """When a PLAN: task moves to In Progress: Claude Code reads the codebase
    and breaks the feature/project into epics and tasks in Jira.

    Flow:
    1. Clone repo
    2. Claude Code analyzes codebase + task description
    3. Outputs JSON with epics and tasks
    4. Pipeline creates epics and tasks in Jira
    5. Original task → Done
    """
    _ensure_description_text(job)
    issue_key = job["issue_key"]
    job_id = job["job_id"]
    work_dir = f"/tmp/pipeline-work/{job_id}"

    try:
        jira_domain = job.get("jira_domain", "")
        from telegram_notifier import _send
        _send(
            f"📝 <b>Planning started</b>\n"
            f"Task: <a href='https://{jira_domain}/browse/{issue_key}'>{issue_key}</a>\n"
            f"{job['summary']}\n"
            f"Claude Code is analyzing the codebase..."
        )

        jira.add_comment(
            issue_key,
            f"🤖 Planning started. Claude Code is reading the codebase "
            f"and breaking down the feature into epics and tasks.\n"
            f"Job: {job_id}",
        )

        from prompts import build_plan_prompt
        prompt = build_plan_prompt(job)

        repo_cfg = _get_repo_config(job)
        os.makedirs(work_dir, exist_ok=True)
        _clone_repo(work_dir, f"plan/{issue_key.lower()}", repo_cfg)

        start = time.time()
        if job.get("cancelled"):
            raise Exception("Cancelled")

        result = _run_claude_with_retry(prompt, work_dir, job)
        duration = int(time.time() - start)

        if result.returncode != 0:
            raise Exception(
                f"Claude Code rc={result.returncode}: {result.stderr[:500]}"
            )

        # Parse the JSON output
        import json
        output = result.stdout.strip()
        # Try to extract JSON from the output (Claude might add text around it)
        json_start = output.find("{")
        json_end = output.rfind("}") + 1
        if json_start == -1 or json_end == 0:
            raise Exception("Claude Code did not output valid JSON")

        plan = json.loads(output[json_start:json_end])

        # Check if business analysis rejected the feature
        if plan.get("rejected"):
            reason = plan.get("reason", "No reason provided.")
            jira.add_comment(
                issue_key,
                f"🤖 **Feature rejected after business analysis.**\n\n"
                f"{reason}\n\n"
                f"No epics or tasks were created.\n"
                f"⏱ {duration // 60}m {duration % 60}s | Job: {job_id}",
            )
            jira.transition(issue_key, STATUS_DONE)
            _send(
                f"🚫 <b>Feature rejected</b>\n"
                f"<a href='https://{jira_domain}/browse/{issue_key}'>{issue_key}</a>\n"
                f"{reason[:200]}"
            )
            logger.info("[%s] Plan rejected: %s", issue_key, reason[:200])
            return

        epics = plan.get("epics", [])

        if not epics:
            jira.add_comment(
                issue_key,
                "🤖 Claude Code could not break this down into epics. "
                "Try adding more detail to the task description.",
            )
            jira.transition(issue_key, STATUS_DONE)
            return

        # Create epics and tasks in Jira
        from config import JIRA_PROJECT_KEY
        created_epics = []
        total_tasks = 0

        for epic_data in epics:
            epic_title = epic_data.get("title", "Untitled epic")
            epic_desc = epic_data.get("description", "")
            tasks = epic_data.get("tasks", [])

            # Create epic
            epic_body = {
                "fields": {
                    "project": {"key": JIRA_PROJECT_KEY},
                    "summary": epic_title,
                    "description": {
                        "version": 1,
                        "type": "doc",
                        "content": [{
                            "type": "paragraph",
                            "content": [{"type": "text", "text": epic_desc}],
                        }],
                    },
                    "issuetype": {"name": "Epic"},
                }
            }
            import httpx as _httpx
            r = _httpx.post(
                f"{jira.base_url}/rest/api/3/issue",
                headers=jira.headers,
                json=epic_body,
                timeout=10,
            )
            if not r.is_success:
                logger.warning("[%s] Failed to create epic '%s': %s",
                               issue_key, epic_title, r.text[:200])
                continue

            epic_key = r.json()["key"]
            created_epics.append({"key": epic_key, "title": epic_title, "tasks": []})

            # Create tasks under this epic
            for task_data in tasks:
                task_title = task_data.get("title", "Untitled task")
                task_desc = task_data.get("description", "")
                task_labels = task_data.get("labels", [])

                task_body = {
                    "fields": {
                        "project": {"key": JIRA_PROJECT_KEY},
                        "parent": {"key": epic_key},
                        "summary": task_title,
                        "description": {
                            "version": 1,
                            "type": "doc",
                            "content": [{
                                "type": "paragraph",
                                "content": [{"type": "text", "text": task_desc}],
                            }],
                        },
                        "issuetype": {"name": "Task"},
                        "labels": task_labels,
                    }
                }
                r = _httpx.post(
                    f"{jira.base_url}/rest/api/3/issue",
                    headers=jira.headers,
                    json=task_body,
                    timeout=10,
                )
                if r.is_success:
                    task_key = r.json()["key"]
                    created_epics[-1]["tasks"].append(task_key)
                    total_tasks += 1
                else:
                    logger.warning("[%s] Failed to create task '%s': %s",
                                   issue_key, task_title, r.text[:200])

        # Build summary comment
        summary_lines = [
            f"🤖 **Planning complete** — {len(created_epics)} epics, "
            f"{total_tasks} tasks created.\n",
        ]
        for epic in created_epics:
            epic_url = f"https://{jira_domain}/browse/{epic['key']}"
            summary_lines.append(
                f"### [{epic['key']}]({epic_url}): {epic['title']}"
            )
            for task_key in epic["tasks"]:
                task_url = f"https://{jira_domain}/browse/{task_key}"
                summary_lines.append(f"  - [{task_key}]({task_url})")
            summary_lines.append("")

        summary_lines.append(
            f"⏱ {duration // 60}m {duration % 60}s | Job: {job_id}\n\n"
            "Move any task to **In Progress** to start the dev pipeline."
        )

        jira.add_comment(issue_key, "\n".join(summary_lines))
        jira.transition(issue_key, STATUS_DONE)

        _send(
            f"📝 <b>Planning complete!</b>\n"
            f"<a href='https://{jira_domain}/browse/{issue_key}'>{issue_key}</a>\n"
            f"{len(created_epics)} epics, {total_tasks} tasks\n"
            f"⏱ {duration // 60}m {duration % 60}s"
        )
        logger.info("[%s] Plan complete: %d epics, %d tasks (%ds)",
                    issue_key, len(created_epics), total_tasks, duration)

    except Exception as e:
        logger.error("[%s] plan FAIL: %s", issue_key, e)
        notify_error(issue_key, "planning", str(e), job.get("jira_domain", ""))
        try:
            jira.add_comment(
                issue_key,
                f"❌ Planning error: {str(e)[:500]}\nJob: {job_id}",
            )
        except Exception:
            pass
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


# ── Setup job: create pipeline subtasks for a parent task ────────────────────

_STAGE_SUMMARIES = {
    "sys-analysis":  "System Analysis",
    "architecture":  "Architecture Decision",
    "development":   "Development",
    "testing":       "Testing",
}


def run_setup_job(job: dict) -> None:
    """When a parent task moves to In Progress: create 4 pipeline subtasks,
    then immediately transition the ones with no prerequisites to In Progress.

    Idempotent: if pipeline subtasks already exist, skips creation.
    """
    issue_key = job["issue_key"]
    job_id = job["job_id"]

    try:
        jira_domain = job.get("jira_domain", "")
        notify_pipeline_started(issue_key, job["summary"], jira_domain)

        # Auto-tag parent with domain/service labels
        description_text = job.get("description_text", "")
        auto_labels = suggest_labels(job["summary"], description_text)
        if auto_labels:
            jira.add_labels(issue_key, auto_labels)

        # Check which stages already have subtasks
        existing = jira.get_subtasks(issue_key)
        from dependency_tracker import get_stage
        existing_stages = {
            get_stage(sub["labels"])
            for sub in existing
            if get_stage(sub["labels"])
        }

        created: dict[str, str] = {}  # stage → subtask key

        for stage in ALL_STAGES:
            if stage in existing_stages:
                logger.info("[%s] subtask for stage %s already exists, skipping", issue_key, stage)
                continue

            label = f"{PIPELINE_LABEL_PREFIX}{stage}"
            summary = f"[{issue_key}] {_STAGE_SUMMARIES.get(stage, stage)}"
            subtask_key = jira.create_subtask(
                parent_key=issue_key,
                summary=summary,
                labels=[label] + (auto_labels or []),
                project_key=JIRA_PROJECT_KEY,
            )
            created[stage] = subtask_key
            logger.info("[%s] created subtask %s for stage %s", issue_key, subtask_key, stage)

        if created:
            notify_subtasks_created(issue_key, list(created.values()),
                                    auto_labels or [], jira_domain)

        jira.add_comment(
            issue_key,
            "🤖 Pipeline ready.\n"
            + "\n".join(
                f"• {_STAGE_SUMMARIES.get(s, s)}: {k}"
                for s, k in created.items()
            )
            + ("\n\nStages with no dependencies start automatically." if created else ""),
        )

        # Trigger stages with no prerequisites → transition to In Progress
        # Jira will fire a webhook for each, which will process them
        all_subtasks = jira.get_subtasks(issue_key)
        for sub in all_subtasks:
            stage = get_stage(sub["labels"])
            if not stage or STAGE_PREREQUISITES.get(stage):
                continue
            sub_status = sub.get("status", "").lower()
            from jira_client import _status_matches

            # Already done or in review → skip
            if (_status_matches(sub["status"], STATUS_DONE)
                    or _status_matches(sub["status"], STATUS_IN_REVIEW)):
                logger.info("[%s] stage %s (%s) already '%s', skipping",
                            issue_key, stage, sub["key"], sub["status"])
                continue

            # "In Progress" but no active job → dead stage, re-launch directly
            if _status_matches(sub["status"], STATUS_IN_PROGRESS):
                from main import jobs
                has_active_job = any(
                    j["issue_key"] == sub["key"] and j["status"] in ("queued", "running")
                    for j in jobs.values()
                )
                if has_active_job:
                    logger.info("[%s] stage %s (%s) has active job, skipping",
                                issue_key, stage, sub["key"])
                    continue
                # Dead stage: already In Progress but no job running — re-launch
                logger.info("[%s] stage %s (%s) is '%s' with no active job, re-launching",
                            issue_key, stage, sub["key"], sub["status"])
                _relaunch_subtask(sub, issue_key, stage)
                continue

            # To Do → transition to In Progress (webhook will trigger job)
            ok = jira.transition(sub["key"], STATUS_IN_PROGRESS)
            if ok:
                logger.info("[%s] auto-started stage %s (%s)", issue_key, stage, sub["key"])
            else:
                available = jira.get_transitions(sub["key"])
                msg = (f"⚠️ Cannot transition {sub['key']} to '{STATUS_IN_PROGRESS}'.\n"
                       f"Available transitions: {available}")
                logger.warning(msg)
                from telegram_notifier import _send
                _send(msg)

        # Also check prerequisite stages: if their deps are all Done,
        # trigger them too (handles restart recovery)
        all_subtasks = jira.get_subtasks(issue_key)
        stage_map = {}
        for sub in all_subtasks:
            s = get_stage(sub["labels"])
            if s:
                stage_map[s] = sub

        for stage, prereqs in STAGE_PREREQUISITES.items():
            if not prereqs:
                continue  # already handled above
            sub = stage_map.get(stage)
            if not sub:
                continue
            # Skip if already done/in review
            if (_status_matches(sub["status"], STATUS_DONE)
                    or _status_matches(sub["status"], STATUS_IN_REVIEW)):
                continue
            # Check if all prerequisites are done
            all_done = all(
                _status_matches(stage_map.get(p, {}).get("status", ""), STATUS_DONE)
                for p in prereqs
            )
            if not all_done:
                continue
            # Prerequisites met — check if needs (re)launch
            if _status_matches(sub["status"], STATUS_IN_PROGRESS):
                from main import jobs as all_jobs
                has_active = any(
                    j["issue_key"] == sub["key"]
                    and j["status"] in ("queued", "running")
                    for j in all_jobs.values()
                )
                if has_active:
                    continue
                logger.info("[%s] re-launching dead stage %s (%s)",
                            issue_key, stage, sub["key"])
                _relaunch_subtask(sub, issue_key, stage)
            else:
                # To Do → start
                logger.info("[%s] prerequisites met for %s, triggering",
                            issue_key, stage)
                ok = jira.transition(sub["key"], STATUS_IN_PROGRESS)
                if ok:
                    logger.info("[%s] auto-started stage %s (%s)",
                                issue_key, stage, sub["key"])

    except Exception as e:
        logger.error("[%s] setup FAIL: %s", issue_key, e)
        try:
            jira.add_comment(issue_key, f"❌ Pipeline setup error: {str(e)[:500]}\nJob: {job_id}")
        except Exception:
            pass


# ── Artifact stage (sys-analysis, architecture) ───────────────────────────────

_ARTIFACT_FILENAMES = {
    "sys-analysis": "SYSTEM_ANALYSIS",
    "architecture": "ARCHITECTURE_DECISION",
}


def _artifact_filename(stage: str, parent_key: str) -> str:
    """E.g. docs/decisions/SYSTEM_ANALYSIS_PROJ-38.md"""
    from config import ARTIFACTS_DIR
    base = _ARTIFACT_FILENAMES.get(stage, stage.upper())
    return f"{ARTIFACTS_DIR}/{base}_{parent_key}.md"


def _ensure_description_text(job: dict) -> None:
    """Enrich job with parent summary and description.

    Subtasks only have a pipeline-generated name like "[PROJ-38] Development".
    We need the parent's actual summary and description to give Claude Code
    enough context.
    """
    from orchestrator import parse_adf_to_text

    # Convert own ADF description
    if not job.get("description_text") and job.get("description"):
        job["description_text"] = parse_adf_to_text(job["description"])

    # For subtasks: fetch parent info (summary + description + epic context)
    if job.get("parent_key") and job["parent_key"] != job.get("issue_key"):
        try:
            parent = jira.get_issue(job["parent_key"])
            parent_fields = parent.get("fields", {})

            # Store parent summary (the actual task name)
            parent_summary = parent_fields.get("summary", "")
            if parent_summary:
                job["parent_summary"] = parent_summary

            # Fetch parent description if we don't have one
            if not job.get("description_text"):
                parent_desc = parent_fields.get("description", {})
                if parent_desc:
                    job["description_text"] = parse_adf_to_text(parent_desc)

            # Try to get epic context too
            epic_ref = parent_fields.get("parent", {})
            if epic_ref and epic_ref.get("key"):
                try:
                    epic = jira.get_issue(epic_ref["key"])
                    epic_fields = epic.get("fields", {})
                    epic_summary = epic_fields.get("summary", "")
                    epic_desc = epic_fields.get("description", {})
                    parts = []
                    if epic_summary:
                        parts.append(f"Epic: {epic_summary}")
                    if epic_desc:
                        parts.append(parse_adf_to_text(epic_desc))
                    if parts:
                        job["epic_context"] = "\n".join(parts)
                except Exception:
                    pass

        except Exception as e:
            logger.warning("Failed to fetch parent info: %s", e)


def run_artifact_stage(job: dict) -> None:
    """Run sys-analysis or architecture via Claude Code (git clone + claude CLI).

    Claude Code reads the codebase and writes SYSTEM_ANALYSIS.md or
    ARCHITECTURE_DECISION.md. Pipeline reads the file, posts it to Jira,
    and marks the subtask Done.
    """
    _ensure_description_text(job)
    issue_key = job["issue_key"]
    parent_key = job["parent_key"]
    stage = job["stage"]
    job_id = job["job_id"]
    work_dir = f"/tmp/pipeline-work/{job_id}"

    try:
        jira.transition(issue_key, STATUS_IN_PROGRESS)
        jira.add_comment(issue_key, f"🤖 Stage {stage} started (Claude Code). Job: {job_id}")
        notify_stage_started(stage, issue_key, parent_key, job.get("jira_domain", ""))

        auto_labels = suggest_labels(job["summary"], job.get("description_text", ""))
        if auto_labels:
            jira.add_labels(issue_key, auto_labels)
            if parent_key != issue_key:
                jira.add_labels(parent_key, auto_labels)

        artifact_context = collect_artifact_context(parent_key, jira)
        prompt = build_stage_prompt(job, artifact_context)

        repo_cfg = _get_repo_config(job)
        logger.info("[%s] Cloning for artifact stage %s (repo: %s)", issue_key, stage, repo_cfg["repo"])
        os.makedirs(work_dir, exist_ok=True)
        _clone_repo(work_dir, f"analysis/{issue_key.lower()}", repo_cfg)

        start = time.time()
        if job.get("cancelled"):
            raise Exception("Cancelled")
        logger.info("[%s] Claude Code: running stage %s", issue_key, stage)
        result = _run_claude_with_retry(prompt, work_dir, job)
        duration = int(time.time() - start)

        if result.returncode != 0:
            raise Exception(
                f"Claude Code rc={result.returncode}: {result.stderr[:500]}"
            )

        artifact_fname = _artifact_filename(stage, parent_key)
        # Ensure artifacts directory exists
        from config import ARTIFACTS_DIR
        os.makedirs(os.path.join(work_dir, ARTIFACTS_DIR), exist_ok=True)
        # Claude writes the generic name; rename to task-specific
        generic_fname = _ARTIFACT_FILENAMES.get(stage, stage.upper()) + ".md"
        generic_path = os.path.join(work_dir, generic_fname)
        artifact_path = os.path.join(work_dir, artifact_fname)
        if os.path.exists(generic_path) and generic_fname != artifact_fname:
            os.rename(generic_path, artifact_path)
        if os.path.exists(artifact_path):
            with open(artifact_path, encoding="utf-8") as fh:
                artifact_text = fh.read()
        else:
            artifact_text = result.stdout.strip() or "Artifact not created — check manually."
            logger.warning("[%s] %s not found, using stdout", issue_key, artifact_fname)
            # Write stdout as artifact so it gets committed
            if artifact_text and artifact_text != "Artifact not created — check manually.":
                with open(artifact_path, "w", encoding="utf-8") as fh:
                    fh.write(artifact_text)

        # Commit and push artifact to feature branch
        branch_name = f"feature/{parent_key.lower()}"
        try:
            subprocess.run(["git", "checkout", "-B", branch_name], cwd=work_dir,
                           check=True, capture_output=True)
            subprocess.run(["git", "add", artifact_fname], cwd=work_dir,
                           check=True, capture_output=True)
            subprocess.run(
                ["git", "commit", "-m",
                 f"{parent_key}: {_STAGE_SUMMARIES.get(stage, stage)} [{stage}]\n\n"
                 "Automated by Claudev"],
                cwd=work_dir, check=True, capture_output=True,
            )
            subprocess.run(
                ["git", "push", "origin", branch_name],
                cwd=work_dir, check=True, capture_output=True, timeout=60,
            )
            logger.info("[%s] pushed artifact %s to %s", issue_key, artifact_fname, branch_name)
        except Exception as e:
            logger.warning("[%s] failed to push artifact: %s", issue_key, e)

        jira_domain = job.get("jira_domain", "")
        parent_url = f"https://{jira_domain}/browse/{parent_key}"
        github_url = f"https://github.com/{repo_cfg['repo']}/blob/{branch_name}/{artifact_fname}"

        jira.add_comment(
            issue_key,
            f"🔗 Task: [{parent_key}]({parent_url})\n\n"
            f"## Stage result: {stage}\n\n{artifact_text[:24000]}\n\n"
            f"---\n⏱ {duration // 60}m {duration % 60}s | Job: {job_id}",
        )
        jira.add_comment(
            parent_key,
            f"✅ Stage **{stage}** complete (Claude Code).\n"
            f"📄 [{artifact_fname}]({github_url})\n"
            f"⏱ {duration // 60}m {duration % 60}s",
        )

        jira.transition(issue_key, STATUS_DONE)
        logger.info("[%s] stage %s done (%ds)", issue_key, stage, duration)
        notify_artifact_done(stage, issue_key, parent_key,
                             job.get("jira_domain", ""), duration)

        triggered = trigger_next_stages(parent_key, stage, jira)
        if triggered:
            jira.add_comment(
                issue_key,
                f"🤖 Automatically triggered stages: {', '.join(triggered)}",
            )

    except Exception as e:
        logger.error("[%s] artifact stage FAIL: %s", issue_key, e)
        notify_error(issue_key, stage, str(e), job.get("jira_domain", ""))
        try:
            jira.add_comment(
                issue_key,
                f"❌ Pipeline error (stage={stage}): {str(e)[:500]}\nJob: {job_id}",
            )
        except Exception:
            pass
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


# ── Code stage (development, testing) ─────────────────────────────────────────

def run_code_stage(job: dict) -> None:
    """Run development or testing stage.

    Claude Code writes actual code. Pipeline creates a PR (development)
    or pushes to dev branch (testing) and transitions to In Review / Done.
    """
    _ensure_description_text(job)
    issue_key = job["issue_key"]
    parent_key = job["parent_key"]
    stage = job["stage"]
    job_id = job["job_id"]
    work_dir = f"/tmp/pipeline-work/{job_id}"

    try:
        jira.transition(issue_key, STATUS_IN_PROGRESS)
        jira.add_comment(issue_key, f"🤖 Stage {stage} started. Job: {job_id}")
        notify_stage_started(stage, issue_key, parent_key, job.get("jira_domain", ""))

        # Auto-tag issue and parent with domain/service labels
        auto_labels = suggest_labels(job["summary"], job.get("description_text", ""))
        if auto_labels:
            jira.add_labels(issue_key, auto_labels)
            if parent_key != issue_key:
                jira.add_labels(parent_key, auto_labels)

        artifact_context = collect_artifact_context(parent_key, jira)
        prompt = build_stage_prompt(job, artifact_context)

        repo_cfg = _get_repo_config(job)
        branch_name = f"feature/{parent_key.lower()}"
        logger.info("[%s] Cloning for code stage %s (branch %s, repo: %s)",
                    issue_key, stage, branch_name, repo_cfg["repo"])
        os.makedirs(work_dir, exist_ok=True)
        _clone_repo_with_branch(work_dir, branch_name, repo_cfg)

        start = time.time()
        if job.get("cancelled"):
            raise Exception("Cancelled")
        logger.info("[%s] Running Claude Code (stage=%s)", issue_key, stage)
        result = _run_claude_with_retry(prompt, work_dir, job)
        duration = int(time.time() - start)
        logger.info("[%s] Claude Code: %ds rc=%d", issue_key, duration, result.returncode)

        if result.returncode != 0:
            raise Exception(
                f"Claude Code rc={result.returncode}: {result.stderr[:500]}"
            )

        changed = _git_changed_files(work_dir)
        if not changed:
            jira.add_comment(
                issue_key,
                "🤖 Claude Code made no changes. Task needs clarification.",
            )
            jira.transition(issue_key, "Ready for Dev")
            return

        analysis = analyze_result(result.stdout, changed)

        logger.info("[%s] Pushing %s", issue_key, branch_name)
        subprocess.run(["git", "add", "-A"], cwd=work_dir, check=True)
        subprocess.run(
            [
                "git", "commit", "-m",
                f"{issue_key}: {job['summary']} [{stage}]\n\n"
                "Automated by Claudev",
            ],
            cwd=work_dir, check=True, capture_output=True,
        )
        subprocess.run(
            ["git", "push", "origin", branch_name],
            cwd=work_dir, check=True, capture_output=True, timeout=60,
        )

        if stage == "development":
            jira_domain = job.get("jira_domain", os.environ.get("JIRA_DOMAIN", "x"))
            files_list = "\n".join(
                "- " + f for f in analysis.get("files_changed", changed)
            )
            pr_body = (
                f"## {issue_key}: {job['summary']}\n\n"
                f"**Jira:** https://{jira_domain}/browse/{parent_key}\n"
                f"**Subtask:** {issue_key} (stage: {stage})\n"
                "**Automated by:** Claudev\n\n"
                f"### Changes\n{analysis.get('summary_ru', 'N/A')}\n\n"
                f"### Files\n{files_list}\n\n"
                f"### Tests: {analysis.get('tests_status', '?')}\n"
            )
            gh = _github_for_repo(repo_cfg)
            pr = gh.create_pr(
                head=branch_name,
                base=STAGE_BRANCH,
                title=f"{issue_key}: {job['summary']}",
                body=pr_body,
            )
            gh.add_labels(pr["number"], ["automated", "claude-code", stage])
            notify_pr_created(issue_key, parent_key, pr["html_url"],
                              job.get("jira_domain", ""), len(changed))

            concerns = (
                "\n⚠️ " + "; ".join(analysis["concerns"])
                if analysis.get("concerns") else ""
            )
            jira.transition(issue_key, STATUS_DONE)
            jira.add_comment(
                issue_key,
                f"🤖 PR created: {pr['html_url']}\n"
                f"Files: {len(changed)} | "
                f"Tests: {analysis.get('tests_status', '?')} | "
                f"Duration: {duration // 60}m {duration % 60}s\n"
                f"{analysis.get('summary_ru', '')}{concerns}",
            )
            logger.info("[%s] Done! PR #%s", issue_key, pr["number"])

        else:  # testing
            jira.transition(issue_key, STATUS_DONE)
            jira.add_comment(
                issue_key,
                f"🤖 Tests written and pushed to {branch_name}.\n"
                f"Files: {len(changed)} | "
                f"Status: {analysis.get('tests_status', '?')} | "
                f"Duration: {duration // 60}m {duration % 60}s\n"
                f"{analysis.get('summary_ru', '')}",
            )
            notify_testing_done(issue_key, parent_key, job.get("jira_domain", ""), duration)
            logger.info("[%s] Testing stage done (%ds)", issue_key, duration)

        if all_stages_done(parent_key, jira):
            # Auto-transition parent to In Review / Ready for Test
            if AUTO_TRANSITION_ON_COMPLETE:
                jira.transition(parent_key, AUTO_TRANSITION_ON_COMPLETE)
                logger.info("[%s] auto-transitioned → %s", parent_key, AUTO_TRANSITION_ON_COMPLETE)
            jira.add_comment(
                parent_key,
                "🎉 All pipeline stages complete!\n"
                "sys-analysis ✅ | architecture ✅ | development ✅ | testing ✅\n"
                "Task is ready for review.",
            )
            notify_all_done(parent_key, job.get("jira_domain", ""))

        triggered = trigger_next_stages(parent_key, stage, jira)
        if triggered:
            jira.add_comment(
                issue_key,
                f"🤖 Automatically triggered stages: {', '.join(triggered)}",
            )

    except Exception as e:
        logger.error("[%s] code stage FAIL: %s", issue_key, e)
        notify_error(issue_key, stage, str(e), job.get("jira_domain", ""))
        try:
            jira.add_comment(
                issue_key,
                f"❌ Pipeline error (stage={stage}): {str(e)[:500]}\nJob: {job_id}",
            )
        except Exception:
            pass
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


# ── Main entry point ──────────────────────────────────────────────────────────

def _create_stage_to_main_pr(gh: GitHubClient, issue_key: str, summary: str) -> None:
    """After merging feature → stage, auto-create PR from stage → main."""
    try:
        # Check if there's already an open stage → main PR
        existing = gh.find_pr(STAGE_BRANCH)
        if existing:
            logger.info("[%s] stage→main PR already exists: #%s",
                        issue_key, existing["number"])
            return

        pr = gh.create_pr(
            head=STAGE_BRANCH,
            base="main",
            title=f"Release: {issue_key} — {summary}",
            body=(
                f"## Auto-release from `{STAGE_BRANCH}` → `main`\n\n"
                f"Triggered by merge of {issue_key}: {summary}\n\n"
                "Created automatically by Claudev."
            ),
        )
        logger.info("[%s] created stage→main PR #%s: %s",
                    issue_key, pr["number"], pr["html_url"])
    except Exception as e:
        logger.warning("[%s] failed to create stage→main PR: %s", issue_key, e)


def run_merge_job(job: dict) -> None:
    """Triggered when parent task moves to STATUS_MERGE ('Ready to Merge').

    Finds the open PR for feature/<issue_key>, merges it into stage,
    then transitions Jira task to Done.
    """
    issue_key = job["issue_key"]
    jira_domain = job.get("jira_domain", "")
    repo_cfg = _get_repo_config(job)
    gh = _github_for_repo(repo_cfg)

    try:
        branch_name = f"feature/{issue_key.lower()}"
        pr = gh.find_pr(branch_name)

        if not pr:
            jira.add_comment(
                issue_key,
                f"⚠️ Pipeline: no open PR found for branch `{branch_name}`. "
                "Please merge manually.",
            )
            return

        pr_number = pr["number"]
        pr_url = pr["html_url"]
        logger.info("[%s] Merging PR #%s into %s", issue_key, pr_number, pr["base"]["ref"])

        merge_result = gh.merge_pr(
            pr_number,
            commit_message=f"{issue_key}: {job['summary']} (auto-merge)",
        )

        if not merge_result.get("merged"):
            raise Exception(f"GitHub merge failed: {merge_result.get('message')}")

        jira.transition(issue_key, STATUS_DONE)
        jira.add_comment(
            issue_key,
            f"🎉 Merged into `{pr['base']['ref']}`!\n"
            f"PR: {pr_url}\n"
            f"Commit: {merge_result.get('sha', '')[:8]}",
        )
        notify_merged(issue_key, pr_url, pr["base"]["ref"], jira_domain)
        logger.info("[%s] merged PR #%s → Done", issue_key, pr_number)

        # Auto-create PR from stage → main
        base_branch = pr["base"]["ref"]  # usually "stage"
        if base_branch == STAGE_BRANCH:
            _create_stage_to_main_pr(gh, issue_key, job["summary"])

    except Exception as e:
        logger.error("[%s] merge FAIL: %s", issue_key, e)
        notify_error(issue_key, "merge", str(e), jira_domain)
        try:
            jira.add_comment(
                issue_key,
                f"❌ Auto-merge failed: {str(e)[:400]}\n"
                "Please merge the PR manually and move the task to Done.",
            )
        except Exception:
            pass


def run_job(job: dict) -> None:
    """Route job to the correct handler.

    PLAN: task (no stage)   → run_plan_job: break feature into epics/tasks
    Parent task (no stage)  → run_setup_job: create subtasks, start first stages
    Sub-task artifact stage → run_artifact_stage: Claude Code writes markdown
    Sub-task code stage     → run_code_stage: Claude Code writes code + PR
    """
    from config import PLAN_PREFIX
    stage = job.get("stage")

    if job.get("trigger") == STATUS_MERGE:
        run_merge_job(job)
    elif stage is None and job.get("summary", "").startswith(PLAN_PREFIX):
        run_plan_job(job)
    elif stage is None:
        run_setup_job(job)
    elif stage in ARTIFACT_STAGES:
        run_artifact_stage(job)
    elif stage in CODE_STAGES:
        run_code_stage(job)
    else:
        logger.warning("[%s] Unknown stage '%s', falling back to legacy", job["issue_key"], stage)
        _run_legacy_job(job)


# ── Legacy single-stage flow (backward compatibility) ─────────────────────────

def _run_legacy_job(job: dict) -> None:
    """Original single-stage worker for tasks without pipeline labels."""
    from orchestrator import (
        parse_adf_to_text,
        classify_issue,
        build_claude_prompt,
        analyze_result as _analyze,
    )

    issue_key = job["issue_key"]
    job_id = job["job_id"]
    work_dir = f"/tmp/pipeline-work/{job_id}"

    try:
        jira.transition(issue_key, "In Progress")
        jira.add_comment(issue_key, f"🤖 Pipeline started. Job: {job_id}")

        description_text = parse_adf_to_text(job.get("description", ""))
        issue = {
            "key": issue_key,
            "summary": job["summary"],
            "description_text": description_text,
            "issue_type": job.get("issue_type", "Task"),
            "priority": job.get("priority", "Medium"),
            "labels": job.get("labels", []),
            "components": job.get("components", []),
        }

        classification = classify_issue(
            issue["summary"], description_text, issue["labels"]
        )
        prompt = build_claude_prompt(issue, classification)

        branch_name = f"feature/{issue_key.lower()}"
        os.makedirs(work_dir, exist_ok=True)
        _clone_repo(work_dir, branch_name)

        start = time.time()
        result = _run_claude_with_retry(prompt, work_dir, job)
        duration = int(time.time() - start)
        logger.info("[%s] Claude Code: %ds rc=%d", issue_key, duration, result.returncode)

        changed = _git_changed_files(work_dir)
        if not changed:
            jira.add_comment(
                issue_key,
                "🤖 Claude Code made no changes. Task needs clarification.",
            )
            jira.transition(issue_key, "Ready for Dev")
            return

        analysis = _analyze(result.stdout, changed)

        subprocess.run(["git", "add", "-A"], cwd=work_dir, check=True)
        subprocess.run(
            [
                "git", "commit", "-m",
                f"{issue_key}: {issue['summary']}\n\nAutomated by Claudev",
            ],
            cwd=work_dir, check=True, capture_output=True,
        )
        subprocess.run(
            ["git", "push", "origin", branch_name],
            cwd=work_dir, check=True, capture_output=True, timeout=60,
        )

        jira_domain = job.get("jira_domain", os.environ.get("JIRA_DOMAIN", "x"))
        files_list = "\n".join(
            "- " + f for f in analysis.get("files_changed", changed)
        )
        pr_body = (
            f"## {issue_key}: {issue['summary']}\n\n"
            f"**Jira:** https://{jira_domain}/browse/{issue_key}\n"
            "**Automated by:** Claudev\n\n"
            f"### Changes\n{analysis.get('summary_ru', 'N/A')}\n\n"
            f"### Files\n{files_list}\n\n"
            f"### Tests: {analysis.get('tests_status', '?')}\n"
        )
        pr = github.create_pr(
            head=branch_name,
            base=STAGE_BRANCH,
            title=f"{issue_key}: {issue['summary']}",
            body=pr_body,
        )
        github.add_labels(pr["number"], ["automated", "claude-code"])

        concerns = (
            "\n⚠️ " + "; ".join(analysis["concerns"])
            if analysis.get("concerns") else ""
        )
        jira.transition(issue_key, "In Review")
        jira.add_comment(
            issue_key,
            f"🤖 PR created: {pr['html_url']}\n"
            f"Files: {len(changed)} | "
            f"Tests: {analysis.get('tests_status', '?')} | "
            f"Duration: {duration // 60}m {duration % 60}s\n"
            f"{analysis.get('summary_ru', '')}{concerns}",
        )
        logger.info("[%s] Done! PR #%s", issue_key, pr["number"])

    except Exception as e:
        logger.error("[%s] FAIL: %s", issue_key, e)
        try:
            jira.add_comment(
                issue_key,
                f"❌ Pipeline error: {str(e)[:500]}\nJob: {job_id}",
            )
        except Exception:
            pass
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)