Run build repair only on regressions

services/hackbot-pulse-listener/app/consumer.py

@@ -5,7 +5,7 @@
 from kombu import Connection, Exchange, Queue
 from kombu.mixins import ConsumerMixin
 
-from app import client, lando, taskcluster, worker
+from app import client, lando, regression, taskcluster, worker
 from app.config import settings
 from app.models import RunContext
 
@@ -15,8 +15,11 @@
 
 EXCHANGES = ("exchange/taskcluster-queue/v1/task-failed",)
 
-# In-memory dedupe of hg revisions already handed to the agent. Only the
-# single consumer thread touches it, so no lock is needed.
+# In-memory dedupe of hg revisions already handed to the agent. A revision is
+# recorded only once we actually trigger a run, so an inherited failure on one
+# build label never suppresses a genuine regression on another label of the
+# same push, while a revision that breaks many builds still triggers only once.
+# Only the single consumer thread touches it, so no lock is needed.
 _seen: TTLCache = TTLCache(
     maxsize=settings.dedupe_max_size, ttl=settings.dedupe_ttl_seconds
 )
@@ -44,7 +47,15 @@ def process(body: dict, executor: Executor) -> str | None:
         return None
 
     if hg_revision in _seen:
-        logger.info("Revision %s already processed; skipping", hg_revision)
+        logger.info("Revision %s already triggered a run; skipping", hg_revision)
+        return None
+
+    if not regression.is_new_build_failure(project, hg_revision, task_label):
+        logger.info(
+            "Build %s at %s inherited from an ancestor push; skipping",
+            task_label,
+            hg_revision,
+        )
         return None
     _seen[hg_revision] = True
 
@@ -109,12 +120,18 @@ def on_message(body, message):
 
 
 def _build_queues(user: str) -> list[Queue]:
