diff --git a/backend/scheduler.py b/backend/scheduler.py
index 15bd979..edc4af7 100644
--- a/backend/scheduler.py
+++ b/backend/scheduler.py
@@ -2,7 +2,7 @@
 
 import asyncio
 import logging
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timezone
 
 from croniter import croniter
 
@@ -10,6 +10,13 @@
 
 _scheduler_task: asyncio.Task | None = None
 
+# AUDIT C2: (schedule_id, cron_expression) pairs we've already warned about,
+# so a permanently-malformed cron doesn't spam a warning every tick. Keyed on
+# the expression too, so editing the schedule to a new (still-bad) value
+# warns again instead of staying silent forever. Process-lifetime cache —
+# intentionally not cleared on scheduler stop/start within the same process.
+_warned_bad_cron: set[tuple[str, str]] = set()
+
 
 async def _get_enabled_schedules() -> list[dict]:
     from sqlalchemy import select
@@ -29,39 +36,115 @@ async def _get_enabled_schedules() -> list[dict]:
                 "source_id": sched.source_id,
                 "cron_expression": sched.cron_expression,
                 "parameters": sched.parameters,
+                "name": sched.name,
             }
             for sched, _ in result.all()
         ]
 
 
-def _is_due(cron_expression: str, now: datetime) -> bool:
-    """Return True if cron fired within the last 60 seconds."""
+def _now() -> datetime:
+    """Thin seam over datetime.now so tests can drive the clock without sleeping."""
+    return datetime.now(timezone.utc)
+
+
+def _fires_in_window(
+    cron_expression: str,
+    schedule_id: str,
+    window_start: datetime,
+    window_end: datetime,
+    *,
+    name: str | None = None,
+) -> int:
+    """Count cron fire times in the half-open interval (window_start, window_end].
+
+    AUDIT C2: a cron_expression croniter can't parse used to be swallowed by
+    a bare `except Exception: return False` — the schedule went permanently
+    silent with zero log trace. Now it warns once per (schedule_id,
+    cron_expression) pair and is treated as "not due" (0 fires) so one bad
+    schedule can't crash the loop or take down the others.
+    """
     try:
-        base = now - timedelta(seconds=61)
-        cron = croniter(cron_expression, base)
+        cron = croniter(cron_expression, window_start)
+    except Exception as exc:
+        warn_key = (schedule_id, cron_expression)
+        if warn_key not in _warned_bad_cron:
+            _warned_bad_cron.add(warn_key)
+            logger.warning(
+                "schedule %s (%s) has an unparseable cron_expression %r; "
+                "skipping until fixed: %s",
+                schedule_id, name or "?", cron_expression, exc,
+            )
+        return 0
+
+    count = 0
+    while True:
         next_fire = cron.get_next(datetime)
-        return next_fire <= now
-    except Exception:
-        return False
+        if next_fire > window_end:
+            break
+        count += 1
+    return count
 
 
 async def _scheduler_loop() -> None:
     logger.info("Local scheduler started")
+    # AUDIT C4: the previous "due within the last 61s" check was a fixed
+    # window decoupled from actual tick cadence (sleep(60) + loop body
+    # time) — drift near the boundary could get one fire dispatched by two
+    # consecutive ticks, and a slow loop body (>1s) could silently miss a
+    # fire. A process-local watermark instead makes consecutive ticks cover
+    # disjoint, gapless (last_tick, now] windows: no fire time can ever fall
+    # in two windows, and a slow tick just widens its own window (catching
+    # up) instead of losing anything.
+    last_tick: datetime | None = None
     while True:
         try:
             await asyncio.sleep(60)
-            now = datetime.now(timezone.utc)
+            now = _now()
+
+            if last_tick is None:
+                # First tick after process start: establish the watermark
+                # without dispatching, so a restart never replays everything
+                # that fired while the process was down.
+                last_tick = now
+                continue
+
             schedules = await _get_enabled_schedules()
             from backend.executor import get_executor
             executor = get_executor()
             for sched in schedules:
-                if _is_due(sched["cron_expression"], now):
-                    logger.info("Firing schedule %s", sched["schedule_id"])
+                fire_count = _fires_in_window(
+                    sched["cron_expression"],
+                    sched["schedule_id"],
+                    last_tick,
+                    now,
+                    name=sched["name"],
+                )
+                if fire_count == 0:
+                    continue
+                if fire_count > 1:
+                    logger.debug(
+                        "schedule %s coalesced %d fire times into one dispatch",
+                        sched["schedule_id"], fire_count,
+                    )
+                logger.info("Firing schedule %s", sched["schedule_id"])
+                try:
                     await executor.dispatch_scheduled_collection(
                         sched["schedule_id"],
                         sched["source_id"],
                         sched["parameters"],
                     )
+                except Exception as exc:
+                    # AUDIT C4: one schedule's dispatch raising must not stop
+                    # the rest of this tick's schedules from being evaluated,
+                    # and must not stall last_tick below — otherwise the next
+                    # tick's re-widened window would re-dispatch schedules
+                    # that already fired successfully earlier in this same
+                    # tick, reopening the double-dispatch bug this fix closes.
+                    logger.warning(
+                        "schedule %s dispatch failed: %s", sched["schedule_id"], exc,
+                    )
+
+            last_tick = now
         except asyncio.CancelledError:
             break
         except Exception as exc:
diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py
index 3ccfee4..f1e7dd0 100644
--- a/tests/unit/test_scheduler.py
+++ b/tests/unit/test_scheduler.py
@@ -6,57 +6,118 @@
 
 import pytest
 
-from backend.scheduler import _is_due, start_scheduler, stop_scheduler
+from backend.scheduler import _fires_in_window, start_scheduler, stop_scheduler
 
 
-# ── _is_due tests ─────────────────────────────────────────────────────────────
+@pytest.fixture(autouse=True)
+def _reset_warned_bad_cron():
+    """AUDIT C2's warn-once cache is a process-lifetime module-level set by
+    design — reset it around every test so test execution order can't leak
+    warn state between cases (see triage doc P3-6 on shared process state)."""
+    import backend.scheduler as sched_module
+
+    sched_module._warned_bad_cron.clear()
+    yield
+    sched_module._warned_bad_cron.clear()
+
+
+# ── _fires_in_window tests ──────────────────────────────────────────────────────
+
+def test_fires_in_window_fires_within_window_counts_once():
+    """A fire time landing inside (window_start, window_end] counts."""
+    last_tick = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    now = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)
+    # "* * * * *" fires every minute; next fire after 12:00:00 is 12:01:00
+    assert _fires_in_window("* * * * *", "sched-1", last_tick, now) == 1
+
+
+def test_fires_in_window_open_left_edge_excludes_prior_fire():
+    """A fire time exactly at window_start (the exclusive edge) is not recounted."""
+    last_tick = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)  # itself a fire instant
+    now = datetime(2024, 6, 1, 12, 2, 0, tzinfo=timezone.utc)
+    assert _fires_in_window("* * * * *", "sched-1", last_tick, now) == 1  # only 12:02:00
+
 
-def test_is_due_every_minute_just_fired():
-    """Cron '* * * * *' fires every minute; should be due within last 60s."""
-    now = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
-    assert _is_due("* * * * *", now) is True
+def test_fires_in_window_consecutive_windows_never_double_count():
+    """Two back-to-back (last_tick, now] windows sharing a boundary each see
+    the boundary fire exactly once between them — this is the C4 guarantee."""
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)  # exact fire instant
+    t2 = datetime(2024, 6, 1, 12, 2, 0, tzinfo=timezone.utc)
 
+    window1 = _fires_in_window("* * * * *", "sched-1", t0, t1)
+    window2 = _fires_in_window("* * * * *", "sched-1", t1, t2)
 
