[CRCR] Add nightly/periodic callback handler (Phase 1) (#8302)

## Summary

Phase 1 of the CRCR nightly/periodic self-report implementation ([RFC
98](pytorch/rfcs#98)).

Adds a new code path in the callback Lambda for `nightly` and `periodic`
event types. Unlike PR/push callbacks which require a `DISPATCHED`
record in Redis and follow a two-step state machine (`in_progress` →
`completed`), nightly/periodic callbacks use a **single-callback
model**:

- Downstream repo self-triggers via cron
- Runs CI against a `pytorch/pytorch` SHA (from `nightly` branch or
`main`)
- Reports the final result in **one callback** — no `in_progress` step

**What's different from PR/push:**
- No Redis writes, no state machine, no zombie tracking
- Status must be `completed` (rejects `in_progress`)
- No upstream check runs (nightly results are informational only)
- Timing metrics (queue_time, execution_time) are `null` — no dispatch
record to measure against

**Changes:**
| File | Change |
|------|--------|
| `callback/callback_handler.py` | New `_handle_nightly_callback()` +
routing in `handle()` |
| `tests/test_callback_handler.py` | 8 new tests covering the nightly
path |

**Remaining phases:**
- Phase 2: Update callback action YAML with `delivery-id` / `event-type`
inputs
- Phase 3: Downstream cron workflow in `TorchedHat/pytorch-redhat-ci`
- Phase 4: ClickHouse query for nightly results
- Phase 5-6: HUD display

## Test plan
- [x] 8 unit tests: forward to HUD, skip Redis, no timing metrics,
periodic variant, reject `in_progress`, allowlist gating, rate limiting,
missing `delivery_id`
- [ ] CI passes on this PR

Author: subinz1

aws/lambda/cross_repo_ci_relay/callback/callback_handler.py

@@ -22,6 +22,23 @@
 
 logger = logging.getLogger(__name__)
 
+_NIGHTLY_EVENT_TYPES = frozenset({"nightly", "periodic"})
+
+
+def _build_trusted(
+    verified_repo: str,
+    repo_level: AllowlistLevel,
+    ci_metrics: dict | None = None,
+) -> dict:
+    """Build the trusted payload block forwarded to HUD."""
+    if ci_metrics is None:
+        ci_metrics = {"queue_time": None, "execution_time": None}
+    return {
+        "ci_metrics": ci_metrics,
+        "verified_repo": verified_repo,
+        "downstream_repo_level": repo_level.value,
+    }
+
 
 def _safe_delta(
     start_ts: float | None, end_ts: float | None, label: str
@@ -234,6 +251,42 @@ def _create_upstream_check_run(
         )
 
 
+def _handle_nightly_callback(
+    config: RelayConfig,
+    body: dict,
+    verified_repo: str,
+    repo_level: AllowlistLevel,
+) -> dict:
+    """Handle a nightly/periodic self-report callback.
+
+    Unlike PR/push callbacks, nightly/periodic have no prior dispatch record
+    and no state machine.  The downstream repo self-triggers via cron, runs CI
+    against a pytorch/pytorch SHA (from the nightly branch or main), and
+    reports the final result in a single callback.
+
+    No Redis writes, no zombie tracking, no upstream check runs.
+    """
+    delivery_id, status, *_ = _parse_callback_body(body)
+
+    if status != "completed":
+        raise HTTPException(
+            400,
+            f"nightly/periodic callbacks must have status 'completed', got {status!r}",
+        )
+
+    trusted = _build_trusted(verified_repo, repo_level)
+    untrusted = {"callback_payload": body}
+
+    forward_to_hud(config, trusted, untrusted)
+    logger.info(
+        "nightly callback forwarded delivery_id=%s repo=%s event_type=%s",
+        delivery_id,
+        verified_repo,
+        body.get("event_type", "unknown"),
+    )
+    return {"ok": True, "status": status}
+
+
 def handle(config: RelayConfig, body: dict, verified_repo: str) -> dict:
     """Forward a downstream callback to HUD.
 
@@ -245,7 +298,11 @@ def handle(config: RelayConfig, body: dict, verified_repo: str) -> dict:
     for allowlist / timing lookups, and surfaced to HUD as ``verified_repo``
     so HUD can trust it over anything self-reported in the body.
 
-    State machine ensures:
+    For nightly/periodic event types, the state machine is bypassed entirely
+    and the result is forwarded to HUD in a single callback (no Redis, no
+    in_progress step, no zombie tracking).
+
+    For PR/push events, the state machine ensures:
     - Callbacks without prior dispatch are rejected
     - Timestamps (started_at, completed_at) are recorded once only
     - Duplicate callbacks are handled gracefully
@@ -256,6 +313,15 @@ def handle(config: RelayConfig, body: dict, verified_repo: str) -> dict:
         return {"ok": True, "status": "ignored"}
     allowlist, repo_level = result
 
+    # NOTE: event_type is untrusted (comes from the callback body, not from the
+    # OIDC token). It only selects among safe code paths — the nightly path
+    # never grants additional capability (no check runs, no Redis writes, no
+    # state machine bypass for PR events).  HUD treats nightly rows as
+    # informational, attributed to the OIDC-verified repo.
+    event_type = body.get("event_type", "")
+    if event_type in _NIGHTLY_EVENT_TYPES:
+        return _handle_nightly_callback(config, body, verified_repo, repo_level)
+
     delivery_id, status, run_id, run_attempt, workflow_name, job_name = (
         _parse_callback_body(body)
     )
@@ -278,14 +344,7 @@ def handle(config: RelayConfig, body: dict, verified_repo: str) -> dict:
     payload = None
     if status == "in_progress":
         payload = {
-            "trusted": {
-                "ci_metrics": {
-                    "queue_time": None,
-                    "execution_time": None,
-                },
-                "verified_repo": verified_repo,
-                "downstream_repo_level": repo_level.value,
-            },
+            "trusted": _build_trusted(verified_repo, repo_level),
             "untrusted": {"callback_payload": body},
         }
 
@@ -360,11 +419,7 @@ def handle(config: RelayConfig, body: dict, verified_repo: str) -> dict:
             config, delivery_id, verified_repo, run_id, run_attempt, job_name=job_name
         )
 
-    trusted = {
-        "ci_metrics": ci_metrics,
-        "verified_repo": verified_repo,
-        "downstream_repo_level": repo_level.value,
-    }
+    trusted = _build_trusted(verified_repo, repo_level, ci_metrics)
     # downstream's payload is untrusted — provide it under the "callback_payload"
     # key so HUD receives it under the expected untrusted namespace.
     untrusted = {"callback_payload": body}

aws/lambda/cross_repo_ci_relay/tests/test_callback_handler.py

@@ -459,5 +459,127 @@ def test_check_run_update_failure_does_not_break_response(self):
         self.assertEqual(result, {"ok": True, "status": "completed"})
 
 
+class TestNightlyCallback(unittest.TestCase):
+    """Nightly/periodic callbacks bypass the state machine entirely."""
+
+    def setUp(self):
+        self.patcher_allowlist = patch("callback.callback_handler.load_allowlist")
+        self.mock_load = self.patcher_allowlist.start()
+        mock_map = MagicMock()
+        mock_map.get_repo_level.return_value = AllowlistLevel.L2
+        self.mock_load.return_value = mock_map
+
+        self.patcher_redis = patch("callback.callback_handler.redis_helper")
+        self.mock_redis = self.patcher_redis.start()
+
+        self.patcher_rate = patch("callback.callback_handler.check_rate_limit")
+        self.mock_rate = self.patcher_rate.start()
+        self.mock_rate.return_value = True
+
+        self.patcher_hud = patch("callback.callback_handler.forward_to_hud")
+        self.mock_hud = self.patcher_hud.start()
+
+    def tearDown(self):
+        self.patcher_allowlist.stop()
+        self.patcher_redis.stop()
+        self.patcher_rate.stop()
+        self.patcher_hud.stop()
+
+    def _nightly_body(
+        self, event_type="nightly", status="completed", conclusion="success"
+    ):
+        return {
+            "event_type": event_type,
+            "delivery_id": "abc123def456",
+            "payload": {},
+            "workflow": {
+                "status": status,
+                "conclusion": conclusion,
+                "name": "CRCR Nightly CI",
+                "url": "https://github.com/org/repo/actions/runs/12345",
+                "run_id": "12345",
+                "run_attempt": "1",
+                "job_name": "nightly-build",
+                "check_run_id": "67890",
+            },
+        }
+
+    def test_nightly_callback_forwards_to_hud(self):
+        body = self._nightly_body()
+        result = handle(_cfg(), body, verified_repo="org/repo")
+
+        self.assertEqual(result, {"ok": True, "status": "completed"})
+        self.mock_hud.assert_called_once()
+        _, trusted, untrusted = self.mock_hud.call_args[0]
+        self.assertEqual(trusted["verified_repo"], "org/repo")
+        self.assertIs(untrusted["callback_payload"], body)
+
+    def test_nightly_callback_skips_redis(self):
+        handle(_cfg(), self._nightly_body(), verified_repo="org/repo")
+
+        self.mock_redis.get_callback_state.assert_not_called()
+        self.mock_redis.set_callback_state.assert_not_called()
+        self.mock_redis.add_in_progress_tracker.assert_not_called()
+        self.mock_redis.remove_in_progress_tracker.assert_not_called()
+
+    def test_nightly_callback_has_no_timing_metrics(self):
+        handle(_cfg(), self._nightly_body(), verified_repo="org/repo")
+
+        _, trusted, _ = self.mock_hud.call_args[0]
+        self.assertIsNone(trusted["ci_metrics"]["queue_time"])
+        self.assertIsNone(trusted["ci_metrics"]["execution_time"])
+
+    def test_periodic_callback_also_works(self):
+        body = self._nightly_body(event_type="periodic")
+        result = handle(_cfg(), body, verified_repo="org/repo")
+
+        self.assertEqual(result, {"ok": True, "status": "completed"})
+        self.mock_hud.assert_called_once()
+
+    def test_nightly_rejects_in_progress_status(self):
+        body = self._nightly_body(status="in_progress")
+        with self.assertRaises(HTTPException) as ctx:
+            handle(_cfg(), body, verified_repo="org/repo")
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertIn("completed", str(ctx.exception.detail))
+
+    def test_nightly_not_in_allowlist_is_ignored(self):
+        mock_map = MagicMock()
+        mock_map.get_repo_level.return_value = None
+        self.mock_load.return_value = mock_map
+
+        result = handle(_cfg(), self._nightly_body(), verified_repo="unknown/repo")
+
+        self.assertEqual(result, {"ok": True, "status": "ignored"})
+        self.mock_hud.assert_not_called()
+
+    def test_nightly_rate_limited(self):
+        self.mock_rate.return_value = False
+
+        with self.assertRaises(HTTPException) as ctx:
+            handle(_cfg(), self._nightly_body(), verified_repo="org/repo")
+        self.assertEqual(ctx.exception.status_code, 429)
+
+    def test_nightly_missing_delivery_id_returns_400(self):
+        body = self._nightly_body()
+        del body["delivery_id"]
+
+        with self.assertRaises(HTTPException) as ctx:
+            handle(_cfg(), body, verified_repo="org/repo")
+        self.assertEqual(ctx.exception.status_code, 400)
+
+    def test_nightly_failure_conclusion_forwards(self):
+        body = self._nightly_body()
+        body["workflow"]["conclusion"] = "failure"
+        result = handle(_cfg(), body, verified_repo="org/repo")
+
+        self.assertEqual(result, {"ok": True, "status": "completed"})
+        self.mock_hud.assert_called_once()
+        _, trusted, untrusted = self.mock_hud.call_args[0]
+        self.assertEqual(
+            untrusted["callback_payload"]["workflow"]["conclusion"], "failure"
+        )
+
+
 if __name__ == "__main__":
     unittest.main()