diff --git a/scripts/ci/prune_merged_worktrees.py b/scripts/ci/prune_merged_worktrees.py index 2de0193e31..42ad6f7355 100644 --- a/scripts/ci/prune_merged_worktrees.py +++ b/scripts/ci/prune_merged_worktrees.py @@ -316,11 +316,27 @@ def lookup_pr_for_branch(branch: str) -> Optional[dict]: def lookup_pr_for_detached_head(wt_path: str) -> Optional[dict]: - """Verdict par contenu pour HEAD detaché : sujet de commit = titre PR ? + """Verdict par contenu pour HEAD detaché (#14476) : PR exacte, ou rien. Le squash-merge efface l'ascendance, donc `git merge-base --is-ancestor` - ne marche pas. On cherche dans les messages de commit du HEAD du - worktree un motif qui ressemble a un titre de PR recente. + ne marche pas. On cherche un match EXACT entre les sujets de commit du + HEAD et une PR reelle, en deux voies : + + 1. **Resolution directe par numero** : un squash-commit sur ce depot a + pour sujet `` (#N)``. On extrait N via + ``re.search(r"\\(#(\\d+)\\)\\s*$", subj)`` et on resout la PR par + ``gh pr view N --json ...`` -- pas de liste, pas d'ambiguite. + C'est la voie nominale (squash-merge preserve le numero de PR + dans le sujet du commit, et c'est le seul invariant mesurable). + + 2. **Egalite normalisee du sujet** : a defaut de numero extractible, + le sujet integral (apres normalisation casse + espaces) doit etre + egal a un titre PR normalise. Toute intersection par jetons est + un faux positif structurel sur ce depot (notebook, guard, + training, slides sont des mots partout) et on l'interdit. + + 3. **Sinon None** : aucun match = aucun verdict. Le fail-CLOSED est + deja le bon defaut (REFUSE downstream). """ log_proc = run_git( wt_path, "log", "HEAD", "--format=%s", "-n", "20", check=False @@ -331,30 +347,59 @@ def lookup_pr_for_detached_head(wt_path: str) -> Optional[dict]: if not subjects: return None - # Match contre titres PR recents (limite 50) pour eviter explosion - pr_proc = run_gh( + # Voie 1 : resolution directe par numero extractible du sujet + # (squash-commit preserve "(#N)" a la fin du sujet). + pr_num_re = re.compile(r"\(#(\d+)\)\s*$") + direct_attempted = False + for subj in subjects: + m = pr_num_re.search(subj) + if not m: + continue + direct_attempted = True + pr_num = int(m.group(1)) + view_proc = run_gh( + "pr", "view", str(pr_num), + "--json", "number,state,url,title", + check=False, + ) + if view_proc.returncode != 0: + continue + try: + data = json.loads(view_proc.stdout) + except json.JSONDecodeError: + continue + if not data or "state" not in data: + continue + return data + # Si des sujets portaient (#N) mais qu'aucun n'a resolu, c'est un + # defaut d'autorite gh -- on n'invente rien, pas de fallback liste. + if direct_attempted: + return None + + # Voie 2 : egalite normalisee du sujet contre titres PR recents. + # Garde-fou : `limit 50` uniquement pour borner le cout d'appel. + list_proc = run_gh( "pr", "list", "--state", "all", "--limit", "50", "--json", "number,state,url,title", check=False, ) - if pr_proc.returncode != 0: + if list_proc.returncode != 0: return None try: - prs = json.loads(pr_proc.stdout) + prs = json.loads(list_proc.stdout) except json.JSONDecodeError: return None - # Heuristique : on prend la 1ere PR dont un mot-clé du titre matche un - # mot du commit subject. Conservateur : on exige au moins 4 chars - # communs, ce qui limite les faux positifs sur "fix", "test", etc. + def _normalize(s: str) -> str: + # strip + lower + collapse whitespace ; retire ponctuation terminale + s = s.strip().lower() + s = re.sub(r"\s+", " ", s) + return s.rstrip(".!?") + + subj_norm_set = {_normalize(s) for s in subjects} for pr in prs: - title_tokens = { - t.lower() for t in re.findall(r"[A-Za-z0-9-]{4,}", pr["title"]) - } - for subj in subjects: - subj_tokens = {t.lower() for t in re.findall(r"[A-Za-z0-9-]{4,}", subj)} - if title_tokens & subj_tokens: - return pr + if _normalize(pr["title"]) in subj_norm_set: + return pr return None @@ -518,32 +563,71 @@ def apply_removal(wt: WorktreeStatus) -> tuple[bool, str]: return False, proc.stderr.strip() -def render_text(statuses: list[WorktreeStatus], dry_run: bool) -> str: - """Rendu texte canonique (lisible humain).""" - verb = "WOULD REMOVE" if dry_run else "REMOVED" +def render_text( + statuses: list[WorktreeStatus], + dry_run: bool, + apply_results: Optional[list[dict]] = None, +) -> str: + """Rendu texte canonique (lisible humain) (#14476). + + `apply_results` est la liste exacte retournee par `apply_removal` : + `[{path, branch, pr_number, applied, stderr}, ...]`. En mode `--apply`, + on imprime `REMOVED` UNIQUEMENT pour les entrees dont `applied=True`. + Si `git worktree remove` a echoue (worktree sale par exemple), on + imprime `FAILED` avec le stderr -- c'est lisible, factuel, et JAMAIS + mensonger sur ce qui a effectivement quitte le disque. + """ + # Indexation par path pour une reconciliation O(1) + applied_by_path = {} + if apply_results is not None: + for r in apply_results: + applied_by_path[r["path"]] = r + lines: list[str] = [] - counts = {"REMOVE": 0, "REFUSE": 0, "SKIP_CURRENT": 0} + counts = {"REMOVE": 0, "REFUSE": 0, "SKIP_CURRENT": 0, "FAILED": 0} for s in statuses: counts[s.decision] = counts.get(s.decision, 0) + 1 if s.decision == "REMOVE": - label = f"{verb} {s.path}" - if s.branch: - label += f" branch={s.branch}" - if s.pr_state and s.pr_number: - label += f" pr=#{s.pr_number}({s.pr_state})" - lines.append(label) + branch_part = f"branch={s.branch}" if s.branch else "no_branch" + pr_part = ( + f"pr=#{s.pr_number}({s.pr_state})" + if s.pr_state and s.pr_number else "" + ) + if dry_run: + lines.append( + f"WOULD REMOVE {s.path} {branch_part} {pr_part}".rstrip() + ) + else: + # Mode --apply : vraie realite du disque. + result = applied_by_path.get(s.path) + if result is None or result.get("applied"): + lines.append( + f"REMOVED {s.path} {branch_part} {pr_part}".rstrip() + ) + else: + # `git worktree remove` a echoue : on dit FAILED + cause. + # Counts['FAILED'] n'est pas une decision de WorktreeStatus, + # c'est un evenement d'application ; ne s'ajoute pas a + # refused qui reste REFUSE semantique. + stderr = result.get("stderr") or "unknown error" + lines.append( + f"FAILED {s.path} {branch_part} {pr_part} " + f"apply_error={stderr[:120]}" + ) + counts["FAILED"] = counts.get("FAILED", 0) + 1 elif s.decision == "REFUSE": branch_part = f"branch={s.branch}" if s.branch else "no_branch" lines.append( - f"REFUSE {s.path} {branch_part} reason={s.refusal_reason}" + f"REFUSE {s.path} {branch_part} reason={s.refusal_reason}" ) elif s.decision == "SKIP_CURRENT": - lines.append(f"SKIP {s.path} reason=current_worktree") + lines.append(f"SKIP {s.path} reason=current_worktree") lines.append("---") lines.append( f"total={len(statuses)} " f"removable={counts.get('REMOVE', 0)} " f"refused={counts.get('REFUSE', 0)} " + f"failed={counts.get('FAILED', 0)} " f"skipped={counts.get('SKIP_CURRENT', 0)}" ) return "\n".join(lines) @@ -634,10 +718,12 @@ def main() -> int: print(json.dumps(out, indent=2, ensure_ascii=False)) else: # Texte - print(render_text(statuses, dry_run=not args.apply)) if args.apply: + print(render_text(statuses, dry_run=False, apply_results=apply_results)) print() print(f"applied={removal_count} errors={error_count}") + else: + print(render_text(statuses, dry_run=True)) # Exit code if error_count > 0: diff --git a/scripts/tests/test_prune_merged_worktrees.py b/scripts/tests/test_prune_merged_worktrees.py index f373e58d75..370ce28ba5 100644 --- a/scripts/tests/test_prune_merged_worktrees.py +++ b/scripts/tests/test_prune_merged_worktrees.py @@ -193,6 +193,240 @@ def test_to_dict_round_trip(self): assert k in d, f"missing key: {k}" +# --------------------------------------------------------------------------- +# Tests lookup_pr_for_detached_head (#14476) -- anti faux-positifs +# --------------------------------------------------------------------------- + + +class TestLookupPRForDetachedHead: + """Tests unitaires du verdict par contenu pour HEAD detaché (#14476). + + Avant #14476 : intersection de jetons `re.findall(r"[A-Za-z0-9-]{4,}")` + entre titre PR et sujet commit -- prenait la 1ere PR dont au moins un + mot >=4 chars matchait. Cause structurelle de faux positifs massifs + (notebook, guard, training, slides sont des mots partout). + + Apres #14476 : 1) resolution directe par numero si ``re.search(r"\\(#\\d+\\)$")`` + extrait du sujet ; 2) sinon egalite normalisee stricte ; 3) sinon None. + """ + + def test_subject_without_pr_number_no_match(self, monkeypatch): + """Sujet sans `(#N)` retourne par defaut le verdict `None` quand la + liste de PRs recentes est vide ou sans egalite. + """ + # Stub run_gh : pas de PR list exploitable + monkeypatch.setattr(pmw, "run_gh", lambda *a, **k: _fake_proc( + returncode=0, + json_payload=[], + )) + # Pas de run_git OK -> pas de sujet exploitable non plus -> None + # est deja valide. Ici on verifie qu'un sujet SANS (#N) et une + # liste vide rendent bien None, jamais une PR par defaut. + import pytest + # Worktree path bidon ; on stub run_git aussi. + monkeypatch.setattr( + pmw, "run_git", + lambda *a, **k: _fake_proc(returncode=128, stdout=""), + ) + result = pmw.lookup_pr_for_detached_head("/tmp/fake") + assert result is None + + def test_subject_with_pr_number_resolves_directly(self, monkeypatch): + """Sujet `fix: chg (#14476)` -> `gh pr view 14476` direct. + + C'est la voie nominale post-#14476 (squash-merge preserve le + numero de PR dans le sujet du commit). On verifie qu'on ne tombe + PAS dans la voie liste -- le PR rendu vient de `gh pr view N`, + pas d'une intersection par jetons. + """ + # run_git retourne un sujet avec `(#14476)` + monkeypatch.setattr( + pmw, "run_git", + lambda *a, **k: _fake_proc( + returncode=0, + stdout="fix(scripts,#14476): prune bug repair (#14476)\n", + ), + ) + + gh_calls: list[list] = [] + + def fake_run_gh(*args, **kwargs): + gh_calls.append(list(args)) + cmd = list(args) + # `gh pr view 14476 --json ...` -- la voie directe + if "view" in cmd and "14476" in cmd: + return _fake_proc( + returncode=0, + json_payload={ + "number": 14476, + "state": "MERGED", + "url": "https://github.com/jsboige/CoursIA/pull/14476", + "title": "fix(scripts,#14476): prune bug repair", + }, + ) + # Liste vide en fallback (au cas où on tomberait dans la voie 2) + return _fake_proc(returncode=0, json_payload=[]) + + monkeypatch.setattr(pmw, "run_gh", fake_run_gh) + + result = pmw.lookup_pr_for_detached_head("/tmp/fake") + assert result is not None + assert result["number"] == 14476 + assert result["state"] == "MERGED" + # On a appelé la vue directe, pas la liste + assert any( + "view" in c and "14476" in c + for c in gh_calls + ), f"expected direct pr view call, got {gh_calls}" + + def test_subject_share_token_returns_none(self, monkeypatch): + """Faux positif élimine : sujet `fix(notebook): ...` partage `notebook` + avec un titre PR récent, mais aucun match par numero ni par egalite + normalisee. AVANT #14476, ce cas retournait la 1ere PR dont le + titre contenait `notebook`, ce qui causait le retrait d'un + worktree encore actif. + """ + # Sujet sans `(#N)` -- force la voie liste + monkeypatch.setattr( + pmw, "run_git", + lambda *a, **k: _fake_proc( + returncode=0, + stdout="fix(notebook): unrelated work in progress\n", + ), + ) + + # PRs recentes qui partagent `notebook` mais ne sont PAS ce commit + monkeypatch.setattr( + pmw, "run_gh", + lambda *a, **k: _fake_proc( + returncode=0, + json_payload=[ + { + "number": 14437, + "state": "MERGED", + "url": "https://github.com/jsboige/CoursIA/pull/14437", + "title": "feat(notebook): something else entirely", + }, + { + "number": 14195, + "state": "OPEN", + "url": "https://github.com/jsboige/CoursIA/pull/14195", + "title": "refactor(scripts): prune merged worktrees", + }, + ], + ), + ) + result = pmw.lookup_pr_for_detached_head("/tmp/fake") + assert result is None, ( + "faux positif elimine : intersection de jetons interdite, " + "egalite normalisee impossible (sujet != titre). Resultat doit " + "etre None, pas une PR partageant `notebook`." + ) + + +def _fake_proc(returncode: int = 0, stdout: str = "", json_payload=None): + """Construit un subprocess.CompletedProcess minimal pour stubbing.""" + import subprocess + out = stdout + if json_payload is not None: + out = json.dumps(json_payload) + return subprocess.CompletedProcess( + args=[], returncode=returncode, stdout=out, stderr="" + ) + + +# --------------------------------------------------------------------------- +# Tests render_text -- fidelite au disque en mode --apply (#14476) +# --------------------------------------------------------------------------- + + +class TestRenderText: + """Le rendu texte ne doit JAMAIS mentir sur ce qui a quitte le disque. + + Avant #14476 : `render_text` lisait depuis `s.decision`, et affichait + `REMOVED` pour tout `decision="REMOVE"` même si `git worktree remove` + avait échoué (worktree sale par exemple). Le compteur `removable` + devenait alors un mensonge structurel. + """ + + def test_apply_results_failed_path_is_not_printed_as_removed(self): + """Fixture : un `apply_results` avec `applied=False` pour un path + dont `WorktreeStatus.decision == "REMOVE"`. La sortie NE DOIT PAS + contenir `REMOVED` pour ce path ; elle DOIT contenir `FAILED` ou + une trace explicite de l'échec. + """ + statuses = [ + pmw.WorktreeStatus( + path="C:/dev/CoursIA-FAILED", + branch="fix/14195-x", + is_current=False, + pr_state="MERGED", + pr_number=14427, + pr_url="https://github.com/jsboige/CoursIA/pull/14427", + ahead_count=0, + has_source_dirty=False, + untracked_paths=[], + decision="REMOVE", + refusal_reason=None, + ), + pmw.WorktreeStatus( + path="C:/dev/CoursIA-OK", + branch="fix/14195-y", + is_current=False, + pr_state="MERGED", + pr_number=14403, + pr_url="https://github.com/jsboige/CoursIA/pull/14403", + ahead_count=0, + has_source_dirty=False, + untracked_paths=[], + decision="REMOVE", + refusal_reason=None, + ), + ] + apply_results = [ + { + "path": "C:/dev/CoursIA-FAILED", + "branch": "fix/14195-x", + "pr_number": 14427, + "applied": False, + "stderr": "fatal: 'C:/dev/CoursIA-FAILED' contains modified or untracked files", + }, + { + "path": "C:/dev/CoursIA-OK", + "branch": "fix/14195-y", + "pr_number": 14403, + "applied": True, + "stderr": "", + }, + ] + out = pmw.render_text( + statuses, dry_run=False, apply_results=apply_results + ) + # Le path FAILED ne doit jamais apparaitre en REMOVED ; il doit + # apparaitre en FAILED avec la cause d'erreur. + failed_lines = [ + l for l in out.splitlines() + if "C:/dev/CoursIA-FAILED" in l + ] + assert failed_lines, "FAILED path missing from output" + for line in failed_lines: + assert "REMOVED" not in line, ( + f"render_text ment : line={line!r} claim REMOVED pour un " + f"path dont applied=False. apply_results gagne, pas decision." + ) + assert "FAILED" in line, ( + f"expected FAILED marker, got: {line!r}" + ) + # Le compteur failed doit etre non-nul dans la sortie + assert "failed=1" in out + # Le path OK doit apparaitre en REMOVED + ok_lines = [ + l for l in out.splitlines() + if "C:/dev/CoursIA-OK" in l and "REMOVED" in l + ] + assert ok_lines, "REMOVED path missing from output" + + # --------------------------------------------------------------------------- # Tests d'integration subprocess sur le repo reel # ---------------------------------------------------------------------------