From 3cb193a1b4f871228c1b9a54d09a8ba0de4bd37c Mon Sep 17 00:00:00 2001
From: Evgeny Pavlov <epavlov@mozilla.com>
Date: Fri, 7 Aug 2026 13:56:06 -0700
Subject: [PATCH 1/5] listener: skip failures Treeherder already ties to an
 intermittent bug

---
 .../hackbot-pulse-listener/app/consumer.py    |  12 ++
 .../hackbot-pulse-listener/app/treeherder.py  |  85 +++++++++-
 .../tests/test_consumer.py                    |  48 ++++++
 .../tests/test_treeherder.py                  | 157 +++++++++++++++++-
 4 files changed, 294 insertions(+), 8 deletions(-)

diff --git a/services/hackbot-pulse-listener/app/consumer.py b/services/hackbot-pulse-listener/app/consumer.py
index 3fb326b256..e2eaeadc0f 100644
--- a/services/hackbot-pulse-listener/app/consumer.py
+++ b/services/hackbot-pulse-listener/app/consumer.py
@@ -273,6 +273,18 @@ def _process_test(body: dict, tags: dict, executor: Executor) -> str | None:
     # once. The same record carries the configuration the regression check compares
     # against.
     job = treeherder.job_for_task(project, task_id)
+    # Populated at ingestion, so this needs no wait for a sheriff to star the job.
+    intermittent = treeherder.intermittent_match(project, job)
+    if intermittent.known:
+        logger.info(
+            "Task %s failed only lines already known in this revision and tracked by "
+            "intermittent bug(s) %s; skipping -- %s",
+            task_id,
+            ", ".join(str(bug) for bug in intermittent.bug_ids),
+            job_link,
+        )
+        return None
+
     reason = treeherder.await_skip_reason(project, task_id, job)
     if reason:
         logger.info(
diff --git a/services/hackbot-pulse-listener/app/treeherder.py b/services/hackbot-pulse-listener/app/treeherder.py
index ccaff7aaa7..6a301c7a98 100644
--- a/services/hackbot-pulse-listener/app/treeherder.py
+++ b/services/hackbot-pulse-listener/app/treeherder.py
@@ -17,6 +17,7 @@
 import logging
 import threading
 import time
+from dataclasses import dataclass, field
 from urllib.parse import quote
 
 import httpx
@@ -37,6 +38,8 @@
 _jobs_cache: TTLCache = TTLCache(maxsize=512, ttl=_TTL_SECONDS)
 # A revision never maps to a different push, so this one is held far longer.
 _push_id_cache: TTLCache = TTLCache(maxsize=512, ttl=6 * 60 * 60)
+# Derived from a parsed log, which no longer changes, so held longer than the rest.
+_bug_suggestions_cache: TTLCache = TTLCache(maxsize=512, ttl=30 * 60)
 _cache_lock = threading.Lock()
 
 # /api/failureclassification/
@@ -89,7 +92,7 @@ def _job(project: str, task_id: str) -> dict | None:
     return results[0] if results else None
 
 
-def _get(path: str) -> dict:
+def _get(path: str) -> dict | list:
     url = f"{settings.treeherder_url.rstrip('/')}/api/project/{path}"
     resp = httpx.get(url, timeout=_TIMEOUT, follow_redirects=True)
     resp.raise_for_status()
@@ -311,3 +314,83 @@ def skip_reason(job: dict | None) -> str | None:
     if classification in _NOT_A_REGRESSION:
         return _CLASSIFICATIONS.get(classification, str(classification))
     return None
+
+
+# Harness noise like "[taskcluster:error] exit status 1" matches unrelated bugs on
+# nearly every failing job, so only real failure lines are judged.
+_FAILURE_LINE_PREFIX = "TEST-UNEXPECTED-"
+_INTERMITTENT_KEYWORD = "intermittent-failure"
+
+
+def bug_suggestions(project: str, job_id: int) -> list[dict]:
+    """Treeherder's bug matches for a job, one entry per parsed failure line."""
+    key = (project, job_id)
+    with _cache_lock:
+        if key in _bug_suggestions_cache:
+            return _bug_suggestions_cache[key]
+
+    suggestions = _get(f"{project}/jobs/{job_id}/bug_suggestions/") or []
+    with _cache_lock:
+        _bug_suggestions_cache[key] = suggestions
+    return suggestions
+
+
+@dataclass(frozen=True)
+class IntermittentMatch:
+    """What Treeherder's bug suggestions say about a failing job.
+
+    ``known`` means every unexpected-failure line is both already seen in this
+    revision and matched to one of ``bug_ids``.
+    """
+
+    bug_ids: list[int] = field(default_factory=list)
+    known: bool = False
+
+
+def intermittent_match(project: str, job: dict | None) -> IntermittentMatch:
+    """Read the bug suggestions of a failing job and judge it a known intermittent.
+
+    Both signals are required: genuine regressions also match open intermittent bugs,
+    so the bug alone would drop them. Fails open on any error or missing data.
+    """
+    job_id = (job or {}).get("id")
+    if job_id is None:
+        return IntermittentMatch()
+
+    try:
+        lines = [
+            line
+            for line in bug_suggestions(project, job_id)
+            if (line.get("search") or "").startswith(_FAILURE_LINE_PREFIX)
+        ]
+        if not lines:
+            return IntermittentMatch()
+
+        bug_ids: list[int] = []
+        known = True
+        for line in lines:
+            matched = _open_intermittent_bugs(line)
+            bug_ids += [bug for bug in matched if bug not in bug_ids]
+            # A missing flag counts as new, never as known.
+            if not matched or line.get("failure_new_in_rev", True):
+                known = False
+        return IntermittentMatch(bug_ids, known)
+    except Exception:
+        logger.exception(
+            "Could not read the bug suggestions of job %s; investigating -- %s",
+            job_id,
+            job_url(project, None, (job or {}).get("task_id") or ""),
+        )
+        return IntermittentMatch()
+
+
+def _open_intermittent_bugs(suggestion: dict) -> list[int]:
+    """Ids of the unresolved intermittent-failure bugs a failure line matches."""
+    bugs = suggestion.get("bugs") or {}
+    matched = []
+    for bug in (bugs.get("open_recent") or []) + (bugs.get("all_others") or []):
+        keywords = (bug.get("keywords") or "").split(",")
+        if bug.get("id") and not bug.get("resolution"):
+            if _INTERMITTENT_KEYWORD in [k.strip() for k in keywords]:
+                matched.append(bug["id"])
+    return matched
diff --git a/services/hackbot-pulse-listener/tests/test_consumer.py b/services/hackbot-pulse-listener/tests/test_consumer.py
index ca69e675f6..0c975461c3 100644
--- a/services/hackbot-pulse-listener/tests/test_consumer.py
+++ b/services/hackbot-pulse-listener/tests/test_consumer.py
@@ -341,6 +341,9 @@ def env(monkeypatch):
         ),
         recheck_skip_reason=MagicMock(return_value=None),
         await_skip_reason=MagicMock(return_value=None),
