zed/script/run-background-agent-mvp-local

#!/usr/bin/env python3
"""Run the background crash-agent MVP pipeline locally.

Default mode is dry-run: generate branch + agent output without commit/push/PR.
"""

import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path


CRASH_ID_PATTERN = re.compile(r"^[A-Za-z0-9]+-[A-Za-z0-9]+$")


REPO_ROOT = Path(__file__).resolve().parents[1]


def run(command: list[str], check: bool = True, capture_output: bool = False) -> subprocess.CompletedProcess:
    return subprocess.run(
        command,
        cwd=REPO_ROOT,
        text=True,
        check=check,
        capture_output=capture_output,
    )


def validate_crash_ids(ids: list[str]) -> list[str]:
    valid = []
    for crash_id in ids:
        if not CRASH_ID_PATTERN.match(crash_id):
            print(f"WARNING: Skipping invalid crash ID format: '{crash_id}'", file=sys.stderr)
            continue
        valid.append(crash_id)
    return valid


MAX_TOP = 100


def prefetch_crash_data(crashes: list[dict[str, str]], output_dir: str) -> None:
    """Fetch crash reports and save to output_dir/crash-{ID}.md.

    Each crash item must contain:
    - crash_id: short ID used by the pipeline (e.g. ZED-202)
    - fetch_id: identifier passed to script/sentry-fetch (short or numeric)
    """
    os.makedirs(output_dir, exist_ok=True)
    for crash in crashes:
        crash_id = crash["crash_id"]
        fetch_id = crash["fetch_id"]
        output_path = os.path.join(output_dir, f"crash-{crash_id}.md")
        result = run(
            ["python3", "script/sentry-fetch", fetch_id],
            check=False,
            capture_output=True,
        )
        if result.returncode != 0:
            print(
                f"WARNING: Failed to fetch crash data for {crash_id} "
                f"(fetch id: {fetch_id}): {result.stderr.strip()}",
                file=sys.stderr,
            )
            continue
        with open(output_path, "w", encoding="utf-8") as f:
            f.write(result.stdout)
        print(
            f"Fetched crash data for {crash_id} (fetch id: {fetch_id}) -> {output_path}",
            file=sys.stderr,
        )


def resolve_crashes(args) -> list[dict[str, str]]:
    if args.crash_ids:
        raw = [item.strip() for item in args.crash_ids.split(",") if item.strip()]
        crash_ids = validate_crash_ids(raw)
        return [{"crash_id": crash_id, "fetch_id": crash_id} for crash_id in crash_ids]

    top = min(args.top, MAX_TOP)
    if args.top > MAX_TOP:
        print(f"Capping --top from {args.top} to {MAX_TOP}", file=sys.stderr)

    with tempfile.NamedTemporaryFile(mode="w+", suffix=".json", delete=False) as file:
        output_path = file.name

    try:
        run(
            [
                "python3",
                "script/select-sentry-crash-candidates",
                "--org",
                args.org,
                "--top",
                str(top),
                "--sample-size",
                str(args.sample_size),
                "--output",
                output_path,
            ],
            capture_output=True,
        )
        with open(output_path, "r", encoding="utf-8") as file:
            payload = json.load(file)
        crashes = []
        for item in payload.get("selected", []):
            crash_id = item.get("short_id")
            issue_id = item.get("issue_id")
            if not crash_id:
                continue
            crashes.append(
                {
                    "crash_id": crash_id,
                    "fetch_id": str(issue_id) if issue_id else crash_id,
                }
            )

        valid_crash_ids = set(validate_crash_ids([item["crash_id"] for item in crashes]))
        return [item for item in crashes if item["crash_id"] in valid_crash_ids]
    finally:
        try:
            os.remove(output_path)
        except OSError:
            pass


