[autorevert] Surface test identity at signal level for advisor JSON

The autorevert AI advisor's signal_pattern JSON does not surface
authoritative test identity (file/classname/name) for TEST signals.
The advisor knows the test from `signal_key` (which is `file::name`),
but `classname` is dropped by the test_id collapse — so when many tests
share the same file or name, the advisor has to re-derive structure
from logs. When the suspect commit's parent workflow_job is still
in_progress at advisor dispatch time, that re-derivation fails and the
advisor rationalizes pending-heavy partitions as flake → garbage. A
30d audit found 40 garbage verdicts on 23 distinct commits later
auto-reverted; 33/37 had a workflow_job for the suspect commit still
in_progress at verdict time.

A TEST signal corresponds to a SINGLE specific test by construction
(one signal per `(workflow, job_base, test_id)`). Every FAILURE event
in that signal IS that specific test failing — no per-event test
identity is needed (`signal_key` + `status` already convey it).

This change surfaces the structured test identity ONCE at the top of
the advisor signal_pattern payload:

    {
      "signal_key": "test/foo.py::test_bar",
      "signal_source": "test",
      "test_file": "test/foo.py",
      "test_classname": "TestFooBar",
      "test_name": "test_bar",
      ...
    }

Implementation:

- `Signal` gains optional `test_file`/`test_classname`/`test_name`
  attributes (default None).
- `_build_test_signals` captures the (file, classname, name) triple
  from any TestRow contributing to the signal (collapses cleanly
  because each TEST signal is a single test by definition) and
  attaches them to the Signal at construction time.
- `_build_signal_pattern_json` emits the three fields as top-level
  payload keys when populated, only for `signal_source == "test"`.
- New regression tests:
  - `test_test_identity_surfaced_for_test_signals` — top-level keys
    present; no per-event leak.
  - `test_test_identity_omitted_for_job_signals` — JOB signals omit
    the keys entirely.

Companion change: pytorch/pytorch prompt PR adding a "Partial
Visibility Warning" section, tightening the `garbage` definition to
require concrete completed-log infra evidence, and adding an
explicit `signal_key`-is-authoritative assertion that aligns with the
top-level test identity exposed here.

Author: Ivan Zaitsev

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

@@ -359,6 +359,9 @@ def __init__(
         job_base_name: Optional[str] = None,
         test_module: Optional[str] = None,
         source: SignalSource = SignalSource.TEST,
+        test_file: Optional[str] = None,
+        test_classname: Optional[str] = None,
+        test_name: Optional[str] = None,
     ):
         self.key = key
         self.workflow_name = workflow_name