+        intermittent_match=MagicMock(
+            return_value=consumer.treeherder.IntermittentMatch()
+        ),
         new_test_failures=MagicMock(
             side_effect=lambda p, r, cfg, groups, abort=None: set(groups)
         ),
@@ -358,6 +361,9 @@ def env(monkeypatch):
     monkeypatch.setattr(
         consumer.treeherder, "await_skip_reason", mocks.await_skip_reason
     )
+    monkeypatch.setattr(
+        consumer.treeherder, "intermittent_match", mocks.intermittent_match
+    )
     monkeypatch.setattr(
         consumer.regression, "new_test_failures", mocks.new_test_failures
     )
@@ -896,3 +902,45 @@ def test_the_group_less_walk_is_also_abandoned(env):
     env.is_new_task_failure.side_effect = consumer.regression.WalkAborted("task at rev")
     assert consumer.process(_test_msg(), env.executor) is None
     env.trigger_run.assert_not_called()
+
+
+def _known_intermittent(*bug_ids):
+    return consumer.treeherder.IntermittentMatch(list(bug_ids), known=True)
+
+
+def test_a_known_intermittent_bug_skips_before_waiting_for_a_verdict(env):
+    env.intermittent_match.return_value = _known_intermittent(2016093)
+    assert consumer.process(_test_msg(), env.executor) is None
+    env.await_skip_reason.assert_not_called()
+    env.failing_groups.assert_not_called()
+    env.trigger_run.assert_not_called()
+
+
+def test_the_skipped_bug_is_logged(env, caplog):
+    env.intermittent_match.return_value = _known_intermittent(2016093)
+    with caplog.at_level(logging.INFO, logger="app.consumer"):
+        assert consumer.process(_test_msg(), env.executor) is None
+    assert "2016093" in caplog.text
+    assert _LINK in caplog.text
+
+
+def test_the_ingested_job_is_what_the_gate_reads(env):
+    assert consumer.process(_test_msg(), env.executor) == "tr-1"
+    assert env.intermittent_match.call_args.args == (
+        "autoland",
+        env.job_for_task.return_value,
+    )
+
+
+def test_a_known_intermittent_does_not_claim_its_push(env):
+    env.intermittent_match.side_effect = [
+        _known_intermittent(2016093),
+        consumer.treeherder.IntermittentMatch(),
+    ]
+    assert consumer.process(_test_msg(task_id="A"), env.executor) is None
+    assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-1"
+
+
+def test_no_intermittent_match_still_runs(env):
+    env.intermittent_match.return_value = consumer.treeherder.IntermittentMatch()
+    assert consumer.process(_test_msg(), env.executor) == "tr-1"
diff --git a/services/hackbot-pulse-listener/tests/test_treeherder.py b/services/hackbot-pulse-listener/tests/test_treeherder.py
index 9201aad70c..b7e7edcd30 100644
--- a/services/hackbot-pulse-listener/tests/test_treeherder.py
+++ b/services/hackbot-pulse-listener/tests/test_treeherder.py
@@ -108,18 +108,16 @@ def test_not_ingested_returns_none(monkeypatch):
 
 @pytest.fixture(autouse=True)
 def _clear_group_cache():
-    for cache in (
+    caches = (
         treeherder._group_cache,
         treeherder._jobs_cache,
         treeherder._push_id_cache,
-    ):
+        treeherder._bug_suggestions_cache,
+    )
+    for cache in caches:
         cache.clear()
     yield
-    for cache in (
-        treeherder._group_cache,
-        treeherder._jobs_cache,
-        treeherder._push_id_cache,
-    ):
+    for cache in caches:
         cache.clear()
 
 
@@ -365,6 +363,151 @@ def test_await_skip_reason_gives_up_and_investigates(monkeypatch):
     assert treeherder.await_skip_reason("autoland", "T1", _job(6)) is None
 
 