-def test_is_due_specific_time_matches():
-    """Cron expression that exactly matches current minute should be due."""
-    # 12:05 UTC on June 1
-    now = datetime(2024, 6, 1, 12, 5, 30, tzinfo=timezone.utc)
-    # fires at 12:05 every day
-    assert _is_due("5 12 * * *", now) is True
+    assert window1 == 1  # counts the 12:01:00 fire
+    assert window2 == 1  # counts only 12:02:00 — 12:01:00 is not double-counted
 
 
-def test_is_due_future_time_not_due():
-    """Cron for next hour should not be due now."""
-    now = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
-    # fires at 13:00 — hasn't fired yet in the last 60s
-    assert _is_due("0 13 * * *", now) is False
+def test_fires_in_window_no_fire_returns_zero():
+    """No fire time in a short window returns 0."""
+    last_tick = datetime(2024, 6, 1, 12, 0, 10, tzinfo=timezone.utc)
+    now = datetime(2024, 6, 1, 12, 0, 40, tzinfo=timezone.utc)
+    assert _fires_in_window("0 * * * *", "sched-1", last_tick, now) == 0
 
 
-def test_is_due_invalid_cron_expression():
-    """Invalid cron expression should return False (no exception raised)."""
+def test_fires_in_window_future_cron_not_due():
+    """Cron for a later time has no fire in a window that ends before it."""
+    last_tick = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    now = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)
+    assert _fires_in_window("0 13 * * *", "sched-1", last_tick, now) == 0
+
+
+def test_fires_in_window_slow_tick_coalesces_missed_fires():
+    """A window spanning 3 missed fire times reports count=3 so the caller
+    can dispatch once instead of once per missed fire."""
+    last_tick = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    now = datetime(2024, 6, 1, 12, 3, 0, tzinfo=timezone.utc)
+    assert _fires_in_window("* * * * *", "sched-1", last_tick, now) == 3
+
+
+def test_fires_in_window_malformed_cron_returns_zero_and_warns_once():
+    """An unparseable cron_expression returns 0 (treated as not-due) and logs
+    exactly one warning across repeated calls with the same
+    (schedule_id, cron_expression) pair."""
     now = datetime.now(timezone.utc)
-    assert _is_due("not-a-cron", now) is False
+    last_tick = now - timedelta(seconds=60)
 
+    with patch("backend.scheduler.logger") as mock_logger:
+        first = _fires_in_window("not-a-cron", "sched-bad", last_tick, now, name="Bad Sched")
+        second = _fires_in_window("not-a-cron", "sched-bad", last_tick, now, name="Bad Sched")
 
-def test_is_due_invalid_cron_too_many_fields():
-    """Malformed cron with extra fields returns False."""
+    assert first == 0
+    assert second == 0
+    mock_logger.warning.assert_called_once()
+    args = mock_logger.warning.call_args.args
+    assert args[1] == "sched-bad"
+    assert args[3] == "not-a-cron"
+
+
+def test_fires_in_window_rewarns_when_expression_changes():
+    """Editing a bad schedule to a different (still-bad) expression re-warns,
+    while repeating the same bad expression does not spam."""
     now = datetime.now(timezone.utc)
-    assert _is_due("* * * * * * *extra", now) is False
+    last_tick = now - timedelta(seconds=60)
+
+    with patch("backend.scheduler.logger") as mock_logger:
+        _fires_in_window("bad-one", "sched-bad", last_tick, now)
+        _fires_in_window("bad-one", "sched-bad", last_tick, now)  # same key: no rewarn
+        _fires_in_window("bad-two", "sched-bad", last_tick, now)  # edited: rewarn
+
+    assert mock_logger.warning.call_count == 2
 
 
-def test_is_due_every_minute_naive_datetime():
-    """Works with naive datetime (no tzinfo)."""
-    now = datetime(2024, 6, 1, 12, 0, 30)
-    # Should not raise; may return True or False but not crash
-    result = _is_due("* * * * *", now)
-    assert isinstance(result, bool)
+def test_fires_in_window_independent_schedules_warn_independently():
+    """Two different schedules sharing the same broken expression each get
+    their own warning (the cache key includes schedule_id)."""
+    now = datetime.now(timezone.utc)
+    last_tick = now - timedelta(seconds=60)
 
+    with patch("backend.scheduler.logger") as mock_logger:
+        _fires_in_window("not-a-cron", "sched-a", last_tick, now)
+        _fires_in_window("not-a-cron", "sched-b", last_tick, now)
 
-def test_is_due_yearly_cron_not_due():
-    """Yearly cron should not be due at an arbitrary moment."""
-    now = datetime(2024, 6, 15, 8, 0, 0, tzinfo=timezone.utc)
-    # fires at midnight Jan 1 only
-    assert _is_due("0 0 1 1 *", now) is False
+    assert mock_logger.warning.call_count == 2
 
 
 # ── start_scheduler / stop_scheduler tests ────────────────────────────────────
