tools/devshell/contrib/min_owners.py - fuchsia - Git at Google

 #!/usr/bin/env fuchsia-vendored-python
 # Copyright 2026 The Fuchsia Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 """Suggests the smallest set of code owners that covers every file in a commit.

 Gets each changed file's owners from the Gerrit code-owners plugin, then
 searches for the fewest owners who together cover every file. When several
 sets are equally small, it prefers owners from closer OWNERS files. Pass --all
 to see every smallest set.

 If more than 5 reviewers are needed, it points to the docs for asking for an
 owners override.

 Ties between equally good owners are broken randomly to spread out review
 load. The randomness is seeded by today's date and the commit author, so
 repeated runs on the same day give the same answer. (Very large changes fall
 back to a simpler search that doesn't do this.)

 Root OWNERS members can approve anything, so they'd always be picked. To avoid
 that, they're only used for files that have no other owners, unless you pass
 --include-root. Owners whose calendar says they're out of office, and who won't
 be back within 4 hours, are skipped the same way. That check uses the prebuilt
 `luci-auth` (or one on PATH), which works without logging in on cloudtops and
 needs `luci-auth login -scopes-gerrit` elsewhere; without it, the check is
 skipped with a note.
 """

 import argparse
 import collections
 import concurrent.futures
 import datetime
 import json
 import os
 import random
 import re
 import shutil
 import subprocess
 import sys
 import time
 import urllib.error
 import urllib.parse
 import urllib.request
 from typing import AbstractSet, Any, Callable, Mapping, TypeVar

 # Gerrit prefixes JSON responses with this to prevent XSSI.
 XSSI_PREFIX = ")]}'"

 # A nonexistent top-level path only matches root OWNERS, so its owners are
 # exactly the root owners.
 ROOT_PROBE_PATH = "__min_owners_nonexistent__"

 HTTP_TIMEOUT_SECS = 30

 GERRIT_OAUTH_SCOPE = "https://www.googleapis.com/auth/gerritcodereview"

 # Pinned in //manifests/prebuilts. This file is in //tools/devshell/contrib.
 PREBUILT_LUCI_AUTH = os.path.join(
     os.path.dirname(os.path.abspath(__file__)),
     "../../../prebuilt/tools/luci-auth/luci-auth",
 )

 # Keeps availability request URLs well under Gerrit's URL length limit.
 AVAILABILITY_BATCH_SIZE = 100

 # Short OOO blocks (appointments, errands) shouldn't stop someone from getting
 # a review they'll see later the same day.
 OOO_GRACE_SECS = 4 * 60 * 60

 # Past this many reviewers, collecting approvals is enough of a burden that an
 # owners override is worth considering, per the OWNERS docs.
 OVERRIDE_SUGGESTION_THRESHOLD = 5
 OWNERS_OVERRIDE_DOCS = (
     "https://fuchsia.dev/fuchsia-src/development/source_code/owners"
     "#owners-override"
 )

 Getter = Callable[[str], Any]
 T = TypeVar("T")


 def git(*args: str) -> str:
     return subprocess.check_output(["git", *args], text=True).strip()


 def parse_name_status(out: str) -> list[str]:
     """Returns the paths in `git diff --name-status -z` output."""
     files: set[str] = set()
     fields = iter(out.split("\0"))
     for status in fields:
         if not status:
             continue
         files.add(next(fields))
         # Renames and copies list a second path. Both need owner approval.
         if status[0] in "RC":
             files.add(next(fields))
     return sorted(files)


 def diff_args(rev: str) -> list[str]:
     """Returns the `git diff` revision args for a commit or range."""
     if ".." not in rev:
         # Gerrit diffs a commit against its first parent, which also gives
         # a useful file list for merge commits (unlike `rev^!`).
         return [f"{rev}^1", rev]
     # Use the merge base so that `main..HEAD` only counts this branch's
     # changes, even if local main has moved on since the branch was cut.
     base, tip = re.split(r"\.\.\.?", rev, maxsplit=1)
     return [f"{base or 'HEAD'}...{tip or 'HEAD'}"]


 def changed_files(rev: str) -> list[str]:
     out = subprocess.check_output(
         ["git", "diff", "--name-status", "-M", "-z", *diff_args(rev)], text=True
     )
     return parse_name_status(out)


 def parse_remote_url(url: str) -> tuple[str, str]:
     """Returns the Gerrit review host URL and project for a git remote URL."""
     # Matches e.g. https://fuchsia.googlesource.com/fuchsia and the short
     # sso://turquoise-internal/vendor/google form.
     m = re.match(
         r"(?:https?|sso|rpc)://([\w-]+)"
         r"(?:\.(?:git\.corp\.google|googlesource)\.com)?"
         r"/(?:a/)?(.+?)(?:\.git)?/?$",
         url,
     )
     if not m:
         raise ValueError(f"Can't figure out the Gerrit host from {url!r}")
     return f"https://{m.group(1)}-review.googlesource.com", m.group(2)


 def parse_json(body: str) -> Any:
     if not body.startswith(XSSI_PREFIX):
         raise RuntimeError(f"Unexpected Gerrit response: {body[:200]}")
     return json.loads(body[len(XSSI_PREFIX) :])


 def gerrit_getter(host: str) -> Getter:
     """Returns a function that GETs a Gerrit REST path and returns its JSON.

     Public hosts like fuchsia-review work anonymously. Private hosts like
     turquoise-internal-review redirect anonymous requests to a login page or
     reject them, so those go through gob-curl, which handles SSO auth on the
     /a/ prefix.
     """

     def anonymous(path: str) -> Any:
         url = host + path
         try:
             with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT_SECS) as resp:
                 return parse_json(resp.read().decode())
         except urllib.error.URLError as e:
             raise RuntimeError(f"GET {url} failed: {e}") from e

     def authenticated(path: str) -> Any:
         url = f"{host}/a{path}"
         proc = subprocess.run(
             ["gob-curl", "--silent", "--show-error", "--fail", url],
             capture_output=True,
             text=True,
         )
         if proc.returncode:
             raise RuntimeError(f"gob-curl {url} failed: {proc.stderr[:200]}")
         return parse_json(proc.stdout)

     probe = host + "/config/server/version"
     try:
         with urllib.request.urlopen(probe, timeout=HTTP_TIMEOUT_SECS) as resp:
             # Hosts that need auth either redirect anonymous requests to a
             # login page or reject them outright.
             needs_auth = resp.url != probe
     except urllib.error.HTTPError as e:
         e.close()
         if e.code not in (401, 403):
             raise
         needs_auth = True
     if not needs_auth:
         return anonymous
     if not shutil.which("gob-curl"):
         sys.exit(f"{host} needs auth, but gob-curl isn't on PATH")
     return authenticated


 def fetch_owners(
     get: Getter,
     project: str,
     branch: str,
     path: str,
     account_ids: dict[str, int] | None = None,
 ) -> dict[str, int] | None:
     """Returns {owner email: distance} for a path, or None if anyone owns it.

     Also adds each owner's Gerrit account ID to `account_ids`, if given.
     """
     data = get(
         "/projects/{}/branches/{}/code_owners/{}?o=DETAILS&limit=1000".format(
             urllib.parse.quote(project, safe=""),
             urllib.parse.quote(branch, safe=""),
             urllib.parse.quote(path, safe=""),
         )
     )
     if data.get("owned_by_all_users"):
         return None
     owners: dict[str, int] = {}
     for o in data.get("code_owners", []):
         account = o["account"]
         if "email" not in account:
             continue
         owners[account["email"]] = o.get("scorings", {}).get("DISTANCE", 0)
         if account_ids is not None and "_account_id" in account:
             account_ids[account["email"]] = account["_account_id"]
     return owners


 def luci_auth_token() -> str:
     # Fall back to PATH for checkouts that haven't fetched the prebuilt yet.
     luci_auth = (
         PREBUILT_LUCI_AUTH
         if os.path.exists(PREBUILT_LUCI_AUTH)
         else shutil.which("luci-auth")
     )
     if not luci_auth:
         raise RuntimeError("luci-auth isn't in prebuilt/tools or on PATH")
     proc = subprocess.run(
         [luci_auth, "token", "-scopes", GERRIT_OAUTH_SCOPE],
         capture_output=True,
         text=True,
     )
     if proc.returncode:
         msg = proc.stderr.strip().splitlines()[-1:] or ["unknown error"]
         raise RuntimeError(f"luci-auth token failed: {msg[0]}")
     return proc.stdout.strip()


 def out_of_office(
     host: str,
     account_ids: Mapping[str, int],
     token: str,
     now: float | None = None,
 ) -> set[str]:
     """Returns the emails of accounts whose calendar says they're OOO.

     People who'll be back within OOO_GRACE_SECS don't count. This uses
     Gerrit's Google-only availability plugin, which needs a Gaia OAuth token.
     gob-curl's credentials don't work for it.
     """
     if now is None:
         now = time.time()
     emails = {i: e for e, i in account_ids.items()}
     ids = sorted(emails)
     ooo: set[str] = set()
     for start in range(0, len(ids), AVAILABILITY_BATCH_SIZE):
         query = urllib.parse.urlencode(
             [("id", i) for i in ids[start : start + AVAILABILITY_BATCH_SIZE]]
         )
         req = urllib.request.Request(
             f"{host}/a/plugins/availability/statuses/?{query}",
             headers={"Authorization": f"Bearer {token}"},
         )
         try:
             with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_SECS) as resp:
                 statuses = parse_json(resp.read().decode())
             for s in statuses:
                 # OUTSIDE_WORKING_HOURS is ignored on purpose: it's often just
                 # a different time zone, and they'll be back within a day.
                 if s["status"] != "OUT_OF_OFFICE":
                     continue
                 back = s.get("return_time", {}).get("seconds")
                 if back is not None and back - now < OOO_GRACE_SECS:
                     continue
                 ooo.add(emails[s["account_id"]])
         # This check is best-effort, so any failure (including timeouts,
         # which aren't URLErrors, and odd responses) just turns it off.
         except (OSError, RuntimeError, ValueError, LookupError, TypeError) as e:
             raise RuntimeError(f"availability lookup failed: {e!r}") from e
     return ooo


 def prefer(
     owners: dict[str, int], keep: Callable[[str], bool]
 ) -> dict[str, int]:
     """Returns the owners that pass `keep`, or all of them if none do."""
     return {o: d for o, d in owners.items() if keep(o)} or owners


 def owners_to_cover(
     file_owners: Mapping[str, Mapping[str, int] | None],
     root_owners: set[str],
     author: str,
     include_root: bool,
     unavailable: AbstractSet[str] = frozenset(),
 ) -> dict[str, dict[str, int]]:
     """Drops files that need no extra approval, and owners we'd rather skip.

     Unavailable owners, and root owners unless `include_root` is set, are
     only kept for files that nobody else can approve.
     """
     result: dict[str, dict[str, int]] = {}
     for f, owners in file_owners.items():
         # None means anyone can approve, and the author implicitly approves
         # files they own, so neither needs another reviewer.
         if owners is None or author in owners:
             continue
         # Availability is checked first because an available root owner is a
         # better pick than one who's out of office.
         kept = prefer(dict(owners), lambda o: o not in unavailable)
         if not include_root:
             kept = prefer(kept, lambda o: o not in root_owners)
         result[f] = kept
     return result


 def plural(n: int, noun: str) -> str:
     return f"{n} {noun}" if n == 1 else f"{n} {noun}s"


 def owner_distances(file_owners: dict[str, dict[str, int]]) -> dict[str, int]:
     """Returns each owner's total distance over the files they own."""
     distance: dict[str, int] = collections.defaultdict(int)
     for owners in file_owners.values():
         for owner, dist in owners.items():
             distance[owner] += dist
     return distance


 def greedy_cover(
     file_owners: dict[str, dict[str, int]],
 ) -> tuple[list[tuple[str, list[str]]], list[str]]:
     """Picks owners until every file is covered.

     Returns the picked owners with the files each one covers, plus any files
     that have no owners at all.
     """
     uncovered = set(file_owners)
     picks = []
     while uncovered:
         covers: dict[str, list[str]] = collections.defaultdict(list)
         distance: dict[str, int] = collections.defaultdict(int)
         for f in uncovered:
             for owner, dist in file_owners[f].items():
                 covers[owner].append(f)
                 distance[owner] += dist
         if not covers:
             break
         owner = min(covers, key=lambda o: (-len(covers[o]), distance[o], o))
         picks.append((owner, sorted(covers[owner])))
         uncovered -= set(covers[owner])
     return picks, sorted(uncovered)


 # A group of interchangeable owners (they cover exactly the same files), and
 # the files they cover.
 OwnerGroup = tuple[list[str], list[str]]


 def all_min_covers(
     file_owners: dict[str, dict[str, int]],
     max_steps: int = 200_000,
 ) -> list[list[OwnerGroup]] | None:
     """Returns every smallest set of owners that covers all ownable files.

     Owners who cover exactly the same files are interchangeable, so they're
     merged into one group. Otherwise a few directories with several owners
     each would multiply into hundreds of near-identical answers. Each answer
     is a list of groups; picking any one owner from each group gives a
     minimal set.

     Finding the minimum is NP-hard, so this gives up and returns None after
     `max_steps` search steps. That only happens for very large changes.
     """
     covers: dict[str, set[str]] = collections.defaultdict(set)
     for f, owners in file_owners.items():
         for owner in owners:
             covers[owner].add(f)
     distance = owner_distances(file_owners)

     group_owners: dict[frozenset[str], list[str]] = collections.defaultdict(
         list
     )
     for owner, owned in covers.items():
         group_owners[frozenset(owned)].append(owner)
     groups = list(group_owners)
     groups_for_file: dict[str, list[int]] = collections.defaultdict(list)
     for i, group in enumerate(groups):
         for f in group:
             groups_for_file[f].append(i)

     # The greedy answer is an upper bound on the minimum size, which prunes
     # most of the search.
     best = len(greedy_cover(file_owners)[0])
     found: set[frozenset[int]] = set()
     largest = max((len(g) for g in groups), default=1)
     steps = 0

     class TooBig(Exception):
         pass

     def search(uncovered: frozenset[str], chosen: frozenset[int]) -> None:
         nonlocal best, steps
         steps += 1
         if steps > max_steps:
             raise TooBig()
         if not uncovered:
             if len(chosen) < best:
                 best = len(chosen)
                 found.clear()
             found.add(chosen)
             return
         # Each extra group covers at most `largest` files.
         if len(chosen) + -(-len(uncovered) // largest) > best:
             return
         # Branch on the file with the fewest options to keep the tree small.
         f = min(uncovered, key=lambda f: len(groups_for_file[f]))
         for i in groups_for_file[f]:
             search(uncovered - groups[i], chosen | {i})

     ownable = frozenset(f for f, owners in file_owners.items() if owners)
     try:
         search(ownable, frozenset())
     # The recursion depth is bounded by the greedy answer's size, so it only
     # gets too deep for changes that would need hundreds of reviewers.
     except (TooBig, RecursionError):
         return None

     def owner_key(o: str) -> tuple[int, str]:
         return distance[o], o

     results = [
         sorted(
             (sorted(group_owners[groups[i]], key=owner_key), sorted(groups[i]))
             for i in cover
         )
         for cover in found
     ]
     # Show answers with closer owners first.
     results.sort(
         key=lambda cover: (
             sum(distance[owners[0]] for owners, _ in cover),
             cover,
         )
     )
     return results


 def pick_reviewers(
     covers: list[list[OwnerGroup]],
     distance: Mapping[str, int],
     rng: random.Random,
 ) -> list[tuple[str, list[str]]]:
     """Picks one owner from each group of one of the smallest covers.

     Closer owners still win, but ties are broken randomly so the same few
     people (e.g. whoever sorts first alphabetically) don't get every review.
     """

     def closest(items: list[T], dist: Callable[[T], int]) -> T:
         best = min(dist(x) for x in items)
         return rng.choice([x for x in items if dist(x) == best])

     cover = closest(
         covers, lambda c: sum(distance[owners[0]] for owners, _ in c)
     )
     picks = [
         (closest(owners, lambda o: distance[o]), covered)
         for owners, covered in cover
     ]
     return sorted(picks, key=lambda p: (-len(p[1]), p[0]))


 def seeded_rng(author: str) -> random.Random:
     # Seeding with the date keeps reruns stable within a day, and the author
     # stops everyone's changes from going to the same people on a given day.
     return random.Random(f"{datetime.date.today().isoformat()} {author}")


 def suggest(
     file_owners: dict[str, dict[str, int]], author: str
 ) -> tuple[list[tuple[str, list[str]]], list[list[OwnerGroup]] | None]:
     """Returns the suggested owners, plus every smallest cover if known."""
     covers = all_min_covers(file_owners)
     if covers is None:
         return greedy_cover(file_owners)[0], None
     rng = seeded_rng(author)
     return pick_reviewers(covers, owner_distances(file_owners), rng), covers


 def main() -> None:
     parser = argparse.ArgumentParser(
         prog="fx min-owners",
         description=__doc__,
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     parser.add_argument(
         "rev",
         nargs="?",
         default="HEAD",
         help="commit, or range like main..HEAD (default: HEAD)",
     )
     parser.add_argument(
         "--branch",
         default="main",
         help="branch whose OWNERS files to use (default: main)",
     )
     parser.add_argument(
         "--include-root",
         action="store_true",
         help="consider root OWNERS members for every file, not just as a fallback",
     )
     parser.add_argument(
         "--author",
         help="email that gets implicit approval (default: the commit author); "
         "set this if your git email differs from your Gerrit account's",
     )
     parser.add_argument(
         "-v",
         "--verbose",
         action="store_true",
         help="list the files each owner covers",
     )
     parser.add_argument(
         "--all",
         action="store_true",
         help="list every smallest set of owners instead of just one",
     )
     args = parser.parse_args()

     try:
         run(args)
     except (
         RuntimeError,
         ValueError,
         urllib.error.URLError,
         subprocess.CalledProcessError,
     ) as e:
         sys.exit(f"Error: {e}")


 def run(args: argparse.Namespace) -> None:
     files = changed_files(args.rev)
     if not files:
         sys.exit("No changed files.")
     if any(f == "OWNERS" or f.endswith("/OWNERS") for f in files):
         print(
             f"Warning: owners come from OWNERS files on the '{args.branch}' "
             "branch, so this commit's own OWNERS changes aren't taken into "
             "account.\n",
             file=sys.stderr,
         )
     host, project = parse_remote_url(git("remote", "get-url", "origin"))
     tip = re.split(r"\.\.\.?", args.rev)[-1] or "HEAD"
     author = args.author or git("log", "-1", "--format=%ae", tip)

     get = gerrit_getter(host)
     # Written from several threads, which is fine because each email always
     # maps to the same ID.
     account_ids: dict[str, int] = {}
     with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
         results = list(
             pool.map(
                 lambda p: fetch_owners(
                     get, project, args.branch, p, account_ids
                 ),
                 files + [ROOT_PROBE_PATH],
             )
         )
     *results, root = results
     root_owners = set(root or {})
     file_owners = dict(zip(files, results))

     to_cover = owners_to_cover(
         file_owners, root_owners, author, args.include_root
     )
     # Check everyone who could approve, not just the preferred owners, since
     # root owners become the fallback if all the others are out.
     to_check = {
         o: account_ids[o]
         for f in to_cover
         for o in file_owners[f] or {}
         if o in account_ids
     }
     ooo: set[str] = set()
     if to_check:
         try:
             ooo = out_of_office(host, to_check, luci_auth_token())
         except RuntimeError as e:
             print(
                 f"Note: couldn't check who's out of office: {e}\n",
                 file=sys.stderr,
             )
     skipped_ooo: list[str] = []
     if ooo:
         before, _ = suggest(to_cover, author)
         to_cover = owners_to_cover(
             file_owners, root_owners, author, args.include_root, ooo
         )
         listed = {o for owners in to_cover.values() for o in owners}
         skipped_ooo = sorted(
             o for o, _ in before if o in ooo and o not in listed
         )
     picks, covers = suggest(to_cover, author)
     unowned = sorted(f for f, owners in to_cover.items() if not owners)

     skipped = len(files) - len(to_cover)
     if skipped:
         verb = "needs" if skipped == 1 else "need"
         print(
             f"({plural(skipped, 'file')} {verb} no extra owner: owned by "
             f"{author} or by everyone)\n"
         )

     def describe(owner: str) -> str:
         if owner in root_owners:
             owner += " [root]"
         if owner in ooo:
             owner += " [OOO]"
         return owner

     if covers is None:
         print(
             "Note: too many files to find the true minimum, so this list may "
             "have more reviewers than needed.\n",
             file=sys.stderr,
         )
     if covers is not None and args.all:
         if covers[0]:
             print(f"The minimum number of owners is {len(covers[0])}.")
             if len(covers) == 1:
                 print("Pick one owner from each group.")
             else:
                 print(
                     f"There are {len(covers)} options; pick one owner from "
                     "each group of any option."
                 )
         for n, cover in enumerate(covers, 1):
             if not cover:
                 continue
             print(f"\nOption {n}:" if len(covers) > 1 else "")
             for owners, covered in cover:
                 if len(owners) == 1:
                     print(f"  - {plural(len(covered), 'file')}:")
                 else:
                     print(f"  - {plural(len(covered), 'file')}, any one of:")
                 for owner in owners:
                     print(f"      {describe(owner)}")
                 if args.verbose:
                     print("    Files:")
                     for f in covered:
                         print(f"      {f}")
     else:
         for owner, covered in picks:
             print(f"{describe(owner)}: {plural(len(covered), 'file')}")
             if args.verbose:
                 for f in covered:
                     print(f"    {f}")
     if unowned:
         print("\nNo owners found for:")
         for f in unowned:
             print(f"    {f}")
     if skipped_ooo:
         print("\nSkipped because they're out of office:")
         for owner in skipped_ooo:
             print(f"    {owner}")
     if len(picks) > OVERRIDE_SUGGESTION_THRESHOLD:
         print(
             f"\nThat's {plural(len(picks), 'reviewer')}. If this change is "
             "mostly mechanical, consider\nasking for an owners override "
             "instead. See:\n"
             f"    {OWNERS_OVERRIDE_DOCS}"
         )


 if __name__ == "__main__":
     main()
	#!/usr/bin/env fuchsia-vendored-python
	# Copyright 2026 The Fuchsia Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.
	"""Suggests the smallest set of code owners that covers every file in a commit.

	Gets each changed file's owners from the Gerrit code-owners plugin, then
	searches for the fewest owners who together cover every file. When several
	sets are equally small, it prefers owners from closer OWNERS files. Pass --all
	to see every smallest set.

	If more than 5 reviewers are needed, it points to the docs for asking for an
	owners override.

	Ties between equally good owners are broken randomly to spread out review
	load. The randomness is seeded by today's date and the commit author, so
	repeated runs on the same day give the same answer. (Very large changes fall
	back to a simpler search that doesn't do this.)

	Root OWNERS members can approve anything, so they'd always be picked. To avoid
	that, they're only used for files that have no other owners, unless you pass
	--include-root. Owners whose calendar says they're out of office, and who won't
	be back within 4 hours, are skipped the same way. That check uses the prebuilt
	`luci-auth` (or one on PATH), which works without logging in on cloudtops and
	needs `luci-auth login -scopes-gerrit` elsewhere; without it, the check is
	skipped with a note.
	"""

	import argparse
	import collections
	import concurrent.futures
	import datetime
	import json
	import os
	import random
	import re
	import shutil
	import subprocess
	import sys
	import time
	import urllib.error
	import urllib.parse
	import urllib.request
	from typing import AbstractSet, Any, Callable, Mapping, TypeVar

	# Gerrit prefixes JSON responses with this to prevent XSSI.
	XSSI_PREFIX = ")]}'"

	# A nonexistent top-level path only matches root OWNERS, so its owners are
	# exactly the root owners.
	ROOT_PROBE_PATH = "__min_owners_nonexistent__"

	HTTP_TIMEOUT_SECS = 30

	GERRIT_OAUTH_SCOPE = "https://www.googleapis.com/auth/gerritcodereview"

	# Pinned in //manifests/prebuilts. This file is in //tools/devshell/contrib.
	PREBUILT_LUCI_AUTH = os.path.join(
	os.path.dirname(os.path.abspath(__file__)),
	"../../../prebuilt/tools/luci-auth/luci-auth",
	)

	# Keeps availability request URLs well under Gerrit's URL length limit.
	AVAILABILITY_BATCH_SIZE = 100

	# Short OOO blocks (appointments, errands) shouldn't stop someone from getting
	# a review they'll see later the same day.
	OOO_GRACE_SECS = 4 * 60 * 60

	# Past this many reviewers, collecting approvals is enough of a burden that an
	# owners override is worth considering, per the OWNERS docs.
	OVERRIDE_SUGGESTION_THRESHOLD = 5
	OWNERS_OVERRIDE_DOCS = (
	"https://fuchsia.dev/fuchsia-src/development/source_code/owners"
	"#owners-override"
	)

	Getter = Callable[[str], Any]
	T = TypeVar("T")


	def git(*args: str) -> str:
	return subprocess.check_output(["git", *args], text=True).strip()


	def parse_name_status(out: str) -> list[str]:
	"""Returns the paths in `git diff --name-status -z` output."""
	files: set[str] = set()
	fields = iter(out.split("\0"))
	for status in fields:
	if not status:
	continue
	files.add(next(fields))
	# Renames and copies list a second path. Both need owner approval.
	if status[0] in "RC":
	files.add(next(fields))
	return sorted(files)


	def diff_args(rev: str) -> list[str]:
	"""Returns the `git diff` revision args for a commit or range."""
	if ".." not in rev:
	# Gerrit diffs a commit against its first parent, which also gives
	# a useful file list for merge commits (unlike `rev^!`).
	return [f"{rev}^1", rev]
	# Use the merge base so that `main..HEAD` only counts this branch's
	# changes, even if local main has moved on since the branch was cut.
	base, tip = re.split(r"\.\.\.?", rev, maxsplit=1)
	return [f"{base or 'HEAD'}...{tip or 'HEAD'}"]


	def changed_files(rev: str) -> list[str]:
	out = subprocess.check_output(
	["git", "diff", "--name-status", "-M", "-z", *diff_args(rev)], text=True
	)
	return parse_name_status(out)


	def parse_remote_url(url: str) -> tuple[str, str]:
	"""Returns the Gerrit review host URL and project for a git remote URL."""
	# Matches e.g. https://fuchsia.googlesource.com/fuchsia and the short
	# sso://turquoise-internal/vendor/google form.
	m = re.match(
	r"(?:https?\|sso\|rpc)://([\w-]+)"
	r"(?:\.(?:git\.corp\.google\|googlesource)\.com)?"
	r"/(?:a/)?(.+?)(?:\.git)?/?$",
	url,
	)
	if not m:
	raise ValueError(f"Can't figure out the Gerrit host from {url!r}")
	return f"https://{m.group(1)}-review.googlesource.com", m.group(2)


	def parse_json(body: str) -> Any:
	if not body.startswith(XSSI_PREFIX):
	raise RuntimeError(f"Unexpected Gerrit response: {body[:200]}")
	return json.loads(body[len(XSSI_PREFIX) :])


	def gerrit_getter(host: str) -> Getter:
	"""Returns a function that GETs a Gerrit REST path and returns its JSON.

	Public hosts like fuchsia-review work anonymously. Private hosts like
	turquoise-internal-review redirect anonymous requests to a login page or
	reject them, so those go through gob-curl, which handles SSO auth on the
	/a/ prefix.
	"""

	def anonymous(path: str) -> Any:
	url = host + path
	try:
	with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT_SECS) as resp:
	return parse_json(resp.read().decode())
	except urllib.error.URLError as e:
	raise RuntimeError(f"GET {url} failed: {e}") from e

	def authenticated(path: str) -> Any:
	url = f"{host}/a{path}"
	proc = subprocess.run(
	["gob-curl", "--silent", "--show-error", "--fail", url],
	capture_output=True,
	text=True,
	)
	if proc.returncode:
	raise RuntimeError(f"gob-curl {url} failed: {proc.stderr[:200]}")
	return parse_json(proc.stdout)

	probe = host + "/config/server/version"
	try:
	with urllib.request.urlopen(probe, timeout=HTTP_TIMEOUT_SECS) as resp:
	# Hosts that need auth either redirect anonymous requests to a
	# login page or reject them outright.
	needs_auth = resp.url != probe
	except urllib.error.HTTPError as e:
	e.close()
	if e.code not in (401, 403):
	raise
	needs_auth = True
	if not needs_auth:
	return anonymous
	if not shutil.which("gob-curl"):
	sys.exit(f"{host} needs auth, but gob-curl isn't on PATH")
	return authenticated


	def fetch_owners(
	get: Getter,
	project: str,
	branch: str,
	path: str,
	account_ids: dict[str, int] \| None = None,
	) -> dict[str, int] \| None:
	"""Returns {owner email: distance} for a path, or None if anyone owns it.

	Also adds each owner's Gerrit account ID to `account_ids`, if given.
	"""
	data = get(
	"/projects/{}/branches/{}/code_owners/{}?o=DETAILS&limit=1000".format(
	urllib.parse.quote(project, safe=""),
	urllib.parse.quote(branch, safe=""),
	urllib.parse.quote(path, safe=""),
	)
	)
	if data.get("owned_by_all_users"):
	return None
	owners: dict[str, int] = {}
	for o in data.get("code_owners", []):
	account = o["account"]
	if "email" not in account:
	continue
	owners[account["email"]] = o.get("scorings", {}).get("DISTANCE", 0)
	if account_ids is not None and "_account_id" in account:
	account_ids[account["email"]] = account["_account_id"]
	return owners


	def luci_auth_token() -> str:
	# Fall back to PATH for checkouts that haven't fetched the prebuilt yet.
	luci_auth = (
	PREBUILT_LUCI_AUTH
	if os.path.exists(PREBUILT_LUCI_AUTH)
	else shutil.which("luci-auth")
	)
	if not luci_auth:
	raise RuntimeError("luci-auth isn't in prebuilt/tools or on PATH")
	proc = subprocess.run(
	[luci_auth, "token", "-scopes", GERRIT_OAUTH_SCOPE],
	capture_output=True,
	text=True,
	)
	if proc.returncode:
	msg = proc.stderr.strip().splitlines()[-1:] or ["unknown error"]
	raise RuntimeError(f"luci-auth token failed: {msg[0]}")
	return proc.stdout.strip()


	def out_of_office(
	host: str,
	account_ids: Mapping[str, int],
	token: str,
	now: float \| None = None,
	) -> set[str]:
	"""Returns the emails of accounts whose calendar says they're OOO.

	People who'll be back within OOO_GRACE_SECS don't count. This uses
	Gerrit's Google-only availability plugin, which needs a Gaia OAuth token.
	gob-curl's credentials don't work for it.
	"""
	if now is None:
	now = time.time()
	emails = {i: e for e, i in account_ids.items()}
	ids = sorted(emails)
	ooo: set[str] = set()
	for start in range(0, len(ids), AVAILABILITY_BATCH_SIZE):
	query = urllib.parse.urlencode(
	[("id", i) for i in ids[start : start + AVAILABILITY_BATCH_SIZE]]
	)
	req = urllib.request.Request(
	f"{host}/a/plugins/availability/statuses/?{query}",
	headers={"Authorization": f"Bearer {token}"},
	)
	try:
	with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_SECS) as resp:
	statuses = parse_json(resp.read().decode())
	for s in statuses:
	# OUTSIDE_WORKING_HOURS is ignored on purpose: it's often just
	# a different time zone, and they'll be back within a day.
	if s["status"] != "OUT_OF_OFFICE":
	continue
	back = s.get("return_time", {}).get("seconds")
	if back is not None and back - now < OOO_GRACE_SECS:
	continue
	ooo.add(emails[s["account_id"]])
	# This check is best-effort, so any failure (including timeouts,
	# which aren't URLErrors, and odd responses) just turns it off.
	except (OSError, RuntimeError, ValueError, LookupError, TypeError) as e:
	raise RuntimeError(f"availability lookup failed: {e!r}") from e
	return ooo


	def prefer(
	owners: dict[str, int], keep: Callable[[str], bool]
	) -> dict[str, int]:
	"""Returns the owners that pass `keep`, or all of them if none do."""
	return {o: d for o, d in owners.items() if keep(o)} or owners


	def owners_to_cover(
	file_owners: Mapping[str, Mapping[str, int] \| None],
	root_owners: set[str],
	author: str,
	include_root: bool,
	unavailable: AbstractSet[str] = frozenset(),
	) -> dict[str, dict[str, int]]:
	"""Drops files that need no extra approval, and owners we'd rather skip.

	Unavailable owners, and root owners unless `include_root` is set, are
	only kept for files that nobody else can approve.
	"""
	result: dict[str, dict[str, int]] = {}
	for f, owners in file_owners.items():
	# None means anyone can approve, and the author implicitly approves
	# files they own, so neither needs another reviewer.
	if owners is None or author in owners:
	continue
	# Availability is checked first because an available root owner is a
	# better pick than one who's out of office.
	kept = prefer(dict(owners), lambda o: o not in unavailable)
	if not include_root:
	kept = prefer(kept, lambda o: o not in root_owners)
	result[f] = kept
	return result


	def plural(n: int, noun: str) -> str:
	return f"{n} {noun}" if n == 1 else f"{n} {noun}s"


	def owner_distances(file_owners: dict[str, dict[str, int]]) -> dict[str, int]:
	"""Returns each owner's total distance over the files they own."""
	distance: dict[str, int] = collections.defaultdict(int)
	for owners in file_owners.values():
	for owner, dist in owners.items():
	distance[owner] += dist
	return distance


	def greedy_cover(
	file_owners: dict[str, dict[str, int]],
	) -> tuple[list[tuple[str, list[str]]], list[str]]:
	"""Picks owners until every file is covered.

	Returns the picked owners with the files each one covers, plus any files
	that have no owners at all.
	"""
	uncovered = set(file_owners)
	picks = []
	while uncovered:
	covers: dict[str, list[str]] = collections.defaultdict(list)
	distance: dict[str, int] = collections.defaultdict(int)
	for f in uncovered:
	for owner, dist in file_owners[f].items():
	covers[owner].append(f)
	distance[owner] += dist
	if not covers:
	break
	owner = min(covers, key=lambda o: (-len(covers[o]), distance[o], o))
	picks.append((owner, sorted(covers[owner])))
	uncovered -= set(covers[owner])
	return picks, sorted(uncovered)


	# A group of interchangeable owners (they cover exactly the same files), and
	# the files they cover.
	OwnerGroup = tuple[list[str], list[str]]


	def all_min_covers(
	file_owners: dict[str, dict[str, int]],
	max_steps: int = 200_000,
	) -> list[list[OwnerGroup]] \| None:
	"""Returns every smallest set of owners that covers all ownable files.

	Owners who cover exactly the same files are interchangeable, so they're
	merged into one group. Otherwise a few directories with several owners
	each would multiply into hundreds of near-identical answers. Each answer
	is a list of groups; picking any one owner from each group gives a
	minimal set.

	Finding the minimum is NP-hard, so this gives up and returns None after
	`max_steps` search steps. That only happens for very large changes.
	"""
	covers: dict[str, set[str]] = collections.defaultdict(set)
	for f, owners in file_owners.items():
	for owner in owners:
	covers[owner].add(f)
	distance = owner_distances(file_owners)

	group_owners: dict[frozenset[str], list[str]] = collections.defaultdict(
	list
	)
	for owner, owned in covers.items():
	group_owners[frozenset(owned)].append(owner)
	groups = list(group_owners)
	groups_for_file: dict[str, list[int]] = collections.defaultdict(list)
	for i, group in enumerate(groups):
	for f in group:
	groups_for_file[f].append(i)

	# The greedy answer is an upper bound on the minimum size, which prunes
	# most of the search.
	best = len(greedy_cover(file_owners)[0])
	found: set[frozenset[int]] = set()
	largest = max((len(g) for g in groups), default=1)
	steps = 0

	class TooBig(Exception):
	pass

	def search(uncovered: frozenset[str], chosen: frozenset[int]) -> None:
	nonlocal best, steps
	steps += 1
	if steps > max_steps:
	raise TooBig()
	if not uncovered:
	if len(chosen) < best:
	best = len(chosen)
	found.clear()
	found.add(chosen)
	return
	# Each extra group covers at most `largest` files.
	if len(chosen) + -(-len(uncovered) // largest) > best:
	return
	# Branch on the file with the fewest options to keep the tree small.
	f = min(uncovered, key=lambda f: len(groups_for_file[f]))
	for i in groups_for_file[f]:
	search(uncovered - groups[i], chosen \| {i})

	ownable = frozenset(f for f, owners in file_owners.items() if owners)
	try:
	search(ownable, frozenset())
	# The recursion depth is bounded by the greedy answer's size, so it only
	# gets too deep for changes that would need hundreds of reviewers.
	except (TooBig, RecursionError):
	return None

	def owner_key(o: str) -> tuple[int, str]:
	return distance[o], o

	results = [
	sorted(
	(sorted(group_owners[groups[i]], key=owner_key), sorted(groups[i]))
	for i in cover
	)
	for cover in found
	]
	# Show answers with closer owners first.
	results.sort(
	key=lambda cover: (
	sum(distance[owners[0]] for owners, _ in cover),
	cover,
	)
	)
	return results


	def pick_reviewers(
	covers: list[list[OwnerGroup]],
	distance: Mapping[str, int],
	rng: random.Random,
	) -> list[tuple[str, list[str]]]:
	"""Picks one owner from each group of one of the smallest covers.

	Closer owners still win, but ties are broken randomly so the same few
	people (e.g. whoever sorts first alphabetically) don't get every review.
	"""

	def closest(items: list[T], dist: Callable[[T], int]) -> T:
	best = min(dist(x) for x in items)
	return rng.choice([x for x in items if dist(x) == best])

	cover = closest(
	covers, lambda c: sum(distance[owners[0]] for owners, _ in c)
	)
	picks = [
	(closest(owners, lambda o: distance[o]), covered)
	for owners, covered in cover
	]
	return sorted(picks, key=lambda p: (-len(p[1]), p[0]))


	def seeded_rng(author: str) -> random.Random:
	# Seeding with the date keeps reruns stable within a day, and the author
	# stops everyone's changes from going to the same people on a given day.
	return random.Random(f"{datetime.date.today().isoformat()} {author}")


	def suggest(
	file_owners: dict[str, dict[str, int]], author: str
	) -> tuple[list[tuple[str, list[str]]], list[list[OwnerGroup]] \| None]:
	"""Returns the suggested owners, plus every smallest cover if known."""
	covers = all_min_covers(file_owners)
	if covers is None:
	return greedy_cover(file_owners)[0], None
	rng = seeded_rng(author)
	return pick_reviewers(covers, owner_distances(file_owners), rng), covers


	def main() -> None:
	parser = argparse.ArgumentParser(
	prog="fx min-owners",
	description=__doc__,
	formatter_class=argparse.RawDescriptionHelpFormatter,
	)
	parser.add_argument(
	"rev",
	nargs="?",
	default="HEAD",
	help="commit, or range like main..HEAD (default: HEAD)",
	)
	parser.add_argument(
	"--branch",
	default="main",
	help="branch whose OWNERS files to use (default: main)",
	)
	parser.add_argument(
	"--include-root",
	action="store_true",
	help="consider root OWNERS members for every file, not just as a fallback",
	)
	parser.add_argument(
	"--author",
	help="email that gets implicit approval (default: the commit author); "
	"set this if your git email differs from your Gerrit account's",
	)
	parser.add_argument(
	"-v",
	"--verbose",
	action="store_true",
	help="list the files each owner covers",
	)
	parser.add_argument(
	"--all",
	action="store_true",
	help="list every smallest set of owners instead of just one",
	)
	args = parser.parse_args()

	try:
	run(args)
	except (
	RuntimeError,
	ValueError,
	urllib.error.URLError,
	subprocess.CalledProcessError,
	) as e:
	sys.exit(f"Error: {e}")


	def run(args: argparse.Namespace) -> None:
	files = changed_files(args.rev)
	if not files:
	sys.exit("No changed files.")
	if any(f == "OWNERS" or f.endswith("/OWNERS") for f in files):
	print(
	f"Warning: owners come from OWNERS files on the '{args.branch}' "
	"branch, so this commit's own OWNERS changes aren't taken into "
	"account.\n",
	file=sys.stderr,
	)
	host, project = parse_remote_url(git("remote", "get-url", "origin"))
	tip = re.split(r"\.\.\.?", args.rev)[-1] or "HEAD"
	author = args.author or git("log", "-1", "--format=%ae", tip)

	get = gerrit_getter(host)
	# Written from several threads, which is fine because each email always
	# maps to the same ID.
	account_ids: dict[str, int] = {}
	with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
	results = list(
	pool.map(
	lambda p: fetch_owners(
	get, project, args.branch, p, account_ids
	),
	files + [ROOT_PROBE_PATH],
	)
	)
	*results, root = results
	root_owners = set(root or {})
	file_owners = dict(zip(files, results))

	to_cover = owners_to_cover(
	file_owners, root_owners, author, args.include_root
	)
	# Check everyone who could approve, not just the preferred owners, since
	# root owners become the fallback if all the others are out.
	to_check = {
	o: account_ids[o]
	for f in to_cover
	for o in file_owners[f] or {}
	if o in account_ids
	}
	ooo: set[str] = set()
	if to_check:
	try:
	ooo = out_of_office(host, to_check, luci_auth_token())
	except RuntimeError as e:
	print(
	f"Note: couldn't check who's out of office: {e}\n",
	file=sys.stderr,
	)
	skipped_ooo: list[str] = []
	if ooo:
	before, _ = suggest(to_cover, author)
	to_cover = owners_to_cover(
	file_owners, root_owners, author, args.include_root, ooo
	)
	listed = {o for owners in to_cover.values() for o in owners}
	skipped_ooo = sorted(
	o for o, _ in before if o in ooo and o not in listed
	)
	picks, covers = suggest(to_cover, author)
	unowned = sorted(f for f, owners in to_cover.items() if not owners)

	skipped = len(files) - len(to_cover)
	if skipped:
	verb = "needs" if skipped == 1 else "need"
	print(
	f"({plural(skipped, 'file')} {verb} no extra owner: owned by "
	f"{author} or by everyone)\n"
	)

	def describe(owner: str) -> str:
	if owner in root_owners:
	owner += " [root]"
	if owner in ooo:
	owner += " [OOO]"
	return owner

	if covers is None:
	print(
	"Note: too many files to find the true minimum, so this list may "
	"have more reviewers than needed.\n",
	file=sys.stderr,
	)
	if covers is not None and args.all:
	if covers[0]:
	print(f"The minimum number of owners is {len(covers[0])}.")
	if len(covers) == 1:
	print("Pick one owner from each group.")
	else:
	print(
	f"There are {len(covers)} options; pick one owner from "
	"each group of any option."
	)
	for n, cover in enumerate(covers, 1):
	if not cover:
	continue
	print(f"\nOption {n}:" if len(covers) > 1 else "")
	for owners, covered in cover:
	if len(owners) == 1:
	print(f" - {plural(len(covered), 'file')}:")
	else:
	print(f" - {plural(len(covered), 'file')}, any one of:")
	for owner in owners:
	print(f" {describe(owner)}")
	if args.verbose:
	print(" Files:")
	for f in covered:
	print(f" {f}")
	else:
	for owner, covered in picks:
	print(f"{describe(owner)}: {plural(len(covered), 'file')}")
	if args.verbose:
	for f in covered:
	print(f" {f}")
	if unowned:
	print("\nNo owners found for:")
	for f in unowned:
	print(f" {f}")
	if skipped_ooo:
	print("\nSkipped because they're out of office:")
	for owner in skipped_ooo:
	print(f" {owner}")
	if len(picks) > OVERRIDE_SUGGESTION_THRESHOLD:
	print(
	f"\nThat's {plural(len(picks), 'reviewer')}. If this change is "
	"mostly mechanical, consider\nasking for an owners override "
	"instead. See:\n"
	f" {OWNERS_OVERRIDE_DOCS}"
	)


	if __name__ == "__main__":
	main()