GitHub action

README.md

@@ -58,7 +58,6 @@ warn_at: 50000          # impact above which to warn
 block_at: 200000        # impact above which to block
 enforcement: warn       # off, warn, or block. Start on warn. Flip to block when ready.
 tolerance: 1.0          # CI-adjustable multiplier on both thresholds. Above 1 is more lenient.
-wmc_context: before     # measure definition. Uses the pre-change container. This is canonical.
 # measure_config: .impact-measure.yml   # optional: ignore globs and language overrides
 ```
 

action.yml

@@ -23,9 +23,6 @@ inputs:
   tolerance:
     description: "Multiplier applied to both thresholds."
     default: ""
-  wmc-context:
-    description: "before (canonical) or after."
-    default: ""
   config:
     description: "Path to an .impact-gate.yml."
     default: ""
@@ -55,7 +52,6 @@ runs:
         INPUT_WARN_AT: ${{ inputs.warn-at }}
         INPUT_BLOCK_AT: ${{ inputs.block-at }}
         INPUT_TOLERANCE: ${{ inputs.tolerance }}
-        INPUT_WMC_CONTEXT: ${{ inputs.wmc-context }}
         INPUT_CONFIG: ${{ inputs.config }}
         DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}
         GITHUB_TOKEN: ${{ inputs.github-token || github.token }}
@@ -74,7 +70,6 @@ runs:
         if [ -n "$INPUT_WARN_AT" ]; then args+=(--warn-at "$INPUT_WARN_AT"); fi
         if [ -n "$INPUT_BLOCK_AT" ]; then args+=(--block-at "$INPUT_BLOCK_AT"); fi
         if [ -n "$INPUT_TOLERANCE" ]; then args+=(--tolerance "$INPUT_TOLERANCE"); fi
-        if [ -n "$INPUT_WMC_CONTEXT" ]; then args+=(--wmc-context "$INPUT_WMC_CONTEXT"); fi
         if [ -n "$INPUT_CONFIG" ]; then args+=(--config "$INPUT_CONFIG"); fi
 
         report="$RUNNER_TEMP/impact-gate.md"

impact_gate/baseline.py

@@ -0,0 +1,208 @@
+"""The project baseline: the repo's own per-commit impact distribution, and the
+empirical-Bayes blend of it with the shipped seed prior.
+
+Scope. The walk starts from the merged base branch (default `main`) and follows its
+first-parent mainline, so only landed work is measured; an unmerged in-flight branch is
+never reached, because the walk only ever follows real merge commits.
+
+One observation = one atomic landed change:
+  * a main non-merge commit -> its own diff vs its parent;
+  * a leaf MR (a merged branch with no MRs inside it) -> the net change from the branch
+    start (merge-base of the merge's parents) to the branch tip;
+  * a parent MR (a merged branch that contains MRs) -> its direct commits netted per run
+    between the merges on its spine, each run based at the preceding synced base; every
+    child MR recurses, and a merge that only syncs the parent/main branch in is skipped.
+A merge is never scored as its own diff, so a long-running roll-up cannot inflate things.
+
+Each observation is one composite impact = (Σ mutation + Σ godclass) * files. The
+distribution of that over all observations is the baseline. The gate always grades the
+composite; mutation cost stays a per-file ranking signal (report side), never a rival
+distribution to gate against.
+
+Grading. A change's grade blends its percentile rank in this project distribution (n
+observations) with its rank against the per-language seed table, weighting the project by
+  w = n / (n + K)
+so a shallow history leans on the seed and a deep one trusts itself. K is curve_prior_weight.
+"""
+from __future__ import annotations
+
+import bisect
+import json
+import re
+from dataclasses import dataclass, field
+
+from . import gitio
+from .core.config import MeasureConfig
+from .core.gitplumb import GitRepo
+from .data import load_defaults, rank_in_table, seed_table
+from .engine import score_change
+
+_BD = load_defaults().get("baseline", {})
+
+
+# --------------------------------------------------------------------------- model
+
+@dataclass
+class Baseline:
+    """The project's composite per-observation distribution (ascending)."""
+    n: int
+    dist: list[int] = field(default_factory=list)
+    head: str | None = None
+    base_ref: str | None = None
+
+    def rank(self, value: float) -> float:
+        return _percentile_rank(self.dist, value)
+
+    def to_dict(self) -> dict:
+        return {"_meta": {"tool": "impact-gate", "n": self.n, "head": self.head,
+                          "base_ref": self.base_ref},
+                "distribution": self.dist}
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "Baseline":
+        meta = d.get("_meta", {})
+        dist = [int(v) for v in (d.get("distribution") or [])]
+        n = int(meta.get("n", len(dist)))
+        return cls(n=n, dist=dist, head=meta.get("head"),
+                   base_ref=meta.get("base_ref"))
+
+
+def _percentile_rank(sorted_vals: list[int], value: float) -> float:
+    """Percentile rank of `value` in an ascending list: fraction below plus half the
+    ties, in [0, 100]. Above every observation -> 100; at or below the smallest -> ~0."""
+    n = len(sorted_vals)
+    if n == 0:
+        return 0.0
+    lo = bisect.bisect_left(sorted_vals, value)
+    hi = bisect.bisect_right(sorted_vals, value)
+    return round((lo + 0.5 * (hi - lo)) / n * 100, 2)
+
+
+# ----------------------------------------------------------------------- the walk
+
+def build_baseline(repo_path: str, mcfg: MeasureConfig | None = None, *,
+                   base_ref: str | None = None, max_commits: int | None = None,
+                   exclude_subject_pattern: str | None = None) -> Baseline:
+    """Walk the merged history of `base_ref` into the project's impact distribution."""
+    mcfg = mcfg or MeasureConfig()
+    base_ref = base_ref or _BD.get("base_ref") or "HEAD"
+    if max_commits is None:
+        max_commits = _BD.get("max_commits")
+    if exclude_subject_pattern is None:
+        exclude_subject_pattern = _BD.get("exclude_subject_pattern")
+    pat = re.compile(exclude_subject_pattern) if exclude_subject_pattern else None
+
+    parents = gitio.rev_parents(repo_path, base_ref)
+    mainline = gitio.mainline_commits(repo_path, base_ref, max_commits)
+    repo = GitRepo(repo_path)
+    comp: list[int] = []
+    seen: set[str] = set()
+
+    def emit(old_rev: str, new_rev: str) -> None:
+        score = score_change(gitio.diff_between(repo, old_rev, new_rev), mcfg)
+        if score.empty:
+            return
+        comp.append(score.impact)
+
+    def walk_mr(merge_sha: str, p1: str, tip: str) -> None:
+        if merge_sha in seen:
+            return
+        seen.add(merge_sha)
+        if pat and pat.search(gitio.commit_subject(repo_path, merge_sha)):
+            return  # naming-convention exclusion: skip this MR entirely
+        start = gitio.merge_base(repo_path, p1, tip) or p1
+        spine = gitio.first_parent_spine(repo_path, start, tip)
+        base = start
+        for c in reversed(spine):                       # oldest first, along the branch
+            cps = parents.get(c) or gitio.rev_parents(repo_path, c).get(c, [])
+            if len(cps) < 2:
+                continue                                # direct commit: extends the run
+            prev = cps[0]                               # branch state just before this merge
+            if prev != base:
+                emit(base, prev)                        # net of the run of direct commits
+            for sec in cps[1:]:
+                if gitio.is_ancestor(repo_path, sec, p1):
+                    continue                            # a sync of parent/main: already counted
+                walk_mr(c, cps[0], sec)                 # a child MR: recurse
+            base = c                                    # advance past the merge
+        if base != tip:
+            emit(base, tip)                             # final run up to the branch tip
+
+    try:
+        for c in mainline:
+            ps = parents.get(c, [])
+            if len(ps) >= 2:                            # an MR landed on the mainline
+                for sec in ps[1:]:
+                    walk_mr(c, ps[0], sec)
+            else:                                       # a direct commit on the mainline
+                emit(ps[0] if ps else gitio.EMPTY_TREE, c)
+    finally:
+        repo.close()
+
+    comp.sort()
+    return Baseline(n=len(comp), dist=comp,
+                    head=gitio.head_sha(repo_path, base_ref), base_ref=base_ref)
+
+
+# ------------------------------------------------------------------- persistence
+
+def save_baseline(baseline: Baseline, path: str) -> None:
+    with open(path, "w", encoding="utf-8") as fh:
+        json.dump(baseline.to_dict(), fh, indent=1)
+
+
+def load_baseline(path: str) -> Baseline | None:
+    """The cached baseline at `path`, or None if it is missing or unreadable."""
+    try:
+        with open(path, encoding="utf-8") as fh:
+            return Baseline.from_dict(json.load(fh))
+    except (OSError, ValueError):
+        return None
+
+
+# ------------------------------------------------------------------------ grading
+
+@dataclass
+class Grade:
+    percentile: float            # the blended grade, 0..100
+    value: int                   # the composite impact that was graded
+    seed_percentile: float       # rank against the shipped per-language seed
+    project_percentile: float | None   # rank against project history (None at cold start)
+    weight: float                # w = n / (n + K): the project's share of the blend
+    n: int                       # project observations behind the grade
+    language: str | None
+
+
+def dominant_language(score) -> str | None:
+    """The language driving the change (highest-cost file), else None -> pooled seed."""
+    best, best_cost = None, -1
+    for f in score.files:
+        if f.lang and f.cost > best_cost:
+            best, best_cost = f.lang, f.cost
+    return best
+
+
+def grade_value(value: int, *, language: str | None,
+                baseline: Baseline | None, prior_weight_K: float) -> Grade:
+    seed_pct, seed_vals = seed_table(language)
+    seed_rank = rank_in_table(value, seed_pct, seed_vals)
+    n = baseline.n if baseline else 0
+    if n <= 0:
+        return Grade(seed_rank, value, seed_rank, None, 0.0, 0, language)
+    project_rank = baseline.rank(value)
+    denom = n + prior_weight_K
+    w = n / denom if denom > 0 else 1.0
+    blended = round(w * project_rank + (1 - w) * seed_rank, 2)
+    return Grade(blended, value, round(seed_rank, 2), round(project_rank, 2),
+                 round(w, 4), n, language)
+
+
+def grade_change(score, *, baseline: Baseline | None = None,
+                 prior_weight_K: float = 200) -> Grade:
+    """Grade a scored change by blending its project and seed percentile ranks.
+
+    The graded value is always the composite impact; the seed table is picked by the
+    change's dominant language.
+    """
+    return grade_value(score.impact, language=dominant_language(score),
+                       baseline=baseline, prior_weight_K=prior_weight_K)

impact_gate/cli.py

@@ -35,14 +35,13 @@ def _add_score_args(p: argparse.ArgumentParser) -> None:
     p.add_argument("--block-at", type=int)
     p.add_argument("--enforcement", choices=ENFORCEMENTS)
     p.add_argument("--tolerance", type=float)
-    p.add_argument("--wmc-context", choices=("before", "after"))
     p.add_argument("--measure-config", help="Surveyor-style YAML for ignore globs etc.")
 
 
 def _resolve_config(args) -> GateConfig:
     cfg = GateConfig.load(args.config, args.repo)
     for attr in ("warn_at", "block_at", "enforcement", "tolerance",
-                 "wmc_context", "measure_config"):
+                 "measure_config"):
         val = getattr(args, attr, None)
         if val is not None:
             setattr(cfg, attr, val)
@@ -59,7 +58,7 @@ def _cmd_score(args) -> int:
         print(f"impact-gate: {e}", file=sys.stderr)
         return 1
 
-    score = score_change(changed, mcfg, cfg.wmc_context)
+    score = score_change(changed, mcfg)
     level = cfg.level(score.impact)
     blocked = cfg.blocks(score.impact)
 

impact_gate/config.py

@@ -1,27 +1,53 @@
 """Gate configuration: thresholds, enforcement mode, CI-adjustable tolerance.
 
 Resolved in layers (later wins): built-in defaults -> `.impact-gate.yml` in the repo
--> CLI flags / CI inputs. For now thresholds are absolute impact numbers (seeded from
-the Surveyor corpus); the per-project grading curve is a later phase and will populate
-`warn_at` / `block_at` automatically from history.
+-> CLI flags / CI inputs. The built-in defaults are not literals here; they are read
+from `impact_gate/data/defaults.json`, so tuning them is a data edit, not a code change.
+
+Two gating modes coexist. Absolute: `warn_at` / `block_at` are raw composite-impact
+numbers. Curve (`curve_enabled`): a change is graded by its percentile against the
+blended seed + project distribution, and `warn_percentile` / `block_percentile` gate on
+that. The curve fields are wired here; `baseline.py` and the percentile gate consume
+them (a later phase). Absolute stays the fallback when the curve is off or no baseline
+exists yet.
 """
 from __future__ import annotations
 
 import os
 from dataclasses import dataclass
 
+from .data import load_defaults
+
 CONFIG_NAMES = (".impact-gate.yml", ".impact-gate.yaml")
 ENFORCEMENTS = ("off", "warn", "block")
 
+# Built-in defaults come from the shipped JSON, never from literals in this file.
+_D = load_defaults()
+_ABS = _D["absolute"]
+_CURVE = _D["curve"]
+
+# The knobs an .impact-gate.yml / CLI flag may override, and their loaders. Absolute and
+# curve knobs share one path so a repo can set either mode's numbers in the same file.
+_SCALAR_KEYS = ("warn_at", "block_at", "enforcement", "tolerance", "measure_config",
+                "curve_enabled", "warn_percentile", "block_percentile",
+                "curve_prior_weight", "baseline_file")
+
 
 @dataclass
 class GateConfig:
-    warn_at: int | None = None       # impact above which to warn (None = never warn)
-    block_at: int | None = None      # impact above which to block (None = never block)
-    enforcement: str = "warn"        # off | warn | block  (block = a too-high change fails)
-    tolerance: float = 1.0           # CI-adjustable multiplier on both thresholds (>1 = looser)
-    wmc_context: str = "before"      # measure definition (canonical: before-context)
-    measure_config: str | None = None  # optional Surveyor-style YAML (ignore globs, etc.)
+    warn_at: int | None = _ABS["warn_at"]     # impact above which to warn (None = never)
+    block_at: int | None = _ABS["block_at"]   # impact above which to block (None = never)
+    enforcement: str = _D["enforcement"]      # off | warn | block (block = too-high fails)
+    tolerance: float = _D["tolerance"]        # CI multiplier on both thresholds (>1 = looser)
+    measure_config: str | None = None         # optional Surveyor-style YAML (ignore globs)
+    # Grading curve (percentile-based). Consumed once baseline.py + the percentile gate land.
+    # The gate always scores the composite (change-level) impact; the mutation cost is a
+    # per-file signal used to rank which files to consider, not a rival gating metric.
+    curve_enabled: bool = _CURVE["enabled"]           # gate on percentile vs absolute
+    warn_percentile: float = _CURVE["warn_percentile"]
+    block_percentile: float = _CURVE["block_percentile"]
+    curve_prior_weight: float = _CURVE["prior_weight_K"]  # K in w = n / (n + K)
+    baseline_file: str = _CURVE["baseline_file"]      # project distribution cache
 
     def effective_warn(self) -> float | None:
         return None if self.warn_at is None else self.warn_at * self.tolerance
@@ -53,8 +79,7 @@ def load(cls, path: str | None = None, repo_path: str = ".") -> "GateConfig":
         import yaml  # optional; only needed when a config file exists
         with open(found) as fh:
             data = yaml.safe_load(fh) or {}
-        for key in ("warn_at", "block_at", "enforcement", "tolerance",
-                    "wmc_context", "measure_config"):
+        for key in _SCALAR_KEYS:
             if key in data and data[key] is not None:
                 setattr(cfg, key, data[key])
         cfg.validate()
@@ -64,10 +89,16 @@ def validate(self) -> None:
         if self.enforcement not in ENFORCEMENTS:
             raise ValueError(f"enforcement must be one of {ENFORCEMENTS}, got "
                              f"{self.enforcement!r}")
-        if self.wmc_context not in ("before", "after"):
-            raise ValueError("wmc_context must be 'before' or 'after'")
         if self.tolerance <= 0:
             raise ValueError("tolerance must be > 0")
+        for name in ("warn_percentile", "block_percentile"):
+            p = getattr(self, name)
+            if not 0 < p < 100:
+                raise ValueError(f"{name} must be within (0, 100), got {p}")
+        if self.warn_percentile > self.block_percentile:
+            raise ValueError("warn_percentile must be <= block_percentile")
+        if self.curve_prior_weight < 0:
+            raise ValueError("curve_prior_weight (K) must be >= 0")
 
 
 def _discover(repo_path: str) -> str | None:

impact_gate/core/config.py

@@ -66,7 +66,16 @@ def language(self, path: str) -> str | None:
         return self.lang_by_ext.get(self.ext(path))
 
     def is_ignored(self, path: str) -> bool:
-        return any(fnmatch.fnmatch(path, pat) for pat in self.ignore)
+        # Match each glob against the path, and also with a leading "**/" stripped, so a
+        # pattern like "**/vendor/**" ignores a TOP-LEVEL vendor/ too. fnmatch's "*" spans
+        # "/", but "**/vendor/**" still requires a parent segment before "vendor", which
+        # let a repo-root vendor/ or node_modules/ slip through.
+        for pat in self.ignore:
+            if fnmatch.fnmatch(path, pat):
+                return True
+            if pat.startswith("**/") and fnmatch.fnmatch(path, pat[3:]):
+                return True
+        return False
 
     def is_test(self, path: str) -> bool:
         return any(fnmatch.fnmatch(path, pat) for pat in self.test_patterns)