1#!/usr/bin/env python3
2"""Run the background crash-agent MVP pipeline locally.
3
4Default mode is dry-run: generate branch + agent output without commit/push/PR.
5"""
6
7import argparse
8import json
9import os
10import re
11import subprocess
12import sys
13import tempfile
14from datetime import datetime
15from pathlib import Path
16
17
18CRASH_ID_PATTERN = re.compile(r"^[A-Za-z0-9]+-[A-Za-z0-9]+$")
19
20
21REPO_ROOT = Path(__file__).resolve().parents[1]
22
23
24def run(command: list[str], check: bool = True, capture_output: bool = False) -> subprocess.CompletedProcess:
25 return subprocess.run(
26 command,
27 cwd=REPO_ROOT,
28 text=True,
29 check=check,
30 capture_output=capture_output,
31 )
32
33
34def validate_crash_ids(ids: list[str]) -> list[str]:
35 valid = []
36 for crash_id in ids:
37 if not CRASH_ID_PATTERN.match(crash_id):
38 print(f"WARNING: Skipping invalid crash ID format: '{crash_id}'", file=sys.stderr)
39 continue
40 valid.append(crash_id)
41 return valid
42
43
44MAX_TOP = 100
45
46
47def prefetch_crash_data(crashes: list[dict[str, str]], output_dir: str) -> None:
48 """Fetch crash reports and save to output_dir/crash-{ID}.md.
49
50 Each crash item must contain:
51 - crash_id: short ID used by the pipeline (e.g. ZED-202)
52 - fetch_id: identifier passed to script/sentry-fetch (short or numeric)
53 """
54 os.makedirs(output_dir, exist_ok=True)
55 for crash in crashes:
56 crash_id = crash["crash_id"]
57 fetch_id = crash["fetch_id"]
58 output_path = os.path.join(output_dir, f"crash-{crash_id}.md")
59 result = run(
60 ["python3", "script/sentry-fetch", fetch_id],
61 check=False,
62 capture_output=True,
63 )
64 if result.returncode != 0:
65 print(
66 f"WARNING: Failed to fetch crash data for {crash_id} "
67 f"(fetch id: {fetch_id}): {result.stderr.strip()}",
68 file=sys.stderr,
69 )
70 continue
71 with open(output_path, "w", encoding="utf-8") as f:
72 f.write(result.stdout)
73 print(
74 f"Fetched crash data for {crash_id} (fetch id: {fetch_id}) -> {output_path}",
75 file=sys.stderr,
76 )
77
78
79def resolve_crashes(args) -> list[dict[str, str]]:
80 if args.crash_ids:
81 raw = [item.strip() for item in args.crash_ids.split(",") if item.strip()]
82 crash_ids = validate_crash_ids(raw)
83 return [{"crash_id": crash_id, "fetch_id": crash_id} for crash_id in crash_ids]
84
85 top = min(args.top, MAX_TOP)
86 if args.top > MAX_TOP:
87 print(f"Capping --top from {args.top} to {MAX_TOP}", file=sys.stderr)
88
89 with tempfile.NamedTemporaryFile(mode="w+", suffix=".json", delete=False) as file:
90 output_path = file.name
91
92 try:
93 run(
94 [
95 "python3",
96 "script/select-sentry-crash-candidates",
97 "--org",
98 args.org,
99 "--top",
100 str(top),
101 "--sample-size",
102 str(args.sample_size),
103 "--output",
104 output_path,
105 ],
106 capture_output=True,
107 )
108 with open(output_path, "r", encoding="utf-8") as file:
109 payload = json.load(file)
110 crashes = []
111 for item in payload.get("selected", []):
112 crash_id = item.get("short_id")
113 issue_id = item.get("issue_id")
114 if not crash_id:
115 continue
116 crashes.append(
117 {
118 "crash_id": crash_id,
119 "fetch_id": str(issue_id) if issue_id else crash_id,
120 }
121 )
122
123 valid_crash_ids = set(validate_crash_ids([item["crash_id"] for item in crashes]))
124 return [item for item in crashes if item["crash_id"] in valid_crash_ids]
125 finally:
126 try:
127 os.remove(output_path)
128 except OSError:
129 pass
130
131
132def write_prompt(crash_id: str, crash_data_file: str | None = None) -> Path:
133 if crash_data_file:
134 fetch_step = (
135 f"The crash report has been pre-fetched and is available at: {crash_data_file}\n"
136 f"Read this file to get the crash data. Do not call script/sentry-fetch.\n"
137 f"\n"
138 f"1. Read the crash report from {crash_data_file}"
139 )
140 else:
141 fetch_step = f"1. Fetch crash data with: script/sentry-fetch {crash_id}"
142
143 prompt = f"""You are running the weekly background crash-fix MVP pipeline for crash {crash_id}.
144
145Required workflow:
146{fetch_step}
1472. Read and follow `.rules`.
1483. Follow .factory/prompts/crash/investigate.md and write ANALYSIS.md
1494. Follow .factory/prompts/crash/link-issues.md and write LINKED_ISSUES.md
1505. Follow .factory/prompts/crash/fix.md to implement a minimal fix with tests
1516. Run validators required by the fix prompt for the affected code paths
1527. Write PR_BODY.md with sections:
153 - Crash Summary
154 - Root Cause
155 - Fix
156 - Validation
157 - Potentially Related Issues (High/Medium/Low from LINKED_ISSUES.md)
158 - Reviewer Checklist
159 - Release Notes (final section, formatted as "Release Notes:", blank line, then one bullet like "- N/A" when not user-facing)
160
161Constraints:
162- Keep changes narrowly scoped to this crash.
163- If the crash is not solvable with available context, write a clear blocker summary to PR_BODY.md.
164"""
165
166 file_path = Path(tempfile.gettempdir()) / f"background-agent-{crash_id}.md"
167 file_path.write_text(prompt, encoding="utf-8")
168 return file_path
169
170
171def run_pipeline_for_crash(args, crash_id: str) -> dict:
172 branch = f"background-agent/mvp-{crash_id.lower()}-{datetime.utcnow().strftime('%Y%m%d')}"
173
174 if args.reset_branch:
175 run(["git", "fetch", "origin", "main"], check=False)
176 run(["git", "checkout", "-B", branch, "origin/main"])
177
178 prompt_file = write_prompt(crash_id)
179 try:
180 droid_command = [
181 args.droid_bin,
182 "exec",
183 "--auto",
184 "medium",
185 "-m",
186 args.model,
187 "-f",
188 str(prompt_file),
189 ]
190 completed = run(droid_command, check=False)
191
192 has_changes = run(["git", "diff", "--quiet"], check=False).returncode != 0
193 return {
194 "crash_id": crash_id,
195 "branch": branch,
196 "droid_exit_code": completed.returncode,
197 "has_changes": has_changes,
198 }
199 finally:
200 try:
201 prompt_file.unlink()
202 except OSError:
203 pass
204
205
206def main() -> int:
207 parser = argparse.ArgumentParser(description="Run local background-agent MVP dry-run workflow")
208 parser.add_argument("--crash-ids", help="Comma-separated crash IDs (e.g. ZED-4VS,ZED-123)")
209 parser.add_argument("--top", type=int, default=3, help="Top N crashes when --crash-ids is omitted (max 100)")
210 parser.add_argument(
211 "--sample-size",
212 type=int,
213 default=100,
214 help="Number of unresolved issues to consider for candidate selection",
215 )
216 parser.add_argument("--org", default="zed-dev", help="Sentry org slug")
217 parser.add_argument(
218 "--select-only",
219 action="store_true",
220 help="Resolve crash IDs and print them comma-separated, then exit. No agent execution.",
221 )
222 parser.add_argument(
223 "--prefetch-dir",
224 help="When used with --select-only, also fetch crash data via script/sentry-fetch "
225 "and save reports to DIR/crash-{ID}.md for each resolved ID.",
226 )
227 parser.add_argument("--model", default=os.environ.get("DROID_MODEL", "claude-opus-4-5-20251101"))
228 parser.add_argument("--droid-bin", default=os.environ.get("DROID_BIN", "droid"))
229 parser.add_argument(
230 "--reset-branch",
231 action="store_true",
232 help="For each crash, checkout a fresh local branch from origin/main",
233 )
234 args = parser.parse_args()
235
236 crashes = resolve_crashes(args)
237 if not crashes:
238 print("No crash IDs were selected.", file=sys.stderr)
239 return 1
240
241 crash_ids = [item["crash_id"] for item in crashes]
242
243 if args.select_only:
244 if args.prefetch_dir:
245 prefetch_crash_data(crashes, args.prefetch_dir)
246 print(",".join(crash_ids))
247 return 0
248
249 print(f"Running local dry-run for crashes: {', '.join(crash_ids)}")
250 results = [run_pipeline_for_crash(args, crash_id) for crash_id in crash_ids]
251
252 print("\nRun summary:")
253 for result in results:
254 print(
255 f"- {result['crash_id']}: droid_exit={result['droid_exit_code']} "
256 f"changes={result['has_changes']} branch={result['branch']}"
257 )
258
259 failures = [result for result in results if result["droid_exit_code"] != 0]
260 return 1 if failures else 0
261
262
263if __name__ == "__main__":
264 sys.exit(main())