Coverage for src/pullapprove/config.py: 95%
529 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-19 16:21 -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: # noqa: SIM102
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: # noqa: SIM102
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: object) -> bool:
705 if not isinstance(other, ScopeModel):
706 return NotImplemented
707 return self.name == other.name
709 def matches_path(self, path: Path) -> bool:
710 return matches_path_patterns(path=path, patterns=self.paths)
712 def matches_code(self, code: str) -> Generator[dict[str, int]]:
713 patterns = getattr(self, "_code_regex_patterns", [])
714 if not patterns:
715 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
716 self._code_regex_patterns = patterns
718 for pattern in patterns:
719 for match in pattern.finditer(code):
720 start_index = match.start()
721 end_index = match.end()
723 start_line = code.count("\n", 0, start_index) + 1
724 start_col = start_index - code.rfind("\n", 0, start_index)
726 end_line = code.count("\n", 0, end_index) + 1
727 end_col = end_index - code.rfind("\n", 0, end_index)
729 yield {
730 "start_line": start_line,
731 "start_col": start_col,
732 "end_line": end_line,
733 "end_col": end_col,
734 }
736 def matches_author(self, author_username: str) -> bool:
737 if not self.authors:
738 # No authors specified, so assume it matches
739 return True
741 author_username_lower = author_username.lower()
743 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
744 authors = [a.lower() for a in self.authors if not a.startswith("!")]
746 if author_username_lower in negated_authors:
747 # If the author is in the negated list, return False
748 return False
750 if not authors:
751 # Negation-only: everyone not negated matches
752 return True
754 return author_username_lower in authors
757class LargeScaleChangeModel(BaseModel):
758 model_config = ConfigDict(extra="forbid")
760 # Note, an LSC only applies to diffs, not raw files,
761 # because we have to know what *changed*.
763 # Pretty similar to a scope, but more manual.
764 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
765 # Negative values are rejected at compile time (_validate_review_counts)
766 require: int = 1
767 reviewers: list[str] = [] # Field(min_length=1)
768 # min_paths: int = 300
769 # min_lines: int = 3000
770 labels: list[str] = []
771 # really need author value too...?
773 @field_validator("reviewers", mode="after")
774 @classmethod
775 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
776 return _validate_team_refs(values)
778 def unsolvable_reason(self, author_username: str) -> str | None:
779 """
780 Explain why this LSC can never pass for a PR authored by this user,
781 or None if it can.
783 Messages must stay under ~130 chars: they flow into the git-host
784 commit status description, which the adapters slice to 140.
785 """
786 if "*" in self.reviewers:
787 # Anyone can review, so any require is satisfiable
788 return None
790 if not self.reviewers:
791 # An empty roster is "configuration required", which
792 # process_large_scale_change reports on its own
793 return None
795 # Count eligible reviewers (excluding author who can't self-approve)
796 eligible_reviewers = {r.lower() for r in self.reviewers} - {
797 author_username.lower()
798 }
799 if self.require > 0 and len(eligible_reviewers) < self.require:
800 if not eligible_reviewers:
801 return "the PR author is the only reviewer and cannot self-approve"
802 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)"
804 return None
807class ConfigModel(BaseModel):
808 model_config = ConfigDict(extra="forbid")
810 # Nothing is technically required
811 extends: list[str] = []
812 template: bool = False
813 aliases: dict[str, list[str]] = {}
814 large_scale_change: LargeScaleChangeModel | None = None
815 scopes: list[ScopeModel] = []
816 agents: list[AgentModel] = []
818 @field_validator("scopes", mode="after")
819 @classmethod
820 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
821 if dup := _first_case_insensitive_duplicate(scope.name for scope in scopes):
822 raise ValueError(f"Duplicate scope name: {dup}")
823 return scopes
825 @field_validator("agents", mode="after")
826 @classmethod
827 def validate_unique_agent_names(cls, agents: list[AgentModel]) -> list[AgentModel]:
828 if dup := _first_case_insensitive_duplicate(agent.name for agent in agents):
829 raise ValueError(f"Duplicate agent name: {dup}")
830 return agents
832 @field_validator("extends", mode="before")
833 @classmethod
834 def validate_extends(cls, extends: list[str]) -> list[str]:
835 for i, path in enumerate(extends):
836 basename = Path(path).name
837 if not basename.startswith(CONFIG_FILENAME_PREFIX):
838 raise ValueError(
839 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
840 )
841 return extends
843 def compiled_config(
844 self,
845 config_path: Path,
846 other_configs: ConfigModels,
847 teams: dict[str, list[str]] | None = None,
848 ) -> ConfigModel:
849 """
850 Resolve `extends` and replace aliases, returning the effective config.
852 Two phases: flatten the whole extends chain into one merged (raw,
853 unexpanded) config, then expand aliases once. Expanding after the full
854 merge is what makes transitive inheritance and cross-chain alias
855 scoping work — an alias defined anywhere in the chain resolves anywhere.
857 `teams` maps team refs (any case, without the leading `@`) to member
858 usernames, and is only consulted for user-list fields (reviewers,
859 alternates, authors, cc, large_scale_change.reviewers). Keys are
860 lowercased here, so callers don't need to normalize case themselves.
861 With `teams=None` (the default), `@org/team` references in those
862 fields pass through unexpanded — the offline/CLI mode, since this
863 library never calls out to GitHub/GitLab itself. That
864 partially-resolved config is only sanctioned for offline use;
865 anything that needs real reviewer usernames must pass a `teams`
866 mapping.
868 Pure function of the raw `self` and `other_configs` (it never reads its
869 own anchored output), so it is safe to call uncached.
870 """
872 if teams is not None:
873 teams = {key.lower(): members for key, members in teams.items()}
875 compiled_data = self._merged_data(config_path, other_configs)
877 # Expand aliases for any aliasable list fields. Team refs only expand
878 # in user-list fields — paths/code/labels keep a leading "@" literal.
879 for scope in compiled_data["scopes"]:
880 for field in [
881 "paths",
882 "code",
883 "authors",
884 "reviewers",
885 "alternates",
886 "cc",
887 "labels",
888 ]:
889 if field in scope:
890 scope[field] = _expand_aliases(
891 scope[field],
892 compiled_data["aliases"],
893 teams=teams,
894 expand_teams=field in USER_LIST_FIELDS,
895 )
897 # An agent's paths are paths like a scope's: `$path-alias` references
898 # resolve, team refs stay literal.
899 for agent in compiled_data["agents"]:
900 agent["paths"] = _expand_aliases(
901 agent["paths"],
902 compiled_data["aliases"],
903 teams=teams,
904 )
906 # Apply compile-time "!" subtraction to roster fields.
907 # `_apply_negations` leaves a field with an unexpanded team ref
908 # (offline compile, i.e. teams=None) untouched.
909 for scope in compiled_data["scopes"]:
910 for field in ROSTER_FIELDS:
911 if field in scope:
912 scope[field] = _apply_negations(scope[field])
913 _reject_bot_reviewers(
914 scope[field],
915 where=f"Scope '{scope['name']}' {field}",
916 )
918 # The "*" wildcard is only meaningful in `reviewers` — everywhere
919 # else it's matched as a literal username and silently does
920 # nothing: in `alternates` the scope stays pending forever, in
921 # `authors` the scope never applies to any PR, in `cc` nobody is
922 # notified. Reject it at compile time (after alias expansion, so
923 # `$alias` indirection can't smuggle it in) rather than as a
924 # ScopeModel validator, because compiled configs stored inside
925 # old processing results must keep parsing.
926 # Messages must stay under ~130 chars: they become the git-host
927 # commit status description, which the adapters slice to 140.
928 for field, hint in (
929 (
930 "authors",
931 "remove it (a scope with no authors applies to any author)",
932 ),
933 (
934 "alternates",
935 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
936 ),
937 ("cc", "remove it"),
938 ):
939 if "*" in scope.get(field, []):
940 raise ValueError(
941 f"Scope '{scope['name']}': \"*\" is not supported in "
942 f"{field} — {hint}"
943 )
945 _validate_review_counts(f"Scope '{scope['name']}'", scope)
947 if large_scale_change := compiled_data.get("large_scale_change"):
948 large_scale_change["reviewers"] = _expand_aliases(
949 large_scale_change["reviewers"],
950 compiled_data["aliases"],
951 teams=teams,
952 expand_teams=True,
953 )
954 large_scale_change["labels"] = _expand_aliases(
955 large_scale_change["labels"],
956 compiled_data["aliases"],
957 )
958 large_scale_change["reviewers"] = _apply_negations(
959 large_scale_change["reviewers"]
960 )
961 _reject_bot_reviewers(
962 large_scale_change["reviewers"],
963 where="large_scale_change reviewers",
964 )
965 _validate_review_counts("large_scale_change", large_scale_change)
967 # Anchor each scope's paths at the directory tagged during flattening
968 # (after alias expansion, so any `$path-alias` is resolved first). The
969 # transient tag is popped so it never reaches the model.
970 # Scopes and agents alike: paths are relative to the config that owns
971 # the entry, `/` for repo-root-absolute.
972 for entry in [*compiled_data["scopes"], *compiled_data["agents"]]:
973 anchor_dir = entry.pop("_anchor_dir", "")
974 entry["paths"] = [_anchor_path(anchor_dir, p) for p in entry["paths"]]
976 # The compiled config is the self-contained effective config: extends
977 # are already merged in and aliases already expanded, so drop both. This
978 # keeps stored results lean and makes the compiled form standalone (it
979 # can never dangle on a missing extends target or re-expand differently).
980 compiled_data["extends"] = []
981 compiled_data["aliases"] = {}
983 return ConfigModel.from_data(
984 data=compiled_data,
985 path=config_path,
986 )
988 def _merged_data(
989 self,
990 config_path: Path,
991 other_configs: ConfigModels,
992 _in_progress: list[str] | None = None,
993 _seen: set[str] | None = None,
994 ) -> dict[str, Any]:
995 """
996 Flatten the `extends` chain into one merged, *unexpanded* config dict.
998 Parents are merged before this config (so a child can specialize), with
999 aliases unioned child-wins and the large-scale-change config taken from
1000 the child if set else the first parent that defines one.
1002 `_in_progress` is the current ancestor path, used to detect circular
1003 extends. `_seen` is every config already merged into this flatten, used
1004 to merge a shared ancestor only once (diamond dedup).
1005 """
1006 if _in_progress is None:
1007 _in_progress = []
1008 if _seen is None:
1009 _seen = set()
1011 config_path_str = str(config_path)
1012 if config_path_str in _in_progress:
1013 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
1014 config_path_str
1015 ]
1016 raise ValueError(
1017 f"Circular reference detected in extends: {' -> '.join(cycle)}"
1018 )
1019 _in_progress.append(config_path_str)
1021 inherited_scopes: list[dict[str, Any]] = []
1022 inherited_agents: list[dict[str, Any]] = []
1023 inherited_aliases: dict[str, list[str]] = {}
1024 inherited_lsc: dict[str, Any] | None = None
1026 for extend_path in self.extends:
1027 resolved_path = _resolve_extends_path(config_path_str, extend_path)
1028 if resolved_path not in other_configs:
1029 raise ValueError(
1030 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
1031 )
1032 if resolved_path in _seen:
1033 # Already merged via another branch (diamond) — skip the dup.
1034 continue
1036 parent_data = other_configs[resolved_path]._merged_data(
1037 Path(resolved_path), other_configs, _in_progress, _seen
1038 )
1039 inherited_scopes = inherited_scopes + parent_data["scopes"]
1040 inherited_agents = inherited_agents + parent_data["agents"]
1041 inherited_aliases = inherited_aliases | parent_data["aliases"]
1042 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
1044 merged = self.model_dump()
1045 merged["scopes"] = inherited_scopes + merged["scopes"]
1046 merged["agents"] = inherited_agents + merged["agents"]
1047 merged["aliases"] = inherited_aliases | merged["aliases"]
1048 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
1050 # Tag each scope with the directory its paths should anchor at. A scope's
1051 # paths are relative to the config that owns it, so the first
1052 # non-template config to consume a scope claims it: a non-template's own
1053 # scopes (and any it inherits from a template) anchor at its directory,
1054 # while a template defers to its consumer. `setdefault` means an
1055 # already-tagged scope (from a non-template ancestor) keeps its anchor.
1056 if not self.template:
1057 base_dir = posixpath.dirname(config_path_str)
1058 for entry in [*merged["scopes"], *merged["agents"]]:
1059 entry.setdefault("_anchor_dir", base_dir)
1061 _seen.add(config_path_str)
1062 _in_progress.pop()
1064 return merged
1066 @classmethod
1067 def from_filesystem(cls, path: Path | str) -> ConfigModel:
1068 with open(path, "rb") as f:
1069 return cls.from_data(tomllib.load(f), path)
1071 @classmethod
1072 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
1073 return cls.from_data(tomllib.loads(content), path)
1075 @classmethod
1076 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
1077 return cls(**data)
1080class _ConfigModelsBase(RootModel):
1081 """Shared storage and accessors for a set of configs keyed by repo path."""
1083 root: dict[str, ConfigModel]
1085 @classmethod
1086 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
1087 """Build from a dict of already-constructed configs keyed by path."""
1088 configs = cls(root={})
1089 for path, config_model in models.items():
1090 configs.root[str(Path(path))] = config_model
1091 return configs
1093 def get_config_models(self) -> dict[str, ConfigModel]:
1094 return dict(self.root.items())
1096 def __bool__(self) -> bool:
1097 return bool(self.root)
1099 def __getitem__(self, key: str) -> ConfigModel:
1100 return self.root[key]
1102 def __contains__(self, key: str) -> bool:
1103 return key in self.root
1105 def __len__(self) -> int:
1106 return len(self.root)
1109class ConfigModels(_ConfigModelsBase):
1110 """Configs exactly as loaded from the repo — extends unresolved, aliases
1111 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
1113 def declared_check_names(self) -> set[str]:
1114 """Every check-run name an effective (non-template) scope's `unless`
1115 references, across the whole config set.
1117 Collected from an offline compile (`teams=None`, the `team_refs`
1118 precedent) so a template's refs count only where a consumer actually
1119 inherits them -- an unconsumed template must not make every pull
1120 request fetch (or error on) checks that nothing evaluates. `unless`
1121 refs are literal strings (aliases and teams never expand inside
1122 them), so the offline compile is exact. Empty for the common case (no
1123 `unless` anywhere), which is what lets the processor skip fetching
1124 check runs entirely.
1125 """
1126 return {
1127 name
1128 for config in self.compiled().root.values()
1129 if not config.template
1130 for scope in config.scopes
1131 for _, name in scope.unless_refs()
1132 }
1134 @classmethod
1135 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
1136 """Load configs from a dict of parsed config data keyed by path."""
1137 configs = cls(root={})
1139 for path, config_data in data.items():
1140 config = ConfigModel.from_data(config_data, Path(path))
1141 configs.add_config(config, Path(path))
1143 return configs
1145 @classmethod
1146 def from_contents(cls, contents: dict[str, str]) -> ConfigModels:
1147 """Load configs from a dict of raw TOML content keyed by path."""
1148 configs = cls(root={})
1150 for path, content in contents.items():
1151 configs.add_config(ConfigModel.from_content(content, path), Path(path))
1153 return configs
1155 def add_config(self, config: ConfigModel, path: Path) -> None:
1156 self.root[str(path)] = config
1158 def team_refs(self) -> set[str]:
1159 """Collect every team ref (lowercase, no leading `@`/`!`) that
1160 `compiled(teams=...)` would actually try to expand: refs written
1161 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
1162 `large_scale_change.reviewers`), plus any refs reachable from those
1163 fields through `$alias`/`!$alias` chains.
1165 Meant for callers that need to know which teams to fetch/sync before
1166 calling `compiled(teams=...)`.
1168 Implemented as an offline compile (`teams=None`): aliases expand but
1169 team refs pass through unexpanded, so whatever refs remain in the
1170 compiled user-list fields are — by construction — exactly the refs a
1171 real compile will try to expand. A ref that only appears in a
1172 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
1173 or inside an alias nothing references never survives into a compiled
1174 user-list field, so it is never collected. Raises the same config
1175 errors `compiled()` would (unknown alias, circular refs, ...), just
1176 earlier.
1177 """
1178 # Cheap pre-check: a team ref can only enter a compile as a literal
1179 # `@`/`!@` value in a user-list field or an alias value. Most repos
1180 # have none, and skipping the compile keeps this near-free for them.
1181 candidate_lists: list[list[str]] = []
1182 for config in self.root.values():
1183 candidate_lists.extend(config.aliases.values())
1184 for scope in config.scopes:
1185 for field in USER_LIST_FIELDS:
1186 candidate_lists.append(getattr(scope, field))
1187 if config.large_scale_change:
1188 candidate_lists.append(config.large_scale_change.reviewers)
1189 if not any(
1190 _split_team_ref(value) is not None
1191 for values in candidate_lists
1192 for value in values
1193 ):
1194 return set()
1196 refs: set[str] = set()
1198 def collect_refs(values: list[str]) -> None:
1199 for value in values:
1200 if (split := _split_team_ref(value)) is not None:
1201 refs.add(split[1].lower())
1203 for config in self.compiled(teams=None).get_config_models().values():
1204 if config.template:
1205 continue
1206 for scope in config.scopes:
1207 for field in USER_LIST_FIELDS:
1208 collect_refs(getattr(scope, field))
1209 if config.large_scale_change:
1210 collect_refs(config.large_scale_change.reviewers)
1212 return refs
1214 def compiled(
1215 self, teams: dict[str, list[str]] | None = None
1216 ) -> CompiledConfigModels:
1217 """Resolve the whole set into its effective, PR-independent form.
1219 Each non-template config is compiled once — extends merged, aliases
1220 expanded, paths anchored. Templates are NOT compiled standalone: a
1221 template scope may reference an alias the consuming config provides, and
1222 its paths anchor at the consumer. They are carried through untouched
1223 (folded into each consumer during that consumer's compile, and kept in
1224 the set for display).
1226 `teams` maps team refs (any case, without the leading `@`) to member
1227 usernames; passed straight through to each config's `compiled_config`
1228 (see there for case normalization and the `teams=None` vs provided
1229 semantics).
1231 The result is an immutable `CompiledConfigModels` — there is no way to
1232 compile it again, so the non-idempotent path anchoring can never
1233 double-apply.
1234 """
1235 effective: dict[str, ConfigModel] = {}
1236 for path, config in self.root.items():
1237 if config.template:
1238 effective[path] = config
1239 else:
1240 effective[path] = config.compiled_config(
1241 config_path=Path(path), other_configs=self, teams=teams
1242 )
1244 return CompiledConfigModels.from_config_models(effective)
1247class CompiledConfigModels(_ConfigModelsBase):
1248 """The effective configs used for matching: every non-template config is
1249 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
1251 def agents_by_config(self) -> dict[str, dict[str, AgentModel]]:
1252 """Every effective config's agents, keyed by config path then name.
1254 Per config, not flattened, because an agent belongs to the config that
1255 declares it and covers that config's files. Two sibling configs may
1256 each declare a `security` agent; those are two agents with one name,
1257 each reviewing its own config's subtree — flattening them into one
1258 by-name dict would silently make the last one win.
1260 Templates are skipped: their agents are already folded into each
1261 consumer's compiled config, and a template governs no files itself.
1262 """
1263 return {
1264 config_path: {agent.name: agent for agent in config.agents}
1265 for config_path, config in self.root.items()
1266 if not config.template
1267 }
1269 def closest_config_path(self, file_path: Path) -> str | None:
1270 """The path of the closest non-template config governing this file,
1271 or None when nothing governs it."""
1272 for parent in file_path.parents:
1273 parent_config_path = str(parent / CONFIG_FILENAME)
1274 config = self.root.get(parent_config_path)
1275 if config is not None and not config.template:
1276 return parent_config_path
1277 return None
1279 def closest_config(self, file_path: Path) -> ConfigModel:
1280 """Return the closest non-template config governing this file."""
1281 config_path = self.closest_config_path(file_path)
1282 if config_path is None:
1283 raise ValueError(f"No config found for {file_path}")
1284 return self.root[config_path]
1286 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
1287 """The primary (repo-root) config's large-scale-change section, if any.
1289 The primary was compiled by `compiled()`, so its reviewers/labels are
1290 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
1291 true` repo root is a misconfiguration (templates are meant to be
1292 extended, not be the primary); it is passed through uncompiled, so its
1293 LSC would read with aliases unexpanded.
1294 """
1295 if CONFIG_FILENAME in self.root and (
1296 lsc := self.root[CONFIG_FILENAME].large_scale_change
1297 ):
1298 return lsc
1300 return LargeScaleChangeModel()
1302 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
1303 """
1304 Overlay PR-dependent scope gating: drop scopes that author rules disable
1305 for this pull request.
1307 This is the only PR-dependent step. The configs are already compiled, so
1308 each config's scopes are self-contained and dropping one is a plain list
1309 filter — no re-inheritance. Templates are passed through (they are never
1310 matched directly; their scopes already live in each consumer).
1311 """
1312 effective: dict[str, ConfigModel] = {}
1313 for config_path, config in self.root.items():
1314 if config.template:
1315 # Templates are never matched directly; pass them through.
1316 effective[config_path] = config
1317 continue
1319 kept_scopes = [
1320 scope
1321 for scope in config.scopes
1322 if scope.matches_author(author_username)
1323 ]
1324 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
1326 return CompiledConfigModels.from_config_models(effective)