Coverage for src/pullapprove/config.py: 96%
530 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-17 21:57 -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
28from .presets import PRESET_NAMES, resolve_preset
31def _resolve_config_filename_prefix() -> str:
32 """The name every config file starts with, for this process.
34 Defaults to CODEREVIEW. An instance can rename it with
35 PULLAPPROVE_CONFIG_PREFIX, which renames the config file itself
36 (REVIEW -> REVIEW.toml, REVIEW.template.toml). Because discovery matches
37 on this prefix, an instance only ever sees files named for its own prefix
38 -- so two instances can watch the same repo without seeing each other's
39 configs, which is how unreleased config features get tried out on a repo
40 that production also watches.
42 Read once at import: this is a per-process constant, not runtime state.
44 Two operational constraints this doesn't (and can't) enforce:
46 Prefixes must not be prefixes of one another. Matching is `startswith`
47 (see is_config_filename), and a suffix after the prefix is legitimate --
48 CODEREVIEW.template.toml and CODEREVIEW-BASE.toml are both real config
49 names -- so a REVIEW instance would also pick up REVIEW_DEV.toml. An
50 instance only knows its own prefix, so pick names that don't overlap
51 (REVIEW and DEV_REVIEW, not REVIEW and REVIEW_DEV).
53 Changing the prefix on a live instance means clearing its cached configs.
54 Config rows are keyed by (repo, sha) with no record of which prefix
55 discovered them, so a sha already processed under the old prefix keeps
56 serving from cache and the new prefix's files are never fetched.
57 """
58 # `or` rather than a get() default: an explicitly empty value
59 # (`PULLAPPROVE_CONFIG_PREFIX=` in a .env or compose file -- a normal way to
60 # write "unset") is not a missing key, and would name the config ".toml".
61 prefix = os.environ.get("PULLAPPROVE_CONFIG_PREFIX", "").strip() or "CODEREVIEW"
63 # A bad prefix would silently match nothing (i.e. every repo looks
64 # unconfigured), so refuse to start instead.
65 if prefix.endswith(".toml"):
66 raise ValueError(
67 f"PULLAPPROVE_CONFIG_PREFIX should be a name without an extension, not {prefix!r}. "
68 f"The config file is named after it (e.g. {prefix[: -len('.toml')]!r} -> {prefix!r})."
69 )
70 if "\\" in prefix or prefix != posixpath.basename(prefix):
71 raise ValueError(
72 f"PULLAPPROVE_CONFIG_PREFIX should be a filename prefix, not a path: {prefix!r}"
73 )
75 return prefix
78CONFIG_FILENAME_PREFIX = _resolve_config_filename_prefix()
79CONFIG_FILENAME = f"{CONFIG_FILENAME_PREFIX}.toml"
82def is_config_filename(basename: str) -> bool:
83 """Whether a filename is a config file (CODEREVIEW.toml,
84 CODEREVIEW.template.toml).
86 The prefix half is what isolates instances: an instance running a renamed
87 prefix (see PULLAPPROVE_CONFIG_PREFIX) never even discovers another
88 instance's configs. The extension half keeps a neighbor like CODEREVIEW.md
89 from being treated as one.
90 """
91 return basename.startswith(CONFIG_FILENAME_PREFIX) and basename.endswith(".toml")
94_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
97def _has_nested_quantifiers(data: Any) -> bool:
98 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
99 for op, av in data:
100 if op in _REPEAT_OPS:
101 if _contains_quantifier(av[2]):
102 return True
103 elif op == sre_parse.SUBPATTERN:
104 if _has_nested_quantifiers(av[-1]):
105 return True
106 elif op == sre_parse.BRANCH:
107 if any(_has_nested_quantifiers(branch) for branch in av[1]):
108 return True
109 return False
112def _contains_quantifier(data: Any) -> bool:
113 for op, av in data:
114 if op in _REPEAT_OPS:
115 return True
116 elif op == sre_parse.SUBPATTERN:
117 if _contains_quantifier(av[-1]):
118 return True
119 elif op == sre_parse.BRANCH:
120 if any(_contains_quantifier(branch) for branch in av[1]):
121 return True
122 return False
125_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*"
126_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$")
128# Fields that hold reviewer identities (plain usernames, `$aliases`, and
129# `@team` refs) rather than paths/code/labels. Team refs only expand here.
130USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc")
132# Roster fields where "!" performs compile-time subtraction (remove from the
133# resolved list) rather than surviving as a match-time predicate. `authors`
134# is deliberately excluded — its "!" entries are consumed by `matches_author`.
135ROSTER_FIELDS = ("reviewers", "alternates", "cc")
138def _split_team_ref(value: str) -> tuple[str, str] | None:
139 """Split a value into `(prefix, ref)` if it has team-reference shape
140 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns
141 `None` for anything else, so callers fall back to their own handling of
142 plain values.
143 """
144 if value.startswith("!@"):
145 return "!", value[2:]
146 if value.startswith("@"):
147 return "", value[1:]
148 return None
151def is_unexpanded_ref(value: str) -> bool:
152 """True if the value is a `$alias` or `@team` reference (optionally
153 negated with a leading `!`) that has not been expanded — as opposed to a
154 plain username. Only an offline compile (`teams=None`) leaves such values
155 in user-list fields.
156 """
157 return value.removeprefix("!").startswith(("$", "@"))
160def _validate_team_refs(values: list[str]) -> list[str]:
161 """Team references (`@org/team`, `!@org/team`) need at least two
162 slash-separated segments. This only checks shape — membership is resolved
163 later, at compile time, against the caller-provided `teams` mapping.
165 Any other value containing "@" is rejected too — that shape is reserved
166 for team references (and, in future, email-style identifiers), so it
167 can't be confused with a plain username.
168 """
169 for value in values:
170 split = _split_team_ref(value)
171 if split is None:
172 if "@" in value:
173 raise ValueError(
174 f"Invalid value '{value}': email addresses are not "
175 "supported here — use the platform username"
176 )
177 continue
179 _prefix, ref = split
180 if not _TEAM_REF_RE.match(ref):
181 raise ValueError(
182 f"Invalid team reference '{value}': team references need the "
183 "'org/team' form"
184 )
185 return values
188def _expand_team_ref(
189 ref: str, teams: dict[str, list[str]] | None, prefix: str
190) -> list[str]:
191 """Expand a single team reference (without its `@`/`!@`) to member usernames.
193 `teams=None` is the offline/CLI mode: the library never calls out to
194 GitHub/GitLab itself, so with no mapping provided the reference passes
195 through unexpanded rather than erroring. With a `teams` mapping (even an
196 empty one), an unresolvable ref is a loud config error, matching how
197 unknown `$aliases` are handled.
198 """
199 if teams is None:
200 return [f"{prefix}@{ref}"]
202 members = teams.get(ref.lower())
203 if members is None:
204 raise ValueError(f"Unknown team: {prefix}@{ref}")
206 return [f"{prefix}{member}" for member in members]
209def _expand_aliases(
210 values: list[str],
211 aliases: dict[str, list[str]],
212 teams: dict[str, list[str]] | None = None,
213 expand_teams: bool = False,
214 _seen: set[str] | None = None,
215 _path: list[str] | None = None,
216) -> list[str]:
217 """Replace alias references in a list with their mapped values recursively.
219 Team references (`@org/team`) are only expanded when `expand_teams` is
220 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere
221 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style
222 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes —
223 their values are plain usernames, never other refs — so expanding one
224 never recurses further.
225 """
226 if _seen is None:
227 _seen = set()
228 if _path is None:
229 _path = []
231 expanded: list[str] = []
232 for value in values:
233 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
234 if value.startswith("!$"):
235 prefix = "!"
236 alias_ref = value[2:]
237 elif value.startswith("$"):
238 prefix = ""
239 alias_ref = value[1:]
240 elif expand_teams and (team_ref := _split_team_ref(value)) is not None:
241 team_prefix, ref = team_ref
242 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix))
243 continue
244 else:
245 expanded.append(value)
246 continue
248 if alias_ref in _seen:
249 # Cycle detected, raise an error with the cycle path
250 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
251 raise ValueError(
252 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
253 )
254 if alias_ref in aliases:
255 _seen.add(alias_ref)
256 _path.append(alias_ref)
257 # Recursively expand the alias values
258 nested_expanded = _expand_aliases(
259 aliases[alias_ref],
260 aliases=aliases,
261 teams=teams,
262 expand_teams=expand_teams,
263 _seen=_seen,
264 _path=_path,
265 )
266 if prefix:
267 expanded.extend(prefix + v for v in nested_expanded)
268 else:
269 expanded.extend(nested_expanded)
270 _path.pop()
271 _seen.remove(alias_ref)
272 else:
273 # Unknown alias — surface it loudly instead of silently dropping the
274 # reference (a typo'd alias would otherwise vanish reviewers/paths).
275 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
277 # Remove duplicates while preserving order
278 return list(dict.fromkeys(expanded))
281def _apply_negations(values: list[str]) -> list[str]:
282 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name"
283 entry removes "name" from the resolved list instead of surviving as a
284 match-time rule (contrast `authors`, handled by `matches_author`).
286 A list still holding an unexpanded `@team` ref (offline compile, i.e.
287 teams=None) is returned untouched — it isn't fully resolved yet, so
288 subtraction can't run, and the partially-resolved config must round-trip
289 unchanged. `$aliases` must already be expanded by the caller.
291 The wildcard+negation error fires BEFORE that early return: it's a check
292 on the written form (`"*"` alongside any `"!"` entry, team ref or not),
293 so an offline compile must reject it the same way the server will — a
294 `pullapprove check` that passes locally can't then error in production.
295 """
296 if "*" in values and any(v.startswith("!") for v in values):
297 raise ValueError(
298 'Negation cannot be combined with "*" in reviewers/alternates/cc '
299 "(wildcard exclusion is not supported)"
300 )
302 if any(_split_team_ref(v) is not None for v in values):
303 return values
305 negations = {v[1:].lower() for v in values if v.startswith("!")}
306 return [v for v in values if not v.startswith("!") and v.lower() not in negations]
309def _validate_review_counts(label: str, data: dict[str, Any]) -> None:
310 """Compile-time rejection of negative require/request/author_value.
311 Shared by scopes and large_scale_change (which has no
312 request/author_value — the `.get` defaults pass trivially there).
314 These are compile-time checks (not field validators) because negative
315 values were previously accepted, so compiled configs stored inside old
316 processing results must keep parsing (where they keep their old
317 behavior: a negative require always passed, a negative request
318 requested nobody).
319 """
320 for field in ("require", "request", "author_value"):
321 if data.get(field, 0) < 0:
322 raise ValueError(f"{label}: {field} cannot be negative")
325def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
326 """Resolve an `extends` reference to a canonical repo-relative config key.
328 - `/x` is repo-root-relative.
329 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
330 file's directory.
332 Raises if the reference escapes above the repo root.
333 """
334 if extends_ref.startswith("/"):
335 resolved = posixpath.normpath(extends_ref.lstrip("/"))
336 else:
337 base_dir = posixpath.dirname(extending_path)
338 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
340 if resolved == ".." or resolved.startswith("../"):
341 raise ValueError(
342 f"Invalid extends path: '{extends_ref}' points above the repo root"
343 )
345 return resolved
348def matches_path_patterns(*, path: Path, patterns: list[str]) -> bool:
349 """Whether `path` matches any of the config's path globs.
351 The one definition of the config's glob semantics — scopes and agents
352 both match through here, so their paths can never mean different things.
353 """
354 # TODO paths shouldn't start with /
355 return glob.globmatch(
356 path,
357 patterns,
358 flags=glob.GLOBSTAR | glob.BRACE | glob.NEGATE | glob.IGNORECASE | glob.DOTGLOB,
359 )
362def _anchor_path(base_dir: str, pattern: str) -> str:
363 """Anchor a scope path glob at `base_dir` (the owning config's directory).
365 Scope paths are written relative to the config they live in. A leading `/`
366 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
367 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
368 """
369 negate = pattern.startswith("!")
370 if negate:
371 pattern = pattern[1:]
373 if pattern.startswith("/"):
374 anchored = pattern.lstrip("/")
375 elif base_dir:
376 anchored = f"{base_dir}/{pattern}"
377 else:
378 anchored = pattern
380 return f"!{anchored}" if negate else anchored
383class OwnershipChoices(StrEnum):
384 EMPTY = ""
385 APPEND = "append"
386 GLOBAL = "global"
389# A GitHub App has two names: the login it posts as (`name[bot]`) and the slug
390# that owns its check runs (`name`). The suffix is how the engine tells a bot
391# from a person -- a `[bot]` account never counts as a human and never sits in
392# a roster, no declaration needed.
393BOT_LOGIN_SUFFIX = "[bot]"
396def is_bot_login(username: str) -> bool:
397 """The engine rule, in one place: a `[bot]` account is never a person."""
398 return username.lower().endswith(BOT_LOGIN_SUFFIX)
401def _split_check_ref(ref: str) -> tuple[str, str]:
402 """An `unless` ref as (producer slug lowercased, check name).
404 Split on the FIRST `/`: App slugs cannot contain one, check names can.
405 Both parts are stripped -- the validator checks THIS function's output, so
406 a ref that validates is exactly a ref that matches at runtime.
407 """
408 producer, _, name = ref.partition("/")
409 return producer.strip().lower(), name.strip()
412def _reject_bot_reviewers(values: list[str], *, where: str) -> None:
413 """Reject any `[bot]` account listed in a roster surface (a scope's
414 reviewers/alternates/cc, or large_scale_change.reviewers).
416 A bot is never a person, so it can never sit in a roster: bots that open
417 pull requests are routed with `authors`, and bots that attest are
418 referenced from `unless`. Checked after alias expansion so `$alias`
419 indirection can't smuggle one in. `where` names the surface; the message
420 becomes the git-host commit status, so it stays under ~130 chars.
421 """
422 for entry in values:
423 if is_bot_login(entry):
424 raise ValueError(
425 f"{where}: '{entry}' is a bot and cannot be listed as a reviewer"
426 )
429def _first_case_insensitive_duplicate(values: Iterable[str]) -> str | None:
430 """The first value whose lowercased form was already seen, else None."""
431 seen: set[str] = set()
432 for value in values:
433 if value.lower() in seen:
434 return value
435 seen.add(value.lower())
436 return None
439_KEBAB_NAME_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")
442class AgentModel(BaseModel):
443 """One agent PullApprove runs against the change.
445 [[agents]]
446 name = "react"
447 preset = "codex-review"
448 paths = ["frontend/**"]
450 An agent names exactly one reviewer: a `preset` — a command PullApprove
451 maintains, see presets.py. Nothing else is configurable — the preset IS
452 the reviewer, and varying it per repo is what presets exist to prevent.
454 `paths` is the agent's own jurisdiction, matched the same way a scope's
455 paths are: it runs when the change touches them, it reviews the touched
456 files that match, and findings it reports elsewhere are recorded but never
457 counted. Required, so "everything" is a line someone wrote (`["**"]`)
458 rather than a blank someone has to interpret.
459 """
461 model_config = ConfigDict(extra="forbid")
463 name: str
465 preset: str = ""
467 paths: list[str] = Field(min_length=1)
469 @field_validator("name", mode="after")
470 @classmethod
471 def validate_name(cls, name: str) -> str:
472 if not _KEBAB_NAME_RE.match(name):
473 raise ValueError(
474 f"Invalid agent name '{name}'. Use lowercase letters, numbers, "
475 "and single hyphens (e.g. 'security-review')."
476 )
477 return name
479 @model_validator(mode="after")
480 def validate_reviewer(self) -> AgentModel:
481 """Every agent names a real preset.
483 Model-level rather than compiled-level: `extends` concatenates whole
484 agents rather than merging their fields, and the compiled data goes
485 back through this model, so an agent is the same object here and there.
486 """
487 if not self.preset:
488 raise ValueError(
489 f"Agent '{self.name}' names no reviewer. Set 'preset' (one of: "
490 f"{', '.join(PRESET_NAMES)})."
491 )
493 try:
494 resolve_preset(self.preset)
495 except ValueError as exc:
496 raise ValueError(f"Agent '{self.name}': {exc}") from exc
498 return self
500 def command_for(self, *, base: str) -> str:
501 """The reviewer command to run, with the base spelled into it.
503 A preset is resolved through the catalog. `validate_reviewer`
504 guarantees a preset is set, so there is no default to fall back to.
505 """
506 return resolve_preset(self.preset).command(base=base)
508 def matches_path(self, path: Path) -> bool:
509 return matches_path_patterns(path=path, patterns=self.paths)
512class ScopeModel(BaseModel):
513 model_config = ConfigDict(extra="forbid")
515 # Required fields
516 name: str = Field(min_length=1)
517 paths: list[str] = Field(min_length=1)
519 # Optional fields
521 # Expanded version of lines could be dict
522 # with fnmatch, regex, exclude patterns, etc?
523 code: list[str] = []
525 # This only filtering field that can't be used with raw diff/files...
526 # If we get into that, the others are:
527 # - labels
528 # - ref
529 # - statuses
530 # - dates
531 # - body
532 # - title
533 # - other scopes
534 # (this is how I ended up with expressions...
535 # I'm not trying to build a general purpose workflow tool,
536 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
537 authors: list[str] = []
539 # (defaults should be the "empty" values)
540 description: str = ""
541 reviewers: list[str] = []
542 alternates: list[str] = []
543 cc: list[str] = []
545 # Review scoring
546 # Negative values are rejected at compile time (_validate_review_counts)
547 require: int = 0
548 author_value: int = 0
550 # The checks that can attest this scope's review requirement away. Each ref
551 # is "producer/check-name" -- the slug of the App that produces the check,
552 # then the check run's name, split on the first "/". When every listed
553 # check has completed successfully on the current head commit, the scope is
554 # waived: its requirement drops to zero and the evidence is stored on the
555 # result. Failed, skipped, pending, missing -- the requirement stands
556 # unchanged; passing is the only state that subtracts.
557 #
558 # Deliberately placed next to the `require` it undermines, so reading a
559 # scope always reveals whether its human review is removable. A list is
560 # implicitly "all must pass" -- OR, thresholds, and 2-of-3 belong in the
561 # producer, which can post one combined check.
562 unless: list[str] = []
564 # How scopes are combined
565 ownership: OwnershipChoices = OwnershipChoices.EMPTY
567 # Actionable items
568 request: int = 0
569 labels: list[str] = []
570 instructions: str = ""
572 # Approval checklist
573 checklist: Checklist | None = None
575 @field_validator("name", mode="after")
576 @classmethod
577 def validate_name(cls, name: str) -> str:
578 if "," in name:
579 raise ValueError("Scope name cannot contain commas")
580 return name
582 @field_validator(*USER_LIST_FIELDS, mode="after")
583 @classmethod
584 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
585 return _validate_team_refs(values)
587 @field_validator("unless", mode="after")
588 @classmethod
589 def validate_unless_refs(cls, values: list[str]) -> list[str]:
590 """A check name alone is worthless -- any workflow with `checks: write`
591 can post any name -- so a bare name is a parse error, not a default.
592 Messages stay under ~130 chars: they become the git-host commit status.
593 """
594 for ref in values:
595 producer, name = _split_check_ref(ref)
596 if "/" not in ref or not producer or not name:
597 raise ValueError(
598 f"unless: '{ref}' must be 'producer/check-name' -- the App "
599 "slug that produces the check, then the check's name"
600 )
601 # To most users PullApprove *is* "the check" on their PRs.
602 # PullApprove never creates these checks, it only reads them -- and
603 # a config waiting on our own status would deadlock politely.
604 if producer == "pullapprove":
605 raise ValueError(
606 f"unless: '{ref}' references PullApprove itself -- "
607 "PullApprove never creates checks, it only reads them"
608 )
609 return values
611 @field_validator("code", mode="after")
612 @classmethod
613 def validate_code_patterns(cls, code: list[str]) -> list[str]:
614 for pattern in code:
615 try:
616 parsed = sre_parse.parse(pattern)
617 except re.error as e:
618 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
619 if _has_nested_quantifiers(parsed):
620 raise ValueError(
621 f"Regex pattern '{pattern}' contains nested quantifiers, "
622 "which can cause catastrophic backtracking."
623 )
624 return code
626 @model_validator(mode="after")
627 def validate_reviewers_for_require(self) -> ScopeModel:
628 all_reviewers = self.reviewers + self.alternates
630 # Skip if wildcard - anyone can review
631 if "*" in all_reviewers:
632 return self
634 # Skip if aliases or team refs (possibly negated with "!") are not yet
635 # expanded. Re-validation after compilation only re-checks refs that
636 # actually resolve — an offline compile (teams=None) leaves refs
637 # unexpanded, so this same skip fires again there too.
638 if any(is_unexpanded_ref(r) for r in all_reviewers):
639 return self
641 if len(all_reviewers) < self.require:
642 raise ValueError(
643 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
644 )
645 return self
647 def author_points(self, author_username: str) -> int:
648 """Author points only count if the author is explicitly listed as a
649 reviewer (a wildcard is not converted to usernames)."""
650 if author_username.lower() in {r.lower() for r in self.reviewers}:
651 return self.author_value
652 return 0
654 def unless_refs(self) -> list[tuple[str, str]]:
655 """Each `unless` entry as (producer slug lowercased, check name)."""
656 return [_split_check_ref(ref) for ref in self.unless]
658 def unsolvable_reason(self, author_username: str) -> str | None:
659 """
660 Explain why this scope can never pass for a PR authored by this user,
661 or None if it can.
663 Messages must stay under ~130 chars: they flow into the git-host
664 commit status description, which the adapters slice to 140.
665 """
666 if "*" in self.reviewers:
667 # Anyone can review, so any require is satisfiable
668 return None
670 # Count eligible reviewers (excluding author who can't self-approve)
671 eligible_reviewers = {r.lower() for r in self.reviewers + self.alternates} - {
672 author_username.lower()
673 }
674 max_possible_points = len(eligible_reviewers) + self.author_points(
675 author_username
676 )
678 if self.require > 0 and max_possible_points < self.require:
679 if not eligible_reviewers:
680 return (
681 "PR author is the only reviewer/alternate and cannot self-approve"
682 )
683 return f"require={self.require} but only {max_possible_points} possible approvals (excluding author)"
685 return None
687 def ownership_marker(self) -> str:
688 """The glyph a non-default ownership puts in front of the scope name.
690 Split out from printed_name so a renderer that styles the marker apart
691 from the name doesn't have to know which glyph goes with which mode.
692 """
693 match self.ownership:
694 case OwnershipChoices.APPEND:
695 return "+"
696 case OwnershipChoices.GLOBAL:
697 return "*"
699 return ""
701 def printed_name(self) -> str:
702 return self.ownership_marker() + self.name
704 def __eq__(self, other: Any) -> bool:
705 return self.name == other.name
707 def matches_path(self, path: Path) -> bool:
708 return matches_path_patterns(path=path, patterns=self.paths)
710 def matches_code(self, code: str) -> Generator[dict[str, int]]:
711 patterns = getattr(self, "_code_regex_patterns", [])
712 if not patterns:
713 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
714 self._code_regex_patterns = patterns
716 for pattern in patterns:
717 for match in pattern.finditer(code):
718 start_index = match.start()
719 end_index = match.end()
721 start_line = code.count("\n", 0, start_index) + 1
722 start_col = start_index - code.rfind("\n", 0, start_index)
724 end_line = code.count("\n", 0, end_index) + 1
725 end_col = end_index - code.rfind("\n", 0, end_index)
727 yield {
728 "start_line": start_line,
729 "start_col": start_col,
730 "end_line": end_line,
731 "end_col": end_col,
732 }
734 def matches_author(self, author_username: str) -> bool:
735 if not self.authors:
736 # No authors specified, so assume it matches
737 return True
739 author_username_lower = author_username.lower()
741 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
742 authors = [a.lower() for a in self.authors if not a.startswith("!")]
744 if author_username_lower in negated_authors:
745 # If the author is in the negated list, return False
746 return False
748 if not authors:
749 # Negation-only: everyone not negated matches
750 return True
752 if author_username_lower in authors:
753 # If the author is in the authors list, return True
754 return True
756 return False
759class LargeScaleChangeModel(BaseModel):
760 model_config = ConfigDict(extra="forbid")
762 # Note, an LSC only applies to diffs, not raw files,
763 # because we have to know what *changed*.
765 # Pretty similar to a scope, but more manual.
766 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
767 # Negative values are rejected at compile time (_validate_review_counts)
768 require: int = 1
769 reviewers: list[str] = [] # Field(min_length=1)
770 # min_paths: int = 300
771 # min_lines: int = 3000
772 labels: list[str] = []
773 # really need author value too...?
775 @field_validator("reviewers", mode="after")
776 @classmethod
777 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
778 return _validate_team_refs(values)
780 def unsolvable_reason(self, author_username: str) -> str | None:
781 """
782 Explain why this LSC can never pass for a PR authored by this user,
783 or None if it can.
785 Messages must stay under ~130 chars: they flow into the git-host
786 commit status description, which the adapters slice to 140.
787 """
788 if "*" in self.reviewers:
789 # Anyone can review, so any require is satisfiable
790 return None
792 if not self.reviewers:
793 # An empty roster is "configuration required", which
794 # process_large_scale_change reports on its own
795 return None
797 # Count eligible reviewers (excluding author who can't self-approve)
798 eligible_reviewers = {r.lower() for r in self.reviewers} - {
799 author_username.lower()
800 }
801 if self.require > 0 and len(eligible_reviewers) < self.require:
802 if not eligible_reviewers:
803 return "the PR author is the only reviewer and cannot self-approve"
804 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)"
806 return None
809class ConfigModel(BaseModel):
810 model_config = ConfigDict(extra="forbid")
812 # Nothing is technically required
813 extends: list[str] = []
814 template: bool = False
815 aliases: dict[str, list[str]] = {}
816 large_scale_change: LargeScaleChangeModel | None = None
817 scopes: list[ScopeModel] = []
818 agents: list[AgentModel] = []
820 @field_validator("scopes", mode="after")
821 @classmethod
822 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
823 if dup := _first_case_insensitive_duplicate(scope.name for scope in scopes):
824 raise ValueError(f"Duplicate scope name: {dup}")
825 return scopes
827 @field_validator("agents", mode="after")
828 @classmethod
829 def validate_unique_agent_names(cls, agents: list[AgentModel]) -> list[AgentModel]:
830 if dup := _first_case_insensitive_duplicate(agent.name for agent in agents):
831 raise ValueError(f"Duplicate agent name: {dup}")
832 return agents
834 @field_validator("extends", mode="before")
835 @classmethod
836 def validate_extends(cls, extends: list[str]) -> list[str]:
837 for i, path in enumerate(extends):
838 basename = Path(path).name
839 if not basename.startswith(CONFIG_FILENAME_PREFIX):
840 raise ValueError(
841 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
842 )
843 return extends
845 def compiled_config(
846 self,
847 config_path: Path,
848 other_configs: ConfigModels,
849 teams: dict[str, list[str]] | None = None,
850 ) -> ConfigModel:
851 """
852 Resolve `extends` and replace aliases, returning the effective config.
854 Two phases: flatten the whole extends chain into one merged (raw,
855 unexpanded) config, then expand aliases once. Expanding after the full
856 merge is what makes transitive inheritance and cross-chain alias
857 scoping work — an alias defined anywhere in the chain resolves anywhere.
859 `teams` maps team refs (any case, without the leading `@`) to member
860 usernames, and is only consulted for user-list fields (reviewers,
861 alternates, authors, cc, large_scale_change.reviewers). Keys are
862 lowercased here, so callers don't need to normalize case themselves.
863 With `teams=None` (the default), `@org/team` references in those
864 fields pass through unexpanded — the offline/CLI mode, since this
865 library never calls out to GitHub/GitLab itself. That
866 partially-resolved config is only sanctioned for offline use;
867 anything that needs real reviewer usernames must pass a `teams`
868 mapping.
870 Pure function of the raw `self` and `other_configs` (it never reads its
871 own anchored output), so it is safe to call uncached.
872 """
874 if teams is not None:
875 teams = {key.lower(): members for key, members in teams.items()}
877 compiled_data = self._merged_data(config_path, other_configs)
879 # Expand aliases for any aliasable list fields. Team refs only expand
880 # in user-list fields — paths/code/labels keep a leading "@" literal.
881 for scope in compiled_data["scopes"]:
882 for field in [
883 "paths",
884 "code",
885 "authors",
886 "reviewers",
887 "alternates",
888 "cc",
889 "labels",
890 ]:
891 if field in scope:
892 scope[field] = _expand_aliases(
893 scope[field],
894 compiled_data["aliases"],
895 teams=teams,
896 expand_teams=field in USER_LIST_FIELDS,
897 )
899 # An agent's paths are paths like a scope's: `$path-alias` references
900 # resolve, team refs stay literal.
901 for agent in compiled_data["agents"]:
902 agent["paths"] = _expand_aliases(
903 agent["paths"],
904 compiled_data["aliases"],
905 teams=teams,
906 )
908 # Apply compile-time "!" subtraction to roster fields.
909 # `_apply_negations` leaves a field with an unexpanded team ref
910 # (offline compile, i.e. teams=None) untouched.
911 for scope in compiled_data["scopes"]:
912 for field in ROSTER_FIELDS:
913 if field in scope:
914 scope[field] = _apply_negations(scope[field])
915 _reject_bot_reviewers(
916 scope[field],
917 where=f"Scope '{scope['name']}' {field}",
918 )
920 # The "*" wildcard is only meaningful in `reviewers` — everywhere
921 # else it's matched as a literal username and silently does
922 # nothing: in `alternates` the scope stays pending forever, in
923 # `authors` the scope never applies to any PR, in `cc` nobody is
924 # notified. Reject it at compile time (after alias expansion, so
925 # `$alias` indirection can't smuggle it in) rather than as a
926 # ScopeModel validator, because compiled configs stored inside
927 # old processing results must keep parsing.
928 # Messages must stay under ~130 chars: they become the git-host
929 # commit status description, which the adapters slice to 140.
930 for field, hint in (
931 (
932 "authors",
933 "remove it (a scope with no authors applies to any author)",
934 ),
935 (
936 "alternates",
937 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
938 ),
939 ("cc", "remove it"),
940 ):
941 if "*" in scope.get(field, []):
942 raise ValueError(
943 f"Scope '{scope['name']}': \"*\" is not supported in "
944 f"{field} — {hint}"
945 )
947 _validate_review_counts(f"Scope '{scope['name']}'", scope)
949 if large_scale_change := compiled_data.get("large_scale_change"):
950 large_scale_change["reviewers"] = _expand_aliases(
951 large_scale_change["reviewers"],
952 compiled_data["aliases"],
953 teams=teams,
954 expand_teams=True,
955 )
956 large_scale_change["labels"] = _expand_aliases(
957 large_scale_change["labels"],
958 compiled_data["aliases"],
959 )
960 large_scale_change["reviewers"] = _apply_negations(
961 large_scale_change["reviewers"]
962 )
963 _reject_bot_reviewers(
964 large_scale_change["reviewers"],
965 where="large_scale_change reviewers",
966 )
967 _validate_review_counts("large_scale_change", large_scale_change)
969 # Anchor each scope's paths at the directory tagged during flattening
970 # (after alias expansion, so any `$path-alias` is resolved first). The
971 # transient tag is popped so it never reaches the model.
972 # Scopes and agents alike: paths are relative to the config that owns
973 # the entry, `/` for repo-root-absolute.
974 for entry in [*compiled_data["scopes"], *compiled_data["agents"]]:
975 anchor_dir = entry.pop("_anchor_dir", "")
976 entry["paths"] = [_anchor_path(anchor_dir, p) for p in entry["paths"]]
978 # The compiled config is the self-contained effective config: extends
979 # are already merged in and aliases already expanded, so drop both. This
980 # keeps stored results lean and makes the compiled form standalone (it
981 # can never dangle on a missing extends target or re-expand differently).
982 compiled_data["extends"] = []
983 compiled_data["aliases"] = {}
985 return ConfigModel.from_data(
986 data=compiled_data,
987 path=config_path,
988 )
990 def _merged_data(
991 self,
992 config_path: Path,
993 other_configs: ConfigModels,
994 _in_progress: list[str] | None = None,
995 _seen: set[str] | None = None,
996 ) -> dict[str, Any]:
997 """
998 Flatten the `extends` chain into one merged, *unexpanded* config dict.
1000 Parents are merged before this config (so a child can specialize), with
1001 aliases unioned child-wins and the large-scale-change config taken from
1002 the child if set else the first parent that defines one.
1004 `_in_progress` is the current ancestor path, used to detect circular
1005 extends. `_seen` is every config already merged into this flatten, used
1006 to merge a shared ancestor only once (diamond dedup).
1007 """
1008 if _in_progress is None:
1009 _in_progress = []
1010 if _seen is None:
1011 _seen = set()
1013 config_path_str = str(config_path)
1014 if config_path_str in _in_progress:
1015 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
1016 config_path_str
1017 ]
1018 raise ValueError(
1019 f"Circular reference detected in extends: {' -> '.join(cycle)}"
1020 )
1021 _in_progress.append(config_path_str)
1023 inherited_scopes: list[dict[str, Any]] = []
1024 inherited_agents: list[dict[str, Any]] = []
1025 inherited_aliases: dict[str, list[str]] = {}
1026 inherited_lsc: dict[str, Any] | None = None
1028 for extend_path in self.extends:
1029 resolved_path = _resolve_extends_path(config_path_str, extend_path)
1030 if resolved_path not in other_configs:
1031 raise ValueError(
1032 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
1033 )
1034 if resolved_path in _seen:
1035 # Already merged via another branch (diamond) — skip the dup.
1036 continue
1038 parent_data = other_configs[resolved_path]._merged_data(
1039 Path(resolved_path), other_configs, _in_progress, _seen
1040 )
1041 inherited_scopes = inherited_scopes + parent_data["scopes"]
1042 inherited_agents = inherited_agents + parent_data["agents"]
1043 inherited_aliases = inherited_aliases | parent_data["aliases"]
1044 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
1046 merged = self.model_dump()
1047 merged["scopes"] = inherited_scopes + merged["scopes"]
1048 merged["agents"] = inherited_agents + merged["agents"]
1049 merged["aliases"] = inherited_aliases | merged["aliases"]
1050 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
1052 # Tag each scope with the directory its paths should anchor at. A scope's
1053 # paths are relative to the config that owns it, so the first
1054 # non-template config to consume a scope claims it: a non-template's own
1055 # scopes (and any it inherits from a template) anchor at its directory,
1056 # while a template defers to its consumer. `setdefault` means an
1057 # already-tagged scope (from a non-template ancestor) keeps its anchor.
1058 if not self.template:
1059 base_dir = posixpath.dirname(config_path_str)
1060 for entry in [*merged["scopes"], *merged["agents"]]:
1061 entry.setdefault("_anchor_dir", base_dir)
1063 _seen.add(config_path_str)
1064 _in_progress.pop()
1066 return merged
1068 @classmethod
1069 def from_filesystem(cls, path: Path | str) -> ConfigModel:
1070 with open(path, "rb") as f:
1071 return cls.from_data(tomllib.load(f), path)
1073 @classmethod
1074 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
1075 return cls.from_data(tomllib.loads(content), path)
1077 @classmethod
1078 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
1079 return cls(**data)
1082class _ConfigModelsBase(RootModel):
1083 """Shared storage and accessors for a set of configs keyed by repo path."""
1085 root: dict[str, ConfigModel]
1087 @classmethod
1088 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
1089 """Build from a dict of already-constructed configs keyed by path."""
1090 configs = cls(root={})
1091 for path, config_model in models.items():
1092 configs.root[str(Path(path))] = config_model
1093 return configs
1095 def get_config_models(self) -> dict[str, ConfigModel]:
1096 return dict(self.root.items())
1098 def __bool__(self) -> bool:
1099 return bool(self.root)
1101 def __getitem__(self, key: str) -> ConfigModel:
1102 return self.root[key]
1104 def __contains__(self, key: str) -> bool:
1105 return key in self.root
1107 def __len__(self) -> int:
1108 return len(self.root)
1111class ConfigModels(_ConfigModelsBase):
1112 """Configs exactly as loaded from the repo — extends unresolved, aliases
1113 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
1115 def declared_check_names(self) -> set[str]:
1116 """Every check-run name an effective (non-template) scope's `unless`
1117 references, across the whole config set.
1119 Collected from an offline compile (`teams=None`, the `team_refs`
1120 precedent) so a template's refs count only where a consumer actually
1121 inherits them -- an unconsumed template must not make every pull
1122 request fetch (or error on) checks that nothing evaluates. `unless`
1123 refs are literal strings (aliases and teams never expand inside
1124 them), so the offline compile is exact. Empty for the common case (no
1125 `unless` anywhere), which is what lets the processor skip fetching
1126 check runs entirely.
1127 """
1128 return {
1129 name
1130 for config in self.compiled().root.values()
1131 if not config.template
1132 for scope in config.scopes
1133 for _, name in scope.unless_refs()
1134 }
1136 @classmethod
1137 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
1138 """Load configs from a dict of parsed config data keyed by path."""
1139 configs = cls(root={})
1141 for path, config_data in data.items():
1142 config = ConfigModel.from_data(config_data, Path(path))
1143 configs.add_config(config, Path(path))
1145 return configs
1147 @classmethod
1148 def from_contents(cls, contents: dict[str, str]) -> ConfigModels:
1149 """Load configs from a dict of raw TOML content keyed by path."""
1150 configs = cls(root={})
1152 for path, content in contents.items():
1153 configs.add_config(ConfigModel.from_content(content, path), Path(path))
1155 return configs
1157 def add_config(self, config: ConfigModel, path: Path) -> None:
1158 self.root[str(path)] = config
1160 def team_refs(self) -> set[str]:
1161 """Collect every team ref (lowercase, no leading `@`/`!`) that
1162 `compiled(teams=...)` would actually try to expand: refs written
1163 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
1164 `large_scale_change.reviewers`), plus any refs reachable from those
1165 fields through `$alias`/`!$alias` chains.
1167 Meant for callers that need to know which teams to fetch/sync before
1168 calling `compiled(teams=...)`.
1170 Implemented as an offline compile (`teams=None`): aliases expand but
1171 team refs pass through unexpanded, so whatever refs remain in the
1172 compiled user-list fields are — by construction — exactly the refs a
1173 real compile will try to expand. A ref that only appears in a
1174 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
1175 or inside an alias nothing references never survives into a compiled
1176 user-list field, so it is never collected. Raises the same config
1177 errors `compiled()` would (unknown alias, circular refs, ...), just
1178 earlier.
1179 """
1180 # Cheap pre-check: a team ref can only enter a compile as a literal
1181 # `@`/`!@` value in a user-list field or an alias value. Most repos
1182 # have none, and skipping the compile keeps this near-free for them.
1183 candidate_lists: list[list[str]] = []
1184 for config in self.root.values():
1185 candidate_lists.extend(config.aliases.values())
1186 for scope in config.scopes:
1187 for field in USER_LIST_FIELDS:
1188 candidate_lists.append(getattr(scope, field))
1189 if config.large_scale_change:
1190 candidate_lists.append(config.large_scale_change.reviewers)
1191 if not any(
1192 _split_team_ref(value) is not None
1193 for values in candidate_lists
1194 for value in values
1195 ):
1196 return set()
1198 refs: set[str] = set()
1200 def collect_refs(values: list[str]) -> None:
1201 for value in values:
1202 if (split := _split_team_ref(value)) is not None:
1203 refs.add(split[1].lower())
1205 for config in self.compiled(teams=None).get_config_models().values():
1206 if config.template:
1207 continue
1208 for scope in config.scopes:
1209 for field in USER_LIST_FIELDS:
1210 collect_refs(getattr(scope, field))
1211 if config.large_scale_change:
1212 collect_refs(config.large_scale_change.reviewers)
1214 return refs
1216 def compiled(
1217 self, teams: dict[str, list[str]] | None = None
1218 ) -> CompiledConfigModels:
1219 """Resolve the whole set into its effective, PR-independent form.
1221 Each non-template config is compiled once — extends merged, aliases
1222 expanded, paths anchored. Templates are NOT compiled standalone: a
1223 template scope may reference an alias the consuming config provides, and
1224 its paths anchor at the consumer. They are carried through untouched
1225 (folded into each consumer during that consumer's compile, and kept in
1226 the set for display).
1228 `teams` maps team refs (any case, without the leading `@`) to member
1229 usernames; passed straight through to each config's `compiled_config`
1230 (see there for case normalization and the `teams=None` vs provided
1231 semantics).
1233 The result is an immutable `CompiledConfigModels` — there is no way to
1234 compile it again, so the non-idempotent path anchoring can never
1235 double-apply.
1236 """
1237 effective: dict[str, ConfigModel] = {}
1238 for path, config in self.root.items():
1239 if config.template:
1240 effective[path] = config
1241 else:
1242 effective[path] = config.compiled_config(
1243 config_path=Path(path), other_configs=self, teams=teams
1244 )
1246 return CompiledConfigModels.from_config_models(effective)
1249class CompiledConfigModels(_ConfigModelsBase):
1250 """The effective configs used for matching: every non-template config is
1251 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
1253 def agents_by_config(self) -> dict[str, dict[str, AgentModel]]:
1254 """Every effective config's agents, keyed by config path then name.
1256 Per config, not flattened, because an agent belongs to the config that
1257 declares it and covers that config's files. Two sibling configs may
1258 each declare a `security` agent; those are two agents with one name,
1259 each reviewing its own config's subtree — flattening them into one
1260 by-name dict would silently make the last one win.
1262 Templates are skipped: their agents are already folded into each
1263 consumer's compiled config, and a template governs no files itself.
1264 """
1265 return {
1266 config_path: {agent.name: agent for agent in config.agents}
1267 for config_path, config in self.root.items()
1268 if not config.template
1269 }
1271 def closest_config_path(self, file_path: Path) -> str | None:
1272 """The path of the closest non-template config governing this file,
1273 or None when nothing governs it."""
1274 for parent in file_path.parents:
1275 parent_config_path = str(parent / CONFIG_FILENAME)
1276 config = self.root.get(parent_config_path)
1277 if config is not None and not config.template:
1278 return parent_config_path
1279 return None
1281 def closest_config(self, file_path: Path) -> ConfigModel:
1282 """Return the closest non-template config governing this file."""
1283 config_path = self.closest_config_path(file_path)
1284 if config_path is None:
1285 raise ValueError(f"No config found for {file_path}")
1286 return self.root[config_path]
1288 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
1289 """The primary (repo-root) config's large-scale-change section, if any.
1291 The primary was compiled by `compiled()`, so its reviewers/labels are
1292 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
1293 true` repo root is a misconfiguration (templates are meant to be
1294 extended, not be the primary); it is passed through uncompiled, so its
1295 LSC would read with aliases unexpanded.
1296 """
1297 if CONFIG_FILENAME in self.root:
1298 if lsc := self.root[CONFIG_FILENAME].large_scale_change:
1299 return lsc
1301 return LargeScaleChangeModel()
1303 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
1304 """
1305 Overlay PR-dependent scope gating: drop scopes that author rules disable
1306 for this pull request.
1308 This is the only PR-dependent step. The configs are already compiled, so
1309 each config's scopes are self-contained and dropping one is a plain list
1310 filter — no re-inheritance. Templates are passed through (they are never
1311 matched directly; their scopes already live in each consumer).
1312 """
1313 effective: dict[str, ConfigModel] = {}
1314 for config_path, config in self.root.items():
1315 if config.template:
1316 # Templates are never matched directly; pass them through.
1317 effective[config_path] = config
1318 continue
1320 kept_scopes = [
1321 scope
1322 for scope in config.scopes
1323 if scope.matches_author(author_username)
1324 ]
1325 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
1327 return CompiledConfigModels.from_config_models(effective)