fix(cron): avoid polling running jobs and rearm after completion

ms_agent/cron/scheduler.py

@@ -67,6 +67,10 @@ def stop(self) -> None:
     def running(self) -> bool:
         return self._running
 
+    def reschedule(self) -> None:
+        """Recompute the next wake after a job state or schedule changes."""
+        self._arm_timer()
+
     def _arm_timer(self) -> None:
         if self._timer_task and not self._timer_task.done():
             self._timer_task.cancel()
@@ -83,10 +87,11 @@ def _arm_timer(self) -> None:
         self._timer_task = asyncio.ensure_future(self._sleep_and_tick(delay_s))
 
     def _get_earliest_due_ms(self) -> Optional[int]:
-        """Find the earliest next_run_at among enabled, non-paused jobs."""
+        """Find the earliest next_run_at among enabled, dispatchable jobs."""
         earliest = None
         for job, state in self._repo.load_all_with_state():
-            if not job.enabled or state.status == 'paused':
+            if not job.enabled or state.status in ('paused', 'running',
+                                                   'completed'):
                 continue
             ms = _iso_to_ms(state.next_run_at)
             if ms is not None and (earliest is None or ms < earliest):

ms_agent/cron/service.py

@@ -290,6 +290,7 @@ async def _execute_job(self, job: CronJobSpec,
             result = await self._executor.execute(job, config)
 
         self._manager.record_result(job, result)
+        self._scheduler.reschedule()
 
         for cb in self.on_job_complete:
             try:
@@ -397,6 +398,7 @@ async def run_job_now(self, job_id: str) -> Optional[ExecutionResult]:
             result = await self._executor.execute(job, config)
 
         self._manager.record_result(job, result)
+        self._scheduler.reschedule()
 
         for cb in self.on_job_complete:
             try:

tests/cron/test_scheduler.py

@@ -169,3 +169,66 @@ async def on_due(jobs):
         scheduler = AsyncScheduler(repo, on_due=on_due, tick_interval=60)
         count = await scheduler.manual_tick()
         assert count == 0
+
+    @pytest.mark.parametrize('status', ['paused', 'running', 'completed'])
+    def test_non_dispatchable_job_does_not_control_next_wake(self, repo, status):
+        past = '2025-01-01T00:00:00+00:00'
+        future = '2099-01-01T00:00:00+00:00'
+        repo.save_job_and_state(
+            CronJobSpec(id='inactive', prompt='test'),
+            CronJobState(status=status, next_run_at=past),
+        )
+        scheduler = AsyncScheduler(repo, on_due=None, tick_interval=60)
+        assert scheduler._get_earliest_due_ms() is None
+
+        repo.save_job_and_state(
+            CronJobSpec(id='future', prompt='test'),
+            CronJobState(status='scheduled', next_run_at=future),
+        )
+        assert scheduler._get_earliest_due_ms() == _iso_to_ms(future)
+
+    @pytest.mark.parametrize('status', ['scheduled', 'error'])
+    def test_dispatchable_job_controls_next_wake(self, repo, status):
+        past = '2025-01-01T00:00:00+00:00'
+        repo.save_job_and_state(
+            CronJobSpec(id='dispatchable', prompt='test'),
+            CronJobState(status=status, next_run_at=past),
+        )
+        scheduler = AsyncScheduler(repo, on_due=None, tick_interval=60)
+        assert scheduler._get_earliest_due_ms() == _iso_to_ms(past)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('status', ['running', 'completed'])
+    async def test_non_dispatchable_job_keeps_normal_poll_interval(
+            self, repo, monkeypatch, status):
+        repo.save_job_and_state(
+            CronJobSpec(id='inactive', prompt='test'),
+            CronJobState(status=status,
+                         next_run_at='2025-01-01T00:00:00+00:00'),
+        )
+        scheduler = AsyncScheduler(repo, on_due=None, tick_interval=60)
+        delays = []
+
+        async def record_delay(delay):
+            delays.append(delay)
+
+        monkeypatch.setattr(scheduler, '_sleep_and_tick', record_delay)
+        await scheduler.start()
+        try:
+            await scheduler._timer_task
+        finally:
+            scheduler.stop()
+        assert delays == [60]
+
+    @pytest.mark.parametrize('status', ['scheduled', 'error'])
+    def test_recovered_job_is_visible_on_next_reload(self, repo, status):
+        job = CronJobSpec(id='recovered', prompt='test')
+        state = CronJobState(status='running',
+                             next_run_at='2025-01-01T00:00:00+00:00')
+        repo.save_job_and_state(job, state)
+        scheduler = AsyncScheduler(repo, on_due=None)
+        assert scheduler._get_earliest_due_ms() is None
+        state.status = status
+        repo.save_state(job.id, state)
+        assert scheduler._get_earliest_due_ms() == _iso_to_ms(state.next_run_at)
+        assert [j.id for j, _ in scheduler._collect_due_jobs()] == [job.id]

tests/cron/test_service.py

@@ -145,3 +145,75 @@ async def on_complete(j, r):
 
         assert len(completed) == 1
         assert completed[0] == (job.id, True)
+
+
+class TestCronServiceSchedulerWake:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('manual', [False, True])
+    async def test_recurring_completion_rearms_before_poll_deadline(
+            self, workspace, monkeypatch, manual):
+        from datetime import datetime, timezone
+        from ms_agent.cron import parser, scheduler
+
+        now = datetime(2099, 1, 1, tzinfo=timezone.utc)
+
+        class FrozenDatetime(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return now.astimezone(tz)
+
+        monkeypatch.setattr(parser, 'datetime', FrozenDatetime)
+        monkeypatch.setattr(scheduler, '_now_ms',
+                            lambda: int(now.timestamp() * 1000))
+        service = CronService(workspace=workspace)
+        job = service.create_job(schedule_str='every 2s', prompt='test')
+        state = service.manager.repo.load_state(job.id)
+        state.next_run_at = '2098-12-31T23:59:59+00:00'
+        state.status = 'running'
+        service.manager.repo.save_state(job.id, state)
+        other = service.create_job(schedule_str='every 2s', prompt='still running')
+        other_state = service.manager.repo.load_state(other.id)
+        other_state.status = 'running'
+        other_state.next_run_at = state.next_run_at
+        service.manager.repo.save_state(other.id, other_state)
+        delays = []
+
+        async def record_delay(delay):
+            delays.append(delay)
+            await asyncio.Event().wait()
+
+        monkeypatch.setattr(service._scheduler, '_sleep_and_tick', record_delay)
+        monkeypatch.setattr(service._executor, 'execute',
+                            AsyncMock(return_value=ExecutionResult(success=True)))
+        await service.start()
+        try:
+            await asyncio.sleep(0)
+            assert delays == [60]
+            if manual:
+                await service.run_job_now(job.id)
+            else:
+                await service._execute_job(job, state)
+            await asyncio.sleep(0)
+            assert service.manager.repo.load_state(job.id).status == 'scheduled'
+            assert delays == [60, 1]
+            assert service.manager.repo.load_state(other.id).status == 'running'
+        finally:
+            await service.stop()
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('manual', [False, True])
+    async def test_completion_does_not_restart_stopped_scheduler(
+            self, workspace, monkeypatch, manual):
+        service = CronService(workspace=workspace)
+        job = service.create_job(schedule_str='every 2s', prompt='test')
+        monkeypatch.setattr(service._executor, 'execute',
+                            AsyncMock(return_value=ExecutionResult(success=True)))
+        await service.start()
+        await service.stop()
+        state = service.manager.repo.load_state(job.id)
+        if manual:
+            await service.run_job_now(job.id)
+        else:
+            await service._execute_job(job, state)
+        assert not service._scheduler.running
+        assert service._scheduler._timer_task is None