Coverage for src/pullapprove/config.py: 94%
487 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-27 14:35 -0500
1from __future__ import annotations
3import os
4import posixpath
5import re
6import tomllib
7import warnings
9with warnings.catch_warnings():
10 warnings.simplefilter("ignore", DeprecationWarning)
11 import sre_parse
12from collections.abc import Generator, Iterable
13from enum import StrEnum
14from pathlib import Path
15from typing import Any, Self
17from pydantic import (
18 BaseModel,
19 ConfigDict,
20 Field,
21 RootModel,
22 field_validator,
23 model_validator,
24)
25from wcmatch import glob
27from .checklists import Checklist
30def _resolve_config_filename_prefix() -> str:
31 """The name every config file starts with, for this process.
33 Defaults to CODEREVIEW. An instance can rename it with
34 PULLAPPROVE_CONFIG_PREFIX, which renames the config file itself
35 (REVIEW -> REVIEW.toml, REVIEW.template.toml). Because discovery matches
36 on this prefix, an instance only ever sees files named for its own prefix
37 -- so two instances can watch the same repo without seeing each other's
38 configs, which is how unreleased config features get tried out on a repo
39 that production also watches.
41 Read once at import: this is a per-process constant, not runtime state.
43 Two operational constraints this doesn't (and can't) enforce:
45 Prefixes must not be prefixes of one another. Matching is `startswith`
46 (see is_config_filename), and a suffix after the prefix is legitimate --
47 CODEREVIEW.template.toml and CODEREVIEW-BASE.toml are both real config
48 names -- so a REVIEW instance would also pick up REVIEW_DEV.toml. An
49 instance only knows its own prefix, so pick names that don't overlap
50 (REVIEW and DEV_REVIEW, not REVIEW and REVIEW_DEV).
52 Changing the prefix on a live instance means clearing its cached configs.
53 Config rows are keyed by (repo, sha) with no record of which prefix
54 discovered them, so a sha already processed under the old prefix keeps
55 serving from cache and the new prefix's files are never fetched.
56 """
57 # `or` rather than a get() default: an explicitly empty value
58 # (`PULLAPPROVE_CONFIG_PREFIX=` in a .env or compose file -- a normal way to
59 # write "unset") is not a missing key, and would name the config ".toml".
60 prefix = os.environ.get("PULLAPPROVE_CONFIG_PREFIX", "").strip() or "CODEREVIEW"
62 # A bad prefix would silently match nothing (i.e. every repo looks
63 # unconfigured), so refuse to start instead.
64 if prefix.endswith(".toml"):
65 raise ValueError(
66 f"PULLAPPROVE_CONFIG_PREFIX should be a name without an extension, not {prefix!r}. "
67 f"The config file is named after it (e.g. {prefix[: -len('.toml')]!r} -> {prefix!r})."
68 )
69 if "\\" in prefix or prefix != posixpath.basename(prefix):
70 raise ValueError(
71 f"PULLAPPROVE_CONFIG_PREFIX should be a filename prefix, not a path: {prefix!r}"
72 )
74 return prefix
77CONFIG_FILENAME_PREFIX = _resolve_config_filename_prefix()
78CONFIG_FILENAME = f"{CONFIG_FILENAME_PREFIX}.toml"
81def is_config_filename(basename: str) -> bool:
82 """Whether a filename is a config file (CODEREVIEW.toml,
83 CODEREVIEW.template.toml).
85 The prefix half is what isolates instances: an instance running a renamed
86 prefix (see PULLAPPROVE_CONFIG_PREFIX) never even discovers another
87 instance's configs. The extension half keeps a neighbor like CODEREVIEW.md
88 from being treated as one.
89 """
90 return basename.startswith(CONFIG_FILENAME_PREFIX) and basename.endswith(".toml")
93_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
96def _has_nested_quantifiers(data: Any) -> bool:
97 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
98 for op, av in data:
99 if op in _REPEAT_OPS:
100 if _contains_quantifier(av[2]):
101 return True
102 elif op == sre_parse.SUBPATTERN:
103 if _has_nested_quantifiers(av[-1]):
104 return True
105 elif op == sre_parse.BRANCH:
106 if any(_has_nested_quantifiers(branch) for branch in av[1]):
107 return True
108 return False
111def _contains_quantifier(data: Any) -> bool:
112 for op, av in data:
113 if op in _REPEAT_OPS:
114 return True
115 elif op == sre_parse.SUBPATTERN:
116 if _contains_quantifier(av[-1]):
117 return True
118 elif op == sre_parse.BRANCH:
119 if any(_contains_quantifier(branch) for branch in av[1]):
120 return True
121 return False
124_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*"
125_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$")
127# Fields that hold reviewer identities (plain usernames, `$aliases`, and
128# `@team` refs) rather than paths/code/labels. Team refs only expand here.
129USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc")
131# Roster fields where "!" performs compile-time subtraction (remove from the
132# resolved list) rather than surviving as a match-time predicate. `authors`
133# is deliberately excluded — its "!" entries are consumed by `matches_author`.
134ROSTER_FIELDS = ("reviewers", "alternates", "cc")
137def _split_team_ref(value: str) -> tuple[str, str] | None:
138 """Split a value into `(prefix, ref)` if it has team-reference shape
139 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns
140 `None` for anything else, so callers fall back to their own handling of
141 plain values.
142 """
143 if value.startswith("!@"):
144 return "!", value[2:]
145 if value.startswith("@"):
146 return "", value[1:]
147 return None
150def is_unexpanded_ref(value: str) -> bool:
151 """True if the value is a `$alias` or `@team` reference (optionally
152 negated with a leading `!`) that has not been expanded — as opposed to a
153 plain username. Only an offline compile (`teams=None`) leaves such values
154 in user-list fields.
155 """
156 return value.removeprefix("!").startswith(("$", "@"))
159def _validate_team_refs(values: list[str]) -> list[str]:
160 """Team references (`@org/team`, `!@org/team`) need at least two
161 slash-separated segments. This only checks shape — membership is resolved
162 later, at compile time, against the caller-provided `teams` mapping.
164 Any other value containing "@" is rejected too — that shape is reserved
165 for team references (and, in future, email-style identifiers), so it
166 can't be confused with a plain username.
167 """
168 for value in values:
169 split = _split_team_ref(value)
170 if split is None:
171 if "@" in value:
172 raise ValueError(
173 f"Invalid value '{value}': email addresses are not "
174 "supported here — use the platform username"
175 )
176 continue
178 _prefix, ref = split
179 if not _TEAM_REF_RE.match(ref):
180 raise ValueError(
181 f"Invalid team reference '{value}': team references need the "
182 "'org/team' form"
183 )
184 return values
187def _expand_team_ref(
188 ref: str, teams: dict[str, list[str]] | None, prefix: str
189) -> list[str]:
190 """Expand a single team reference (without its `@`/`!@`) to member usernames.
192 `teams=None` is the offline/CLI mode: the library never calls out to
193 GitHub/GitLab itself, so with no mapping provided the reference passes
194 through unexpanded rather than erroring. With a `teams` mapping (even an
195 empty one), an unresolvable ref is a loud config error, matching how
196 unknown `$aliases` are handled.
197 """
198 if teams is None:
199 return [f"{prefix}@{ref}"]
201 members = teams.get(ref.lower())
202 if members is None:
203 raise ValueError(f"Unknown team: {prefix}@{ref}")
205 return [f"{prefix}{member}" for member in members]
208def _expand_aliases(
209 values: list[str],
210 aliases: dict[str, list[str]],
211 teams: dict[str, list[str]] | None = None,
212 expand_teams: bool = False,
213 _seen: set[str] | None = None,
214 _path: list[str] | None = None,
215) -> list[str]:
216 """Replace alias references in a list with their mapped values recursively.
218 Team references (`@org/team`) are only expanded when `expand_teams` is
219 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere
220 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style
221 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes —
222 their values are plain usernames, never other refs — so expanding one
223 never recurses further.
224 """
225 if _seen is None:
226 _seen = set()
227 if _path is None:
228 _path = []
230 expanded: list[str] = []
231 for value in values:
232 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
233 if value.startswith("!$"):
234 prefix = "!"
235 alias_ref = value[2:]
236 elif value.startswith("$"):
237 prefix = ""
238 alias_ref = value[1:]
239 elif expand_teams and (team_ref := _split_team_ref(value)) is not None:
240 team_prefix, ref = team_ref
241 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix))
242 continue
243 else:
244 expanded.append(value)
245 continue
247 if alias_ref in _seen:
248 # Cycle detected, raise an error with the cycle path
249 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
250 raise ValueError(
251 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
252 )
253 if alias_ref in aliases:
254 _seen.add(alias_ref)
255 _path.append(alias_ref)
256 # Recursively expand the alias values
257 nested_expanded = _expand_aliases(
258 aliases[alias_ref],
259 aliases=aliases,
260 teams=teams,
261 expand_teams=expand_teams,
262 _seen=_seen,
263 _path=_path,
264 )
265 if prefix:
266 expanded.extend(prefix + v for v in nested_expanded)
267 else:
268 expanded.extend(nested_expanded)
269 _path.pop()
270 _seen.remove(alias_ref)
271 else:
272 # Unknown alias — surface it loudly instead of silently dropping the
273 # reference (a typo'd alias would otherwise vanish reviewers/paths).
274 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
276 # Remove duplicates while preserving order
277 return list(dict.fromkeys(expanded))
280def _apply_negations(values: list[str]) -> list[str]:
281 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name"
282 entry removes "name" from the resolved list instead of surviving as a
283 match-time rule (contrast `authors`, handled by `matches_author`).
285 A list still holding an unexpanded `@team` ref (offline compile, i.e.
286 teams=None) is returned untouched — it isn't fully resolved yet, so
287 subtraction can't run, and the partially-resolved config must round-trip
288 unchanged. `$aliases` must already be expanded by the caller.
290 The wildcard+negation error fires BEFORE that early return: it's a check
291 on the written form (`"*"` alongside any `"!"` entry, team ref or not),
292 so an offline compile must reject it the same way the server will — a
293 `pullapprove check` that passes locally can't then error in production.
294 """
295 if "*" in values and any(v.startswith("!") for v in values):
296 raise ValueError(
297 'Negation cannot be combined with "*" in reviewers/alternates/cc '
298 "(wildcard exclusion is not supported)"
299 )
301 if any(_split_team_ref(v) is not None for v in values):
302 return values
304 negations = {v[1:].lower() for v in values if v.startswith("!")}
305 return [v for v in values if not v.startswith("!") and v.lower() not in negations]
308def _validate_review_counts(label: str, data: dict[str, Any]) -> None:
309 """Compile-time rejection of negative require/request/author_value.
310 Shared by scopes and large_scale_change (which has no
311 request/author_value — the `.get` defaults pass trivially there).
313 These are compile-time checks (not field validators) because negative
314 values were previously accepted, so compiled configs stored inside old
315 processing results must keep parsing (where they keep their old
316 behavior: a negative require always passed, a negative request
317 requested nobody).
318 """
319 for field in ("require", "request", "author_value"):
320 if data.get(field, 0) < 0:
321 raise ValueError(f"{label}: {field} cannot be negative")
324def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
325 """Resolve an `extends` reference to a canonical repo-relative config key.
327 - `/x` is repo-root-relative.
328 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
329 file's directory.
331 Raises if the reference escapes above the repo root.
332 """
333 if extends_ref.startswith("/"):
334 resolved = posixpath.normpath(extends_ref.lstrip("/"))
335 else:
336 base_dir = posixpath.dirname(extending_path)
337 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
339 if resolved == ".." or resolved.startswith("../"):
340 raise ValueError(
341 f"Invalid extends path: '{extends_ref}' points above the repo root"
342 )
344 return resolved
347def _anchor_path(base_dir: str, pattern: str) -> str:
348 """Anchor a scope path glob at `base_dir` (the owning config's directory).
350 Scope paths are written relative to the config they live in. A leading `/`
351 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
352 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
353 """
354 negate = pattern.startswith("!")
355 if negate:
356 pattern = pattern[1:]
358 if pattern.startswith("/"):
359 anchored = pattern.lstrip("/")
360 elif base_dir:
361 anchored = f"{base_dir}/{pattern}"
362 else:
363 anchored = pattern
365 return f"!{anchored}" if negate else anchored
368class OwnershipChoices(StrEnum):
369 EMPTY = ""
370 APPEND = "append"
371 GLOBAL = "global"
374def _validate_agent_username(username: str) -> None:
375 """A declared agent is always one concrete account name -- not a ref,
376 wildcard, or negation. There is no meaningful "any bot" and nothing to
377 subtract from, so the roster syntax doesn't apply here.
378 """
379 if username == "*" or username.startswith("!") or is_unexpanded_ref(username):
380 raise ValueError(
381 f"Agent username '{username}' must be a plain account username"
382 )
383 # Every `@` value left is one _validate_team_refs already rejects (a plain
384 # username can't be a team ref), so reuse it rather than restate it.
385 _validate_team_refs([username])
388def _reject_declared_agents(
389 values: list[str], declared_agents: set[str], *, where: str
390) -> None:
391 """Reject any declared agent listed in a roster surface (a scope's
392 reviewers/alternates/cc, or large_scale_change.reviewers).
394 A declared agent is never a person, so it can never sit in a roster.
395 Matched case-insensitively and run after alias expansion, so `$alias`
396 indirection can't smuggle one in. `where` names the surface; the message
397 becomes the git-host commit status, so it stays under ~130 chars.
398 """
399 for entry in values:
400 if entry.lower() in declared_agents:
401 raise ValueError(
402 f"{where}: '{entry}' is declared as an agent and cannot be "
403 "listed as a reviewer"
404 )
407def _first_case_insensitive_duplicate(values: Iterable[str]) -> str | None:
408 """The first value whose lowercased form was already seen, else None.
409 Shared by the case-insensitive uniqueness checks (agents, scope names)."""
410 seen: set[str] = set()
411 for value in values:
412 if value.lower() in seen:
413 return value
414 seen.add(value.lower())
415 return None
418class ScopeModel(BaseModel):
419 model_config = ConfigDict(extra="forbid")
421 # Required fields
422 name: str = Field(min_length=1)
423 paths: list[str] = Field(min_length=1)
425 # Optional fields
427 # Expanded version of lines could be dict
428 # with fnmatch, regex, exclude patterns, etc?
429 code: list[str] = []
431 # This only filtering field that can't be used with raw diff/files...
432 # If we get into that, the others are:
433 # - labels
434 # - ref
435 # - statuses
436 # - dates
437 # - body
438 # - title
439 # - other scopes
440 # (this is how I ended up with expressions...
441 # I'm not trying to build a general purpose workflow tool,
442 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
443 authors: list[str] = []
445 # (defaults should be the "empty" values)
446 description: str = ""
447 reviewers: list[str] = []
448 alternates: list[str] = []
449 cc: list[str] = []
451 # Review scoring
452 # Negative values are rejected at compile time (_validate_review_counts)
453 require: int = 0
454 author_value: int = 0
456 # How scopes are combined
457 ownership: OwnershipChoices = OwnershipChoices.EMPTY
459 # Actionable items
460 request: int = 0
461 labels: list[str] = []
462 instructions: str = ""
464 # Approval checklist
465 checklist: Checklist | None = None
467 @field_validator("name", mode="after")
468 @classmethod
469 def validate_name(cls, name: str) -> str:
470 if "," in name:
471 raise ValueError("Scope name cannot contain commas")
472 return name
474 @field_validator(*USER_LIST_FIELDS, mode="after")
475 @classmethod
476 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
477 return _validate_team_refs(values)
479 @field_validator("code", mode="after")
480 @classmethod
481 def validate_code_patterns(cls, code: list[str]) -> list[str]:
482 for pattern in code:
483 try:
484 parsed = sre_parse.parse(pattern)
485 except re.error as e:
486 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
487 if _has_nested_quantifiers(parsed):
488 raise ValueError(
489 f"Regex pattern '{pattern}' contains nested quantifiers, "
490 "which can cause catastrophic backtracking."
491 )
492 return code
494 @model_validator(mode="after")
495 def validate_reviewers_for_require(self) -> ScopeModel:
496 all_reviewers = self.reviewers + self.alternates
498 # Skip if wildcard - anyone can review
499 if "*" in all_reviewers:
500 return self
502 # Skip if aliases or team refs (possibly negated with "!") are not yet
503 # expanded. Re-validation after compilation only re-checks refs that
504 # actually resolve — an offline compile (teams=None) leaves refs
505 # unexpanded, so this same skip fires again there too.
506 if any(is_unexpanded_ref(r) for r in all_reviewers):
507 return self
509 if len(all_reviewers) < self.require:
510 raise ValueError(
511 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
512 )
513 return self
515 def author_points(self, author_username: str) -> int:
516 """Author points only count if the author is explicitly listed as a
517 reviewer (a wildcard is not converted to usernames)."""
518 if author_username.lower() in {r.lower() for r in self.reviewers}:
519 return self.author_value
520 return 0
522 def unsolvable_reason(self, author_username: str) -> str | None:
523 """
524 Explain why this scope can never pass for a PR authored by this user,
525 or None if it can.
527 Messages must stay under ~130 chars: they flow into the git-host
528 commit status description, which the adapters slice to 140.
529 """
530 if "*" in self.reviewers:
531 # Anyone can review, so any require is satisfiable
532 return None
534 # Count eligible reviewers (excluding author who can't self-approve)
535 eligible_reviewers = {r.lower() for r in self.reviewers + self.alternates} - {
536 author_username.lower()
537 }
538 max_possible_points = len(eligible_reviewers) + self.author_points(
539 author_username
540 )
542 if self.require > 0 and max_possible_points < self.require:
543 if not eligible_reviewers:
544 return (
545 "PR author is the only reviewer/alternate and cannot self-approve"
546 )
547 return f"require={self.require} but only {max_possible_points} possible approvals (excluding author)"
549 return None
551 def printed_name(self) -> str:
552 match self.ownership:
553 case OwnershipChoices.APPEND:
554 return "+" + self.name
555 case OwnershipChoices.GLOBAL:
556 return "*" + self.name
558 return self.name
560 def __eq__(self, other: Any) -> bool:
561 return self.name == other.name
563 def matches_path(self, path: Path) -> bool:
564 # TODO paths shouldn't start with /
565 return glob.globmatch(
566 path,
567 self.paths,
568 flags=glob.GLOBSTAR
569 | glob.BRACE
570 | glob.NEGATE
571 | glob.IGNORECASE
572 | glob.DOTGLOB,
573 )
575 def matches_code(self, code: str) -> Generator[dict[str, int]]:
576 patterns = getattr(self, "_code_regex_patterns", [])
577 if not patterns:
578 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
579 self._code_regex_patterns = patterns
581 for pattern in patterns:
582 for match in pattern.finditer(code):
583 start_index = match.start()
584 end_index = match.end()
586 start_line = code.count("\n", 0, start_index) + 1
587 start_col = start_index - code.rfind("\n", 0, start_index)
589 end_line = code.count("\n", 0, end_index) + 1
590 end_col = end_index - code.rfind("\n", 0, end_index)
592 yield {
593 "start_line": start_line,
594 "start_col": start_col,
595 "end_line": end_line,
596 "end_col": end_col,
597 }
599 def matches_author(self, author_username: str) -> bool:
600 if not self.authors:
601 # No authors specified, so assume it matches
602 return True
604 author_username_lower = author_username.lower()
606 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
607 authors = [a.lower() for a in self.authors if not a.startswith("!")]
609 if author_username_lower in negated_authors:
610 # If the author is in the negated list, return False
611 return False
613 if not authors:
614 # Negation-only: everyone not negated matches
615 return True
617 if author_username_lower in authors:
618 # If the author is in the authors list, return True
619 return True
621 return False
624class LargeScaleChangeModel(BaseModel):
625 model_config = ConfigDict(extra="forbid")
627 # Note, an LSC only applies to diffs, not raw files,
628 # because we have to know what *changed*.
630 # Pretty similar to a scope, but more manual.
631 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
632 # Negative values are rejected at compile time (_validate_review_counts)
633 require: int = 1
634 reviewers: list[str] = [] # Field(min_length=1)
635 # min_paths: int = 300
636 # min_lines: int = 3000
637 labels: list[str] = []
638 # really need author value too...?
640 @field_validator("reviewers", mode="after")
641 @classmethod
642 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
643 return _validate_team_refs(values)
645 def unsolvable_reason(self, author_username: str) -> str | None:
646 """
647 Explain why this LSC can never pass for a PR authored by this user,
648 or None if it can.
650 Messages must stay under ~130 chars: they flow into the git-host
651 commit status description, which the adapters slice to 140.
652 """
653 if "*" in self.reviewers:
654 # Anyone can review, so any require is satisfiable
655 return None
657 if not self.reviewers:
658 # An empty roster is "configuration required", which
659 # process_large_scale_change reports on its own
660 return None
662 # Count eligible reviewers (excluding author who can't self-approve)
663 eligible_reviewers = {r.lower() for r in self.reviewers} - {
664 author_username.lower()
665 }
666 if self.require > 0 and len(eligible_reviewers) < self.require:
667 if not eligible_reviewers:
668 return "the PR author is the only reviewer and cannot self-approve"
669 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)"
671 return None
674class ConfigModel(BaseModel):
675 model_config = ConfigDict(extra="forbid")
677 # Nothing is technically required
678 extends: list[str] = []
679 template: bool = False
680 aliases: dict[str, list[str]] = {}
681 large_scale_change: LargeScaleChangeModel | None = None
682 scopes: list[ScopeModel] = []
684 # Accounts whose review output PullApprove reads -- review bots (CodeRabbit,
685 # Copilot, CI reviewers). A declared agent is never treated as a person: it
686 # is never counted or requested, and listing one in reviewers/alternates/cc
687 # is a compile-time error (enforced in compiled_config, where the union
688 # across the whole config set is known). Concrete usernames only.
689 agents: list[str] = []
691 @field_validator("agents", mode="after")
692 @classmethod
693 def validate_agents(cls, agents: list[str]) -> list[str]:
694 for username in agents:
695 _validate_agent_username(username)
696 if dup := _first_case_insensitive_duplicate(agents):
697 raise ValueError(f"Agent '{dup}' is listed more than once")
698 return agents
700 @field_validator("scopes", mode="after")
701 @classmethod
702 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
703 if dup := _first_case_insensitive_duplicate(scope.name for scope in scopes):
704 raise ValueError(f"Duplicate scope name: {dup}")
705 return scopes
707 @field_validator("extends", mode="before")
708 @classmethod
709 def validate_extends(cls, extends: list[str]) -> list[str]:
710 for i, path in enumerate(extends):
711 basename = Path(path).name
712 if not basename.startswith(CONFIG_FILENAME_PREFIX):
713 raise ValueError(
714 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
715 )
716 return extends
718 def compiled_config(
719 self,
720 config_path: Path,
721 other_configs: ConfigModels,
722 teams: dict[str, list[str]] | None = None,
723 ) -> ConfigModel:
724 """
725 Resolve `extends` and replace aliases, returning the effective config.
727 Two phases: flatten the whole extends chain into one merged (raw,
728 unexpanded) config, then expand aliases once. Expanding after the full
729 merge is what makes transitive inheritance and cross-chain alias
730 scoping work — an alias defined anywhere in the chain resolves anywhere.
732 `teams` maps team refs (any case, without the leading `@`) to member
733 usernames, and is only consulted for user-list fields (reviewers,
734 alternates, authors, cc, large_scale_change.reviewers). Keys are
735 lowercased here, so callers don't need to normalize case themselves.
736 With `teams=None` (the default), `@org/team` references in those
737 fields pass through unexpanded — the offline/CLI mode, since this
738 library never calls out to GitHub/GitLab itself. That
739 partially-resolved config is only sanctioned for offline use;
740 anything that needs real reviewer usernames must pass a `teams`
741 mapping.
743 Pure function of the raw `self` and `other_configs` (it never reads its
744 own anchored output), so it is safe to call uncached.
745 """
747 if teams is not None:
748 teams = {key.lower(): members for key, members in teams.items()}
750 compiled_data = self._merged_data(config_path, other_configs)
752 # Expand aliases for any aliasable list fields. Team refs only expand
753 # in user-list fields — paths/code/labels keep a leading "@" literal.
754 for scope in compiled_data["scopes"]:
755 for field in [
756 "paths",
757 "code",
758 "authors",
759 "reviewers",
760 "alternates",
761 "cc",
762 "labels",
763 ]:
764 if field in scope:
765 scope[field] = _expand_aliases(
766 scope[field],
767 compiled_data["aliases"],
768 teams=teams,
769 expand_teams=field in USER_LIST_FIELDS,
770 )
772 # A declared agent is never a person, so it can't sit in a roster
773 # field. The set is the union across the whole config set (agents
774 # declared in any file, including a shared template), matched
775 # case-insensitively and checked after alias expansion so `$alias`
776 # indirection can't smuggle one in.
777 declared_agents = other_configs.agent_usernames()
779 # Apply compile-time "!" subtraction to roster fields.
780 # `_apply_negations` leaves a field with an unexpanded team ref
781 # (offline compile, i.e. teams=None) untouched.
782 for scope in compiled_data["scopes"]:
783 for field in ROSTER_FIELDS:
784 if field in scope:
785 scope[field] = _apply_negations(scope[field])
786 _reject_declared_agents(
787 scope[field],
788 declared_agents,
789 where=f"Scope '{scope['name']}' {field}",
790 )
792 # The "*" wildcard is only meaningful in `reviewers` — everywhere
793 # else it's matched as a literal username and silently does
794 # nothing: in `alternates` the scope stays pending forever, in
795 # `authors` the scope never applies to any PR, in `cc` nobody is
796 # notified. Reject it at compile time (after alias expansion, so
797 # `$alias` indirection can't smuggle it in) rather than as a
798 # ScopeModel validator, because compiled configs stored inside
799 # old processing results must keep parsing.
800 # Messages must stay under ~130 chars: they become the git-host
801 # commit status description, which the adapters slice to 140.
802 for field, hint in (
803 (
804 "authors",
805 "remove it (a scope with no authors applies to any author)",
806 ),
807 (
808 "alternates",
809 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
810 ),
811 ("cc", "remove it"),
812 ):
813 if "*" in scope.get(field, []):
814 raise ValueError(
815 f"Scope '{scope['name']}': \"*\" is not supported in "
816 f"{field} — {hint}"
817 )
819 _validate_review_counts(f"Scope '{scope['name']}'", scope)
821 if large_scale_change := compiled_data.get("large_scale_change"):
822 large_scale_change["reviewers"] = _expand_aliases(
823 large_scale_change["reviewers"],
824 compiled_data["aliases"],
825 teams=teams,
826 expand_teams=True,
827 )
828 large_scale_change["labels"] = _expand_aliases(
829 large_scale_change["labels"],
830 compiled_data["aliases"],
831 )
832 large_scale_change["reviewers"] = _apply_negations(
833 large_scale_change["reviewers"]
834 )
835 _reject_declared_agents(
836 large_scale_change["reviewers"],
837 declared_agents,
838 where="large_scale_change reviewers",
839 )
840 _validate_review_counts("large_scale_change", large_scale_change)
842 # Anchor each scope's paths at the directory tagged during flattening
843 # (after alias expansion, so any `$path-alias` is resolved first). The
844 # transient tag is popped so it never reaches the model.
845 for scope in compiled_data["scopes"]:
846 anchor_dir = scope.pop("_anchor_dir", "")
847 scope["paths"] = [_anchor_path(anchor_dir, p) for p in scope["paths"]]
849 # The compiled config is the self-contained effective config: extends
850 # are already merged in and aliases already expanded, so drop both. This
851 # keeps stored results lean and makes the compiled form standalone (it
852 # can never dangle on a missing extends target or re-expand differently).
853 compiled_data["extends"] = []
854 compiled_data["aliases"] = {}
856 return ConfigModel.from_data(
857 data=compiled_data,
858 path=config_path,
859 )
861 def _merged_data(
862 self,
863 config_path: Path,
864 other_configs: ConfigModels,
865 _in_progress: list[str] | None = None,
866 _seen: set[str] | None = None,
867 ) -> dict[str, Any]:
868 """
869 Flatten the `extends` chain into one merged, *unexpanded* config dict.
871 Parents are merged before this config (so a child can specialize), with
872 aliases unioned child-wins and the large-scale-change config taken from
873 the child if set else the first parent that defines one.
875 `_in_progress` is the current ancestor path, used to detect circular
876 extends. `_seen` is every config already merged into this flatten, used
877 to merge a shared ancestor only once (diamond dedup).
878 """
879 if _in_progress is None:
880 _in_progress = []
881 if _seen is None:
882 _seen = set()
884 config_path_str = str(config_path)
885 if config_path_str in _in_progress:
886 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
887 config_path_str
888 ]
889 raise ValueError(
890 f"Circular reference detected in extends: {' -> '.join(cycle)}"
891 )
892 _in_progress.append(config_path_str)
894 inherited_scopes: list[dict[str, Any]] = []
895 inherited_aliases: dict[str, list[str]] = {}
896 inherited_agents: list[str] = []
897 inherited_lsc: dict[str, Any] | None = None
899 for extend_path in self.extends:
900 resolved_path = _resolve_extends_path(config_path_str, extend_path)
901 if resolved_path not in other_configs:
902 raise ValueError(
903 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
904 )
905 if resolved_path in _seen:
906 # Already merged via another branch (diamond) — skip the dup.
907 continue
909 parent_data = other_configs[resolved_path]._merged_data(
910 Path(resolved_path), other_configs, _in_progress, _seen
911 )
912 inherited_scopes = inherited_scopes + parent_data["scopes"]
913 inherited_aliases = inherited_aliases | parent_data["aliases"]
914 inherited_agents = inherited_agents + parent_data["agents"]
915 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
917 merged = self.model_dump()
918 merged["scopes"] = inherited_scopes + merged["scopes"]
919 merged["aliases"] = inherited_aliases | merged["aliases"]
920 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
922 # Union the declared agents across the chain, deduped case-insensitively
923 # (first casing wins) so a parent and child naming the same agent don't
924 # re-trip the duplicate validator when the merged config is rebuilt.
925 merged_agents: list[str] = []
926 seen_agents: set[str] = set()
927 for username in inherited_agents + merged["agents"]:
928 if username.lower() not in seen_agents:
929 seen_agents.add(username.lower())
930 merged_agents.append(username)
931 merged["agents"] = merged_agents
933 # Tag each scope with the directory its paths should anchor at. A scope's
934 # paths are relative to the config that owns it, so the first
935 # non-template config to consume a scope claims it: a non-template's own
936 # scopes (and any it inherits from a template) anchor at its directory,
937 # while a template defers to its consumer. `setdefault` means an
938 # already-tagged scope (from a non-template ancestor) keeps its anchor.
939 if not self.template:
940 base_dir = posixpath.dirname(config_path_str)
941 for scope in merged["scopes"]:
942 scope.setdefault("_anchor_dir", base_dir)
944 _seen.add(config_path_str)
945 _in_progress.pop()
947 return merged
949 @classmethod
950 def from_filesystem(cls, path: Path | str) -> ConfigModel:
951 with open(path, "rb") as f:
952 return cls.from_data(tomllib.load(f), path)
954 @classmethod
955 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
956 return cls.from_data(tomllib.loads(content), path)
958 @classmethod
959 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
960 return cls(**data)
963class _ConfigModelsBase(RootModel):
964 """Shared storage and accessors for a set of configs keyed by repo path."""
966 root: dict[str, ConfigModel]
968 @classmethod
969 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
970 """Build from a dict of already-constructed configs keyed by path."""
971 configs = cls(root={})
972 for path, config_model in models.items():
973 configs.root[str(Path(path))] = config_model
974 return configs
976 def get_config_models(self) -> dict[str, ConfigModel]:
977 return dict(self.root.items())
979 def __bool__(self) -> bool:
980 return bool(self.root)
982 def __getitem__(self, key: str) -> ConfigModel:
983 return self.root[key]
985 def __contains__(self, key: str) -> bool:
986 return key in self.root
988 def __len__(self) -> int:
989 return len(self.root)
992class ConfigModels(_ConfigModelsBase):
993 """Configs exactly as loaded from the repo — extends unresolved, aliases
994 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
996 def agent_usernames(self) -> set[str]:
997 """Every account declared as an agent in any config, lowercased --
998 the union across the whole set.
1000 Read off the uncompiled configs on purpose: `agents` entries are
1001 literal account names -- no aliases, teams, or negation to resolve --
1002 so the set is known before compilation, which is what lets a caller
1003 fetch an agent's review output before evaluating scopes against it.
1004 """
1005 return {
1006 username.lower()
1007 for config in self.root.values()
1008 for username in config.agents
1009 }
1011 @classmethod
1012 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
1013 """Load configs from a dict of parsed config data keyed by path."""
1014 configs = cls(root={})
1016 for path, config_data in data.items():
1017 config = ConfigModel.from_data(config_data, Path(path))
1018 configs.add_config(config, Path(path))
1020 return configs
1022 @classmethod
1023 def from_contents(cls, contents: dict[str, str]) -> ConfigModels:
1024 """Load configs from a dict of raw TOML content keyed by path."""
1025 configs = cls(root={})
1027 for path, content in contents.items():
1028 configs.add_config(ConfigModel.from_content(content, path), Path(path))
1030 return configs
1032 def add_config(self, config: ConfigModel, path: Path) -> None:
1033 self.root[str(path)] = config
1035 def team_refs(self) -> set[str]:
1036 """Collect every team ref (lowercase, no leading `@`/`!`) that
1037 `compiled(teams=...)` would actually try to expand: refs written
1038 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
1039 `large_scale_change.reviewers`), plus any refs reachable from those
1040 fields through `$alias`/`!$alias` chains.
1042 Meant for callers that need to know which teams to fetch/sync before
1043 calling `compiled(teams=...)`.
1045 Implemented as an offline compile (`teams=None`): aliases expand but
1046 team refs pass through unexpanded, so whatever refs remain in the
1047 compiled user-list fields are — by construction — exactly the refs a
1048 real compile will try to expand. A ref that only appears in a
1049 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
1050 or inside an alias nothing references never survives into a compiled
1051 user-list field, so it is never collected. Raises the same config
1052 errors `compiled()` would (unknown alias, circular refs, ...), just
1053 earlier.
1054 """
1055 # Cheap pre-check: a team ref can only enter a compile as a literal
1056 # `@`/`!@` value in a user-list field or an alias value. Most repos
1057 # have none, and skipping the compile keeps this near-free for them.
1058 candidate_lists: list[list[str]] = []
1059 for config in self.root.values():
1060 candidate_lists.extend(config.aliases.values())
1061 for scope in config.scopes:
1062 for field in USER_LIST_FIELDS:
1063 candidate_lists.append(getattr(scope, field))
1064 if config.large_scale_change:
1065 candidate_lists.append(config.large_scale_change.reviewers)
1066 if not any(
1067 _split_team_ref(value) is not None
1068 for values in candidate_lists
1069 for value in values
1070 ):
1071 return set()
1073 refs: set[str] = set()
1075 def collect_refs(values: list[str]) -> None:
1076 for value in values:
1077 if (split := _split_team_ref(value)) is not None:
1078 refs.add(split[1].lower())
1080 for config in self.compiled(teams=None).get_config_models().values():
1081 if config.template:
1082 continue
1083 for scope in config.scopes:
1084 for field in USER_LIST_FIELDS:
1085 collect_refs(getattr(scope, field))
1086 if config.large_scale_change:
1087 collect_refs(config.large_scale_change.reviewers)
1089 return refs
1091 def compiled(
1092 self, teams: dict[str, list[str]] | None = None
1093 ) -> CompiledConfigModels:
1094 """Resolve the whole set into its effective, PR-independent form.
1096 Each non-template config is compiled once — extends merged, aliases
1097 expanded, paths anchored. Templates are NOT compiled standalone: a
1098 template scope may reference an alias the consuming config provides, and
1099 its paths anchor at the consumer. They are carried through untouched
1100 (folded into each consumer during that consumer's compile, and kept in
1101 the set for display).
1103 `teams` maps team refs (any case, without the leading `@`) to member
1104 usernames; passed straight through to each config's `compiled_config`
1105 (see there for case normalization and the `teams=None` vs provided
1106 semantics).
1108 The result is an immutable `CompiledConfigModels` — there is no way to
1109 compile it again, so the non-idempotent path anchoring can never
1110 double-apply.
1111 """
1112 effective: dict[str, ConfigModel] = {}
1113 for path, config in self.root.items():
1114 if config.template:
1115 effective[path] = config
1116 else:
1117 effective[path] = config.compiled_config(
1118 config_path=Path(path), other_configs=self, teams=teams
1119 )
1121 return CompiledConfigModels.from_config_models(effective)
1124class CompiledConfigModels(_ConfigModelsBase):
1125 """The effective configs used for matching: every non-template config is
1126 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
1128 def closest_config(self, file_path: Path) -> ConfigModel:
1129 """Return the closest non-template config governing this file."""
1130 for parent in file_path.parents:
1131 parent_config_path = str(parent / CONFIG_FILENAME)
1133 if parent_config_path in self.root:
1134 config = self.root[parent_config_path]
1136 if config.template:
1137 # Skip templates
1138 continue
1140 return config
1142 raise ValueError(f"No config found for {file_path}")
1144 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
1145 """The primary (repo-root) config's large-scale-change section, if any.
1147 The primary was compiled by `compiled()`, so its reviewers/labels are
1148 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
1149 true` repo root is a misconfiguration (templates are meant to be
1150 extended, not be the primary); it is passed through uncompiled, so its
1151 LSC would read with aliases unexpanded.
1152 """
1153 if CONFIG_FILENAME in self.root:
1154 if lsc := self.root[CONFIG_FILENAME].large_scale_change:
1155 return lsc
1157 return LargeScaleChangeModel()
1159 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
1160 """
1161 Overlay PR-dependent scope gating: drop scopes that author rules disable
1162 for this pull request.
1164 This is the only PR-dependent step. The configs are already compiled, so
1165 each config's scopes are self-contained and dropping one is a plain list
1166 filter — no re-inheritance. Templates are passed through (they are never
1167 matched directly; their scopes already live in each consumer).
1168 """
1169 effective: dict[str, ConfigModel] = {}
1170 for config_path, config in self.root.items():
1171 if config.template:
1172 # Templates are never matched directly; pass them through.
1173 effective[config_path] = config
1174 continue
1176 kept_scopes = [
1177 scope
1178 for scope in config.scopes
1179 if scope.matches_author(author_username)
1180 ]
1181 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
1183 return CompiledConfigModels.from_config_models(effective)