+def _bug(bug_id, resolution="", keywords="intermittent-failure,intermittent-testcase"):
+    return {
+        "id": bug_id,
+        "status": "NEW",
+        "resolution": resolution,
+        "keywords": keywords,
+    }
+
+
+def _suggestion(search, new_in_rev, open_recent=(), all_others=()):
+    return {
+        "search": search,
+        "failure_new_in_rev": new_in_rev,
+        "bugs": {"open_recent": list(open_recent), "all_others": list(all_others)},
+    }
+
+
+_FAIL_LINE = "TEST-UNEXPECTED-FAIL | test_dataChannel.html | Test timed out."
+
+
+@pytest.fixture
+def suggestions(monkeypatch):
+    """Stub the bug_suggestions fetch; call it with the entries to return."""
+
+    def use(*entries):
+        monkeypatch.setattr(
+            treeherder, "bug_suggestions", lambda project, job_id: list(entries)
+        )
+
+    return use
+
+
+def _failing_job():
+    return {"id": 42, "task_id": "TT", "failure_classification_id": 1}
+
+
+def test_known_intermittent_needs_both_signals(suggestions):
+    suggestions(_suggestion(_FAIL_LINE, False, open_recent=[_bug(2016093)]))
+    match = treeherder.intermittent_match("autoland", _failing_job())
+    assert match.known is True
+    assert match.bug_ids == [2016093]
+
+
+def test_a_new_failure_with_an_intermittent_bug_still_runs(suggestions):
+    # Genuine regressions also match open intermittent bugs, so the bug alone
+    # must never be enough to skip.
+    suggestions(_suggestion(_FAIL_LINE, True, open_recent=[_bug(1828735)]))
+    match = treeherder.intermittent_match("autoland", _failing_job())
+    assert match.known is False
+    assert match.bug_ids == [1828735]
+
+
+def test_an_old_failure_without_a_bug_still_runs(suggestions):
+    suggestions(_suggestion(_FAIL_LINE, False))
+    assert treeherder.intermittent_match("autoland", _failing_job()).known is False
+
+
+def test_one_unknown_line_keeps_the_whole_job(suggestions):
+    suggestions(
+        _suggestion(_FAIL_LINE, False, open_recent=[_bug(2016093)]),
+        _suggestion("TEST-UNEXPECTED-FAIL | browser_startup.js | Got 1", True),
+    )
+    assert treeherder.intermittent_match("autoland", _failing_job()).known is False
+
+
+def test_harness_noise_does_not_decide(suggestions):
+    # Harness noise matches junk bugs on nearly every job.
+    suggestions(
+        _suggestion("[taskcluster:error] exit status 1", False, [_bug(2034259)]),
+        _suggestion(_FAIL_LINE, True),
+    )
+    match = treeherder.intermittent_match("autoland", _failing_job())
+    assert match.known is False
+    assert match.bug_ids == []
+
+
+def test_resolved_bugs_are_not_evidence(suggestions):
+    suggestions(
+        _suggestion(_FAIL_LINE, False, all_others=[_bug(1798750, "INCOMPLETE")])
+    )
+    match = treeherder.intermittent_match("autoland", _failing_job())
+    assert match.known is False
+    assert match.bug_ids == []
+
+
+def test_a_bug_without_the_keyword_is_not_evidence(suggestions):
+    suggestions(_suggestion(_FAIL_LINE, False, [_bug(2055984, keywords="regression")]))
+    assert treeherder.intermittent_match("autoland", _failing_job()).known is False
+
+
+def test_no_suggestions_fails_open(suggestions):
+    suggestions()
+    assert treeherder.intermittent_match("autoland", _failing_job()).known is False
+
+
+def test_a_suggestions_error_fails_open(monkeypatch):
+    def boom(project, job_id):
+        raise RuntimeError("treeherder down")
+
+    monkeypatch.setattr(treeherder, "bug_suggestions", boom)
+    assert treeherder.intermittent_match("autoland", _failing_job()).known is False
+
+
+def test_a_job_without_an_id_fails_open(monkeypatch):
+    monkeypatch.setattr(
+        treeherder,
+        "bug_suggestions",
+        lambda *_: pytest.fail("nothing to look up without a job id"),
+    )
+    assert treeherder.intermittent_match("autoland", None).known is False
+    assert treeherder.intermittent_match("autoland", {"task_id": "TT"}).known is False
+
+
+def test_a_missing_new_in_rev_flag_is_treated_as_new(suggestions):
+    entry = _suggestion(_FAIL_LINE, False, [_bug(2016093)])
+    del entry["failure_new_in_rev"]
+    suggestions(entry)
+    assert treeherder.intermittent_match("autoland", _failing_job()).known is False
+
+
+def test_bug_ids_are_deduped_across_lines(suggestions):
+    suggestions(
+        _suggestion(_FAIL_LINE, False, [_bug(2016093)]),
+        _suggestion(_FAIL_LINE + " (retry)", False, [_bug(2016093)]),
+    )
+    assert treeherder.intermittent_match("autoland", _failing_job()).bug_ids == [
+        2016093
+    ]
+
+
+def test_bug_suggestions_are_fetched_once_per_job(monkeypatch):
+    calls = []
+
+    def fake_get(url, **kwargs):
+        calls.append(url)
+        return _response([_suggestion(_FAIL_LINE, False)])
+
+    monkeypatch.setattr(treeherder.httpx, "get", fake_get)
+    assert treeherder.bug_suggestions("autoland", 42) == treeherder.bug_suggestions(
+        "autoland", 42
+    )
+    assert len(calls) == 1
+    assert calls[0].endswith("/project/autoland/jobs/42/bug_suggestions/")
+
+
 def test_push_url_points_at_the_push():
     assert treeherder.push_url("autoland", "abc123") == (
         "https://treeherder.mozilla.org/#/jobs?repo=autoland&revision=abc123"

From 2bc9d4198894debd3237c0d225f712a7eb713e0a Mon Sep 17 00:00:00 2001
From: Evgeny Pavlov <epavlov@mozilla.com>
Date: Fri, 7 Aug 2026 13:57:04 -0700
Subject: [PATCH 2/5] listener: dedupe test-repair by manifest as well as by
 push

---
 services/hackbot-pulse-listener/app/config.py |  2 +
 .../hackbot-pulse-listener/app/consumer.py    | 38 ++++++++
 .../tests/test_consumer.py                    | 90 ++++++++++++++++++-
 3 files changed, 127 insertions(+), 3 deletions(-)

diff --git a/services/hackbot-pulse-listener/app/config.py b/services/hackbot-pulse-listener/app/config.py
index 8f3bcdb9fb..a6ef64ee4d 100644
--- a/services/hackbot-pulse-listener/app/config.py
+++ b/services/hackbot-pulse-listener/app/config.py
@@ -47,6 +47,8 @@ class Settings(BaseSettings):
     # Dedupe (in-memory, by hg revision)
     dedupe_ttl_seconds: int = 6 * 60 * 60
     dedupe_max_size: int = 4096
+    # Dedupe by failing manifest, on top of the per-push one above.
+    group_dedupe_ttl_seconds: int = 12 * 60 * 60
 
     # Cap on test-repair runs started in any rolling 24 hours. Each run clones and
     # builds Firefox in its own container, so a bad day on autoland -- a broken
diff --git a/services/hackbot-pulse-listener/app/consumer.py b/services/hackbot-pulse-listener/app/consumer.py
index e2eaeadc0f..f55fcfbf91 100644
--- a/services/hackbot-pulse-listener/app/consumer.py
+++ b/services/hackbot-pulse-listener/app/consumer.py
@@ -50,6 +50,13 @@
 )
 _seen_tests_lock = threading.Lock()
 
