[autorevert] Surface test identity at signal level for advisor JSON (#8031)

## Summary

Surfaces authoritative TEST signal identity (`test_file` /
`test_classname` / `test_name`) once at the top of the advisor
`signal_pattern` payload, so the AI advisor has structured ground truth
for which test failed without re-deriving it from logs. A TEST signal
corresponds to a single test by construction (one signal per `(workflow,
job_base, test_id)`), so the identity is emitted once at the Signal
level, not per event.

## Classname uniqueness

`test_id` is `file::name` — classname is dropped from the id, so one
`file::name` can span multiple test classes. `test_classname` is
therefore surfaced **only when a single distinct classname is observed**
for the `test_id`; when ambiguous it is omitted (`None`) rather than
guessed. (A first-seen capture with no ordering would be
nondeterministic and could name a different class than the one that
failed.) `test_file` / `test_name` are the stable components of the
signal key.

## Changes

- `signal.py`: `Signal` gains optional `test_file` / `test_classname` /
`test_name`, threaded through `Signal.replace` so they survive the
dedup/filter passes.
- `signal_extraction.py`: `_build_test_signals` collects the distinct
classnames per `test_id` and emits one only when unambiguous.
- `signal_actions.py::_build_signal_pattern_json`: emits the three
fields for TEST signals only, when populated; JOB signals omit them.
- Tests: classname surfaced when unique, omitted when ambiguous;
identity preserved through dedup.

## Payload shape

```json
{
  "signal_key": "test/foo.py::test_bar",
  "signal_source": "test",
  "test_file": "test/foo.py",
  "test_classname": "TestFooBar",
  "test_name": "test_bar",
  "workflow_name": "trunk",
  "job_base_name": "linux-jammy / test",
  "suspect_commit": "abc...",
  "commits": [ ... ]
}
```

## Companion change

pytorch/pytorch#182176 — the advisor prompt rewrite that consumes this
structured identity.

## Test plan

- [x] CI runs the existing + new unit tests under
`pytorch_auto_revert/tests/`
- [x] Manual: observe the three keys at the top of TEST advisor
`signal_pattern` JSONs after merge

Author: izaitsevfb

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

@@ -383,6 +383,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
@@ -393,6 +396,16 @@ 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 and surfaced at the top of the advisor
+        # signal_pattern JSON. test_file/test_name are authoritative — they
+        # are the components of the signal key ("file::name"). test_classname
+        # is only populated when a single classname was observed for this
+        # test_id; it is None when the same file::name spans multiple classes
+        # (ambiguous — better omitted than guessed).
+        self.test_file = test_file
+        self.test_classname = test_classname
+        self.test_name = test_name
 
     def replace(self, **changes) -> "Signal":
         """Return a copy with selected fields replaced (`dataclasses.replace`-style).
@@ -407,6 +420,9 @@ def replace(self, **changes) -> "Signal":
             "job_base_name": changes.pop("job_base_name", self.job_base_name),
             "test_module": changes.pop("test_module", self.test_module),
             "source": changes.pop("source", self.source),
+            "test_file": changes.pop("test_file", self.test_file),
+            "test_classname": changes.pop("test_classname", self.test_classname),
+            "test_name": changes.pop("test_name", self.test_name),
         }
         if changes:
             raise TypeError(

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 (
@@ -863,15 +864,27 @@ def _partition_label(sha: str) -> str:
                 }
             )
 
-        payload = {
+        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,
-            "commits": commits_json,
         }
+        # 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
         if dispatch_advisor.is_born_red:
             payload["pattern_context"] = _BORN_RED_PATTERN_CONTEXT
         return json.dumps(payload)

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

@@ -351,6 +351,13 @@ def _build_test_signals(
         failing_tests_by_job_base_name: Set[
             Tuple[WorkflowName, JobBaseName, TestId]
         ] = set()
+        # Capture structured test identity 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, so one test_id may span
+        # multiple classes; collect all distinct non-empty classnames and
+        # only surface one later if it is unambiguous.
+        test_file_name_by_test_id: Dict[TestId, Tuple[str, str]] = {}
+        test_classnames_by_test_id: Dict[TestId, Set[str]] = {}
         for tr in test_rows:
             job = jobs_by_id.get(tr.job_id)
             job_base_name = job.base_name
@@ -377,6 +384,11 @@ def _build_test_signals(
                 outcome = existing
 
             tests_by_group_attempt[key] = outcome
+            test_file_name_by_test_id.setdefault(tr.test_id, (tr.file, tr.name))
+            if tr.classname:
+                test_classnames_by_test_id.setdefault(tr.test_id, set()).add(
+                    tr.classname
+                )
 
             # Track keys that have at least one persistent failure (no retry success)
             if outcome.failure_runs > 0 and outcome.success_runs == 0:
@@ -490,6 +502,13 @@ def _build_test_signals(
                     test_module = test_id.split("::")[0].replace(".py", "")
                 else:
                     test_module = None
+                test_file, test_name = test_file_name_by_test_id.get(test_id, ("", ""))
+                # Classname is only trustworthy when a single distinct value
+                # was seen for this test_id; otherwise omit rather than guess.
+                classnames = test_classnames_by_test_id.get(test_id, set())
+                test_classname = (
+                    next(iter(classnames)) if len(classnames) == 1 else None
+                )
 
                 signals.append(
                     Signal(
@@ -499,6 +518,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.py

@@ -1129,6 +1129,9 @@ def setUp(self) -> None:
         "job_base_name": "linux-jammy / test",
         "test_module": "test_foo",
         "source": SignalSource.JOB,
+        "test_file": "test/foo.py",
+        "test_classname": "TestFooBar",
+        "test_name": "test_bar",
     }
 
     def _init_params(self):

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

@@ -1100,6 +1100,130 @@ 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."""
@@ -1404,6 +1528,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)
+
 
 class TestGroupActionsTestsToInclude(unittest.TestCase):
     """Coalescing rules around `tests_to_include` in `group_actions`.