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. Follow .factory/prompts/crash/investigate.md and write ANALYSIS.md
1483. Follow .factory/prompts/crash/link-issues.md and write LINKED_ISSUES.md
1494. Follow .factory/prompts/crash/fix.md to implement a minimal fix with tests
1505. Run validators required by the fix prompt for the affected code paths
1516. Write PR_BODY.md with sections:
152 - Crash Summary
153 - Root Cause
154 - Fix
155 - Validation
156 - Potentially Related Issues (High/Medium/Low from LINKED_ISSUES.md)
157 - Reviewer Checklist
158
159Constraints:
160- Keep changes narrowly scoped to this crash.
161- If the crash is not solvable with available context, write a clear blocker summary to PR_BODY.md.
162"""
163
164 file_path = Path(tempfile.gettempdir()) / f"background-agent-{crash_id}.md"
165 file_path.write_text(prompt, encoding="utf-8")
166 return file_path
167
168
169def run_pipeline_for_crash(args, crash_id: str) -> dict:
170 branch = f"background-agent/mvp-{crash_id.lower()}-{datetime.utcnow().strftime('%Y%m%d')}"
171
172 if args.reset_branch:
173 run(["git", "fetch", "origin", "main"], check=False)
174 run(["git", "checkout", "-B", branch, "origin/main"])
175
176 prompt_file = write_prompt(crash_id)
177 try:
178 droid_command = [
179 args.droid_bin,
180 "exec",
181 "--auto",
182 "medium",
183 "-m",
184 args.model,
185 "-f",
186 str(prompt_file),
187 ]
188 completed = run(droid_command, check=False)
189
190 has_changes = run(["git", "diff", "--quiet"], check=False).returncode != 0
191 return {
192 "crash_id": crash_id,
193 "branch": branch,
194 "droid_exit_code": completed.returncode,
195 "has_changes": has_changes,
196 }
197 finally:
198 try:
199 prompt_file.unlink()
200 except OSError:
201 pass
202
203
204def main() -> int:
205 parser = argparse.ArgumentParser(description="Run local background-agent MVP dry-run workflow")
206 parser.add_argument("--crash-ids", help="Comma-separated crash IDs (e.g. ZED-4VS,ZED-123)")
207 parser.add_argument("--top", type=int, default=3, help="Top N crashes when --crash-ids is omitted (max 100)")
208 parser.add_argument(
209 "--sample-size",
210 type=int,
211 default=100,
212 help="Number of unresolved issues to consider for candidate selection",
213 )
214 parser.add_argument("--org", default="zed-dev", help="Sentry org slug")
215 parser.add_argument(
216 "--select-only",
217 action="store_true",
218 help="Resolve crash IDs and print them comma-separated, then exit. No agent execution.",
219 )
220 parser.add_argument(
221 "--prefetch-dir",
222 help="When used with --select-only, also fetch crash data via script/sentry-fetch "
223 "and save reports to DIR/crash-{ID}.md for each resolved ID.",
224 )
225 parser.add_argument("--model", default=os.environ.get("DROID_MODEL", "claude-opus-4-5-20251101"))
226 parser.add_argument("--droid-bin", default=os.environ.get("DROID_BIN", "droid"))
227 parser.add_argument(
228 "--reset-branch",
229 action="store_true",
230 help="For each crash, checkout a fresh local branch from origin/main",
231 )
232 args = parser.parse_args()
233
234 crashes = resolve_crashes(args)
235 if not crashes:
236 print("No crash IDs were selected.", file=sys.stderr)
237 return 1
238
239 crash_ids = [item["crash_id"] for item in crashes]
240
241 if args.select_only:
242 if args.prefetch_dir:
243 prefetch_crash_data(crashes, args.prefetch_dir)
244 print(",".join(crash_ids))
245 return 0
246
247 print(f"Running local dry-run for crashes: {', '.join(crash_ids)}")
248 results = [run_pipeline_for_crash(args, crash_id) for crash_id in crash_ids]
249
250 print("\nRun summary:")
251 for result in results:
252 print(
253 f"- {result['crash_id']}: droid_exit={result['droid_exit_code']} "
254 f"changes={result['has_changes']} branch={result['branch']}"
255 )
256
257 failures = [result for result in results if result["droid_exit_code"] != 0]
258 return 1 if failures else 0
259
260
261if __name__ == "__main__":
262 sys.exit(main())