+    # Both local and prod authenticate as the same pulse user, so the queue name
+    # must also vary by environment; otherwise both consumers bind to the same
+    # durable queue and steal each other's messages. Production keeps the plain
+    # name for continuity.
+    env = settings.environment
+    env_segment = "" if env == "production" else f"{env}-"
     queues = []
     for exchange in EXCHANGES:
         suffix = exchange.rsplit("/", 1)[-1]
         queues.append(
             Queue(
-                name=f"queue/{user}/build-repair-{suffix}",
+                name=f"queue/{user}/build-repair-{env_segment}{suffix}",
                 exchange=Exchange(exchange, type="topic", no_declare=True),
                 routing_key="#",
                 durable=True,

services/hackbot-pulse-listener/app/regression.py

@@ -0,0 +1,78 @@
+import logging
+
+from mozci.errors import ParentPushNotFound
+from mozci.push import Push
+
+logger = logging.getLogger(__name__)
+
+# Maximum number of ancestor pushes to walk back over coalescing gaps before
+# giving up. Mirrors mozci's own MAX_DEPTH.
+MAX_DEPTH = 20
+
+# mozci results vary by data source (Taskcluster vs Treeherder), so match both
+# vocabularies. Only genuine build failures count as failed; infra results
+# ("exception", "canceled", "superseded", ...) are deliberately non-decisive so
+# they never suppress a real regression.
+_PASSED_RESULTS = ("passed", "success")
+_FAILED_RESULTS = ("busted", "failed")
+
+
+def _build_status(push: Push, label: str) -> str | None:
+    """Return 'passed', 'failed', or None for a build label on a push.
+
+    None means the build produced no decisive result here: it was coalesced (not
+    scheduled), is still running, or only hit infra errors. Retriggers are
+    collapsed: any green run means 'passed'; only a genuine build failure
+    (never an infra exception) counts as 'failed'.
+    """
+    results = [
+        t.result for t in push.tasks if t.label == label and t.state == "completed"
+    ]
+    if any(r in _PASSED_RESULTS for r in results):
+        return "passed"
+    if any(r in _FAILED_RESULTS for r in results):
+        return "failed"
+    return None
+
+
+def is_new_build_failure(branch: str, rev: str, label: str) -> bool:
+    """Return True if this push introduced the failure, False if it inherited it.
+
+    Walks back over pushes that did not run the build (coalescing) until it
+    finds the nearest ancestor that did. Fails open (returns True) on any
+    mozci/network error or if no ancestor within MAX_DEPTH ran the build, so we
+    never silently drop a real regression.
+    """
+    try:
+        ancestor = Push(rev, branch=branch)
+        for _ in range(MAX_DEPTH):
+            try:
+                ancestor = ancestor.parent
+            except ParentPushNotFound:
+                break
+            status = _build_status(ancestor, label)
+            if status is None:
+                continue
+            if status == "failed":
+                logger.info(
+                    "Build %s already failing at %s; inherited failure at %s",
+                    label,
+                    ancestor.rev,
+                    rev,
+                )
+                return False
+            logger.info(
+                "Build %s passed at %s; new failure introduced at %s",
+                label,
+                ancestor.rev,
+                rev,
+            )
+            return True
+    except Exception:
+        logger.exception("Regression check failed for %s@%s; running agent", label, rev)
+        return True
+
+    logger.warning(
+        "No ancestor within %s pushes ran build %s; running agent", MAX_DEPTH, label
+    )
+    return True

services/hackbot-pulse-listener/deploy.sh

@@ -100,6 +100,7 @@ gcloud builds submit "${ROOT_DIR}" \
 
 echo "==> Deploying worker pool"
 ENV_VARS="HACKBOT_API_URL=${HACKBOT_API_URL},HACKBOT_UI_URL=${HACKBOT_UI_URL}"
+ENV_VARS="${ENV_VARS},ENVIRONMENT=production"
 ENV_VARS="${ENV_VARS},PULSE_USER=${PULSE_USER},WATCHED_REPOS=${WATCHED_REPOS}"
 ENV_VARS="${ENV_VARS},NOTIFICATION_SENDER=${NOTIFICATION_SENDER}"
 ENV_VARS="${ENV_VARS},NOTIFICATION_TEAM_EMAIL=${NOTIFICATION_TEAM_EMAIL}"

services/hackbot-pulse-listener/pyproject.toml

@@ -5,13 +5,17 @@ description = "Listens to Taskcluster build failures and triggers the build-repa
 requires-python = ">=3.12"
 dependencies = [
     "kombu>=5.6,<6",
+    "mozci~=2.4.8",
     "taskcluster>=97.1,<100.4",
     "httpx>=0.26.0",
     "pydantic-settings>=2.1.0",
     "sendgrid>=6.12.5",
     "markdown2>=2.4.0",
     "cachetools>=5.3.0",
     "sentry-sdk>=2.51.0",
+    # mozci imports zstandard eagerly (default cache serializer) but only
+    # declares it under its optional `cache` extra, so pull it in explicitly.
+    "zstandard~=0.25.0",
 ]
 
 [project.optional-dependencies]

services/hackbot-pulse-listener/tests/test_consumer.py

@@ -58,6 +58,7 @@ def test_build_failure_triggers_run_and_submits_poll():
     with (
         patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
         patch.object(consumer.lando, "hg_to_git", return_value="deadbeef"),
+        patch.object(consumer.regression, "is_new_build_failure", return_value=True),
         patch.object(consumer.client, "trigger_run", return_value="run-1") as trigger,
     ):
         run_id = consumer.process(_build_msg(), executor)
@@ -83,6 +84,7 @@ def test_same_revision_triggers_once():
     with (
         patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
         patch.object(consumer.lando, "hg_to_git", return_value="deadbeef"),
+        patch.object(consumer.regression, "is_new_build_failure", return_value=True),
         patch.object(consumer.client, "trigger_run", return_value="run-1") as trigger,
     ):
         consumer.process(_build_msg(task_id="T1"), executor)
@@ -91,6 +93,63 @@ def test_same_revision_triggers_once():
     trigger.assert_called_once()
 
 
+def test_inherited_failure_is_skipped_before_mapping():
+    executor = MagicMock()
+    with (
+        patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
+        patch.object(consumer.regression, "is_new_build_failure", return_value=False),
+        patch.object(consumer.lando, "hg_to_git") as hg_to_git,
+        patch.object(consumer.client, "trigger_run") as trigger,
+    ):
+        assert consumer.process(_build_msg(), executor) is None
+
+    hg_to_git.assert_not_called()
+    trigger.assert_not_called()
+    executor.submit.assert_not_called()
+
+
+def test_multiple_builds_same_revision_trigger_once():
+    executor = MagicMock()
+    with (
+        patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
+        patch.object(consumer.lando, "hg_to_git", return_value="deadbeef"),
+        patch.object(consumer.regression, "is_new_build_failure", return_value=True),
+        patch.object(consumer.client, "trigger_run", return_value="run-1") as trigger,
+    ):
+        consumer.process(_build_msg(task_id="T1", label="build-linux64/opt"), executor)
+        consumer.process(_build_msg(task_id="T2", label="build-macosx64/opt"), executor)
+
+    trigger.assert_called_once()
+
+
+def test_inherited_label_does_not_suppress_new_label_on_same_revision():
+    executor = MagicMock()
+    with (
+        patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
+        patch.object(consumer.lando, "hg_to_git", return_value="deadbeef"),
+        patch.object(
+            consumer.regression, "is_new_build_failure", side_effect=[False, True]
+        ),
+        patch.object(consumer.client, "trigger_run", return_value="run-1") as trigger,
+    ):
+        # Inherited failure on the first label must not mark the revision seen.
+        assert (
+            consumer.process(
+                _build_msg(task_id="T1", label="build-linux64/opt"), executor
+            )
+            is None
+        )
+        # A genuine regression on another label of the same push still runs.
+        assert (
+            consumer.process(
+                _build_msg(task_id="T2", label="build-macosx64/opt"), executor
+            )
+            == "run-1"
+        )
+
+    trigger.assert_called_once()
+
+
 def test_unwatched_project_skipped_before_api_call():
     executor = MagicMock()
     with (
@@ -107,6 +166,7 @@ def test_unmappable_revision_skipped():
     executor = MagicMock()
     with (
         patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
+        patch.object(consumer.regression, "is_new_build_failure", return_value=True),
         patch.object(consumer.lando, "hg_to_git", return_value=None),
         patch.object(consumer.client, "trigger_run") as trigger,
     ):
@@ -121,6 +181,7 @@ def test_trigger_failure_releases_revision_for_retry():
     with (
         patch.object(consumer.taskcluster, "get_hg_revision", return_value="hgrev"),
         patch.object(consumer.lando, "hg_to_git", return_value="deadbeef"),
+        patch.object(consumer.regression, "is_new_build_failure", return_value=True),
         patch.object(
             consumer.client, "trigger_run", side_effect=[RuntimeError("boom"), "run-2"]
         ) as trigger,
@@ -130,3 +191,15 @@ def test_trigger_failure_releases_revision_for_retry():
         assert consumer.process(_build_msg(task_id="T2"), executor) == "run-2"
 
     assert trigger.call_count == 2
+
+
+def test_queue_name_includes_non_production_environment():
+    with patch.object(consumer.settings, "environment", "development"):
+        (queue,) = consumer._build_queues("guest")
+    assert queue.name == "queue/guest/build-repair-development-task-failed"
+
+
+def test_queue_name_omits_production_environment():
+    with patch.object(consumer.settings, "environment", "production"):
+        (queue,) = consumer._build_queues("guest")
+    assert queue.name == "queue/guest/build-repair-task-failed"

services/hackbot-pulse-listener/tests/test_regression.py

@@ -0,0 +1,119 @@
+from unittest.mock import patch
+
+from app import regression
+from mozci.errors import ParentPushNotFound
+
+LABEL = "build-linux64/opt"
+
+
+class FakeTask:
+    def __init__(self, label=LABEL, state="completed", result="passed"):
+        self.label = label
+        self.state = state
+        self.result = result
+
+
+class FakePush:
+    def __init__(self, rev, tasks, parent=None):
+        self.rev = rev
+        self.tasks = tasks
+        self._parent = parent
+
+    @property
+    def parent(self):
+        if self._parent is None:
+            raise ParentPushNotFound("no parent", rev=self.rev, branch="autoland")
+        return self._parent
+
+
+def _passed(label=LABEL):
+    return [FakeTask(label=label, result="passed")]
+
+
+def _failed(label=LABEL):
+    return [FakeTask(label=label, result="failed")]
+
+
+def _infra(label=LABEL):
+    return [FakeTask(label=label, state="exception", result="exception")]
+
+
+def _running(label=LABEL):
+    return [FakeTask(label=label, state="running", result=None)]
+
+
+def _chain(*task_lists):
+    """Build a parent chain: task_lists[0] is the observed push, [1] its parent, ..."""
+    push = None
+    for i, tasks in enumerate(reversed(task_lists)):
+        push = FakePush(rev=f"rev{len(task_lists) - 1 - i}", tasks=tasks, parent=push)
+    return push
+
+
+def _run(observed):
+    with patch.object(regression, "Push", return_value=observed):
+        return regression.is_new_build_failure("autoland", observed.rev, LABEL)
+
+
+def test_parent_passed_is_new_failure():
+    assert _run(_chain(_failed(), _passed())) is True
+
+
+def test_parent_failed_is_inherited():
+    assert _run(_chain(_failed(), _failed())) is False
+
+
+def test_parent_intermittent_is_new_failure():
+    parent_tasks = [
+        FakeTask(result="failed"),
+        FakeTask(result="passed"),
+    ]
+    assert _run(_chain(_failed(), parent_tasks)) is True
+
+
+def test_coalesced_parent_then_green_grandparent_is_new_failure():
+    assert _run(_chain(_failed(), [], _passed())) is True
+
+
+def test_coalesced_parent_then_failed_grandparent_is_inherited():
+    assert _run(_chain(_failed(), [], _failed())) is False
+
+
+def test_infra_only_parent_is_skipped_then_green():
+    assert _run(_chain(_failed(), _infra(), _passed())) is True
+
+
+def test_running_parent_is_skipped_then_failed():
+    assert _run(_chain(_failed(), _running(), _failed())) is False
+
+
+def test_no_parent_runs_agent():
+    assert _run(_chain(_failed())) is True
+
+
+def test_no_decisive_ancestor_runs_agent():
+    empties = [_failed()] + [[] for _ in range(regression.MAX_DEPTH + 2)]
+    assert _run(_chain(*empties)) is True
+
+
+def test_mozci_error_runs_agent():
+    with patch.object(regression, "Push", side_effect=RuntimeError("boom")):
+        assert regression.is_new_build_failure("autoland", "rev", LABEL) is True
+
+
+def test_other_label_on_parent_ignored():
+    parent_tasks = _failed(label="build-macosx64/opt")  # different build, not ours
+    assert _run(_chain(_failed(), parent_tasks)) is True
+
+
+def test_completed_exception_parent_is_not_treated_as_failed():
+    # A completed task with an infra result must not suppress a real regression:
+    # it is non-decisive, so we walk past it to the green grandparent.
+    exception_parent = [FakeTask(state="completed", result="exception")]
+    assert _run(_chain(_failed(), exception_parent, _passed())) is True
+
+
+def test_treeherder_result_vocabulary():
+    # success/busted are the Treeherder-source spellings of passed/failed.
+    assert _run(_chain(_failed(), [FakeTask(result="success")])) is True
+    assert _run(_chain(_failed(), [FakeTask(result="busted")])) is False