feat: read AI advisor verdicts from CH and use in autorevert decisions

When advisor verdicts are available in misc.autorevert_advisor_verdicts,
autorevert now uses them to make faster decisions:

- revert → produce AutorevertPattern immediately (skip restarts)
- not_related → early exit (Ineligible)
- garbage → suppress signal for 2 hours since verdict timestamp
- unsure → continue normal restart-to-confirm flow

Key changes:
- AIAdvisorResult dataclass + AdvisorVerdict enum in signal.py
- SignalCommit gets optional advisor_result field
- fetch_advisor_verdicts() query in signal_extraction_datasource.py
- _attach_advisor_verdicts() in signal_extraction.py wires CH data to SignalCommits
- Flakiness check moved AFTER advisor check/dispatch so advisor
  can still evaluate flaky-looking signals
- 8 new tests covering all verdict types, expiry, signal key matching

Author: izaitsevfb

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/signal.py

@@ -6,6 +6,29 @@
 from .bisection_planner import GapBisectionPlanner
 
 
+class AdvisorVerdict(Enum):
+    """Verdict from the AI advisor workflow."""
+
+    REVERT = "revert"
+    UNSURE = "unsure"
+    NOT_RELATED = "not_related"
+    GARBAGE = "garbage"
+
+
+@dataclass(frozen=True)
+class AIAdvisorResult:
+    """Result from a previously-dispatched AI advisor run, read from ClickHouse.
+
+    Attached to SignalCommit so that process_valid_autorevert_pattern() can
+    use the advisor's analysis to make faster revert/skip decisions.
+    """
+
+    verdict: AdvisorVerdict
+    confidence: float
+    timestamp: datetime  # when the verdict was produced
+    signal_key: str  # the signal key this verdict applies to
+
+
 class SignalStatus(Enum):
     """signal status enum"""
 
@@ -76,6 +99,8 @@ class IneligibleReason(Enum):
         "insufficient_successes"  # not enough successes to make call
     )
     PENDING_GAP = "pending_gap"  # unknown/pending commits present
+    ADVISOR_NOT_RELATED = "advisor_not_related"  # AI advisor says not related
+    ADVISOR_GARBAGE = "advisor_garbage"  # AI advisor says signal is garbage
 
 
 @dataclass
@@ -128,7 +153,13 @@ def is_failure(self) -> bool:
 class SignalCommit:
     """All events for a single commit, ordered oldest → newest by start time."""
 
-    def __init__(self, head_sha: str, timestamp: datetime, events: List[SignalEvent]):
+    def __init__(
+        self,
+        head_sha: str,
+        timestamp: datetime,
+        events: List[SignalEvent],
+        advisor_result: Optional[AIAdvisorResult] = None,
+    ):
         self.head_sha = head_sha
         self.timestamp = timestamp
         # enforce events ordered by time, then by wf_run_id (oldest first)