@@ -143,6 +204,7 @@ async def test_get_enabled_schedules_returns_list():
     mock_sched.source_id = "src-1"
     mock_sched.cron_expression = "*/5 * * * *"
     mock_sched.parameters = {"limit": 10}
+    mock_sched.name = "Test Schedule"
 
     mock_source = MagicMock()
 
@@ -164,20 +226,55 @@ async def test_get_enabled_schedules_returns_list():
     assert result[0]["source_id"] == "src-1"
     assert result[0]["cron_expression"] == "*/5 * * * *"
     assert result[0]["parameters"] == {"limit": 10}
+    assert result[0]["name"] == "Test Schedule"
 
 
 # ── _scheduler_loop tests ──────────────────────────────────────────────────────
 
 @pytest.mark.asyncio
-async def test_scheduler_loop_fires_due_schedule():
-    """_scheduler_loop fires dispatch_scheduled_collection for due schedules."""
+async def test_scheduler_loop_first_tick_establishes_watermark_no_dispatch():
+    """First tick after process start only establishes last_tick; it must not
+    fetch schedules or dispatch anything (no catch-up storm on restart)."""
     from backend.scheduler import _scheduler_loop
 
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+
+    iteration = 0
+
+    async def mock_sleep(seconds):
+        nonlocal iteration
+        iteration += 1
+        if iteration >= 2:
+            raise asyncio.CancelledError()
+
+    mock_get_schedules = AsyncMock(return_value=[])
+
+    with (
+        patch("asyncio.sleep", side_effect=mock_sleep),
+        patch("backend.scheduler._now", side_effect=[t0]),
+        patch("backend.scheduler._get_enabled_schedules", mock_get_schedules),
+        patch("backend.executor.get_executor", return_value=AsyncMock()),
+    ):
+        await _scheduler_loop()
+
+    mock_get_schedules.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_scheduler_loop_dispatches_once_when_fire_in_window():
+    """Once last_tick is established, a schedule with a fire time in
+    (last_tick, now] is dispatched exactly once."""
+    from backend.scheduler import _scheduler_loop
+
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)
+
     due_schedule = {
         "schedule_id": "sched-fire",
         "source_id": "src-1",
-        "cron_expression": "* * * * *",  # always due
+        "cron_expression": "* * * * *",
         "parameters": {},
+        "name": "every minute",
     }
 
     mock_executor = AsyncMock()
@@ -188,13 +285,14 @@ async def test_scheduler_loop_fires_due_schedule():
     async def mock_sleep(seconds):
         nonlocal iteration
         iteration += 1
-        if iteration >= 2:
-            # After first loop iteration, cancel
+        if iteration >= 3:
             raise asyncio.CancelledError()
 
     with (
         patch("asyncio.sleep", side_effect=mock_sleep),
+        patch("backend.scheduler._now", side_effect=[t0, t1]),
         patch("backend.scheduler._get_enabled_schedules", new=AsyncMock(return_value=[due_schedule])),
+        patch("backend.scheduler._fires_in_window", return_value=1) as mock_fires,
         patch("backend.executor.get_executor", return_value=mock_executor),
     ):
         await _scheduler_loop()
@@ -202,18 +300,25 @@ async def mock_sleep(seconds):
     mock_executor.dispatch_scheduled_collection.assert_called_once_with(
         "sched-fire", "src-1", {}
     )
+    mock_fires.assert_called_once_with(
+        "* * * * *", "sched-fire", t0, t1, name="every minute"
+    )
 
 
 @pytest.mark.asyncio
-async def test_scheduler_loop_skips_non_due_schedule():
-    """_scheduler_loop does not fire schedules that aren't due."""
+async def test_scheduler_loop_skips_when_no_fire_in_window():
+    """_scheduler_loop does not dispatch schedules with zero fires in the window."""
     from backend.scheduler import _scheduler_loop
 
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)
+
     non_due_schedule = {
         "schedule_id": "sched-skip",
         "source_id": "src-1",
-        "cron_expression": "0 0 1 1 *",  # never due except Jan 1
+        "cron_expression": "0 0 1 1 *",
         "parameters": {},
+        "name": "yearly",
     }
 
     mock_executor = AsyncMock()
