diff --git a/services/scheduler/service.py b/services/scheduler/service.py
index 1b8d9d5da..f142e505d 100644
--- a/services/scheduler/service.py
+++ b/services/scheduler/service.py
@@ -955,9 +955,13 @@ def drain_webhook_outbox(self) -> None:
         """
         try:
             counter = self.notifier.retry_outbox_once()
-        except Exception as exc:  # pragma: no cover — telemetry must not crash
+        except Exception as exc:
+            # #2697 convention (SEL-516): log, then RE-RAISE so APScheduler
+            # fires EVENT_JOB_ERROR and the #1144 listener records the
+            # failure in scheduler_job_runs; returning recorded success
+            # while the outbox was not draining.
             logger.exception("webhook_outbox_drain_failed", error=str(exc))
-            return
+            raise
         if counter["attempted"]:
             logger.info("webhook_outbox_drain_tick", **counter)
 
diff --git a/services/scheduler/tests/test_qa_scheduler_gaps.py b/services/scheduler/tests/test_qa_scheduler_gaps.py
index 941293036..8f0645fb1 100644
--- a/services/scheduler/tests/test_qa_scheduler_gaps.py
+++ b/services/scheduler/tests/test_qa_scheduler_gaps.py
@@ -227,16 +227,18 @@ def test_stops_at_first_failure(self):
 
 
 class TestDrainWebhookOutboxContract:
-    """The drain tick is telemetry: it must never crash the scheduler."""
+    """#2697 / SEL-516: a failing drain must surface so it is recorded as failed."""
 
-    def test_notifier_exception_does_not_propagate(self):
+    def test_notifier_exception_is_reraised(self):
         from services.scheduler.service import SchedulerService
 
         service = SchedulerService.__new__(SchedulerService)
         service.notifier = MagicMock()
         service.notifier.retry_outbox_once.side_effect = RuntimeError("disk gone")
 
-        assert service.drain_webhook_outbox() is None
+        # Swallowing here made APScheduler record a phantom success.
+        with pytest.raises(RuntimeError, match="disk gone"):
+            service.drain_webhook_outbox()
 
     def test_quiet_drain_stays_quiet(self):
         from services.scheduler.service import SchedulerService