diff --git a/mise-tasks/branch-cleanup b/mise-tasks/branch-cleanup index 57e9fca..72ec073 100755 --- a/mise-tasks/branch-cleanup +++ b/mise-tasks/branch-cleanup @@ -1,9 +1,10 @@ #!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.12" -# dependencies = ["rich>=13.0.0", "typer>=0.15.0"] +# dependencies = ["httpx>=0.28.0", "rich>=13.0.0", "typer>=0.15.0"] # /// #MISE description="Delete branches that have been merged into main (local and remote)" +#MISE alias="bc" #USAGE flag "--dry-run" help="Show what would be deleted without deleting" #USAGE flag "--yes" help="Skip confirmation prompt" #USAGE flag "--local-only" help="Only clean up local branches" @@ -11,11 +12,15 @@ #USAGE flag "--cutoff " default="30" help="Only delete branches whose HEAD commit is older than N days (default 30)" """Clean up merged branches locally and on the Forgejo remote. -Finds branches that have been fully merged into main whose HEAD commit -is older than a cutoff (default 30 days), then offers to delete them. -Uses git operations directly (SSH for remote), no API token needed. +Detects merged branches via two methods: + 1. git branch --merged (catches fast-forward merges) + 2. Forgejo API (catches squash-merged PRs) -Protected branches (main) are never deleted. +For remote branches, deletes those confirmed merged by either method. +For local branches, deletes those that match a merged remote branch or +are detected as merged by git. + +Warns about stale local branches that couldn't be confirmed as merged. Usage: mise run branch-cleanup # interactive cleanup (30-day cutoff) @@ -28,11 +33,14 @@ import subprocess from datetime import datetime, timezone from typing import Annotated +import httpx import typer from rich.console import Console from rich.table import Table PROTECTED_BRANCHES = {"main", "master"} +FORGE_API = "https://forge.ops.eblu.me/api/v1" +REPO = "eblume/blumeops" def run_git(*args: str) -> str: @@ -58,37 +66,101 @@ def branch_head_age_days(ref: str) -> int | None: return None -def get_merged_local_branches() -> list[str]: - """Get local branches that are fully merged into main.""" - output = run_git("branch", "--merged", "main") - branches = [] +def get_git_merged_local_branches() -> set[str]: + """Get local branches that are fully merged into main (fast-forward).""" + try: + output = run_git("branch", "--merged", "main") + except subprocess.CalledProcessError: + return set() + branches = set() for line in output.splitlines(): name = line.strip().lstrip("* ") if name and name not in PROTECTED_BRANCHES: - branches.append(name) - return sorted(branches) + branches.add(name) + return branches -def get_merged_remote_branches() -> list[str]: - """Get remote branches that are fully merged into origin/main.""" - output = run_git("branch", "-r", "--merged", "origin/main") - branches = [] +def get_git_merged_remote_branches() -> set[str]: + """Get remote branches that are fully merged into origin/main (fast-forward).""" + try: + output = run_git("branch", "-r", "--merged", "origin/main") + except subprocess.CalledProcessError: + return set() + branches = set() for line in output.splitlines(): name = line.strip() if " -> " in name: - continue # skip HEAD pointer + continue if not name.startswith("origin/"): continue short = name.removeprefix("origin/") if short not in PROTECTED_BRANCHES: - branches.append(short) - return sorted(branches) + branches.add(short) + return branches + + +def get_all_local_branches() -> set[str]: + """Get all local branch names.""" + output = run_git("branch") + branches = set() + for line in output.splitlines(): + name = line.strip().lstrip("* ") + if name and name not in PROTECTED_BRANCHES: + branches.add(name) + return branches + + +def get_all_remote_branches() -> set[str]: + """Get all remote branch names (short form, without origin/ prefix).""" + output = run_git("branch", "-r") + branches = set() + for line in output.splitlines(): + name = line.strip() + if " -> " in name: + continue + if not name.startswith("origin/"): + continue + short = name.removeprefix("origin/") + if short not in PROTECTED_BRANCHES: + branches.add(short) + return branches + + +def get_squash_merged_branches(console: Console) -> set[str]: + """Query Forgejo API for branch names from squash-merged PRs.""" + merged_branches: set[str] = set() + page = 1 + limit = 50 + + try: + with httpx.Client(timeout=15) as client: + while True: + resp = client.get( + f"{FORGE_API}/repos/{REPO}/pulls", + params={"state": "closed", "limit": limit, "page": page}, + ) + resp.raise_for_status() + prs = resp.json() + if not prs: + break + for pr in prs: + if pr.get("merged"): + ref = pr.get("head", {}).get("ref", "") + if ref and ref not in PROTECTED_BRANCHES: + merged_branches.add(ref) + page += 1 + except httpx.HTTPError as e: + console.print(f"[yellow]Warning:[/yellow] Failed to query Forgejo API: {e}") + console.print("[dim]Falling back to git-only merge detection[/dim]") + + return merged_branches def delete_local_branch(branch: str) -> tuple[bool, str]: """Delete a local branch. Returns (success, message).""" try: - run_git("branch", "-d", branch) + # Use -D since squash-merged branches aren't git-ancestors of main + run_git("branch", "-D", branch) return True, "deleted" except subprocess.CalledProcessError as e: return False, e.stderr.strip() @@ -140,76 +212,125 @@ def main( console.print(f"[red]Failed to fetch:[/red] {e.stderr}") raise typer.Exit(1) + # Gather merge info from both sources + console.print("[dim]Checking Forgejo for squash-merged PRs...[/dim]") + api_merged = get_squash_merged_branches(console) + git_merged_local = get_git_merged_local_branches() + git_merged_remote = get_git_merged_remote_branches() + + # Union of all confirmed-merged branch names + all_confirmed_merged = api_merged | git_merged_local | git_merged_remote + do_local = not remote_only do_remote = not local_only - local_merged = get_merged_local_branches() if do_local else [] - remote_merged = get_merged_remote_branches() if do_remote else [] + # Remote: delete if confirmed merged by either method + all_remote = get_all_remote_branches() if do_remote else set() + remote_to_delete = sorted(all_remote & all_confirmed_merged) - # Compute ages and filter by cutoff - all_names = sorted(set(local_merged) | set(remote_merged)) + # Local: delete if confirmed merged (git --merged OR matched a merged PR) + all_local = get_all_local_branches() if do_local else set() + local_to_delete = sorted(all_local & all_confirmed_merged) + + # Compute ages for all candidates + all_candidates = sorted(set(local_to_delete) | set(remote_to_delete)) branch_ages: dict[str, int | None] = {} - - for name in all_names: - # Prefer remote ref for age (more authoritative), fall back to local + for name in all_candidates: age = None - if name in remote_merged: + if name in all_remote: age = branch_head_age_days(f"origin/{name}") - if age is None and name in local_merged: + if age is None and name in all_local: age = branch_head_age_days(name) branch_ages[name] = age # Apply cutoff filter - local_branches = [ - b for b in local_merged - if (branch_ages.get(b) is not None and branch_ages[b] >= cutoff) # type: ignore[operator] + local_to_delete = [ + b for b in local_to_delete + if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator] ] - remote_branches = [ - b for b in remote_merged - if (branch_ages.get(b) is not None and branch_ages[b] >= cutoff) # type: ignore[operator] + remote_to_delete = [ + b for b in remote_to_delete + if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator] ] - skipped = [ - b for b in all_names - if b not in local_branches and b not in remote_branches - ] + filtered_candidates = sorted(set(local_to_delete) | set(remote_to_delete)) - if not local_branches and not remote_branches: - console.print("[green]No merged branches older than " - f"{cutoff} days to clean up.[/green]") - if skipped: - console.print(f"[dim]({len(skipped)} merged branches skipped — " - f"newer than {cutoff} days)[/dim]") - raise typer.Exit(0) + skipped_count = len(all_candidates) - len(filtered_candidates) - # Build summary table - table = Table(title=f"Merged branches to delete (older than {cutoff} days)") - table.add_column("Branch") - table.add_column("Age (days)", justify="right") - table.add_column("Local") - table.add_column("Remote") + if filtered_candidates: + # Build summary table + table = Table(title=f"Merged branches to delete (older than {cutoff} days)") + table.add_column("Branch") + table.add_column("Age (days)", justify="right") + table.add_column("Method") + table.add_column("Local") + table.add_column("Remote") - candidates = sorted(set(local_branches) | set(remote_branches)) - for branch in candidates: - has_local = branch in local_branches - has_remote = branch in remote_branches - age = branch_ages.get(branch) - age_str = str(age) if age is not None else "?" - table.add_row( - branch, - age_str, - "[yellow]delete[/yellow]" if has_local else "[dim]-[/dim]", - "[yellow]delete[/yellow]" if has_remote else "[dim]-[/dim]", + for branch in filtered_candidates: + has_local = branch in local_to_delete + has_remote = branch in remote_to_delete + age = branch_ages.get(branch) + age_str = str(age) if age is not None else "?" + + methods = [] + if branch in git_merged_local or branch in git_merged_remote: + methods.append("git") + if branch in api_merged: + methods.append("api") + method_str = "+".join(methods) + + table.add_row( + branch, + age_str, + method_str, + "[yellow]delete[/yellow]" if has_local else "[dim]-[/dim]", + "[yellow]delete[/yellow]" if has_remote else "[dim]-[/dim]", + ) + + console.print(table) + console.print( + f"\n[bold]{len(local_to_delete)}[/bold] local, " + f"[bold]{len(remote_to_delete)}[/bold] remote branches to delete" ) + if skipped_count: + console.print(f"[dim]({skipped_count} merged branches skipped — " + f"newer than {cutoff} days)[/dim]") + else: + console.print(f"[green]No merged branches older than " + f"{cutoff} days to clean up.[/green]") + if skipped_count: + console.print(f"[dim]({skipped_count} merged branches skipped — " + f"newer than {cutoff} days)[/dim]") - console.print(table) - console.print( - f"\n[bold]{len(local_branches)}[/bold] local, " - f"[bold]{len(remote_branches)}[/bold] remote branches to delete" - ) - if skipped: - console.print(f"[dim]({len(skipped)} merged branches skipped — " - f"newer than {cutoff} days)[/dim]") + # Warn about stale unmerged local branches + if do_local: + unmerged_local = all_local - all_confirmed_merged + stale_unmerged = [] + for name in sorted(unmerged_local): + age = branch_head_age_days(name) + if age is not None and age >= cutoff: + stale_unmerged.append((name, age)) + + if stale_unmerged: + console.print() + warn_table = Table( + title=f"[yellow]Warning:[/yellow] Stale unmerged local branches " + f"(older than {cutoff} days)", + title_style="", + ) + warn_table.add_column("Branch") + warn_table.add_column("Age (days)", justify="right") + for name, age in stale_unmerged: + warn_table.add_row(f"[yellow]{name}[/yellow]", str(age)) + console.print(warn_table) + console.print( + f"[dim]These {len(stale_unmerged)} branches have no merged PR on " + f"Forgejo and are not git-ancestors of main.\n" + f"They may contain work-in-progress — inspect manually before deleting.[/dim]" + ) + + if not filtered_candidates: + raise typer.Exit(0) if dry_run: console.print("\n[dim]Dry run — no branches were deleted.[/dim]") @@ -220,26 +341,26 @@ def main( console.print("[dim]Aborted.[/dim]") raise typer.Exit(0) - # Delete local branches - if local_branches: - console.print("\n[bold]Deleting local branches...[/bold]") - for branch in local_branches: - ok, msg = delete_local_branch(branch) - if ok: - console.print(f" [green]✓[/green] {branch}") - else: - console.print(f" [red]✗[/red] {branch}: {msg}") - - # Delete remote branches - if remote_branches: + # Delete remote branches first (source of truth) + if remote_to_delete: console.print("\n[bold]Deleting remote branches...[/bold]") - for branch in remote_branches: + for branch in remote_to_delete: ok, msg = delete_remote_branch(branch) if ok: console.print(f" [green]✓[/green] origin/{branch}") else: console.print(f" [red]✗[/red] origin/{branch}: {msg}") + # Delete local branches + if local_to_delete: + console.print("\n[bold]Deleting local branches...[/bold]") + for branch in local_to_delete: + ok, msg = delete_local_branch(branch) + if ok: + console.print(f" [green]✓[/green] {branch}") + else: + console.print(f" [red]✗[/red] {branch}: {msg}") + console.print("\n[green]Done.[/green]")