@@ -224,12 +329,14 @@ async def test_scheduler_loop_skips_non_due_schedule():
     async def mock_sleep(seconds):
         nonlocal iteration
         iteration += 1
-        if iteration >= 2:
+        if iteration >= 3:
             raise asyncio.CancelledError()
 
     with (
         patch("asyncio.sleep", side_effect=mock_sleep),
+        patch("backend.scheduler._now", side_effect=[t0, t1]),
         patch("backend.scheduler._get_enabled_schedules", new=AsyncMock(return_value=[non_due_schedule])),
+        patch("backend.scheduler._fires_in_window", return_value=0),
         patch("backend.executor.get_executor", return_value=mock_executor),
     ):
         await _scheduler_loop()
@@ -237,27 +344,188 @@ async def mock_sleep(seconds):
     mock_executor.dispatch_scheduled_collection.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_scheduler_loop_consecutive_ticks_use_disjoint_windows():
+    """last_tick advances to the previous tick's `now`, so tick N+1's window
+    starts exactly where tick N's window ended — proving two consecutive
+    ticks can never both dispatch for the same fire instant (AUDIT C4)."""
+    from backend.scheduler import _scheduler_loop
+
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)
+    t2 = datetime(2024, 6, 1, 12, 2, 0, tzinfo=timezone.utc)
+
+    sched = {
+        "schedule_id": "sched-1",
+        "source_id": "src-1",
+        "cron_expression": "* * * * *",
+        "parameters": {},
+        "name": "n",
+    }
+
+    iteration = 0
+
+    async def mock_sleep(seconds):
+        nonlocal iteration
+        iteration += 1
+        if iteration >= 4:
+            raise asyncio.CancelledError()
+
+    with (
+        patch("asyncio.sleep", side_effect=mock_sleep),
+        patch("backend.scheduler._now", side_effect=[t0, t1, t2]),
+        patch("backend.scheduler._get_enabled_schedules", new=AsyncMock(return_value=[sched])),
+        patch("backend.scheduler._fires_in_window", return_value=0) as mock_fires,
+        patch("backend.executor.get_executor", return_value=AsyncMock()),
+    ):
+        await _scheduler_loop()
+
+    assert mock_fires.call_args_list[0].args[2:4] == (t0, t1)
+    assert mock_fires.call_args_list[1].args[2:4] == (t1, t2)
+
+
+@pytest.mark.asyncio
+async def test_scheduler_loop_coalesces_multiple_fires_into_single_dispatch():
+    """A window spanning several missed fire times (fire_count > 1) still
+    dispatches exactly once, and logs the coalesced count at debug level."""
+    from backend.scheduler import _scheduler_loop
+
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 3, 0, tzinfo=timezone.utc)
+
+    sched = {
+        "schedule_id": "sched-1",
+        "source_id": "src-1",
+        "cron_expression": "* * * * *",
+        "parameters": {},
+        "name": "n",
+    }
+
+    mock_executor = AsyncMock()
+    mock_executor.dispatch_scheduled_collection = AsyncMock()
+
+    iteration = 0
+
+    async def mock_sleep(seconds):
+        nonlocal iteration
+        iteration += 1
+        if iteration >= 3:
+            raise asyncio.CancelledError()
+
+    with (
+        patch("asyncio.sleep", side_effect=mock_sleep),
+        patch("backend.scheduler._now", side_effect=[t0, t1]),
+        patch("backend.scheduler._get_enabled_schedules", new=AsyncMock(return_value=[sched])),
+        patch("backend.scheduler._fires_in_window", return_value=3),
+        patch("backend.executor.get_executor", return_value=mock_executor),
+        patch("backend.scheduler.logger") as mock_logger,
+    ):
+        await _scheduler_loop()
+
+    mock_executor.dispatch_scheduled_collection.assert_called_once()
+    mock_logger.debug.assert_called_once()
+    assert mock_logger.debug.call_args.args[1:] == ("sched-1", 3)
+
+
 @pytest.mark.asyncio
 async def test_scheduler_loop_handles_exception_and_continues():
