fix(scheduler): a task marked not to run on startup waits an interval after boot (#5431)

_DueTracker counted a task that had never run from 0 on the loop clock,
which is host uptime, so every run_on_startup=False task (hourly data
cleanup with the Docker log prune, the weekly report check, alert
reconcile, agent age, Ansible schedules) fired on the first tick after
each deploy, while agents were flushing their backlog. Agent ingest was
slow or dropped for a few minutes after every deploy.

start_scheduler now calls tracker.begin(): such a task is first due one
interval after the loop starts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/core/scheduler.py

@@ -42,6 +42,16 @@ class _DueTracker:
             ready.append(task)
         return ready
 
+    def begin(self, tasks: list[ScheduledTask], now: float) -> None:
+        """Start the clock: a task that doesn't run on startup is first due
+        one interval from now. Without this a task that had never run counted
+        from 0 on the loop's clock (host uptime), so every one of them fired
+        on the first tick after a deploy, the hourly cleanup with its Docker
+        log prune among them, holding up agent ingest for minutes (#5431)."""
+        for task in tasks:
+            if not task.run_on_startup:
+                self.last_run.setdefault(task.name, now)
+
     def mark_started(self, task: ScheduledTask, now: float) -> None:
         self.in_flight.add(task.name)
         self.last_run[task.name] = now
@@ -58,10 +68,12 @@ async def start_scheduler(tasks: list[ScheduledTask]) -> None:
         tracker.mark_started(task, now)
         asyncio.create_task(_run_task(task, tracker))
 
+    started = asyncio.get_event_loop().time()
+    tracker.begin(tasks, started)
     for task in tasks:
         if task.run_on_startup:
             logger.info(f"Startup task: {task.name}")
-            _spawn(task, asyncio.get_event_loop().time())
+            _spawn(task, started)
 
     while True:
         now = asyncio.get_event_loop().time()

tests/core/test_scheduler_overlap.py

@@ -56,3 +56,20 @@ def test_independent_tasks_do_not_block_each_other():
     tr = _DueTracker()
     tr.mark_started(a, now=0)             # a hangs
     assert tr.due([a, b], now=10) == [b]  # b still fires
+
+
+def test_a_task_not_run_on_startup_waits_one_interval_after_boot():
+    """#5431: the loop's clock is host uptime, far past every interval. A task
+    marked not to run on startup used to count from 0 and fire on the first
+    tick anyway; it now waits a whole interval from when the loop began."""
+    boot = 1_000_000.0
+    hourly = _task("cleanup", 3600)
+    eager = ScheduledTask(name="poll", coro_factory=lambda: None, interval_seconds=60,
+                          run_on_startup=True)
+    tr = _DueTracker()
+    tr.begin([hourly, eager], now=boot)
+    assert tr.due([hourly], now=boot + 1) == []
+    assert tr.due([hourly], now=boot + 3599) == []
+    assert tr.due([hourly], now=boot + 3600) == [hourly]
+    assert "poll" not in tr.last_run      # started by start_scheduler itself
+