Add a pure RegressionReport service

docs/results.md

@@ -4,6 +4,24 @@ date = 2024-12-10T07:07:07+01:00
 description = 'Fetch results from the KernelCI ecosystem.'
 +++
 
+## Regression comparison and CI gates
+
+```shell
+kci-dev results compare --giturl URL --branch BRANCH --format json BASE HEAD
+kci-dev results gate --giturl URL --branch BRANCH --base BASE --head HEAD \
+  --fail-on regression --format json
+kci-dev results compare --giturl URL --branch BRANCH --include-issues BASE HEAD
+```
+
+Reports classify executions as `regression`, `fixed`, `unstable`,
+`persistent_fail`, `new`, or `missing`, while preserving duplicates. Identity
+includes origin, platform, architecture, compiler, configuration, and path.
+Exit status is 0 without a policy violation, 1 for a policy violation, and 2
+when API or infrastructure failures make the result incomplete.
+
+Known-issue lookup is opt-in with `--include-issues`, since it makes one
+additional Dashboard request for each regression and persistent failure.
+
 `kci-dev` pulls from our Dashboard API. As of now, it is an EXPERIMENTAL tooling under development with close collaboration from Linux kernel maintainers.
 
 > KNOWN ISSUE: The Dashboard endpoint we are using returns a file of a few megabytes in size, so download may take

kcidev/api.py

@@ -55,6 +55,7 @@
     send_jobretry,
     send_patchset,
 )
+from kcidev.libs.regression import RegressionReport
 from kcidev.main import get_cli
 
 
@@ -419,22 +420,22 @@ def get_hardware_tests(self, name, origin):
             True,
         )
 
-    def get_build_issues(self, build_id):
+    def get_build_issues(self, build_id, error_verbose=True):
         return self._dashboard_request(
             "Dashboard build issues request failed",
             dashboard_fetch_build_issues,
             build_id,
             True,
-            True,
+            error_verbose,
         )
 
-    def get_boot_issues(self, test_id):
+    def get_boot_issues(self, test_id, error_verbose=True):
         return self._dashboard_request(
             "Dashboard boot issues request failed",
             dashboard_fetch_boot_issues,
             test_id,
             True,
-            True,
+            error_verbose,
         )
 
     def get_issue_list(self, origin=None, days=7):