@@ -139,6 +170,8 @@ def __init__(self, head_sha: str, timestamp: datetime, events: List[SignalEvent]
         self.statuses = {}
         for e in self.events:
             self.statuses[e.status] = self.statuses.get(e.status, 0) + 1
+        # Optional AI advisor result for this (commit, signal) pair
+        self.advisor_result = advisor_result
 
     @property
     def has_pending(self) -> bool:
@@ -384,8 +417,10 @@ def partition_by_autorevert_pattern(self) -> Optional[PartitionedCommits]:
         picking_failed = True  # simple state machine
 
         # first broadly partition into failed and successful
+        # A commit with both success and failure (flaky) stays in the failed
+        # partition — only pure successes trigger the transition.
         for c in self.commits:
-            if c.has_success:
+            if c.has_success and not c.has_failure:
                 picking_failed = False
             elif c.has_failure and not picking_failed:
                 # encountered a failure after the streak of successes
@@ -409,24 +444,93 @@ def partition_by_autorevert_pattern(self) -> Optional[PartitionedCommits]:
 
         return PartitionedCommits(failed=failed, unknown=unknown, successful=successful)
 
+    # Minimum confidence threshold for acting on advisor verdicts
+    ADVISOR_CONFIDENCE_THRESHOLD = 0.9
+
+    def _build_autorevert_pattern(
+        self, partition: "PartitionedCommits"
+    ) -> AutorevertPattern:
+        """Build an AutorevertPattern from a validated partition."""
+        suspected = partition.failed[-1]
+        newer_failures = [c.head_sha for c in partition.failed[:-1]]
+        failure_event = next(
+            (e for e in suspected.events if e.is_failure and e.job_id is not None),
+            None,
+        )
+        return AutorevertPattern(
+            workflow_name=self.workflow_name,
+            newer_failing_commits=newer_failures,
+            suspected_commit=suspected.head_sha,
+            older_successful_commit=partition.successful[0].head_sha,
+            wf_run_id=failure_event.wf_run_id if failure_event else None,
+            job_id=failure_event.job_id if failure_event else None,
+        )
+
+    def _check_advisor_verdict(
+        self, partition: "PartitionedCommits"
+    ) -> Optional[Union[AutorevertPattern, Ineligible]]:
+        """Check if an AI advisor verdict is available for the suspect commit.
+
+        Returns:
+            - AutorevertPattern if advisor says "revert" with sufficient confidence
+            - Ineligible if advisor says "not_related" or "garbage" (within 2h window)
+            - None if no verdict, verdict is "unsure", or confidence below threshold
+        """
+        suspected = partition.failed[-1]
+        result = suspected.advisor_result
+        if result is None:
+            return None
+
+        # Only act on the verdict if it matches this signal
+        if result.signal_key != self.key:
+            return None
+
+        # Ignore low-confidence verdicts
+        if result.confidence < self.ADVISOR_CONFIDENCE_THRESHOLD:
+            return None
+
+        if result.verdict == AdvisorVerdict.REVERT:
+            return self._build_autorevert_pattern(partition)
+
+        if result.verdict == AdvisorVerdict.NOT_RELATED:
+            return Ineligible(
+                IneligibleReason.ADVISOR_NOT_RELATED,
+                f"AI advisor says not related (confidence={result.confidence:.2f})",
+            )
+
+        if result.verdict == AdvisorVerdict.GARBAGE:
+            # Garbage verdict blocks the signal for 2 hours since the verdict timestamp
+            from datetime import timezone
+
+            now = datetime.now(timezone.utc)
+            verdict_age = now - result.timestamp.replace(tzinfo=timezone.utc)
+            if verdict_age.total_seconds() < 2 * 3600:
+                return Ineligible(
+                    IneligibleReason.ADVISOR_GARBAGE,
+                    f"AI advisor says garbage signal, suppressed for 2h "
+                    f"(confidence={result.confidence:.2f}, "
+                    f"age={int(verdict_age.total_seconds() / 60)}min)",
+                )
+            # Garbage verdict expired — fall through to normal processing
+
+        # "unsure" or expired garbage → continue normally
+        return None
+
     def process_valid_autorevert_pattern(
         self, *, bisection_limit: Optional[int] = None
     ) -> Union[AutorevertPattern, RestartCommits, Ineligible]:
         """
         Detect valid autorevert pattern in the Signal.
 
-        Validates all invariants before checking for the pattern.
+        Validates invariants, checks AI advisor verdicts, and determines
+        appropriate action.
 
         Returns one of:
             - AutorevertPattern: a confirmed pattern ready for action
             - RestartCommits: a suggested set of commits to restart to reduce uncertainty
             - Ineligible: reason + optional message when no pattern is actionable yet
         """
-        if self.detect_flaky():
-            return Ineligible(
-                IneligibleReason.FLAKY,
-                "signal is flaky (mixed outcomes on same commit)",
-            )
+        # Early exits that prevent partition computation
         if self.detect_fixed():
             return Ineligible(
                 IneligibleReason.FIXED, "signal appears recovered at head"
@@ -443,7 +547,12 @@ def process_valid_autorevert_pattern(
                 "insufficient history to form failed/unknown/successful partitions",
             )
 
-        # Advisor eligibility: partition exists with sufficient data and
+        # --- AI advisor verdict check (before flakiness and other heuristics) ---
+        advisor_decision = self._check_advisor_verdict(partition)
+        if advisor_decision is not None:
+            return advisor_decision
+
+        # Advisor dispatch eligibility: partition exists with sufficient data and
         # no unknown commits between failed and successful partitions.
         # Unknown commits mean the partition boundary is uncertain —
         # wait for restarts to resolve before dispatching advisor.
@@ -459,6 +568,15 @@ def process_valid_autorevert_pattern(
                 successful_commits=tuple(c.head_sha for c in partition.successful),
             )
 
+        # Flakiness check — placed after advisor check/dispatch so that
+        # advisor can still evaluate flaky-looking signals
+        if self.detect_flaky():
+            return Ineligible(
+                IneligibleReason.FLAKY,
+                "signal is flaky (mixed outcomes on same commit)",
+                advisor=advisor,
+            )
+
         restart_commits = set()
 
         # Cover the unknown gap (between failed and successful partitions)
@@ -554,21 +672,4 @@ def process_valid_autorevert_pattern(
             )
 
         # all invariants validated, confirmed not infra, pattern exists
-        # failed is newest -> older; the last element is the suspected commit
-        suspected = partition.failed[-1]
-        newer_failures = [c.head_sha for c in partition.failed[:-1]]
-
-        # Extract job_id and wf_run_id from a failing event on the suspected commit
-        failure_event = next(
-            (e for e in suspected.events if e.is_failure and e.job_id is not None),
-            None,
-        )
-
-        return AutorevertPattern(
-            workflow_name=self.workflow_name,
-            newer_failing_commits=newer_failures,
-            suspected_commit=suspected.head_sha,
-            older_successful_commit=partition.successful[0].head_sha,
-            wf_run_id=failure_event.wf_run_id if failure_event else None,
-            job_id=failure_event.job_id if failure_event else None,
-        )
+        return self._build_autorevert_pattern(partition)

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/signal_extraction.py

@@ -12,7 +12,15 @@
 from typing import Dict, Iterable, List, Optional, Set, Tuple
 
 from .job_agg_index import JobAggIndex, JobMeta, SignalStatus as AggStatus
-from .signal import Signal, SignalCommit, SignalEvent, SignalSource, SignalStatus
+from .signal import (
+    AdvisorVerdict,
+    AIAdvisorResult,
+    Signal,
+    SignalCommit,
+    SignalEvent,
+    SignalSource,
+    SignalStatus,
+)
 from .signal_extraction_datasource import SignalExtractionDatasource
 from .signal_extraction_types import (
     JobBaseName,
@@ -107,7 +115,10 @@ def extract(self) -> List[Signal]:
 
         # Inject synthetic PENDING events for workflow runs that are known to be
         # pending but have no events in a given signal (e.g. multi-stage workflows).
-        return self._inject_pending_workflow_events(signals, jobs)
+        signals = self._inject_pending_workflow_events(signals, jobs)
+
+        # Attach AI advisor verdicts to SignalCommit objects
+        return self._attach_advisor_verdicts(signals, commits)
 
     # -----------------------------
     # Deduplication (GitHub-specific)
@@ -230,6 +241,71 @@ def _inject_pending_workflow_events(
             )
         return out
 
+    # -----------------------------
+    # Advisor verdict attachment
+    # -----------------------------
+    def _attach_advisor_verdicts(
+        self,
+        signals: List[Signal],
+        commits: List[Tuple[Sha, datetime]],
+    ) -> List[Signal]:
+        """Fetch advisor verdicts from CH and attach to SignalCommit objects.
+
+        For each (commit_sha, signal_key) pair that has a verdict in
+        misc.autorevert_advisor_verdicts, sets the advisor_result field
+        on the corresponding SignalCommit.
+        """
+        head_shas = [sha for sha, _ in commits]
+        signal_keys = list({s.key for s in signals})
+        verdicts = self._datasource.fetch_advisor_verdicts(
+            repo_full_name=self.repo_full_name,
+            head_shas=head_shas,
+            signal_keys=signal_keys,
+            lookback_hours=self.lookback_hours,
+        )
+        if not verdicts:
+            return signals
+
+        out: List[Signal] = []
+        for s in signals:
+            new_commits: List[SignalCommit] = []
+            for c in s.commits:
+                key = (c.head_sha.strip(), s.key)
+                v = verdicts.get(key)
+                if v is not None:
+                    verdict_str, confidence, ts = v
+                    try:
+                        advisor_verdict = AdvisorVerdict(verdict_str)
+                    except ValueError:
+                        advisor_verdict = AdvisorVerdict.UNSURE
+                    advisor_result = AIAdvisorResult(
+                        verdict=advisor_verdict,
+                        confidence=confidence,
+                        timestamp=ts,
+                        signal_key=s.key,
+                    )
+                    new_commits.append(
+                        SignalCommit(
+                            head_sha=c.head_sha,
+                            timestamp=c.timestamp,
+                            events=c.events,
+                            advisor_result=advisor_result,
+                        )
+                    )
+                else:
+                    new_commits.append(c)
+            out.append(
+                Signal(
+                    key=s.key,
+                    workflow_name=s.workflow_name,
+                    commits=new_commits,
+                    job_base_name=s.job_base_name,
+                    test_module=s.test_module,
+                    source=s.source,
+                )
+            )
+        return out
+
     # -----------------------------
     # Phase B — Tests (tests.all_test_runs only)
     # -----------------------------

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/signal_extraction_datasource.py

@@ -313,6 +313,64 @@ def fetch_tests_for_job_ids(
         )
         return rows
 
+    def fetch_advisor_verdicts(
+        self,
+        *,
+        repo_full_name: str,
+        head_shas: List[Sha],
+        signal_keys: List[str],
+        lookback_hours: int,
+    ) -> Dict[tuple[str, str], tuple[str, float, datetime]]:
+        """Fetch AI advisor verdicts from misc.autorevert_advisor_verdicts.
+
+        Queries by both commit SHAs AND signal keys to minimize data transferred.
+
+        Returns a dict keyed by (commit_sha, signal_key) → (verdict, confidence, timestamp).
+        When multiple verdicts exist for the same (commit, signal), the most recent is used.
+        """
+        if not head_shas or not signal_keys:
+            return {}
+
+        log = logging.getLogger(__name__)
+        t0 = time.perf_counter()
+        query = """
+        SELECT
+            toString(suspect_commit) AS suspect_commit,
+            signal_key,
+            verdict,
+            confidence,
+            timestamp
+        FROM misc.autorevert_advisor_verdicts
+        WHERE repo = {repo:String}
+          AND suspect_commit IN {shas:Array(String)}
+          AND signal_key IN {keys:Array(String)}
+          AND timestamp > now() - INTERVAL {hours:UInt32} HOUR
+        ORDER BY suspect_commit, signal_key, timestamp DESC
+        """
+        params = {
+            "repo": repo_full_name,
+            "shas": [str(s) for s in head_shas],
+            "keys": signal_keys,
+            "hours": lookback_hours,
+        }
+        results: Dict[tuple[str, str], tuple[str, float, datetime]] = {}
+        for attempt in RetryWithBackoff():
+            with attempt:
+                res = CHCliFactory().client.query(query, parameters=params)
+                for r in res.result_rows:
+                    key = (str(r[0]).strip(), str(r[1]))
+                    if key not in results:  # keep most recent (ORDER BY ... DESC)
+                        results[key] = (str(r[2]), float(r[3]), r[4])
+
+        dt = time.perf_counter() - t0
+        log.info(
+            "[extract] Advisor verdicts fetched: %d for %d commits in %.2fs",
+            len(results),
+            len(head_shas),
+            dt,
+        )
+        return results
+
     def fetch_autorevert_state_rows(
         self,
         *,