+# A manifest stays broken on the pushes following the one that broke it, so per-push
+# dedupe alone still mails a near-identical analysis for each.
+_seen_groups: TTLCache = TTLCache(
+    maxsize=settings.dedupe_max_size, ttl=settings.group_dedupe_ttl_seconds
+)
+_seen_groups_lock = threading.Lock()
+
 # When each test-repair run of the last day was started, oldest first, capping how
 # many may run in any rolling 24 hours. A slot is taken at the moment of triggering
 # and given back if the trigger fails, so only runs that really started count.
@@ -273,6 +280,7 @@ def _process_test(body: dict, tags: dict, executor: Executor) -> str | None:
     # once. The same record carries the configuration the regression check compares
     # against.
     job = treeherder.job_for_task(project, task_id)
+
     # Populated at ingestion, so this needs no wait for a sheriff to star the job.
     intermittent = treeherder.intermittent_match(project, job)
     if intermittent.known:
@@ -363,6 +371,15 @@ def claimed_elsewhere() -> bool:
         )
         return None
 
+    if _groups_claimed(project, fresh):
+        logger.info(
+            "Every failing group of task %s was already investigated on a recent "
+            "push; skipping -- %s",
+            task_id,
+            job_link,
+        )
+        return None
+
     # One last cheap look before spending a run. The gate above already waited for a
     # verdict, so this catches one that landed during the walk: a sheriff's
     # classification, or autoclassification once a retrigger came back green.
@@ -407,6 +424,24 @@ def _claim_push(hg_revision: str) -> bool:
         return True
 
 
