[autorevert] non nullable dates & dedup (#7167)

**Signal event deduplication and timestamp handling:**

* Added a deduplication step in `SignalExtractor` to remove duplicate
signal events within commits, based on identical `(started_at,
wf_run_id)` pairs. This addresses issues with "rerun failed" jobs in
GitHub workflows that reuse the same underlying job (but reports them
with different job ids)

* For test-track signals, extract start_date from the specific job that
hosted the test (when available)

* Changed all job and signal timestamp fields (`started_at`,
`created_at`) to be non-optional and default

Author: izaitsevfb

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

@@ -49,7 +49,7 @@ class JobMeta:
       job signals accounted for by per-test signals.
     """
 
-    started_at: Optional[datetime] = None
+    started_at: datetime = datetime.min
     is_pending: bool = False
     is_cancelled: bool = False
     has_failures: bool = False
@@ -164,9 +164,7 @@ def stats(self, key: KeyT) -> JobMeta:
         jrows = self._groups[key]
 
         # Inline aggregations (only used here)
-        started_at: Optional[datetime]
-        times = [r.started_at for r in jrows if r.started_at is not None]
-        started_at = min(times) if times else None
+        started_at = min(r.started_at for r in jrows)
 
         meta = JobMeta(
             started_at=started_at,

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

@@ -71,12 +71,14 @@ def __init__(
         name: str,
         status: SignalStatus,
         started_at: datetime,
+        wf_run_id: int,
         ended_at: Optional[datetime] = None,
     ):
         self.name = name
         self.status = status
         self.started_at = started_at
         self.ended_at = ended_at
+        self.wf_run_id = wf_run_id
 
     @property
     def is_pending(self) -> bool:
@@ -96,8 +98,10 @@ class SignalCommit:
 
     def __init__(self, head_sha: str, events: List[SignalEvent]):
         self.head_sha = head_sha
-        # enforce events ordered by time, oldest first
-        self.events = sorted(events, key=lambda e: e.started_at) if events else []
+        # enforce events ordered by time, then by wf_run_id (oldest first)
+        self.events = (
+            sorted(events, key=lambda e: (e.started_at, e.wf_run_id)) if events else []
+        )
         # counts by status
         self.statuses = {}
         for e in self.events:

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

@@ -9,7 +9,7 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Dict, Iterable, List, Set, Tuple
+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, SignalStatus
@@ -31,6 +31,7 @@
 class TestOutcome:
     failing: bool
     errored: bool
+    started_at: datetime
 
 
 class SignalExtractor:
@@ -77,7 +78,34 @@ def extract(self) -> List[Signal]:
 
         test_signals = self._build_test_signals(jobs, test_rows)
         job_signals = self._build_non_test_signals(jobs)
-        return test_signals + job_signals
+        # Deduplicate events within commits across all signals as a final step
+        # GitHub-specific behavior like "rerun failed" can reuse job instances for reruns.
+        # When that happens, the jobs have identical timestamps by DIFFERENT job ids.
+        # But since they are still the same job logically, we want to deduplicate them
+        # for the purposes of signal events.
+        return self._dedup_signal_events(test_signals + job_signals)
+
+    # -----------------------------
+    # Deduplication (GitHub-specific)
+    # -----------------------------
+    def _dedup_signal_events(self, signals: List[Signal]) -> List[Signal]:
+        deduped: List[Signal] = []
+        for s in signals:
+            new_commits: List[SignalCommit] = []
+            for c in s.commits:
+                filtered: List[SignalEvent] = []
+                prev_key: Optional[Tuple[datetime, int]] = None
+                for e in c.events:  # already sorted by (started_at, wf_run_id)
+                    key = (e.started_at, e.wf_run_id)
+                    if key == prev_key:
+                        continue
+                    filtered.append(e)
+                    prev_key = key
+                new_commits.append(SignalCommit(head_sha=c.head_sha, events=filtered))
+            deduped.append(
+                Signal(key=s.key, workflow_name=s.workflow_name, commits=new_commits)
+            )
+        return deduped
 
     # -----------------------------
     # Phase B — Tests (test_run_s3 only)
@@ -174,9 +202,13 @@ def _build_test_signals(
                 tr.test_id,
             )
             prev = tests_by_group_attempt.get(key)
+            started_at = min(
+                (prev.started_at if prev else job.started_at), job.started_at
+            )
             outcome = TestOutcome(
                 failing=(prev.failing if prev else False) or bool(tr.failing),
                 errored=(prev.errored if prev else False) or bool(tr.errored),
+                started_at=started_at,
             )
             tests_by_group_attempt[key] = outcome
             if outcome.failing or outcome.errored:
@@ -226,8 +258,8 @@ def _build_test_signals(
                             wf_run_id=wf_run_id,
                             run_attempt=run_attempt,
                         ),
-                        "started_at": meta.started_at or datetime.min,
                         "ended_at": None,
+                        "wf_run_id": int(wf_run_id),
                     }
 
                     if verdict:
@@ -236,12 +268,17 @@ def _build_test_signals(
                                 status=SignalStatus.FAILURE
                                 if (verdict.failing or verdict.errored)
                                 else SignalStatus.SUCCESS,
+                                started_at=verdict.started_at,
                                 **event_common,
                             )
                         )
                     elif meta.is_pending:
                         events.append(
-                            SignalEvent(status=SignalStatus.PENDING, **event_common)
+                            SignalEvent(
+                                status=SignalStatus.PENDING,
+                                started_at=meta.started_at,
+                                **event_common,
+                            )
                         )
                     # else: missing (no event)
 
@@ -330,8 +367,9 @@ def _build_non_test_signals(self, jobs: List[JobRow]) -> List[Signal]:
                                 run_attempt=run_attempt,
                             ),
                             status=ev_status,
-                            started_at=meta.started_at or datetime.min,
+                            started_at=meta.started_at,
                             ended_at=None,
+                            wf_run_id=int(wf_run_id),
                         )
                     )
 

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

@@ -105,6 +105,10 @@ def fetch_jobs_for_workflows(
             created_at,
             rule,
         ) in res.result_rows:
+            # Guard against placeholder started_at by using the later of
+            # started_at and created_at as the effective start.
+            # Both columns are non-NULL in ClickHouse.
+            effective_started = max(started_at, created_at)
             rows.append(
                 JobRow(
                     head_sha=Sha(head_sha),
@@ -115,7 +119,7 @@ def fetch_jobs_for_workflows(
                     name=JobName(str(name or "")),
                     status=str(status or ""),
                     conclusion=str(conclusion or ""),
-                    started_at=started_at,
+                    started_at=effective_started,
                     created_at=created_at,
                     rule=str(rule or ""),
                 )

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

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from functools import cached_property
-from typing import NewType, Optional, Set
+from typing import NewType, Set
 
 
 # Default classification rules that indicate test failures.
@@ -35,8 +35,8 @@ class JobRow:
     name: JobName
     status: str
     conclusion: str
-    started_at: Optional[datetime]
-    created_at: Optional[datetime]
+    started_at: datetime
+    created_at: datetime
     rule: str
 
     @cached_property

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

@@ -23,7 +23,12 @@ def setUp(self) -> None:
         self.t0 = datetime(2025, 8, 19, 12, 0, 0)
 
     def _ev(self, name: str, status: SignalStatus, minute: int) -> SignalEvent:
-        return SignalEvent(name=name, status=status, started_at=ts(self.t0, minute))
+        return SignalEvent(
+            name=name,
+            status=status,
+            started_at=ts(self.t0, minute),
+            wf_run_id=1,
+        )
 
     def test_detect_recovered_first_non_pending_success(self):
         # Newest commit has success (even with pending present) -> recovered

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

@@ -0,0 +1,92 @@
+import unittest
+from datetime import datetime, timedelta
+
+from pytorch_auto_revert.signal import Signal, SignalCommit, SignalEvent, SignalStatus
+from pytorch_auto_revert.signal_extraction import SignalExtractor
+
+
+def ts(base: datetime, minutes: int) -> datetime:
+    return base + timedelta(minutes=minutes)
+
+
+class TestSignalDedup(unittest.TestCase):
+    def setUp(self) -> None:
+        self.t0 = datetime(2025, 8, 20, 12, 0, 0)
+
+    def test_dedup_removes_adjacent_duplicates(self):
+        # Two events with identical (started_at, wf_run_id) within a single commit
+        e1 = SignalEvent(
+            name="job-a",
+            status=SignalStatus.FAILURE,
+            started_at=ts(self.t0, 1),
+            wf_run_id=100,
+        )
+        e2 = SignalEvent(
+            name="job-b",
+            status=SignalStatus.SUCCESS,
+            started_at=ts(self.t0, 1),
+            wf_run_id=100,
+        )
+        commit = SignalCommit(head_sha="sha", events=[e1, e2])
+        s = Signal(key="k", workflow_name="wf", commits=[commit])
+
+        ex = SignalExtractor(workflows=["wf"], lookback_hours=24)
+        out = ex._dedup_signal_events([s])
+        self.assertEqual(len(out), 1)
+        self.assertEqual(len(out[0].commits[0].events), 1)
+        # keeps the first encountered event for that pair
+        self.assertEqual(out[0].commits[0].events[0].name, "job-a")
+        self.assertEqual(out[0].commits[0].events[0].status, SignalStatus.FAILURE)
+
+    def test_dedup_keeps_non_duplicates(self):
+        e1 = SignalEvent(
+            name="job-a",
+            status=SignalStatus.FAILURE,
+            started_at=ts(self.t0, 1),
+            wf_run_id=1,
+        )
+        e2 = SignalEvent(
+            name="job-b",
+            status=SignalStatus.SUCCESS,
+            started_at=ts(self.t0, 1),
+            wf_run_id=2,
+        )
+        e3 = SignalEvent(
+            name="job-c",
+            status=SignalStatus.PENDING,
+            started_at=ts(self.t0, 2),
+            wf_run_id=1,
+        )
+        commit = SignalCommit(head_sha="sha", events=[e1, e2, e3])
+        s = Signal(key="k", workflow_name="wf", commits=[commit])
+
+        ex = SignalExtractor(workflows=["wf"], lookback_hours=24)
+        out = ex._dedup_signal_events([s])
+        self.assertEqual(len(out[0].commits[0].events), 3)
+
+    def test_dedup_applies_per_commit(self):
+        # Duplicates in different commits are not cross-deduped
+        e1 = SignalEvent(
+            name="job-a",
+            status=SignalStatus.FAILURE,
+            started_at=ts(self.t0, 1),
+            wf_run_id=100,
+        )
+        e2 = SignalEvent(
+            name="job-b",
+            status=SignalStatus.SUCCESS,
+            started_at=ts(self.t0, 1),
+            wf_run_id=100,
+        )
+        c1 = SignalCommit(head_sha="sha1", events=[e1, e2])
+        c2 = SignalCommit(head_sha="sha2", events=[e1, e2])
+        s = Signal(key="k", workflow_name="wf", commits=[c1, c2])
+
+        ex = SignalExtractor(workflows=["wf"], lookback_hours=24)
+        out = ex._dedup_signal_events([s])
+        # Both commits should each have one event after dedup
+        self.assertEqual([len(c.events) for c in out[0].commits], [1, 1])
+
+
+if __name__ == "__main__":
+    unittest.main()