diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index f3b04eae9..86f77f34a 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -163,9 +163,13 @@ class CronService: self._store: CronStore | None = None self._timer_task: asyncio.Task[None] | None = None self._running = False - self._timer_active = False + self._active_executions = 0 self.max_sleep_ms = max_sleep_ms + def _should_persist_store(self) -> bool: + """Return whether this instance currently owns the live store.""" + return self._running or self._active_executions > 0 + def _is_unbound_agent_job(self, job: CronJob) -> bool: return job.payload.kind == "agent_turn" and not is_bound_cron_job(job) @@ -278,23 +282,24 @@ class CronService: logger.exception("load action line error") continue self._store.jobs = list(jobs_map.values()) # pyright: ignore[reportOptionalMemberAccess] - if self._running and changed: + if self._should_persist_store() and changed: self._action_path.write_text("", encoding="utf-8") self._save_store() return - def _load_store(self) -> CronStore | None: + def _load_store(self, *, reload_during_execution: bool = False) -> CronStore | None: """Load jobs from disk. Reloads automatically if file was modified externally. - Reload every time because it needs to merge operations on the jobs object from other instances. - - During _on_timer execution, return the existing store to prevent concurrent + - During job execution, return the existing store to prevent concurrent _load_store calls (e.g. from list_jobs polling) from replacing it mid-execution. + The first execution explicitly reloads once when it takes ownership. - When the on-disk store exists but is unreadable: keep using the previous in-memory ``self._store`` if we already have one (so a transient corruption does not drop live jobs); only the very first load (during ``start``) can return ``None`` to signal an unrecoverable state to the caller. """ - if self._timer_active and self._store: + if self._active_executions > 0 and self._store and not reload_during_execution: return self._store loaded = self._load_jobs() if loaded is None: @@ -307,12 +312,12 @@ class CronService: jobs, version = loaded self._store = CronStore(version=version, jobs=jobs) self._merge_action() - if self._enforce_store_agent_bindings() and self._running: + if self._enforce_store_agent_bindings() and self._should_persist_store(): self._save_store() return self._store - def _require_store(self) -> CronStore: + def _require_store(self, *, reload_during_execution: bool = False) -> CronStore: """Return a usable store or raise a clear error. ``_load_store`` deliberately returns ``None`` when the first load sees @@ -322,7 +327,7 @@ class CronService: ``AttributeError`` and, more importantly, prevents follow-up saves from treating a corrupt store as an empty one. """ - store = self._load_store() + store = self._load_store(reload_during_execution=reload_during_execution) if store is None: raise RuntimeError( f"cron store at {self.store_path} could not be loaded and was preserved " @@ -504,19 +509,20 @@ class CronService: async def _on_timer(self) -> None: """Handle timer tick - run due jobs.""" - self._load_store() - # If a hot reload found a corrupt store on disk, ``self._store`` may - # still hold the previous, known-good in-memory snapshot. Keep using - # it rather than crashing the timer or wiping live jobs. - if not self._store: - self._arm_timer() - return - - self._timer_active = True + reload_store = self._active_executions == 0 + self._active_executions += 1 try: + store = self._load_store(reload_during_execution=reload_store) + # If a hot reload found a corrupt store on disk, ``self._store`` may + # still hold the previous, known-good in-memory snapshot. Keep using + # it rather than crashing the timer or wiping live jobs. + if store is None: + self._arm_timer() + return + now = _now_ms() due_jobs = [ - j for j in self._store.jobs + j for j in store.jobs if j.enabled and j.state.next_run_at_ms and now >= j.state.next_run_at_ms ] @@ -525,7 +531,7 @@ class CronService: self._save_store() finally: - self._timer_active = False + self._active_executions -= 1 self._arm_timer() async def _execute_job(self, job: CronJob) -> None: @@ -657,7 +663,7 @@ class CronService: ) _normalize_agent_turn_job(job) self._enforce_agent_binding(job) - if self._running: + if self._should_persist_store(): store = self._require_store() store.jobs.append(job) self._save_store() @@ -697,7 +703,7 @@ class CronService: removed = len(store.jobs) < before if removed: - if self._running: + if self._should_persist_store(): self._save_store() self._arm_timer() else: @@ -719,7 +725,7 @@ class CronService: job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms()) else: job.state.next_run_at_ms = None - if self._running: + if self._should_persist_store(): self._save_store() self._arm_timer() else: @@ -775,7 +781,7 @@ class CronService: else: job.state.next_run_at_ms = None - if self._running: + if self._should_persist_store(): self._save_store() self._arm_timer() else: @@ -786,10 +792,10 @@ class CronService: async def run_job(self, job_id: str, force: bool = False) -> bool: """Manually run a job without disturbing the service's running state.""" - was_running = self._running - self._running = True + reload_store = self._active_executions == 0 + self._active_executions += 1 try: - store = self._require_store() + store = self._require_store(reload_during_execution=reload_store) for job in store.jobs: if job.id == job_id: if self._is_unbound_agent_job(job): @@ -803,8 +809,8 @@ class CronService: return True return False finally: - self._running = was_running - if was_running: + self._active_executions -= 1 + if self._running and self._active_executions == 0: self._arm_timer() def get_job(self, job_id: str) -> CronJob | None: diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index 36079e0ae..7623fdd41 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -600,6 +600,83 @@ async def test_run_job_preserves_running_service_state(tmp_path) -> None: service.stop() +@pytest.mark.asyncio +async def test_manual_run_persists_completion_when_callback_lists_jobs(tmp_path) -> None: + store_path = tmp_path / "cron" / "jobs.json" + + async def on_job(_job) -> None: + service.list_jobs(include_disabled=True) + await asyncio.sleep(0) + + service = CronService(store_path, on_job=on_job) + job = service.add_job( + name="manual", + schedule=CronSchedule(kind="every", every_ms=60_000), + message="hello", + **_bound_chat(), + ) + + assert await service.run_job(job.id) is True + + state = json.loads(store_path.read_text())["jobs"][0]["state"] + assert state["lastStatus"] == "ok" + assert state["lastError"] is None + assert len(state["runHistory"]) == 1 + assert state["runHistory"][0]["status"] == "ok" + + +@pytest.mark.asyncio +async def test_overlapping_manual_runs_preserve_stopped_service_state(tmp_path) -> None: + store_path = tmp_path / "cron" / "jobs.json" + entered = [asyncio.Event(), asyncio.Event()] + release = [asyncio.Event(), asyncio.Event()] + call_count = 0 + + async def on_job(_job) -> None: + nonlocal call_count + call_index = call_count + call_count += 1 + entered[call_index].set() + await release[call_index].wait() + + service = CronService(store_path, on_job=on_job) + jobs = [ + service.add_job( + name=f"manual-{index}", + schedule=CronSchedule(kind="every", every_ms=60_000), + message="hello", + **_bound_chat(str(index)), + ) + for index in range(2) + ] + + first = asyncio.create_task(service.run_job(jobs[0].id)) + await entered[0].wait() + second = asyncio.create_task(service.run_job(jobs[1].id)) + try: + await entered[1].wait() + release[0].set() + assert await first is True + assert service._running is False + + release[1].set() + assert await second is True + assert service._running is False + assert service._timer_task is None + + states = { + item["name"]: item["state"] + for item in json.loads(store_path.read_text())["jobs"] + } + assert states["manual-0"]["lastStatus"] == "ok" + assert states["manual-1"]["lastStatus"] == "ok" + finally: + release[0].set() + release[1].set() + await asyncio.gather(first, second, return_exceptions=True) + service.stop() + + @pytest.mark.asyncio async def test_running_service_honors_external_disable(tmp_path) -> None: store_path = tmp_path / "cron" / "jobs.json"