+def _groups_claimed(project: str, groups: list[str]) -> bool:
+    """Whether a recent run already covered every one of these groups.
+
+    All, not any: a task that also broke an unanalysed manifest is still worth a run.
+    """
+    with _seen_groups_lock:
+        return bool(groups) and all((project, g) in _seen_groups for g in groups)
+
+
+def _claim_groups(project: str, groups: list[str]) -> list[tuple[str, str]]:
+    """Record the groups a run covers; returns the keys to release if it fails."""
+    keys = [(project, group) for group in groups]
+    with _seen_groups_lock:
+        for key in keys:
+            _seen_groups[key] = True
+    return keys
+
+
 def _test_runs_today() -> int:
     """Runs started in the window, dropping those that have aged out of it.
 
@@ -500,6 +535,8 @@ def _trigger_test_repair(
         _release_test_run()
         return None
 
+    group_keys = _claim_groups(project, test_groups)
+
     try:
         run_id = client.trigger_run(
             {"failure_tasks": {label: task_id}},
@@ -510,6 +547,7 @@ def _trigger_test_repair(
             "Failed to trigger test-repair run for task %s -- %s", task_id, job_link
         )
         _release(_seen_tests, _seen_tests_lock, [hg_revision])
+        _release(_seen_groups, _seen_groups_lock, group_keys)
         _release_test_run()
         return None
 
diff --git a/services/hackbot-pulse-listener/tests/test_consumer.py b/services/hackbot-pulse-listener/tests/test_consumer.py
index 0c975461c3..9ee068396d 100644
--- a/services/hackbot-pulse-listener/tests/test_consumer.py
+++ b/services/hackbot-pulse-listener/tests/test_consumer.py
@@ -15,6 +15,7 @@
 def setup_function():
     consumer._seen.clear()
     consumer._seen_tests.clear()
+    consumer._seen_groups.clear()
     consumer._test_run_times.clear()
 
 
@@ -584,11 +585,11 @@ def test_backfill_in_a_new_task_group_is_deduped(env):
 
 
 def test_different_pushes_are_not_deduped(env):
-    # Dedupe is per push: the same manifest newly failing on a later push is a
+    # Dedupe is per push: a different manifest newly failing on a later push is a
     # separate regression and must be investigated again.
-    revisions = iter(["rev-one", "rev-two"])
-    env.get_task.side_effect = lambda task_id: _task_def(next(revisions))
+    _consecutive_pushes(env, "rev-one", "rev-two")
     consumer.process(_test_msg(task_id="A"), env.executor)
+    env.failing_groups.return_value = ["other/test/mochitest.ini"]
     consumer.process(_test_msg(task_id="B"), env.executor)
     assert env.trigger_run.call_count == 2
 
@@ -704,6 +705,7 @@ def test_runs_stop_at_the_daily_limit(env, monkeypatch):
     monkeypatch.setattr(consumer.settings, "max_test_repairs_per_day", 2)
     revisions = iter(["rev-1", "rev-2", "rev-3"])
     env.get_task.side_effect = lambda task_id: _task_def(next(revisions))
+    _distinct_groups(env)
 
     assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
     assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-1"
@@ -716,6 +718,7 @@ def test_the_limit_is_a_rolling_window(env, monkeypatch):
     monkeypatch.setattr(consumer.settings, "max_test_repairs_per_day", 1)
     revisions = iter(["rev-1", "rev-2"])
     env.get_task.side_effect = lambda task_id: _task_def(next(revisions))
+    _distinct_groups(env)
 
     assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
     consumer._test_run_times[0] -= consumer._RATE_WINDOW_SECONDS + 1
@@ -752,6 +755,7 @@ def test_a_budget_blocked_task_does_not_claim_its_push(env, monkeypatch):
     monkeypatch.setattr(consumer.settings, "max_test_repairs_per_day", 1)
     revisions = iter(["rev-1", "rev-2", "rev-2"])
     env.get_task.side_effect = lambda task_id: _task_def(next(revisions))
+    env.failing_groups.side_effect = [[_GROUP], ["other/mochitest.ini"]] * 2
 
     assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
     assert consumer.process(_test_msg(task_id="B"), env.executor) is None
@@ -944,3 +948,83 @@ def test_a_known_intermittent_does_not_claim_its_push(env):
 def test_no_intermittent_match_still_runs(env):
     env.intermittent_match.return_value = consumer.treeherder.IntermittentMatch()
     assert consumer.process(_test_msg(), env.executor) == "tr-1"
+
+
+def _consecutive_pushes(env, *revisions):
+    """Make each message look like it came from a different push."""
+    it = iter(revisions)
+    env.get_task.side_effect = lambda task_id: _task_def(next(it))
+
+
+def test_the_same_manifest_on_later_pushes_is_deduped(env):
+    _consecutive_pushes(env, "rev-one", "rev-two", "rev-three")
+    assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
+    assert consumer.process(_test_msg(task_id="B"), env.executor) is None
+    assert consumer.process(_test_msg(task_id="C"), env.executor) is None
+    env.trigger_run.assert_called_once()
+
+
+def test_a_new_manifest_on_a_later_push_still_runs(env):
+    _consecutive_pushes(env, "rev-one", "rev-two")
+    assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
+    env.failing_groups.return_value = ["layout/style/test/mochitest.toml"]
+    assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-1"
+    assert env.trigger_run.call_count == 2
+
+
+def test_one_unseen_manifest_is_enough_to_run(env):
+    _consecutive_pushes(env, "rev-one", "rev-two")
+    assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
+    env.failing_groups.return_value = [_GROUP, "layout/style/test/mochitest.toml"]
+    assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-1"
+
+
+def test_manifest_dedupe_is_per_project():
+    consumer._claim_groups("autoland", [_GROUP])
+    assert consumer._groups_claimed("autoland", [_GROUP]) is True
+    assert consumer._groups_claimed("mozilla-central", [_GROUP]) is False
+
+
+def test_a_skipped_task_does_not_claim_its_manifests(env):
+    _consecutive_pushes(env, "rev-one", "rev-two")
+    env.await_skip_reason.side_effect = ["intermittent", None]
+    assert consumer.process(_test_msg(task_id="A"), env.executor) is None
+    assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-1"
+
+
+def test_a_failed_trigger_releases_the_manifests(env):
+    _consecutive_pushes(env, "rev-one", "rev-two")
+    env.trigger_run.side_effect = [RuntimeError("boom"), "tr-2"]
+    assert consumer.process(_test_msg(task_id="A"), env.executor) is None
+    assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-2"
+
+
+def test_a_group_less_task_is_not_suppressed_by_the_manifest_cache(env):
+    _consecutive_pushes(env, "rev-one", "rev-two")
+    assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
+    env.failing_groups.side_effect = consumer.treeherder.GroupResultsUnavailable("none")
+    assert consumer.process(_test_msg(task_id="B"), env.executor) == "tr-1"
+
+
+def test_the_deduped_task_is_logged_with_a_treeherder_link(env, caplog):
+    _consecutive_pushes(env, "rev-one", "hgrev")
+    assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
+    with caplog.at_level(logging.INFO, logger="app.consumer"):
+        assert consumer.process(_test_msg(task_id="TT"), env.executor) is None
+    assert "already investigated on a recent push" in caplog.text
+    assert _LINK in caplog.text
+
+
+def test_a_manifest_dedupe_costs_no_recheck(env):
+    _consecutive_pushes(env, "rev-one", "rev-two")
+    assert consumer.process(_test_msg(task_id="A"), env.executor) == "tr-1"
+    env.recheck_skip_reason.reset_mock()
+    assert consumer.process(_test_msg(task_id="B"), env.executor) is None
+    env.recheck_skip_reason.assert_not_called()
+
+
+def _distinct_groups(env):
+    """Give every task its own failing manifest, so only the gate under test applies."""
+    env.failing_groups.side_effect = lambda project, rev, task_id: [
+        f"{task_id}/mochitest.ini"
+    ]

From af09d347a479e1fd217dd4331db2509fee7fa4c0 Mon Sep 17 00:00:00 2001
From: Evgeny Pavlov <epavlov@mozilla.com>
Date: Fri, 7 Aug 2026 13:57:44 -0700
Subject: [PATCH 3/5] notify: make the headline the sheriff's action, the patch
 the developer's

---
 services/hackbot-pulse-listener/app/notify.py | 18 +++++++++--
 .../tests/test_notify.py                      | 30 +++++++++++++++++++
 2 files changed, 45 insertions(+), 3 deletions(-)

diff --git a/services/hackbot-pulse-listener/app/notify.py b/services/hackbot-pulse-listener/app/notify.py
index 40ec85de87..57c19e1076 100644
--- a/services/hackbot-pulse-listener/app/notify.py
+++ b/services/hackbot-pulse-listener/app/notify.py
@@ -147,15 +147,16 @@ def _recipients(primary: str | None, secondary: str | None = None) -> list[str]:
     return recipients
 
 
+# The headline names the sheriff's action, which is always a backout.
 _RECOMMENDATION_BANNER = {
     "backout": "BACK OUT the culprit",
     "do_not_backout": "DO NOT back out (intermittent)",
-    "land_fix": "LAND the proposed fix",
+    "land_fix": "BACK OUT the culprit, reland with the proposed fix squashed in",
 }
 
 
 def _banner(findings: dict) -> str:
-    """The recommendation as a human-readable headline."""
+    """The recommendation as a human-readable headline, or the raw value."""
     recommendation = findings.get("recommendation")
     return _RECOMMENDATION_BANNER.get(recommendation, recommendation or "analysis")
 
@@ -216,10 +217,21 @@ def _build_test_repair_body(
         lines.append(f"- **Bug:** [{bug}]({_bug_url(bug)})")
 
     lines += _run_details(ctx) + _analysis_sections(findings) + _patch_section(patch)
-    lines += _team_footer()
+    lines += _patch_advice(patch) + _team_footer()
     return "\n".join(lines)
 
 
+def _patch_advice(patch: str | None) -> list[str]:
+    """Say who the patch is for, next to the patch itself."""
+    if not patch:
+        return []
+    return [
+        "",
+        "_For the author: squash this into your existing patches and reland. It is a "
+        "suggestion, not a follow-up to land on its own._",
+    ]
+
+
 def _run_details(ctx: RunContext) -> list[str]:
     if not settings.hackbot_ui_url:
         return []
diff --git a/services/hackbot-pulse-listener/tests/test_notify.py b/services/hackbot-pulse-listener/tests/test_notify.py
index 1f715dbce7..8862591ab1 100644
--- a/services/hackbot-pulse-listener/tests/test_notify.py
+++ b/services/hackbot-pulse-listener/tests/test_notify.py
@@ -521,3 +521,33 @@ def test_attaches_patch_file(monkeypatch):
     assert len(attachments) == 1
     assert attachments[0]["filename"] == "changes.patch"
     assert base64.b64decode(attachments[0]["content"]).decode() == "DIFF-CONTENT"
+
+
+def test_the_headline_names_the_sheriffs_action_not_a_landing():
+    findings = _test_repair_findings(recommendation="land_fix")
+    body = notify._build_test_repair_body(_test_repair_ctx(), findings, None, None)
+    assert "LAND the proposed fix" not in body
+    assert "BACK OUT the culprit, reland with the proposed fix squashed in" in body
+
+
+def test_the_patch_is_presented_as_advice_for_a_squashed_reland():
+    findings = _test_repair_findings(recommendation="land_fix")
+    body = notify._build_test_repair_body(
+        _test_repair_ctx(), findings, "--- a/f\n+++ b/f\n", None
+    )
+    assert "squash this into your existing patches and reland" in body
+    assert "not a follow-up to land on its own" in body
+
+
+def test_no_patch_advice_without_a_patch():
+    body = notify._build_test_repair_body(
+        _test_repair_ctx(), _test_repair_findings(), None, None
+    )
+    assert "squash this into your existing patches" not in body
+
+
+def test_an_unknown_recommendation_is_shown_verbatim():
+    assert notify._banner({"recommendation": "backout_and_reland"}) == (
+        "backout_and_reland"
+    )
+    assert notify._banner({}) == "analysis"

From 386a00d8c30291f9510ee3b9d1f16b41b42789d2 Mon Sep 17 00:00:00 2001
From: Evgeny Pavlov <epavlov@mozilla.com>
Date: Fri, 7 Aug 2026 13:58:07 -0700
Subject: [PATCH 4/5] listener: say so when a sheriff acted while the run was
 working

---
 services/hackbot-pulse-listener/app/notify.py | 36 ++++++++--
 services/hackbot-pulse-listener/app/worker.py | 28 +++++++-
 .../tests/test_notify.py                      | 62 +++++++++++++++++
 .../tests/test_worker.py                      | 68 ++++++++++++++++++-
 4 files changed, 186 insertions(+), 8 deletions(-)

diff --git a/services/hackbot-pulse-listener/app/notify.py b/services/hackbot-pulse-listener/app/notify.py
index 57c19e1076..8a7ed4de4c 100644
--- a/services/hackbot-pulse-listener/app/notify.py
+++ b/services/hackbot-pulse-listener/app/notify.py
@@ -12,18 +12,23 @@
 MAX_PATCH_LINES = 400
 
 
-def send_email(ctx: RunContext, run_doc: dict) -> None:
+def send_email(
+    ctx: RunContext, run_doc: dict, already_actioned: str | None = None
+) -> None:
     """Email the failure analysis. Only succeeded runs are notified.
 
     Routes on the agent that produced the run: test-repair sends a
     verdict-led body to the test-repair notification address; build-repair keeps its
     existing behavior.
