#!/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 argparseimport jsonimport osimport reimport subprocessimport sysimport tempfilefrom datetime import datetimefrom 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.md4. Follow .factory/prompts/crash/link-issues.md and write LINKED_ISSUES.md5. Follow .factory/prompts/crash/fix.md to implement a minimal fix with tests6. Run validators required by the fix prompt for the affected code paths7. 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())
#!/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())