@@ -369,6 +372,15 @@ def __init__(
         self.test_module = test_module
         # Track the origin of the signal (test-track or job-track).
         self.source = source
+        # For TEST signals: structured identity of the failing test, sourced
+        # from tests.all_test_runs. Surfaced once at the top of the advisor
+        # signal_pattern JSON so the AI advisor has authoritative ground
+        # truth for which test failed (file/classname/name) without
+        # re-deriving from logs. Every FAILURE event in this signal IS this
+        # specific test failing.
+        self.test_file = test_file
+        self.test_classname = test_classname
+        self.test_name = test_name
 
     def detect_fixed(self) -> bool:
         """

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

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from enum import Enum
-from typing import Dict, FrozenSet, Iterable, List, Optional, Tuple, Union
+from typing import Any, Dict, FrozenSet, Iterable, List, Optional, Tuple, Union
 
 import github
 
@@ -18,6 +18,7 @@
     Ineligible,
     RestartCommits,
     Signal,
+    SignalSource,
 )
 from .signal_extraction_types import RunContext
 from .utils import (
@@ -807,17 +808,28 @@ def _partition_label(sha: str) -> str:
                 }
             )
 
-        return json.dumps(
-            {
-                "signal_key": signal.key,
-                "signal_source": signal.source.value if signal.source else "unknown",
-                "workflow_name": signal.workflow_name,
-                "job_base_name": signal.job_base_name,
-                "commit_order": "newest_first",
-                "suspect_commit": dispatch_advisor.suspect_commit,
-                "commits": commits_json,
-            }
-        )
+        payload: Dict[str, Any] = {
+            "signal_key": signal.key,
+            "signal_source": signal.source.value if signal.source else "unknown",
+            "workflow_name": signal.workflow_name,
+            "job_base_name": signal.job_base_name,
+            "commit_order": "newest_first",
+            "suspect_commit": dispatch_advisor.suspect_commit,
+        }
+        # For TEST signals, surface authoritative test identity once at the
+        # top of the payload (file/classname/name from tests.all_test_runs).
+        # Every FAILURE event in this signal IS this specific test failing —
+        # the AI advisor does not need to re-derive from logs. Only emitted
+        # when populated to keep the payload backward-compatible.
+        if signal.source == SignalSource.TEST:
+            if signal.test_file:
+                payload["test_file"] = signal.test_file
+            if signal.test_classname:
+                payload["test_classname"] = signal.test_classname
+            if signal.test_name:
+                payload["test_name"] = signal.test_name
+        payload["commits"] = commits_json
+        return json.dumps(payload)
 
     def _commit_message_check_pr_is_revert(
         self, commit_message: str, ctx: RunContext

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

@@ -151,6 +151,9 @@ def _dedup_signal_events(self, signals: List[Signal]) -> List[Signal]:
                     job_base_name=s.job_base_name,
                     test_module=s.test_module,
                     source=s.source,
+                    test_file=s.test_file,
+                    test_classname=s.test_classname,
+                    test_name=s.test_name,
                 )
             )
         return deduped
@@ -237,6 +240,9 @@ def _inject_pending_workflow_events(
                     job_base_name=s.job_base_name,
                     test_module=s.test_module,
                     source=s.source,
+                    test_file=s.test_file,
+                    test_classname=s.test_classname,
+                    test_name=s.test_name,
                 )
             )
         return out
@@ -302,6 +308,9 @@ def _attach_advisor_verdicts(
                     job_base_name=s.job_base_name,
                     test_module=s.test_module,
                     source=s.source,
+                    test_file=s.test_file,
+                    test_classname=s.test_classname,
+                    test_name=s.test_name,
                 )
             )
         return out
@@ -393,6 +402,11 @@ def _build_test_signals(
         failing_tests_by_job_base_name: Set[
             Tuple[WorkflowName, JobBaseName, TestId]
         ] = set()
+        # Capture structured test identity (file/classname/name) per test_id so
+        # we can attach it once at the Signal level. The TestRow `test_id`
+        # property collapses to "file::name" and drops classname; we keep the
+        # full triple here for the advisor signal_pattern JSON.
+        test_identity_by_test_id: Dict[TestId, Tuple[str, str, str]] = {}
         for tr in test_rows:
             job = jobs_by_id.get(tr.job_id)
             job_base_name = job.base_name
@@ -419,6 +433,9 @@ def _build_test_signals(
                 outcome = existing
 
             tests_by_group_attempt[key] = outcome
+            test_identity_by_test_id.setdefault(
+                tr.test_id, (tr.file, tr.classname, tr.name)
+            )
 
             # Track keys that have at least one persistent failure (no retry success)
             if outcome.failure_runs > 0 and outcome.success_runs == 0:
@@ -523,6 +540,9 @@ def _build_test_signals(
                 # Extract test module from test_id (format: "file.py::test_name")
                 # Result: "file" or "path/to/file" without .py extension
                 test_module = test_id.split("::")[0].replace(".py", "")
+                test_file, test_classname, test_name = test_identity_by_test_id.get(
+                    test_id, ("", "", "")
+                )
 
                 signals.append(
                     Signal(
@@ -532,6 +552,9 @@ def _build_test_signals(
                         job_base_name=str(job_base_name),
                         test_module=test_module,
                         source=SignalSource.TEST,
+                        test_file=test_file or None,
+                        test_classname=test_classname or None,
+                        test_name=test_name or None,
                     )
                 )
 

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/tests/test_signal_actions.py

@@ -966,6 +966,132 @@ def test_signal_pattern_sanity_check(self):
         reparsed = json.loads(json.dumps(result))
         self.assertEqual(reparsed, result)
 
+    def test_test_identity_surfaced_for_test_signals(self):
+        """TEST signals surface authoritative test identity (file/classname/name)
+        once at the top of the payload. Every FAILURE event in the signal IS
+        this specific test failing — no duplication per event."""
+        import json
+
+        from pytorch_auto_revert.signal import (
+            DispatchAdvisor,
+            Signal,
+            SignalCommit,
+            SignalEvent,
+            SignalSource,
+            SignalStatus,
+        )
+
+        t0 = datetime(2025, 8, 19, 12, 0, 0)
+        t1 = datetime(2025, 8, 19, 11, 0, 0)
+
+        c_fail = SignalCommit(
+            head_sha="sha_fail",
+            timestamp=t0,
+            events=[
+                SignalEvent(
+                    "test", SignalStatus.FAILURE, t0, wf_run_id=100, job_id=200
+                ),
+            ],
+        )
+        c_base = SignalCommit(
+            head_sha="sha_base",
+            timestamp=t1,
+            events=[
+                SignalEvent(
+                    "test", SignalStatus.SUCCESS, t1, wf_run_id=99, job_id=199
+                ),
+            ],
+        )
+        signal = Signal(
+            key="test/inductor/test_comm_analysis.py::test_nccl_estimate_device_resolution_gpu",
+            workflow_name="trunk",
+            commits=[c_fail, c_base],
+            source=SignalSource.TEST,
+            job_base_name="linux-jammy / test",
+            test_file="test/inductor/test_comm_analysis.py",
+            test_classname="TestNcclEstimateDeviceResolution",
+            test_name="test_nccl_estimate_device_resolution_gpu",
+        )
+        advisor = DispatchAdvisor(
+            suspect_commit="sha_fail",
+            failed_commits=("sha_fail",),
+            successful_commits=("sha_base",),
+        )
+
+        result = json.loads(
+            SignalActionProcessor._build_signal_pattern_json(
+                signal=signal,
+                dispatch_advisor=advisor,
+                repo_full_name="pytorch/pytorch",
+            )
+        )
+        # Top-level test identity is present, populated, and authoritative
+        self.assertEqual(result["test_file"], "test/inductor/test_comm_analysis.py")
+        self.assertEqual(result["test_classname"], "TestNcclEstimateDeviceResolution")
+        self.assertEqual(
+            result["test_name"], "test_nccl_estimate_device_resolution_gpu"
+        )
+
+        # No per-event test_failures emission (would be redundant with signal_key)
+        for c in result["commits"]:
+            for ev in c["events"]:
+                self.assertNotIn("test_failures", ev)
+                self.assertNotIn("test_file", ev)
+                self.assertNotIn("test_classname", ev)
+                self.assertNotIn("test_name", ev)
+
+    def test_test_identity_omitted_for_job_signals(self):
+        """JOB signals (and TEST signals without identity populated) must not
+        emit the test_* top-level keys."""
+        import json
+
+        from pytorch_auto_revert.signal import (
+            DispatchAdvisor,
+            Signal,
+            SignalCommit,
+            SignalEvent,
+            SignalSource,
+            SignalStatus,
+        )
+
+        t0 = datetime(2025, 8, 19, 12, 0, 0)
+
+        c_fail = SignalCommit(
+            head_sha="sha_fail",
+            timestamp=t0,
+            events=[
+                SignalEvent("j", SignalStatus.FAILURE, t0, wf_run_id=1, job_id=1),
+            ],
+        )
+        c_base = SignalCommit(
+            head_sha="sha_base",
+            timestamp=t0,
+            events=[
+                SignalEvent("j", SignalStatus.SUCCESS, t0, wf_run_id=2, job_id=2),
+            ],
+        )
+        signal = Signal(
+            key="lint",
+            workflow_name="trunk",
+            commits=[c_fail, c_base],
+            source=SignalSource.JOB,
+        )
+        advisor = DispatchAdvisor(
+            suspect_commit="sha_fail",
+            failed_commits=("sha_fail",),
+            successful_commits=("sha_base",),
+        )
+
+        result = json.loads(
+            SignalActionProcessor._build_signal_pattern_json(
+                signal=signal,
+                dispatch_advisor=advisor,
+                repo_full_name="pytorch/pytorch",
+            )
+        )
+        for k in ("test_file", "test_classname", "test_name"):
+            self.assertNotIn(k, result)
+
 
 class TestDispatchAdvisorsMethod(unittest.TestCase):
     """Tests for SignalActionProcessor.dispatch_advisors."""
@@ -1270,6 +1396,108 @@ def test_invalid_verdict_string_defaults_to_unsure(self):
             result[0].commits[0].advisor_result.verdict, AdvisorVerdict.UNSURE
         )
 
+    def test_preserves_test_identity_on_attach(self):
+        # Signal-level test_file/test_classname/test_name must survive the
+        # Signal reconstruction inside _attach_advisor_verdicts.
+        from pytorch_auto_revert.signal import (
+            Signal,
+            SignalCommit,
+            SignalEvent,
+            SignalSource,
+            SignalStatus,
+        )
+        from pytorch_auto_revert.signal_extraction import SignalExtractor
+        from pytorch_auto_revert.signal_extraction_types import Sha
+
+        t0 = datetime(2025, 8, 19, 12, 0, 0)
+        c1 = SignalCommit(
+            "sha_aaa",
+            t0,
+            [SignalEvent("j", SignalStatus.FAILURE, t0, wf_run_id=1, job_id=10)],
+        )
+        signal = Signal(
+            key="test/foo.py::test_bar",
+            workflow_name="trunk",
+            commits=[c1],
+            source=SignalSource.TEST,
+            test_file="test/foo.py",
+            test_classname="TestFooBar",
+            test_name="test_bar",
+        )
+        extractor = SignalExtractor(workflows=["trunk"], lookback_hours=16)
+        extractor._datasource = Mock()
+        extractor._datasource.fetch_advisor_verdicts.return_value = {
+            ("sha_aaa", "test/foo.py::test_bar"): ("revert", 0.95, t0),
+        }
+        out = extractor._attach_advisor_verdicts([signal], [(Sha("sha_aaa"), t0)])
+        self.assertEqual(out[0].test_file, "test/foo.py")
+        self.assertEqual(out[0].test_classname, "TestFooBar")
+        self.assertEqual(out[0].test_name, "test_bar")
+
+
+class TestInjectPendingWorkflowEvents(unittest.TestCase):
+    """Tests for SignalExtractor._inject_pending_workflow_events."""
+
+    def test_preserves_test_identity_on_inject(self):
+        # Signal-level test_file/test_classname/test_name must survive the
+        # Signal reconstruction inside _inject_pending_workflow_events.
+        from pytorch_auto_revert.signal import (
+            Signal,
+            SignalCommit,
+            SignalEvent,
+            SignalSource,
+            SignalStatus,
+        )
+        from pytorch_auto_revert.signal_extraction import SignalExtractor
+        from pytorch_auto_revert.signal_extraction_types import (
+            JobBaseName,
+            JobId,
+            JobName,
+            JobRow,
+            RunAttempt,
+            Sha,
+            WfRunId,
+            WorkflowName,
+        )
+
+        t0 = datetime(2025, 8, 19, 12, 0, 0)
+        c1 = SignalCommit(
+            "sha_aaa",
+            t0,
+            [SignalEvent("j", SignalStatus.FAILURE, t0, wf_run_id=1, job_id=10)],
+        )
+        signal = Signal(
+            key="test/foo.py::test_bar",
+            workflow_name="trunk",
+            commits=[c1],
+            source=SignalSource.TEST,
+            test_file="test/foo.py",
+            test_classname="TestFooBar",
+            test_name="test_bar",
+        )
+        # One pending JobRow on a different wf_run_id triggers synthesis on c1.
+        pending_job = JobRow(
+            head_sha=Sha("sha_aaa"),
+            workflow_name=WorkflowName("trunk"),
+            wf_run_id=WfRunId(2),
+            job_id=JobId(20),
+            run_attempt=RunAttempt(1),
+            name=JobName("j"),
+            status="in_progress",
+            conclusion="",
+            started_at=t0,
+            created_at=t0,
+            rule="",
+        )
+        extractor = SignalExtractor(workflows=["trunk"], lookback_hours=16)
+        out = extractor._inject_pending_workflow_events([signal], [pending_job])
+        self.assertEqual(out[0].test_file, "test/foo.py")
+        self.assertEqual(out[0].test_classname, "TestFooBar")
+        self.assertEqual(out[0].test_name, "test_bar")
+        # Synthesis actually fired (otherwise the test wouldn't exercise
+        # the reconstruction path)
+        self.assertGreater(len(out[0].commits[0].events), 1)
+
 
 if __name__ == "__main__":
     unittest.main()