-    """_scheduler_loop catches non-cancel exceptions and keeps running."""
+    """_scheduler_loop catches non-cancel exceptions and keeps running,
+    leaving last_tick untouched from before the failed iteration so the next
+    successful tick's window naturally widens to catch up."""
     from backend.scheduler import _scheduler_loop
 
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 5, 0, tzinfo=timezone.utc)
+
     iteration = 0
 
     async def mock_sleep(seconds):
         nonlocal iteration
         iteration += 1
-        if iteration == 1:
+        if iteration == 2:
             raise RuntimeError("transient error")
-        if iteration >= 3:
+        if iteration >= 4:
             raise asyncio.CancelledError()
 
+    mock_get_schedules = AsyncMock(return_value=[])
+
     with (
         patch("asyncio.sleep", side_effect=mock_sleep),
-        patch("backend.scheduler._get_enabled_schedules", new=AsyncMock(return_value=[])),
+        patch("backend.scheduler._now", side_effect=[t0, t1]),
+        patch("backend.scheduler._get_enabled_schedules", mock_get_schedules),
         patch("backend.executor.get_executor", return_value=AsyncMock()),
     ):
         await _scheduler_loop()
 
-    # Should have run 3 sleep calls (1 error + 2 normal + cancel)
-    assert iteration == 3
+    # 1: first tick (sets last_tick=t0) 2: RuntimeError (last_tick untouched)
+    # 3: recovers with now=t1, window (t0, t1] 4: cancels
+    assert iteration == 4
+    mock_get_schedules.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_scheduler_loop_dispatch_failure_does_not_stall_watermark():
+    """If one schedule's dispatch raises, (1) the other due schedule in the
+    same tick is still dispatched, and (2) last_tick still advances — so the
+    next tick's window does not re-widen and re-dispatch a schedule that
+    already fired successfully earlier in this same tick.
+
+    Uses the real _fires_in_window (not mocked): "1 12 * * *" fires exactly
+    once, at 12:01:00. If last_tick incorrectly stalled at t0=12:00:00 after
+    the failure, tick 3's window would be the stale, wider (12:00:00, 12:02:00]
+    instead of the correct (12:01:00, 12:02:00] — and would wrongly re-count
+    the 12:01:00 fire, re-dispatching sched-ok a second time.
+    """
+    from backend.scheduler import _scheduler_loop
+
+    t0 = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2024, 6, 1, 12, 1, 0, tzinfo=timezone.utc)  # "1 12 * * *" fires here
+    t2 = datetime(2024, 6, 1, 12, 2, 0, tzinfo=timezone.utc)
+
+    sched_ok = {
+        "schedule_id": "sched-ok",
+        "source_id": "src-ok",
+        "cron_expression": "1 12 * * *",
+        "parameters": {},
+        "name": "ok",
+    }
+    sched_fail = {
+        "schedule_id": "sched-fail",
+        "source_id": "src-fail",
+        "cron_expression": "1 12 * * *",
+        "parameters": {},
+        "name": "fail",
+    }
+
+    async def dispatch_side_effect(schedule_id, source_id, parameters):
+        if schedule_id == "sched-fail":
+            raise RuntimeError("boom")
+
+    mock_executor = AsyncMock()
+    mock_executor.dispatch_scheduled_collection = AsyncMock(side_effect=dispatch_side_effect)
+
+    iteration = 0
+
+    async def mock_sleep(seconds):
+        nonlocal iteration
+        iteration += 1
+        if iteration >= 4:
+            raise asyncio.CancelledError()
+
+    with (
+        patch("asyncio.sleep", side_effect=mock_sleep),
+        patch("backend.scheduler._now", side_effect=[t0, t1, t2]),
+        patch(
+            "backend.scheduler._get_enabled_schedules",
+            new=AsyncMock(return_value=[sched_ok, sched_fail]),
+        ),
+        patch("backend.executor.get_executor", return_value=mock_executor),
+    ):
+        await _scheduler_loop()
+
+    calls = mock_executor.dispatch_scheduled_collection.call_args_list
+    ok_calls = [c for c in calls if c.args[0] == "sched-ok"]
+    fail_calls = [c for c in calls if c.args[0] == "sched-fail"]
+
+    assert len(ok_calls) == 1, "sched-ok must not be re-dispatched after sched-fail's failure"
+    assert len(fail_calls) == 1