@@ -484,8 +485,8 @@ def get_tree_report(
         git_url,
         test_path=None,
         history_size=10,
-        max_age_in_hours=None,
-        min_age_in_hours=None,
+        max_age_in_hours=24,
+        min_age_in_hours=0,
     ):
         return self._dashboard_request(
             "Dashboard tree report request failed",
@@ -500,6 +501,68 @@ def get_tree_report(
             min_age_in_hours,
         )
 
+    def compare_results(
+        self, base, head, giturl, branch, origin="maestro", include_issues=False
+    ):
+        """Compare two dashboard checkouts and return a CI-grade report dict.
+
+        Issue lookup is disabled by default because it requires one additional
+        Dashboard request for every regression and persistent failure.
+        """
+
+        def checkout(commit):
+            return {
+                "builds": self.get_builds(origin, giturl, branch, commit).get(
+                    "builds", []
+                ),
+                "boots": self.get_boots(origin, giturl, branch, commit).get(
+                    "boots", []
+                ),
+                "tests": self.get_tests(origin, giturl, branch, commit).get(
+                    "tests", []
+                ),
+            }
+
+        base_results, head_results = checkout(base), checkout(head)
+        # tree-report supplies history-aware unstable/regression decisions.  If
+        # HEAD is not the newest checkout the raw transition remains useful.
+        history_incomplete = False
+        try:
+            history = self.get_tree_report(origin, branch, giturl)
+        except KciDevError:
+            # History improves classification, but is not required to compare
+            # the two explicitly requested checkouts.
+            history = None
+            history_incomplete = True
+        report = RegressionReport.compare(
+            base, head, base_results, head_results, history
+        )
+        report.incomplete = history_incomplete
+        if include_issues:
+            for item in report.items:
+                if item["category"] not in ("regression", "persistent_fail"):
+                    continue
+                result_id = item["head_id"]
+                if not result_id:
+                    continue
+                try:
+                    issues = (
+                        self.get_build_issues(result_id, error_verbose=False)
+                        if item["identity"]["kind"] == "build"
+                        else self.get_boot_issues(result_id, error_verbose=False)
+                    )
+                except KciDevError as exc:
+                    if "no issues" in str(exc).lower():
+                        issues = []
+                    else:
+                        report.incomplete = True
+                        continue
+                item["known_issues"] = [
+                    issue.get("id", issue) if isinstance(issue, dict) else issue
+                    for issue in issues
+                ]
+        return report.to_dict()
+
     def _instance_setting(self, key, *, human_readable_key=None):
         value = ((self.cfg or {}).get(self.instance) or {}).get(key)
         if not value:

kcidev/libs/regression.py

@@ -0,0 +1,169 @@
+"""Pure regression classification and reporting.
+
+This module deliberately has no Click or HTTP dependencies.  It can therefore
+be used by the command line, the public Python client, and the MCP server.
+"""
+
+from collections import Counter, defaultdict
+from dataclasses import dataclass, field
+
+CATEGORIES = ("regression", "fixed", "unstable", "persistent_fail", "new", "missing")
+FAIL_STATUSES = {"FAIL", "ERROR"}
+
+
+def _value(item, *names, default="unknown"):
+    for name in names:
+        value = item.get(name)
+        if value not in (None, ""):
+            return value
+    return default
+
+
+def result_identity(item, kind="test"):
+    """Return the complete, stable identity of a result (never its result id)."""
+    environment = item.get("environment_misc") or {}
+    return {
+        "kind": kind,
+        "origin": _value(item, "origin"),
+        "platform": _value(
+            item, "platform", "hardware", default=environment.get("platform", "unknown")
+        ),
+        "architecture": _value(item, "architecture", "arch"),
+        "compiler": _value(item, "compiler"),
+        "config": _value(item, "config", "config_name"),
+        "path": _value(
+            item, "path", "test_path", default="build" if kind == "build" else "unknown"
+        ),
+    }
+
+
+def _key(item, kind):
+    identity = result_identity(item, kind)
+    # A tree report groups results by their execution identity, regardless of
+    # whether the result came from a build, boot, or test endpoint.
+    identity.pop("kind")
+    return tuple(identity.values())
+
+
+def _duplicate_sort_key(item):
+    """Order otherwise-identical results before pairing them across revisions."""
+    return (
+        str(item.get("status") or "UNKNOWN").upper(),
+        str(item.get("id") or ""),
+    )
+
+
+def _tree_report_keys(tree_report, category):
+    keys = set()
+    for platform, configs in (tree_report or {}).get(category, {}).items():
+        for config, arch_compilers in configs.items():
+            for arch_compiler, paths in arch_compilers.items():
+                architecture, _, compiler = arch_compiler.partition("/")
+                for path, tests in paths.items():
+                    for test in tests or [{}]:
+                        identity = result_identity(
+                            {
+                                **test,
+                                "origin": (tree_report or {}).get("origin"),
+                                "platform": platform,
+                                "config": config,
+                                "architecture": architecture,
+                                "compiler": compiler,
+                                "path": path,
+                            },
+                            "test",
+                        )
+                        identity.pop("kind")
+                        keys.add(tuple(identity.values()))
+    return keys
+
+
+@dataclass
+class RegressionReport:
+    """A deterministic comparison report which preserves duplicate results."""
+
+    base: str
+    head: str
+    items: list = field(default_factory=list)
+    incomplete: bool = False
+
+    @classmethod
+    def compare(cls, base, head, base_results, head_results, tree_report=None):
+        report = cls(base=base, head=head)
+        history = {
+            "regression": _tree_report_keys(tree_report, "possible_regressions"),
+            "fixed": _tree_report_keys(tree_report, "fixed_regressions"),
+            "unstable": _tree_report_keys(tree_report, "unstable_tests"),
+        }
+        for kind in ("build", "boot", "test"):
+            old = defaultdict(list)
+            new = defaultdict(list)
+            for item in base_results.get(kind + "s", []):
+                old[_key(item, kind)].append(item)
+            for item in head_results.get(kind + "s", []):
+                new[_key(item, kind)].append(item)
+            for key in sorted(set(old) | set(new)):
+                # The API does not guarantee result order.
+                old[key].sort(key=_duplicate_sort_key)
+                new[key].sort(key=_duplicate_sort_key)
+                count = max(len(old[key]), len(new[key]))
+                for occurrence in range(count):
+                    before = (
+                        old[key][occurrence] if occurrence < len(old[key]) else None
+                    )
+                    after = new[key][occurrence] if occurrence < len(new[key]) else None
+                    category = cls._classify(before, after, key, history)
+                    if category is None:
+                        continue
+                    chosen = after if after is not None else before
+                    report.items.append(
+                        {
+                            "category": category,
+                            "identity": result_identity(chosen, kind),
+                            "occurrence": occurrence,
+                            "base_status": before.get("status") if before else None,
+                            "head_status": after.get("status") if after else None,
+                            "base_id": before.get("id") if before else None,
+                            "head_id": after.get("id") if after else None,
+                            "known_issues": [],
+                        }
+                    )
+        return report
+
+    @staticmethod
+    def _classify(before, after, key, history):
+        if before is None:
+            return "new"
+        if after is None:
+            return "missing"
+        old = str(before.get("status") or "UNKNOWN").upper()
+        new = str(after.get("status") or "UNKNOWN").upper()
+        if key in history["unstable"]:
+            return "unstable"
+        if new in FAIL_STATUSES and (key in history["regression"] or old == "PASS"):
+            return "regression"
+        if new == "PASS" and (key in history["fixed"] or old in FAIL_STATUSES):
+            return "fixed"
+        if old in FAIL_STATUSES and new in FAIL_STATUSES:
+            return "persistent_fail"
+        if old != new:
+            return "unstable"
+        return None
+
+    @property
+    def counts(self):
+        counts = Counter(item["category"] for item in self.items)
+        return {category: counts[category] for category in CATEGORIES}
+
+    def has_violation(self, fail_on="regression"):
+        policies = {part.strip() for part in fail_on.split(",") if part.strip()}
+        return any(self.counts.get(policy, 0) for policy in policies)
+
+    def to_dict(self):
+        return {
+            "base": self.base,
+            "head": self.head,
+            "counts": self.counts,
+            "incomplete": self.incomplete,
+            "items": self.items,
+        }

kcidev/mcp/tools_dashboard.py

@@ -78,6 +78,26 @@ def get_summary(
     }
 
 
+@tool_errors
+def compare_checkouts(
+    giturl: str,
+    branch: str,
+    base: str,
+    head: str,
+    origin: str = "maestro",
+    include_issues: bool = False,
+):
+    """Compare two checkouts, classifying regressions, fixes and unstable tests.
+    The returned report preserves duplicate executions. Set ``include_issues``
+    to look up known issue ids for failing/regressing results; this can require
+    one additional request per result. ``incomplete`` means the report must not
+    be treated as a successful CI gate.
+    """
+    return _current_client().compare_results(
+        base, head, giturl, branch, origin, include_issues=include_issues
+    )
+
+
 @tool_errors
 def list_commits(giturl: str, branch: str, commit: str, origin: str = "maestro"):
     """List recent checkouts of a tree with per-commit result counts.
@@ -294,6 +314,7 @@ def get_issue_tests(
 READ_ONLY_TOOLS = (
     list_trees,
     get_summary,
+    compare_checkouts,
     list_commits,
     list_builds,
     list_boots,