+
+    ``already_actioned`` is Treeherder's classification when a sheriff has already
+    dealt with the failure.
     """
     if run_doc.get("status") != "succeeded":
         logger.info("Run %s did not succeed; skipping notification", ctx.run_id)
         return
     if ctx.agent == settings.test_repair_agent_name:
-        _send_test_repair_email(ctx, run_doc)
+        _send_test_repair_email(ctx, run_doc, already_actioned)
     else:
         _send_build_repair_email(ctx, run_doc)
 
@@ -52,7 +57,9 @@ def _send_build_repair_email(ctx: RunContext, run_doc: dict) -> None:
     _deliver(subject, body_md, recipients, patch)
 
 
-def _send_test_repair_email(ctx: RunContext, run_doc: dict) -> None:
+def _send_test_repair_email(
+    ctx: RunContext, run_doc: dict, already_actioned: str | None = None
+) -> None:
     findings = (run_doc.get("summary") or {}).get("findings") or {}
     culprit = findings.get("culprit_commit")
     culprit_author = (
@@ -77,10 +84,15 @@ def _send_test_repair_email(ctx: RunContext, run_doc: dict) -> None:
         return
 
     patch = _fetch_patch(ctx.run_id, run_doc)
+    # In the subject too, so a sheriff can skip it from the inbox.
+    prefix = "[already actioned] " if already_actioned else ""
     subject = (
-        f"[test-repair] {_banner(findings)} - {_test_groups_label(ctx)} ({ctx.repo})"
+        f"[test-repair] {prefix}{_banner(findings)} - "
+        f"{_test_groups_label(ctx)} ({ctx.repo})"
+    )
+    body_md = _build_test_repair_body(
+        ctx, findings, patch, culprit_author, already_actioned
     )
-    body_md = _build_test_repair_body(ctx, findings, patch, culprit_author)
     _deliver(subject, body_md, recipients, patch)
 
 
@@ -169,14 +181,28 @@ def _test_groups_label(ctx: RunContext) -> str:
     return f"{first} (+{len(rest)} more)" if rest else first
 
 
+def _already_actioned_banner(reason: str | None) -> list[str]:
+    """Say up front that the tree has been dealt with, when it has."""
+    if not reason:
+        return []
+    return [
+        f"> **Already actioned by a sheriff.** Treeherder now classifies this job as "
+        f"_{reason}_, so the tree has been dealt with and this analysis needs no "
+        f"action from a sheriff. It is sent for the developer's reland.",
+        "",
+    ]
+
+
 def _build_test_repair_body(
     ctx: RunContext,
     findings: dict,
     patch: str | None,
     culprit_author: str | None,
+    already_actioned: str | None = None,
 ) -> str:
     groups = ", ".join(f"`{g}`" for g in ctx.test_groups) or "not resolved"
     lines = [
+        *_already_actioned_banner(already_actioned),
         "# Test failure analysis",
         "",
         f"- **Recommendation:** {_banner(findings)}",
diff --git a/services/hackbot-pulse-listener/app/worker.py b/services/hackbot-pulse-listener/app/worker.py
index 188ea81376..d0d715aef9 100644
--- a/services/hackbot-pulse-listener/app/worker.py
+++ b/services/hackbot-pulse-listener/app/worker.py
@@ -1,7 +1,7 @@
 import logging
 import time
 
-from app import client, notify
+from app import client, notify, treeherder
 from app.config import settings
 from app.models import RunContext
 
@@ -30,11 +30,35 @@ def poll_and_notify(ctx: RunContext) -> None:
         return
 
     try:
-        notify.send_email(ctx, run_doc)
+        notify.send_email(ctx, run_doc, _already_actioned(ctx))
     except Exception:
         logger.exception("Failed to send notification for run %s", ctx.run_id)
 
 
+def _already_actioned(ctx: RunContext) -> str | None:
+    """Treeherder's verdict now that the run has finished, or None.
+
+    A sheriff often acts while a run works. Never raises: the email goes out unmarked.
+    """
+    try:
+        reason = treeherder.recheck_skip_reason(ctx.repo, ctx.task_id)
+    except Exception:
+        logger.exception(
+            "Could not re-check the classification of task %s before notifying",
+            ctx.task_id,
+        )
+        return None
+    if reason:
+        logger.info(
+            "Task %s was classified as %s while run %s was working; "
+            "the notification will say so",
+            ctx.task_id,
+            reason,
+            ctx.run_id,
+        )
+    return reason
+
+
 def _poll_until_terminal(run_id: str) -> dict | None:
     deadline = time.monotonic() + settings.run_max_age_minutes * 60
     while True:
diff --git a/services/hackbot-pulse-listener/tests/test_notify.py b/services/hackbot-pulse-listener/tests/test_notify.py
index 8862591ab1..8b36d5f49f 100644
--- a/services/hackbot-pulse-listener/tests/test_notify.py
+++ b/services/hackbot-pulse-listener/tests/test_notify.py
@@ -551,3 +551,65 @@ def test_an_unknown_recommendation_is_shown_verbatim():
         "backout_and_reland"
     )
     assert notify._banner({}) == "analysis"
+
+
+def test_already_actioned_body_leads_with_the_banner():
+    body = notify._build_test_repair_body(
+        _test_repair_ctx(),
+        _test_repair_findings(),
+        None,
+        None,
+        already_actioned="fixed by commit",
+    )
+    banner, _ = body.split("# Test failure analysis", 1)
+    assert "Already actioned by a sheriff" in banner
+    assert "fixed by commit" in banner
+    assert "BACK OUT the culprit" in body
+    assert "## Analysis" in body
+
+
+def test_an_unactioned_body_has_no_banner():
+    body = notify._build_test_repair_body(
+        _test_repair_ctx(), _test_repair_findings(), None, None
+    )
+    assert "Already actioned" not in body
+    assert body.startswith("# Test failure analysis")
+
+
+def test_already_actioned_is_marked_in_the_subject(monkeypatch):
+    monkeypatch.setattr(notify.settings, "sendgrid_api_key", "key")
+    monkeypatch.setattr(notify.settings, "notification_sender", "from@mozilla.com")
+    monkeypatch.setattr(
+        notify.settings, "notification_override_email", "me@mozilla.com"
+    )
+
+    run_doc = {"status": "succeeded", "summary": {"findings": _test_repair_findings()}}
+    fake_client = MagicMock()
+    fake_client.send.return_value = MagicMock(status_code=202)
+    with (
+        patch("sendgrid.SendGridAPIClient", return_value=fake_client),
+        patch.object(notify.github, "commit_author_email", return_value=None),
+    ):
+        notify.send_email(_test_repair_ctx(), run_doc, "fixed by commit")
+
+    message = fake_client.send.call_args.kwargs["message"].get()
+    assert message["subject"].startswith("[test-repair] [already actioned] ")
+    assert "Already actioned by a sheriff" in message["content"][0]["value"]
+
+
+def test_build_repair_ignores_the_actioned_flag(monkeypatch):
+    monkeypatch.setattr(notify.settings, "sendgrid_api_key", "key")
+    monkeypatch.setattr(notify.settings, "notification_sender", "from@mozilla.com")
+    monkeypatch.setattr(
+        notify.settings, "notification_override_email", "me@mozilla.com"
+    )
+    monkeypatch.setattr(notify.settings, "notify_only_with_patch", False)
+
+    run_doc = {"status": "succeeded", "summary": {"findings": {}}}
+    fake_client = MagicMock()
+    fake_client.send.return_value = MagicMock(status_code=202)
+    with patch("sendgrid.SendGridAPIClient", return_value=fake_client):
+        notify.send_email(_ctx(), run_doc, "fixed by commit")
+
+    message = fake_client.send.call_args.kwargs["message"].get()
+    assert "already actioned" not in message["subject"]
diff --git a/services/hackbot-pulse-listener/tests/test_worker.py b/services/hackbot-pulse-listener/tests/test_worker.py
index d1533be3d6..4c50306715 100644
--- a/services/hackbot-pulse-listener/tests/test_worker.py
+++ b/services/hackbot-pulse-listener/tests/test_worker.py
@@ -1,5 +1,6 @@
 from unittest.mock import patch
 
+import pytest
 from app import worker
 from app.models import RunContext
 
@@ -13,6 +14,15 @@
 )
 
 
+@pytest.fixture(autouse=True)
+def unactioned():
+    """Keep the pre-notification re-check off the network; no sheriff acted."""
+    with patch.object(
+        worker.treeherder, "recheck_skip_reason", return_value=None
+    ) as recheck:
+        yield recheck
+
+
 def test_terminal_run_notifies_once():
     run_doc = {"status": "succeeded", "summary": {}}
     with (
@@ -22,7 +32,7 @@ def test_terminal_run_notifies_once():
         worker.poll_and_notify(CTX)
 
     get_run.assert_called_once()
-    notify.send_email.assert_called_once_with(CTX, run_doc)
+    notify.send_email.assert_called_once_with(CTX, run_doc, None)
 
 
 def test_gives_up_after_max_age(monkeypatch):
@@ -37,3 +47,59 @@ def test_gives_up_after_max_age(monkeypatch):
 
     get_run.assert_called_once()
     notify.send_email.assert_not_called()
+
+
+def test_a_late_sheriff_action_marks_the_notification():
+    run_doc = {"status": "succeeded", "summary": {}}
+    with (
+        patch.object(worker.client, "get_run", return_value=run_doc),
+        patch.object(
+            worker.treeherder, "recheck_skip_reason", return_value="fixed by commit"
+        ),
+        patch.object(worker, "notify") as notify,
+    ):
+        worker.poll_and_notify(CTX)
+
+    notify.send_email.assert_called_once_with(CTX, run_doc, "fixed by commit")
+
+
+def test_an_unactioned_failure_notifies_unmarked():
+    run_doc = {"status": "succeeded", "summary": {}}
+    with (
+        patch.object(worker.client, "get_run", return_value=run_doc),
+        patch.object(worker.treeherder, "recheck_skip_reason", return_value=None),
+        patch.object(worker, "notify") as notify,
+    ):
+        worker.poll_and_notify(CTX)
+
+    notify.send_email.assert_called_once_with(CTX, run_doc, None)
+
+
+def test_the_analysis_is_still_sent_after_a_sheriff_acted():
+    run_doc = {"status": "succeeded", "summary": {}}
+    with (
+        patch.object(worker.client, "get_run", return_value=run_doc),
+        patch.object(
+            worker.treeherder, "recheck_skip_reason", return_value="fixed by commit"
+        ),
+        patch.object(worker, "notify") as notify,
+    ):
+        worker.poll_and_notify(CTX)
+
+    notify.send_email.assert_called_once()
+
+
+def test_a_failed_recheck_does_not_block_the_notification():
+    run_doc = {"status": "succeeded", "summary": {}}
+    with (
+        patch.object(worker.client, "get_run", return_value=run_doc),
+        patch.object(
+            worker.treeherder,
+            "recheck_skip_reason",
+            side_effect=RuntimeError("treeherder down"),
+        ),
+        patch.object(worker, "notify") as notify,
+    ):
+        worker.poll_and_notify(CTX)
+
+    notify.send_email.assert_called_once_with(CTX, run_doc, None)

From f1fccca962d192980b913c05e69c4c243235af53 Mon Sep 17 00:00:00 2001
From: Evgeny Pavlov <epavlov@mozilla.com>
Date: Fri, 7 Aug 2026 14:31:55 -0700
Subject: [PATCH 5/5] test-repair-agent: change notification wording

---
 .../hackbot_agents/test_repair/notify.py        | 17 ++++++++++++++---
 agents/test-repair/tests/test_notify.py         | 16 ++++++++++------
 2 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/agents/test-repair/hackbot_agents/test_repair/notify.py b/agents/test-repair/hackbot_agents/test_repair/notify.py
index 73be58c666..86021d49c1 100644
--- a/agents/test-repair/hackbot_agents/test_repair/notify.py
+++ b/agents/test-repair/hackbot_agents/test_repair/notify.py
@@ -9,7 +9,7 @@
 then visible in the hackbot UI before it lands, and the apply step delivers it at
 most once (see ``hackbot_runtime.actions.slack``).
 
-Five lines of context, then the verdict in full. Every identifier a sheriff would
+A few lines of context, then the verdict in full. Every identifier a sheriff would
 otherwise have to look up -- revisions, task, bug, run -- is a link, the way the
 pulse listener's email does it
 (``services/hackbot-pulse-listener/app/notify.py``); unlike the email this stays
@@ -120,11 +120,19 @@ def _culprit_line(result: TestRepairResult, culprit_author: str | None) -> str:
     bug = result.culprit_bug or result.intermittent_bug
     if bug:
         line += f" ({_bug_link(bug)})"
-    if result.proposed_patch:
-        line += ", patch attached"
     return line
 
 
+def _patch_line(result: TestRepairResult) -> str | None:
+    """Who the attached patch is for, so it is not read as an alternative action."""
+    if not result.proposed_patch:
+        return None
+    return (
+        "Patch attached for the author: squash it into the existing patches and"
+        " reland, rather than landing it as a follow-up. The backout still stands."
+    )
+
+
 def build_message(
     result: TestRepairResult,
     investigation: Investigation,
@@ -144,6 +152,9 @@ def build_message(
         _culprit_line(result, culprit_author),
         _link(RUN_URL.format(run_id=run_id), "Hackbot run details"),
     ]
+    patch = _patch_line(result)
+    if patch:
+        lines.insert(-1, patch)
     if result.summary.strip():
         lines += ["", result.summary.strip()]
     return "\n".join(lines)
diff --git a/agents/test-repair/tests/test_notify.py b/agents/test-repair/tests/test_notify.py
index 74996ad270..5e129b3e26 100644
--- a/agents/test-repair/tests/test_notify.py
+++ b/agents/test-repair/tests/test_notify.py
@@ -122,12 +122,16 @@ def test_names_a_known_intermittent():
     )
 
 
-def test_mentions_a_proposed_patch():
-    assert (
-        _message(_result(proposed_patch=True))
-        .splitlines()[4]
-        .endswith(", patch attached")
-    )
+def test_a_patch_is_advice_for_the_author_not_an_alternative_action():
+    message = _message(_result(proposed_patch=True))
+    assert "*test-repair: BACK OUT the culprit*" in message.splitlines()[0]
+    assert "squash it into the existing patches and reland" in message
+    assert "rather than landing it as a follow-up" in message
+    assert "The backout still stands." in message
+
+
+def test_no_patch_line_without_a_patch():
+    assert "Patch attached" not in _message()
 
 
 def test_lists_every_failing_group():