#!/usr/bin/env python3
"""Conservative, recoverable branch and worktree cleanup. No third-party modules."""

import argparse
import datetime
import fnmatch
import functools
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import uuid


if sys.version_info < (3, 11):
    sys.exit('git cleanup: Python 3.11 or newer is required')


class CleanupError(Exception):
    pass


def environment():
    env = os.environ.copy()
    # Always operate on the repository in cwd, never an inherited index/fixture.
    selectors = {
        'GIT_DIR', 'GIT_WORK_TREE', 'GIT_COMMON_DIR', 'GIT_INDEX_FILE',
        'GIT_OBJECT_DIRECTORY', 'GIT_ALTERNATE_OBJECT_DIRECTORIES',
        'GIT_NAMESPACE', 'GIT_SHALLOW_FILE', 'GIT_REPLACE_REF_BASE',
        'GIT_GRAFT_FILE', 'GIT_PREFIX', 'GIT_INTERNAL_SUPER_PREFIX',
    }
    for key in list(env):
        if key in selectors:
            del env[key]
    env.update(GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1', GIT_NO_LAZY_FETCH='1',
               GIT_LITERAL_PATHSPECS='1', LC_ALL='C')
    return env


ENV = environment()


def execute(argv, *, cwd=None, data=None, allowed=(0,), env=None):
    result = subprocess.run(argv, cwd=cwd, env=ENV if env is None else env, input=data,
                            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    if result.returncode not in allowed:
        detail = result.stderr.decode(errors='replace').strip()
        raise CleanupError(f'{argv[0]} failed ({result.returncode}): {detail}')
    return result


def git(*args, cwd=None, data=None, allowed=(0,), settings=()):
    env = ENV
    if settings:
        env = ENV.copy()
        try:
            count = int(env.get('GIT_CONFIG_COUNT', '0'))
        except ValueError as error:
            raise CleanupError('invalid GIT_CONFIG_COUNT') from error
        if count < 0:
            raise CleanupError('invalid GIT_CONFIG_COUNT')
        for key, value in settings:
            env[f'GIT_CONFIG_KEY_{count}'] = key
            env[f'GIT_CONFIG_VALUE_{count}'] = value
            count += 1
        env['GIT_CONFIG_COUNT'] = str(count)
    return execute(['git', '--no-replace-objects', '-c', 'color.ui=false',
                    *args], cwd=cwd, data=data, allowed=allowed, env=env)


def transport_git(endpoint, command, *options, references=()):
    # get-url already applies URL rewriting. Feeding that URL back to Git can
    # rewrite it a second time, with different fetch and push destinations.
    # An explicit pushurl and a unique exact rewrite perform one pinned mapping.
    remote = 'cleanup-transport-' + uuid.uuid4().hex
    token = 'cleanup-endpoint-' + uuid.uuid4().hex
    settings = [(f'remote.{remote}.url', token),
                (f'remote.{remote}.pushurl', token),
                (f'url.{endpoint}.insteadOf', token)]
    return git(command, *options, '--', remote, *references, settings=settings)


def output(*args, **kwargs):
    # Whitespace can be part of a filesystem path. Remove only Git's record
    # terminator, or a sibling repository can be mistaken for the current one.
    return git(*args, **kwargs).stdout.decode(errors='surrogateescape').removesuffix('\n')


def config(key, *, many=False):
    result = git('config', '--get-all' if many else '--get', key, allowed=(0, 1))
    values = result.stdout.decode(errors='surrogateescape').splitlines()
    return values if many else (values[-1] if values else None)


def resolve(ref):
    result = git('rev-parse', '--verify', '--quiet', '--end-of-options',
                 ref + '^{commit}', allowed=(0, 1))
    return result.stdout.decode().strip() if result.returncode == 0 else None


def full_ref(ref):
    return output('rev-parse', '--symbolic-full-name', '--verify',
                  '--end-of-options', ref)


@functools.lru_cache(maxsize=None)
def ancestor(old, new):
    result = git('merge-base', '--is-ancestor', old, new, allowed=(0, 1))
    return result.returncode == 0


def worktrees():
    rows = []
    for record in git('worktree', 'list', '--porcelain', '-z').stdout.split(b'\0\0'):
        if not record:
            continue
        fields = {}
        for field in record.split(b'\0'):
            key, _, value = field.partition(b' ')
            fields[key.decode()] = os.fsdecode(value)
        rows.append(fields)
    return rows


def refs():
    rows = {}
    fmt = '%(refname)%00%(objectname)%00%(symref)'
    for row in output('for-each-ref', '--format=' + fmt,
                      'refs/heads/', 'refs/remotes/').splitlines():
        name, oid, symbolic = row.split('\0')
        rows[name] = {'ref': name, 'oid': oid, 'symbolic': symbolic}
    return rows


def short(ref):
    for prefix in ('refs/heads/', 'refs/remotes/'):
        if ref.startswith(prefix):
            return ref[len(prefix):]
    return ref


def worktree_problem(wt, primary, current):
    path = Path(wt['worktree'])
    if path.resolve() == primary:
        return 'primary worktree'
    if path.resolve() == current:
        return 'current worktree'
    if 'locked' in wt:
        return 'locked worktree'
    if 'prunable' in wt or not path.is_dir():
        return 'missing worktree; inspect registration manually'
    actual = Path(output('rev-parse', '--show-toplevel', cwd=path)).resolve()
    if actual != path.resolve():
        return 'worktree repository mismatch'
    for marker in ('MERGE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD', 'BISECT_LOG',
                   'rebase-merge', 'rebase-apply', 'sequencer', 'index.lock'):
        marker_path = output('rev-parse', '--path-format=absolute', '--git-path',
                             marker, cwd=path)
        if Path(marker_path).exists():
            return 'worktree has an operation in progress'
    # Git status and worktree remove both trust these index flags; either can
    # hide real edits that no commit/recovery ref could restore after removal.
    index = git('-c', 'core.fsmonitor=false', 'ls-files', '-v', '-z', cwd=path).stdout
    if any(row and (row[:1].islower() or row[:1] == b'S')
           for row in index.split(b'\0')):
        return 'worktree has assume-unchanged or skip-worktree index entries'
    status = git('-c', 'core.fsmonitor=false', '-c', 'core.untrackedCache=false',
                 '-c', 'core.trustctime=true', '-c', 'core.checkStat=default',
                 '-c', 'core.filemode=true', '-c', 'core.ignoreStat=false',
                 'status', '--porcelain=v1', '-z', '--untracked-files=all',
                 '--ignored=matching', '--ignore-submodules=none', cwd=path).stdout
    if status:
        return 'worktree has modified, untracked, or ignored files'
    return None


@functools.lru_cache(maxsize=None)
def source_paths(source, common):
    """Reject ambiguous source histories before inferring squash incorporation."""
    commits = output('rev-list', '--max-count=1001', common + '..' + source).splitlines()
    if not commits or len(commits) > 1000:
        return ()
    if output('rev-list', '--merges', common + '..' + source):
        return ()
    trees = output('log', '--format=%T', common + '..' + source).splitlines()
    trees.append(output('rev-parse', common + '^{tree}'))
    if len(trees) != len(set(trees)):
        return ()
    changed = set(git('diff', '--no-ext-diff', '--no-textconv',
                      '--ignore-submodules=none', '--no-renames', '--name-only',
                      '-z', common, source, '--').stdout.split(b'\0')) - {b''}
    touched = set()
    for commit in commits:
        touched.update(git('diff-tree', '--no-commit-id', '--no-renames',
                           '--name-only', '-z', '-r', commit).stdout.split(b'\0'))
    touched.discard(b'')
    if not changed or touched != changed:
        return ()
    return tuple(os.fsdecode(path) for path in sorted(changed))


@functools.lru_cache(maxsize=None)
def content_matches(source, target):
    # A branch restoring an older result is new work, not evidence of landing.
    if ancestor(target, source):
        return False
    result = git('merge-base', '--all', source, target, allowed=(0, 1))
    common = result.stdout.decode().splitlines()
    if len(common) != 1:
        return False
    paths = source_paths(source, common[0])
    if not paths:
        return False
    for start in range(0, len(paths), 100):
        result = git('diff', '--quiet', '--no-ext-diff', '--no-textconv',
                     '--ignore-submodules=none', source, target, '--',
                     *paths[start:start + 100], allowed=(0, 1))
        if result.returncode:
            return False
    return True


class Cleanup:
    def __init__(self, args):
        self.args = args
        self.root = Path(output('rev-parse', '--show-toplevel')).resolve()
        os.chdir(self.root)
        self.common = Path(output('rev-parse', '--path-format=absolute',
                                  '--git-common-dir'))
        # Older supported Git versions do not honor GIT_NO_LAZY_FETCH. Reject
        # promisor repositories before any object reads or explicit fetches.
        partial = git('config', '--get-regexp',
                      r'^(extensions\.partialclone|remote\..*\.promisor)$',
                      allowed=(0, 1))
        if partial.returncode == 0:
            raise CleanupError('partial clones are unsupported; use a complete clone')
        if output('rev-parse', '--is-shallow-repository') == 'true':
            raise CleanupError('shallow history cannot establish cleanup proof')
        if (self.common / 'info/grafts').exists():
            raise CleanupError('legacy grafts cannot establish cleanup proof')
        self.remote = None
        self.fetch_url = None
        self.push_url = None
        base = args.base if args.base is not None else config('cleanup.base')
        if base == '':
            raise CleanupError('cleanup.base must not be empty')
        self.main_setting = config('cleanup.mainBranch')
        if self.main_setting == '':
            raise CleanupError('cleanup.mainBranch must not be empty')
        remotes = output('remote').splitlines()
        if not args.local:
            if args.remote is not None:
                if args.remote not in remotes:
                    raise CleanupError(f'unknown remote: {args.remote}')
                self.remote = args.remote
            elif 'origin' in remotes:
                self.remote = 'origin'
            elif len(remotes) == 1:
                self.remote = remotes[0]
            elif remotes:
                raise CleanupError('choose --remote NAME or --local for multiple remotes')
        if self.remote:
            self.fetch_url = output('remote', 'get-url', self.remote)
            pushes = output('remote', 'get-url', '--push', '--all', self.remote).splitlines()
            if pushes == [self.fetch_url]:
                self.push_url = pushes[0]
            if args.fetch or args.apply:
                transport_git(self.fetch_url, 'fetch', '--no-tags', '--no-prune-tags',
                              '--recurse-submodules=no', '--no-write-fetch-head', '--prune',
                              references=(f'+refs/heads/*:refs/remotes/{self.remote}/*',))
        self.refs = refs()
        self.wts = worktrees()
        self.primary = Path(self.wts[0]['worktree']).resolve()
        self.main = self.integration_branch()
        if base is None:
            candidates = []
            if self.remote:
                candidates.append(f'refs/remotes/{self.remote}/{self.main}')
            candidates.append(f'refs/heads/{self.main}')
            base = next((ref for ref in candidates if ref in self.refs), None)
        if not base or not resolve(base):
            raise CleanupError('cannot identify integration baseline; use --base REF')
        self.base = self.anchor(base)
        self.anchors = [self.base]
        for ref in args.into:
            anchor = self.anchor(ref)
            if anchor not in self.anchors:
                self.anchors.append(anchor)
        self.protected = {f'refs/heads/{name}' for name in ('main', 'master', self.main)}
        if self.remote:
            self.protected.update(f'refs/remotes/{self.remote}/{name}'
                                  for name in ('main', 'master', self.main))
            default = self.refs.get(f'refs/remotes/{self.remote}/HEAD', {}).get('symbolic')
            if default:
                self.protected.add(default)
                self.protected.add('refs/heads/' + default.split('/', 3)[-1])
        self.protected.update(a['ref'] for a in self.anchors if a['ref'])
        self.patterns = config('cleanup.protect', many=True) + args.protect
        self.obsolete = set()
        for value in args.obsolete:
            candidates = [value] if value.startswith('refs/') else [
                'refs/heads/' + value, 'refs/remotes/' + value]
            found = [r for r in candidates if r in self.refs]
            if len(found) != 1:
                raise CleanupError(f'obsolete branch is missing or ambiguous: {value}')
            self.obsolete.add(found[0])
        self.prs = self.github_prs() if args.github else []

    def integration_branch(self):
        name = self.main_setting
        if name:
            git('check-ref-format', 'refs/heads/' + name)
            return name
        if self.remote:
            target = self.refs.get(f'refs/remotes/{self.remote}/HEAD', {}).get('symbolic')
            prefix = f'refs/remotes/{self.remote}/'
            if target and target.startswith(prefix) and target in self.refs:
                return target[len(prefix):]
        for name in ('main', 'master'):
            if f'refs/heads/{name}' in self.refs or (self.remote and
                    f'refs/remotes/{self.remote}/{name}' in self.refs):
                return name
        return ''

    def anchor(self, value):
        ref = full_ref(value)
        ref_oid = output('rev-parse', '--verify', '--end-of-options', ref or value)
        # Peel the captured object, never the mutable ref a second time.
        oid = resolve(ref_oid)
        if not oid:
            raise CleanupError(f'recipient does not resolve to a commit: {value}')
        return {'ref': ref, 'oid': oid, 'ref_oid': ref_oid, 'label': value}

    def github_prs(self):
        if not self.remote:
            raise CleanupError('--github requires a selected remote (not --local)')
        # Pin the repository; GH_REPO and unrelated remotes must not redirect it.
        match = re.fullmatch(r'(?:https://github\.com/|ssh://git@github\.com/|git@github\.com:)([^/]+/[^/]+?)(?:\.git)?', self.fetch_url)
        if not match:
            raise CleanupError('--github currently requires a github.com remote URL')
        try:
            prs = []
            for state in ('open', 'merged'):
                result = execute(['gh', 'pr', 'list', '--repo', 'github.com/' + match[1], '--state', state,
                                  '--limit', '1000', '--json',
                                  'number,state,headRefName,headRefOid,mergeCommit,isCrossRepository'])
                page = json.loads(result.stdout)
                required = {'number', 'state', 'headRefName', 'headRefOid',
                            'mergeCommit', 'isCrossRepository'}
                if not isinstance(page, list) or any(not isinstance(p, dict) or
                        not required <= p.keys() for p in page):
                    raise ValueError('expected complete PR records')
                if state == 'open' and len(page) >= 1000:
                    raise ValueError('open PR limit reached; cannot protect all open branches')
                prs.extend(page)
            return [p for p in prs if p.get('isCrossRepository') is False]
        except (ValueError, TypeError) as error:
            raise CleanupError(f'cannot parse GitHub PR evidence: {error}') from error

    def remote_branch(self, ref):
        prefix = f'refs/remotes/{self.remote}/'
        return ref[len(prefix):] if self.remote and ref.startswith(prefix) else None

    def proof(self, row, anchors):
        for anchor in anchors:
            if row['ref'].startswith('refs/remotes/') and not self.remote_branch(anchor['ref']):
                continue
            target = anchor['oid']
            if ancestor(row['oid'], target):
                return {'kind': 'ancestor', 'anchor': anchor, 'landing': target}
            name = self.remote_branch(row['ref']) or short(row['ref'])
            for pr in self.prs:
                merge = pr.get('mergeCommit') or {}
                landing = merge.get('oid')
                if (pr.get('state') == 'MERGED' and pr.get('headRefName') == name
                        and pr.get('headRefOid') == row['oid'] and landing
                        and resolve(landing) and ancestor(landing, target)):
                    return {'kind': f'merged PR #{pr["number"]}', 'anchor': anchor,
                            'landing': landing}
            targets = [target]
            if self.args.scan_history:
                targets = output('rev-list', f'--max-count={self.args.scan_history}',
                                 target).splitlines()
            for landing in targets:
                if content_matches(row['oid'], landing):
                    return {'kind': 'exact changed-path contents', 'anchor': anchor,
                            'landing': landing}
        return None

    def plan(self):
        entries = []
        selected = set()
        for ref, row in self.refs.items():
            if not ref.startswith('refs/heads/') and self.remote_branch(ref) is None:
                continue
            selected.add(ref)
            entry = {**row, 'action': 'keep', 'reason': '', 'worktrees': [], 'proof': None}
            associated = [wt for wt in self.wts if wt.get('branch') == ref]
            entry['worktrees'] = [wt['worktree'] for wt in associated]
            if row['symbolic']:
                entry['reason'] = 'symbolic ref'
            elif ref in self.protected or any(fnmatch.fnmatchcase(name, p)
                    for p in self.patterns
                    for name in (ref, short(ref), self.remote_branch(ref) or short(ref))):
                entry['reason'] = 'protected branch or recipient'
            else:
                for wt in associated:
                    problem = worktree_problem(wt, self.primary, self.root)
                    if problem:
                        entry['reason'] = problem
                        break
            name = self.remote_branch(ref) or short(ref)
            if not entry['reason'] and any(pr.get('state') == 'OPEN' and
                    pr.get('headRefName') == name for pr in self.prs):
                entry['reason'] = 'open GitHub PR'
            if not entry['reason'] and self.remote_branch(ref) and not self.push_url:
                entry['reason'] = 'remote has different or multiple fetch/push URLs'
            if not entry['reason']:
                if ref in self.obsolete:
                    entry['action'] = 'delete'
                    entry['reason'] = 'explicitly obsolete'
                else:
                    proof = self.proof(row, self.anchors)
                    if proof:
                        entry.update(action='delete', proof=proof,
                                     reason=self.describe(proof))
            entries.append(entry)
        if self.obsolete - selected:
            raise CleanupError('obsolete selection is outside the selected local/remote scope')
        # Only recipients that will survive this plan can justify another deletion.
        # Pin each chosen recipient, preventing cycles between equivalent tips.
        pinned = set()
        for entry in entries:
            if entry['reason'] or not entry['ref'].startswith('refs/heads/'):
                continue
            if entry['ref'] in pinned:
                continue
            targets = sorted((e for e in entries if e['action'] == 'keep'
                              and e['ref'].startswith('refs/heads/')
                              and e['ref'] != entry['ref'] and not e['symbolic']),
                             key=lambda e: (e['ref'] not in self.protected, e['ref']))
            for target in targets:
                if ancestor(entry['oid'], target['oid']):
                    anchor = {'ref': target['ref'], 'oid': target['oid'],
                              'ref_oid': target['oid'],
                              'label': target['ref']}
                    proof = {'kind': 'ancestor', 'anchor': anchor, 'landing': target['oid']}
                    entry.update(action='delete', proof=proof, reason=self.describe(proof))
                    pinned.add(target['ref'])
                    target['reason'] = 'retained recipient for ' + short(entry['ref'])
                    break
        for entry in entries:
            if not entry['reason']:
                entry['reason'] = 'no complete incorporation proof'
        return {'mode': 'apply' if self.args.apply else 'preview', 'base': self.base,
                'remote': self.remote, 'entries': entries,
                'detached_worktrees': [wt['worktree'] for wt in self.wts if 'detached' in wt]}

    @staticmethod
    def describe(proof):
        return (f'{proof["kind"]} in {short(proof["anchor"]["ref"] or proof["anchor"]["label"])} '
                f'at {proof["landing"][:12]}')

    def recheck(self, entry):
        actual = refs().get(entry['ref'])
        if not actual or actual['symbolic'] or actual['oid'] != entry['oid']:
            raise CleanupError(f'{short(entry["ref"])} moved since inspection')
        proof = entry['proof']
        if proof and proof['anchor']['ref']:
            anchor = proof['anchor']
            current = output('rev-parse', '--verify', '--end-of-options', anchor['ref'])
            if current != anchor['ref_oid']:
                raise CleanupError(f'recipient {anchor["ref"]} moved since inspection')
        associated = [wt for wt in worktrees() if wt.get('branch') == entry['ref']]
        if {wt['worktree'] for wt in associated} != set(entry['worktrees']):
            raise CleanupError(f'worktree registrations changed for {short(entry["ref"])}')
        for wt in associated:
            problem = worktree_problem(wt, self.primary, self.root)
            if problem:
                raise CleanupError(f'{wt["worktree"]}: {problem}')
        return associated

    def remote_snapshot(self, names):
        result = transport_git(self.push_url, 'ls-remote', '--heads',
                               references=tuple('refs/heads/' + name for name in names))
        data = result.stdout.decode(errors='surrogateescape')
        return dict((ref, oid) for oid, ref in
                    (line.split('\t') for line in data.splitlines()))

    def delete_local_ref(self, entry):
        commands = []
        anchor = entry['proof']['anchor'] if entry['proof'] else None
        if anchor and anchor['ref']:
            commands.append(f'verify {anchor["ref"]} {anchor["ref_oid"]}')
        commands.append(f'delete {entry["ref"]} {entry["oid"]}')
        # OID verification also accepts symbolic refs with matching referents.
        # Inspect their identity only after prepare holds all transaction locks.
        # The CLI option persists; stdin's `option no-deref` is one-command-only.
        with tempfile.TemporaryFile() as errors:
            process = subprocess.Popen(
                ['git', '--no-replace-objects', 'update-ref', '--no-deref', '--stdin'],
                env=ENV, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=errors)

            def exchange(message, expected):
                try:
                    process.stdin.write(message.encode() + b'\n')
                    process.stdin.flush()
                    response = process.stdout.readline()
                except BrokenPipeError:
                    response = b''
                if response != expected.encode() + b'\n':
                    errors.seek(0)
                    detail = errors.read().decode(errors='replace').strip()
                    raise CleanupError(f'ref transaction failed: {detail or response!r}')

            try:
                exchange('start', 'start: ok')
                exchange('\n'.join(commands + ['prepare']), 'prepare: ok')
                for ref in [entry['ref']] + ([anchor['ref']] if anchor and anchor['ref'] else []):
                    symbolic = git('symbolic-ref', '--quiet', ref, allowed=(0, 1))
                    if symbolic.returncode == 0:
                        raise CleanupError(f'{ref} became symbolic during cleanup')
                exchange('commit', 'commit: ok')
            finally:
                # EOF aborts any uncommitted transaction and releases its locks.
                # stderr is a file so a verbose reference-transaction hook cannot
                # block while the parent waits for a protocol response.
                try:
                    process.stdin.close()
                except BrokenPipeError:
                    pass
                process.wait()
                process.stdout.close()
            if process.returncode:
                errors.seek(0)
                raise CleanupError('ref transaction failed: ' + errors.read().decode(errors='replace'))

    def apply(self, plan):
        deletions = [e for e in plan['entries'] if e['action'] == 'delete']
        if not deletions:
            return
        for entry in deletions:
            self.recheck(entry)
        run = datetime.datetime.now(datetime.UTC).strftime('%Y%m%dT%H%M%SZ-') + uuid.uuid4().hex[:8]
        prefix = f'refs/cleanup/{run}/'
        journal = self.common / 'git-cleanup' / (run + '.json')
        plan['recovery'] = prefix
        plan['journal'] = str(journal)
        commands = ['start']
        anchors = {}
        for entry in deletions:
            commands.append(f'create {prefix}{entry["ref"][5:]} {entry["oid"]}')
            if entry['proof']:
                a = entry['proof']['anchor']
                anchors[(a['ref'], a['oid'])] = a
        for index, anchor in enumerate(anchors.values()):
            commands.append(f'create {prefix}anchors/{index} {anchor["oid"]}')
        commands.extend(['prepare', 'commit'])
        git('update-ref', '--no-deref', '--stdin', data=('\n'.join(commands) + '\n').encode())
        journal.parent.mkdir(parents=True, exist_ok=True)

        def save():
            temporary = journal.with_suffix('.tmp')
            temporary.write_text(json.dumps(plan, indent=2, ensure_ascii=True) + '\n')
            temporary.replace(journal)

        save()
        try:
            for entry in deletions:
                associated = self.recheck(entry)
                name = self.remote_branch(entry['ref'])
                if name is not None:
                    names = [name]
                    anchor = entry['proof']['anchor'] if entry['proof'] else None
                    recipient = self.remote_branch(anchor['ref']) if anchor else None
                    if recipient:
                        names.append(recipient)
                    live = self.remote_snapshot(names)
                    if live.get('refs/heads/' + name) != entry['oid']:
                        raise CleanupError(f'remote branch {name} changed since inspection')
                    if recipient and live.get('refs/heads/' + recipient) != anchor['oid']:
                        raise CleanupError(f'remote recipient {recipient} changed since inspection')
                    entry['result'] = 'remote deletion started; inspect remote if interrupted'
                    save()
                    transport_git(self.push_url, 'push', '--porcelain', '--no-follow-tags',
                                  '--recurse-submodules=no',
                                  f'--force-with-lease=refs/heads/{name}:{entry["oid"]}',
                                  references=(':refs/heads/' + name,))
                    # Git may already have removed the matching tracking ref.
                    # Preserve a concurrently refreshed or recreated tracking ref.
                    tracking = refs().get(entry['ref'])
                    if tracking and not tracking['symbolic'] and tracking['oid'] == entry['oid']:
                        git('update-ref', '--no-deref', '-d', entry['ref'], entry['oid'])
                else:
                    for wt in associated:
                        git('worktree', 'remove', '--', wt['worktree'])
                    if any(wt.get('branch') == entry['ref'] for wt in worktrees()):
                        raise CleanupError('branch was checked out during cleanup')
                    self.delete_local_ref(entry)
                entry['result'] = 'deleted'
                save()
        except (CleanupError, OSError, KeyboardInterrupt) as error:
            plan['error'] = str(error) or 'interrupted'
            save()
            raise


def parse_args():
    def nonempty(value):
        if value == '':
            raise argparse.ArgumentTypeError('must not be empty')
        return value

    parser = argparse.ArgumentParser(prog='git cleanup', description=__doc__)
    parser.add_argument('--apply', action='store_true', help='back up and execute the cleanup plan')
    parser.add_argument('--base', type=nonempty, metavar='REF', help='retained integration baseline')
    scope = parser.add_mutually_exclusive_group()
    scope.add_argument('--local', action='store_true', help='local branches only; no network')
    scope.add_argument('--remote', type=nonempty, metavar='NAME', help='remote to inspect and clean')
    parser.add_argument('--fetch', action='store_true', help='refresh remote refs before preview (apply does this automatically)')
    parser.add_argument('--into', type=nonempty, action='append', default=[], metavar='REF', help='additional retained incorporation recipient')
    parser.add_argument('--protect', type=nonempty, action='append', default=[], metavar='GLOB', help='retain matching branches')
    parser.add_argument('--obsolete', type=nonempty, action='append', default=[], metavar='BRANCH', help='explicitly abandon a local branch or remote/branch')
    parser.add_argument('--github', action='store_true', help='use exact merged PR evidence and retain open PR branches')
    parser.add_argument('--scan-history', type=int, default=0, metavar='N', help='opt into matching historical recipient contents (latest N commits)')
    parser.add_argument('--json', action='store_true', help='print the plan as JSON')
    parser.add_argument('--color', nargs='?', const='always', default='auto',
                        choices=('auto', 'always', 'never'), metavar='WHEN',
                        help='color human-readable output: auto, always, never (default: auto)')
    parser.add_argument('--no-color', dest='color', action='store_const', const='never',
                        help='disable color (same as --color=never)')
    args = parser.parse_args()
    if args.scan_history < 0:
        parser.error('--scan-history must be nonnegative')
    if args.local and (args.fetch or args.github):
        parser.error('--local cannot be combined with --fetch or --github')
    return args


def display(plan, as_json, color='auto'):
    if as_json:
        print(json.dumps(plan, indent=2, ensure_ascii=True))
        return
    colored = color == 'always' or (color == 'auto' and sys.stdout.isatty()
                                   and os.environ.get('TERM') != 'dumb'
                                   and not os.environ.get('NO_COLOR'))

    def style(text, code):
        return f'\033[{code}m{text}\033[0m' if colored else text

    base = plan['base']
    heading = style(f'git cleanup: {plan["mode"]}', '1;36')
    print(f'{heading}; baseline {base["label"]} ({base["oid"][:12]})')
    for entry in plan['entries']:
        action = style(f'{entry["action"]:6}', '31' if entry['action'] == 'delete' else '32')
        branch = style(json.dumps(short(entry['ref']), ensure_ascii=True), '1')
        print(f'{action} {branch}: {entry["reason"]}')
        for path in entry['worktrees']:
            print(f'       {style("worktree", "36")} {json.dumps(path, ensure_ascii=True)}')
    for path in plan['detached_worktrees']:
        print(f'{style("keep  ", "32")} detached worktree {json.dumps(path, ensure_ascii=True)}')
    count = sum(e['action'] == 'delete' for e in plan['entries'])
    print(style(f'{count} branch(es) selected.', '1') + (' Run with --apply to execute.' if plan['mode'] == 'preview' else ''))
    if plan.get('recovery'):
        print(f'{style("Recovery refs:", "1;33")} {plan["recovery"]}\n'
              f'{style("Journal:", "1;33")} {plan["journal"]}')


def main():
    args = parse_args()
    plan = None
    try:
        cleanup = Cleanup(args)
        plan = cleanup.plan()
        if args.apply:
            cleanup.apply(plan)
        display(plan, args.json, args.color)
        return 0
    except (CleanupError, OSError, KeyboardInterrupt) as error:
        if plan:
            plan['error'] = str(error) or 'interrupted'
            display(plan, args.json, args.color)
        print(f'git cleanup: {error or "interrupted"}', file=sys.stderr)
        return 1


if __name__ == '__main__':
    sys.exit(main())