def write_prompt(crash_id: str, crash_data_file: str | None = None) -> Path:
    if crash_data_file:
        fetch_step = (
            f"The crash report has been pre-fetched and is available at: {crash_data_file}\n"
            f"Read this file to get the crash data. Do not call script/sentry-fetch.\n"
            f"\n"
            f"1. Read the crash report from {crash_data_file}"
        )
    else:
        fetch_step = f"1. Fetch crash data with: script/sentry-fetch {crash_id}"

    prompt = f"""You are running the weekly background crash-fix MVP pipeline for crash {crash_id}.

Required workflow:
{fetch_step}
2. Read and follow `.rules`.
3. Follow .factory/prompts/crash/investigate.md and write ANALYSIS.md
4. Follow .factory/prompts/crash/link-issues.md and write LINKED_ISSUES.md
5. Follow .factory/prompts/crash/fix.md to implement a minimal fix with tests
6. Run validators required by the fix prompt for the affected code paths
7. Write PR_BODY.md with sections:
   - Crash Summary
   - Root Cause
   - Fix
   - Validation
   - Potentially Related Issues (High/Medium/Low from LINKED_ISSUES.md)
   - Reviewer Checklist
   - Release Notes (final section, formatted as "Release Notes:", blank line, then one bullet like "- N/A" when not user-facing)

Constraints:
- Keep changes narrowly scoped to this crash.
- If the crash is not solvable with available context, write a clear blocker summary to PR_BODY.md.
"""

    file_path = Path(tempfile.gettempdir()) / f"background-agent-{crash_id}.md"
    file_path.write_text(prompt, encoding="utf-8")
    return file_path


def run_pipeline_for_crash(args, crash_id: str) -> dict:
    branch = f"background-agent/mvp-{crash_id.lower()}-{datetime.utcnow().strftime('%Y%m%d')}"

    if args.reset_branch:
        run(["git", "fetch", "origin", "main"], check=False)
        run(["git", "checkout", "-B", branch, "origin/main"])

    prompt_file = write_prompt(crash_id)
    try:
        droid_command = [
            args.droid_bin,
            "exec",
            "--auto",
            "medium",
            "-m",
            args.model,
            "-f",
            str(prompt_file),
        ]
        completed = run(droid_command, check=False)

        has_changes = run(["git", "diff", "--quiet"], check=False).returncode != 0
        return {
            "crash_id": crash_id,
            "branch": branch,
            "droid_exit_code": completed.returncode,
            "has_changes": has_changes,
        }
    finally:
        try:
            prompt_file.unlink()
        except OSError:
            pass


def main() -> int:
    parser = argparse.ArgumentParser(description="Run local background-agent MVP dry-run workflow")
    parser.add_argument("--crash-ids", help="Comma-separated crash IDs (e.g. ZED-4VS,ZED-123)")
    parser.add_argument("--top", type=int, default=3, help="Top N crashes when --crash-ids is omitted (max 100)")
    parser.add_argument(
        "--sample-size",
        type=int,
        default=100,
        help="Number of unresolved issues to consider for candidate selection",
    )
    parser.add_argument("--org", default="zed-dev", help="Sentry org slug")
    parser.add_argument(
        "--select-only",
        action="store_true",
        help="Resolve crash IDs and print them comma-separated, then exit. No agent execution.",
    )
    parser.add_argument(
        "--prefetch-dir",
        help="When used with --select-only, also fetch crash data via script/sentry-fetch "
        "and save reports to DIR/crash-{ID}.md for each resolved ID.",
    )
    parser.add_argument("--model", default=os.environ.get("DROID_MODEL", "claude-opus-4-5-20251101"))
    parser.add_argument("--droid-bin", default=os.environ.get("DROID_BIN", "droid"))
    parser.add_argument(
        "--reset-branch",
        action="store_true",
        help="For each crash, checkout a fresh local branch from origin/main",
    )
    args = parser.parse_args()

    crashes = resolve_crashes(args)
    if not crashes:
        print("No crash IDs were selected.", file=sys.stderr)
        return 1

    crash_ids = [item["crash_id"] for item in crashes]

    if args.select_only:
        if args.prefetch_dir:
            prefetch_crash_data(crashes, args.prefetch_dir)
        print(",".join(crash_ids))
        return 0

    print(f"Running local dry-run for crashes: {', '.join(crash_ids)}")
    results = [run_pipeline_for_crash(args, crash_id) for crash_id in crash_ids]

    print("\nRun summary:")
    for result in results:
        print(
            f"- {result['crash_id']}: droid_exit={result['droid_exit_code']} "
            f"changes={result['has_changes']} branch={result['branch']}"
        )

    failures = [result for result in results if result["droid_exit_code"] != 0]
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())