Skills › Compound Engineering › ce-babysit-pr › intake-overlap
intake-overlap
227 lines · skills/ce-babysit-pr/scripts/intake-overlap at 3449230
1#!/usr/bin/env python3
2"""One-shot overlap check for ce-babysit-pr's stale-PR intake.
4Answers "did the base move under this PR's paths since it branched off" from
5freshly fetched refs:
7 intake-overlap --repo-dir DIR --remote NAME --base BRANCH --pr N
9Fetches the base branch and refs/pull/<N>/head into refs/ce-babysit/intake/<N>/,
10takes their merge-base, and compares the PR's changed paths with the base's
11changes since then. Overlap kinds:
13 same-path a base change (either rename name) is a PR path (either name)
14 directory-gone a PR path's ancestor directory existed at the merge-base and
15 is absent at the base tip (nearest such ancestor reported)
16 sibling a base change sits in the same non-root directory as a PR path
18Prints one JSON object. "overlap" is true, false, or "unknown" when the question
19cannot be answered (fetch failure, no merge-base); unknown sets "error" and still
20exits 0. Only usage errors exit non-zero.
21"""
22import argparse
23import json
24import os
25import posixpath
26import re
27import subprocess
28import sys
30FETCH_TIMEOUT_SECONDS = 300
31GIT_TIMEOUT_SECONDS = 120
33STATUS_NAMES = {
34 "A": "added",
35 "C": "copied",
36 "D": "deleted",
37 "M": "modified",
38 "R": "renamed",
39 "T": "type-changed",
40}
42_BAD_BRANCH = re.compile(r"[\x00-\x20\x7f~^:?*\[\\]|\.\.|@\{|//|/\.|\.lock$|^/|/$|\.$")
45class GitError(Exception):
46 pass
49def git(repo_dir, *args, timeout=GIT_TIMEOUT_SECONDS, check=True):
50 env = dict(os.environ, GIT_TERMINAL_PROMPT="0")
51 try:
52 r = subprocess.run(
53 ["git", "-C", repo_dir, *args],
54 stdin=subprocess.DEVNULL,
55 stdout=subprocess.PIPE,
56 stderr=subprocess.PIPE,
57 env=env,
58 timeout=timeout,
59 )
60 except subprocess.TimeoutExpired:
61 raise GitError(f"git {args[0]} timed out after {timeout}s")
62 except OSError as e:
63 raise GitError(f"could not run git: {e}")
64 out = r.stdout.decode("utf-8", errors="replace")
65 if check and r.returncode != 0:
66 err = r.stderr.decode("utf-8", errors="replace").strip()
67 raise GitError(f"git {args[0]} failed ({r.returncode}): {err}")
68 return r.returncode, out
71def refs_for(pr):
72 prefix = f"refs/ce-babysit/intake/{pr}"
73 return f"{prefix}/base", f"{prefix}/head"
76def fetch(repo_dir, remote, base, pr, extra=()):
77 base_ref, head_ref = refs_for(pr)
78 git(
79 repo_dir, "fetch", "--no-tags", "--quiet", *extra, remote,
80 f"+refs/heads/{base}:{base_ref}",
81 f"+refs/pull/{pr}/head:{head_ref}",
82 timeout=FETCH_TIMEOUT_SECONDS,
83 )
84 return git(repo_dir, "rev-parse", "--verify", base_ref)[1].strip(), \
85 git(repo_dir, "rev-parse", "--verify", head_ref)[1].strip()
88def merge_base(repo_dir, a, b):
89 code, out = git(repo_dir, "merge-base", a, b, check=False)
90 return out.strip() if code == 0 and out.strip() else None
93def changes(repo_dir, a, b):
94 """[(status, old, new)] for `git diff -M a b`; old == new unless renamed/copied."""
95 out = git(repo_dir, "-c", "diff.renameLimit=0", "diff", "--no-ext-diff",
96 "--name-status", "-M", "-z", a, b)[1]
97 tokens = out.split("\0")
98 result = []
99 i = 0
100 while i < len(tokens) and tokens[i]:
101 letter = tokens[i][0]
102 if letter in ("R", "C"):
103 old, new = tokens[i + 1], tokens[i + 2]
104 i += 3
105 else:
106 old = new = tokens[i + 1]
107 i += 2
108 result.append((STATUS_NAMES.get(letter, letter), old, new))
109 return result
112def directories(repo_dir, commit):
113 out = git(repo_dir, "ls-tree", "-r", "-d", "--name-only", "-z", commit)[1]
114 return {d for d in out.split("\0") if d}
117def ancestors(path):
118 parent = posixpath.dirname(path)
119 while parent:
120 yield parent
121 parent = posixpath.dirname(parent)
124def overlap_reasons(pr_changes, base_changes, dirs_at_mb, dirs_at_base):
125 pr_paths = sorted({p for _, old, new in pr_changes for p in (old, new)})
127 by_path = {}
128 by_parent = {}
129 for change in base_changes:
130 _, old, new = change
131 for name in {old, new}:
132 by_path.setdefault(name, []).append(change)
133 parent = posixpath.dirname(name)
134 if parent:
135 by_parent.setdefault(parent, []).append(change)
137 renamed_from = [old for status, old, _ in base_changes if status == "renamed"]
139 reasons = set()
140 for p in pr_paths:
141 same = by_path.get(p, [])
142 for status, _, new in same:
143 reasons.add(("same-path", p, new, status))
144 parent = posixpath.dirname(p)
145 for change in by_parent.get(parent, []) if parent else []:
146 if change not in same:
147 status, _, new = change
148 reasons.add(("sibling", p, new, status))
149 for d in ancestors(p):
150 if d in dirs_at_mb and d not in dirs_at_base:
151 moved = any(old.startswith(d + "/") for old in renamed_from)
152 reasons.add(("directory-gone", p, d, "renamed" if moved else "deleted"))
153 break
155 return [
156 {"kind": k, "pr_path": p, "base_path": b, "base_status": s}
157 for k, p, b, s in sorted(reasons)
158 ]
161def assess(repo_dir, remote, base, pr):
162 result = {
163 "base_oid": None,
164 "head_oid": None,
165 "merge_base": None,
166 "overlap": "unknown",
167 "reasons": [],
168 "error": None,
169 }
170 try:
171 base_oid, head_oid = fetch(repo_dir, remote, base, pr)
172 result.update(base_oid=base_oid, head_oid=head_oid)
174 mb = merge_base(repo_dir, base_oid, head_oid)
175 if mb is None:
176 shallow = git(repo_dir, "rev-parse", "--is-shallow-repository")[1].strip() == "true"
177 if not shallow:
178 result["error"] = "no merge-base between base and PR head (unrelated histories)"
179 return result
180 base_oid, head_oid = fetch(repo_dir, remote, base, pr, extra=("--unshallow",))
181 result.update(base_oid=base_oid, head_oid=head_oid)
182 mb = merge_base(repo_dir, base_oid, head_oid)
183 if mb is None:
184 result["error"] = "no merge-base between base and PR head after unshallowing"
185 return result
186 result["merge_base"] = mb
188 if mb == base_oid:
189 result["overlap"] = False
190 return result
192 reasons = overlap_reasons(
193 changes(repo_dir, mb, head_oid),
194 changes(repo_dir, mb, base_oid),
195 directories(repo_dir, mb),
196 directories(repo_dir, base_oid),
197 )
198 result["reasons"] = reasons
199 result["overlap"] = bool(reasons)
200 return result
201 except GitError as e:
202 result["error"] = str(e)
203 return result
206def main():
207 p = argparse.ArgumentParser(prog="intake-overlap")
208 p.add_argument("--repo-dir", required=True)
209 p.add_argument("--remote", required=True)
210 p.add_argument("--base", required=True)
211 p.add_argument("--pr", type=int, required=True)
212 args = p.parse_args()
214 if args.pr <= 0:
215 p.error("--pr must be a positive integer")
216 if not args.remote or args.remote.startswith("-"):
217 p.error("--remote must be a remote name or URL")
218 if not args.base or args.base.startswith("-") or _BAD_BRANCH.search(args.base):
219 p.error("--base must be a valid branch name")
220 if not os.path.isdir(args.repo_dir):
221 p.error("--repo-dir must be an existing directory")
223 print(json.dumps(assess(args.repo_dir, args.remote, args.base, args.pr)))
226if __name__ == "__main__":
227 main()