fix(cron): atomic write for jobs.json + don't silently overwrite corrupt store

Two related bugs that together caused scheduled jobs to disappear after
a container restart:

1. `_save_store()` used `Path.write_text(...)`, which truncates the
   destination in place.  A SIGKILL or shutdown mid-write left
   `jobs.json` either truncated or corrupt.

2. `_load_jobs()` caught any parse error, logged at WARNING, and
   returned an empty list.  `start()` then called `_save_store()`
   immediately, overwriting the corrupt-but-recoverable file with an
   empty job array.  Every scheduled job was silently lost with only a
   single warning line in the log.

Reproduction in production: container restart at 18:08, after which a
job that had fired correctly for two consecutive days never fired
again.  jobs.json on disk was missing the job entirely.

Fix:
- `_save_store()` now writes via temp file + `os.replace` + `fsync`
  (matches the session manager pattern from 512bf59,
  "fix(session): fsync sessions on graceful shutdown to prevent data
  loss").  An interrupted write cannot corrupt the live file.
- `_load_jobs()` now moves a corrupt store aside as
  `jobs.json.corrupt-<ts>` and returns `None` instead of `[]`.
- `start()` aborts with a `RuntimeError` when the on-disk store is
  corrupt, instead of starting empty and overwriting.
- `_load_store()` falls back to the previous in-memory snapshot when
  a hot reload encounters a corrupt file, so a transient corruption
  after start does not drop live jobs.

Tests cover the atomic-write path, the corrupt-file preservation,
the start-time refusal, the in-memory fallback, and a basic save/load
round trip across two service instances.  Existing 79 cron tests and
full suite (2553 tests) still pass.

nanobot/cron/service.py

@@ -2,8 +2,10 @@
 
 import asyncio
 import json
+import os
 import time
 import uuid
+from contextlib import suppress
 from dataclasses import asdict
 from datetime import datetime
 from pathlib import Path
@@ -83,8 +85,20 @@ class CronService:
         self._timer_active = False
         self.max_sleep_ms = max_sleep_ms
 
-    def _load_jobs(self) -> tuple[list[CronJob], int]:
-        jobs = []
+    def _load_jobs(self) -> tuple[list[CronJob], int] | None:
+        """Load jobs from disk.
+
+        Returns:
+            ``(jobs, version)`` tuple on success or when no store file exists
+            (in which case an empty list and version 1 are returned).
+            ``None`` when the store file exists but cannot be parsed; the
+            corrupt file is preserved with a ``.corrupt-<ts>`` suffix so the
+            caller can decide whether to overwrite or bail out.  Returning a
+            sentinel here is important: silently treating a parse error as an
+            empty job list would cause the next ``_save_store`` to wipe every
+            job from disk.
+        """
+        jobs: list[CronJob] = []
         version = 1
         if self.store_path.exists():
             try:
@@ -136,7 +150,22 @@ class CronService:
                         delete_after_run=j.get("deleteAfterRun", False),
                     ))
             except Exception as e:
-                logger.warning("Failed to load cron store: {}", e)
+                # Preserve the corrupt file for forensic recovery instead of
+                # letting the next save overwrite it with an empty job list.
+                backup = self.store_path.with_suffix(
+                    self.store_path.suffix + f".corrupt-{int(time.time())}"
+                )
+                with suppress(OSError):
+                    self.store_path.rename(backup)
+                logger.error(
+                    "Failed to load cron store at {}: {}. "
+                    "Corrupt file preserved at {}. "
+                    "Refusing to overwrite to avoid data loss.",
+                    self.store_path,
+                    e,
+                    backup,
+                )
+                return None
         return jobs, version
 
     def _merge_action(self):
@@ -175,15 +204,28 @@ class CronService:
                 self._save_store()
         return
 
-    def _load_store(self) -> CronStore:
+    def _load_store(self) -> 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
           _load_store calls (e.g. from list_jobs polling) from replacing it mid-execution.
+        - 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:
             return self._store
-        jobs, version = self._load_jobs()
+        loaded = self._load_jobs()
+        if loaded is None:
+            # Corrupt store on disk.  Prefer the last good in-memory snapshot
+            # over wiping live jobs; ``_load_jobs`` has already moved the
+            # corrupt file aside with a ``.corrupt-<ts>`` suffix.
+            if self._store is not None:
+                return self._store
+            return None
+        jobs, version = loaded
         self._store = CronStore(version=version, jobs=jobs)
         self._merge_action()
 
@@ -242,12 +284,56 @@ class CronService:
             ]
         }
 
-        self.store_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+        self._atomic_write(self.store_path, json.dumps(data, indent=2, ensure_ascii=False))
+
+    @staticmethod
+    def _atomic_write(path: Path, content: str) -> None:
+        """Write *content* to *path* atomically with fsync.
+
+        Uses a temp-file + ``os.replace`` + ``fsync`` pattern so a crash or
+        SIGKILL mid-write cannot leave the destination truncated or invalid.
+        Mirrors ``nanobot.session.manager.SessionManager.save`` (see
+        commit 512bf59, ``fix(session): fsync sessions on graceful shutdown
+        to prevent data loss``).  Without this, ``jobs.json`` could be
+        corrupted on container shutdown and silently re-created empty on
+        next start, wiping every scheduled job.
+        """
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp_path = path.with_suffix(path.suffix + ".tmp")
+        try:
+            with open(tmp_path, "w", encoding="utf-8") as f:
+                f.write(content)
+                f.flush()
+                os.fsync(f.fileno())
+            os.replace(tmp_path, path)
+            # fsync the parent directory so the rename itself is durable.
+            # Skip on Windows where opening a directory raises PermissionError;
+            # NTFS journals metadata synchronously so this is a no-op there.
+            with suppress(PermissionError):
+                fd = os.open(str(path.parent), os.O_RDONLY)
+                try:
+                    os.fsync(fd)
+                finally:
+                    os.close(fd)
+        except BaseException:
+            tmp_path.unlink(missing_ok=True)
+            raise
 
     async def start(self) -> None:
         """Start the cron service."""
         self._running = True
-        self._load_store()
+        loaded = self._load_store()
+        if loaded is None:
+            # Store file existed but was corrupt and has been preserved with
+            # a ``.corrupt-<ts>`` suffix.  Bail out instead of starting with
+            # an empty store; that would call ``_save_store`` and overwrite
+            # the now-renamed (but still recoverable) data with [].
+            self._running = False
+            raise RuntimeError(
+                f"cron store at {self.store_path} is corrupt and was preserved; "
+                "refusing to start with an empty job list. "
+                "Inspect the .corrupt-<ts> backup and restore manually."
+            )
         self._recompute_next_runs()
         self._save_store()
         self._arm_timer()
@@ -301,7 +387,11 @@ class CronService:
 
     async def _on_timer(self) -> None:
         """Handle timer tick - run due jobs."""
-        self._load_store()
+        loaded = self._load_store()
+        # If a hot reload found a corrupt store on disk, ``loaded`` is
+        # ``None`` but ``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

tests/cron/test_cron_persistence.py

@@ -0,0 +1,167 @@
+"""Persistence tests for ``nanobot.cron.service.CronService``.
+
+These tests target the specific failure mode where a corrupt or partially
+written ``jobs.json`` would silently turn into an empty job list on the next
+start, deleting every scheduled job.  See ``fix(cron): atomic write for
+jobs.json + don't silently overwrite corrupt store``.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+
+import pytest
+
+from nanobot.cron.service import CronService
+from nanobot.cron.types import CronSchedule
+
+
+def _seeded_store(tmp_path: Path) -> tuple[CronService, Path]:
+    """Build a service with one persisted job on disk and return both the
+    service and the resolved store path.  Adds the job via the action log
+    (the path used when the service is not running) and then triggers a
+    merge so ``jobs.json`` is written, mirroring the persisted on-disk
+    state seen in production."""
+    store_path = tmp_path / "cron" / "jobs.json"
+    service = CronService(store_path)
+    service.add_job(
+        name="Daily Loving Message",
+        schedule=CronSchedule(kind="cron", expr="0 10 * * *", tz="Asia/Kuwait"),
+        message="hello",
+    )
+    # add_job appended to action.jsonl; flush to jobs.json by toggling
+    # ``_running`` long enough for ``_merge_action`` to do its rewrite.
+    service._running = True
+    try:
+        service._load_store()
+    finally:
+        service._running = False
+    assert store_path.exists()
+    return service, store_path
+
+
+def test_save_store_is_atomic(tmp_path: Path) -> None:
+    """``_save_store`` must use temp-file + rename so an interrupted write
+    cannot leave the destination truncated or invalid."""
+    service, store_path = _seeded_store(tmp_path)
+
+    # Simulate an arbitrary save and confirm the result parses cleanly and
+    # no orphan ``.tmp`` is left behind.
+    service._save_store()
+    data = json.loads(store_path.read_text(encoding="utf-8"))
+    assert len(data["jobs"]) == 1
+
+    tmp_files = list(store_path.parent.glob("*.tmp"))
+    assert tmp_files == [], f"unexpected temp files left behind: {tmp_files}"
+
+
+def test_save_store_failure_does_not_corrupt_existing_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """If writing the temp file blows up partway through, the previous
+    ``jobs.json`` must remain readable.  This is the regression we are
+    actually fixing: pre-fix, ``write_text`` would truncate the destination
+    in place and leave it corrupt."""
+    service, store_path = _seeded_store(tmp_path)
+    original = store_path.read_bytes()
+
+    # Inject a failure inside the temp-file write.  ``os.replace`` should
+    # never run; the destination must keep its previous content.
+    real_open = open
+
+    def boom(path, *args, **kwargs):  # type: ignore[no-untyped-def]
+        if str(path).endswith(".tmp"):
+            raise OSError("simulated disk full")
+        return real_open(path, *args, **kwargs)
+
+    monkeypatch.setattr("builtins.open", boom)
+
+    with pytest.raises(OSError, match="simulated disk full"):
+        service._save_store()
+
+    assert store_path.read_bytes() == original
+
+
+def test_load_jobs_preserves_corrupt_store_and_returns_none(
+    tmp_path: Path,
+) -> None:
+    """A corrupt ``jobs.json`` must not be silently treated as an empty
+    list.  The loader returns ``None`` and the corrupt file is moved aside
+    with a ``.corrupt-<ts>`` suffix so an operator can recover it."""
+    store_path = tmp_path / "cron" / "jobs.json"
+    store_path.parent.mkdir(parents=True)
+    store_path.write_text("{not valid json", encoding="utf-8")
+
+    service = CronService(store_path)
+    assert service._load_jobs() is None
+
+    # Original path is gone; a ``.corrupt-<ts>`` backup exists alongside it.
+    assert not store_path.exists()
+    backups = list(store_path.parent.glob("jobs.json.corrupt-*"))
+    assert len(backups) == 1
+    assert backups[0].read_text(encoding="utf-8") == "{not valid json"
+
+
+def test_start_refuses_to_overwrite_corrupt_store(tmp_path: Path) -> None:
+    """``start`` must abort instead of running ``_save_store`` against an
+    empty in-memory state when the on-disk store is corrupt.  Otherwise the
+    next save would overwrite the (recoverable) corrupt file with an empty
+    job list and the user's jobs would be unrecoverable."""
+    store_path = tmp_path / "cron" / "jobs.json"
+    store_path.parent.mkdir(parents=True)
+    store_path.write_text("{still not json", encoding="utf-8")
+
+    service = CronService(store_path)
+    import asyncio
+
+    with pytest.raises(RuntimeError, match="corrupt"):
+        asyncio.run(service.start())
+
+    # Service is left in a stopped state so the operator notices.
+    assert service._running is False
+
+    # And the corrupt file is still recoverable from the .corrupt-<ts> copy.
+    backups = list(store_path.parent.glob("jobs.json.corrupt-*"))
+    assert len(backups) == 1
+
+
+def test_load_store_falls_back_to_in_memory_on_corruption_after_start(
+    tmp_path: Path,
+) -> None:
+    """If the store file becomes corrupt *after* a successful start (e.g. a
+    rclone-mounted Drive returns a partial read), the service must keep
+    using its existing in-memory snapshot instead of dropping every job."""
+    service, store_path = _seeded_store(tmp_path)
+    # Force load so ``self._store`` is populated.
+    service._load_store()
+    snapshot = service._store
+    assert snapshot is not None and len(snapshot.jobs) == 1
+
+    # Now corrupt the file on disk.
+    store_path.write_text("\x00garbage\x00", encoding="utf-8")
+
+    # Subsequent reload returns the in-memory snapshot, not None or empty.
+    result = service._load_store()
+    assert result is snapshot
+    assert len(result.jobs) == 1
+    assert result.jobs[0].name == "Daily Loving Message"
+
+
+def test_full_round_trip_survives_repeated_save_load(tmp_path: Path) -> None:
+    """Sanity check: jobs survive add → save → reload across fresh
+    ``CronService`` instances pointing at the same store."""
+    store_path = tmp_path / "cron" / "jobs.json"
+
+    s1 = CronService(store_path)
+    s1.add_job(
+        name="Daily Loving Message",
+        schedule=CronSchedule(kind="cron", expr="0 10 * * *", tz="Asia/Kuwait"),
+        message="hello",
+    )
+
+    s2 = CronService(store_path)
+    s2._load_store()
+    assert s2._store is not None
+    assert [j.name for j in s2._store.jobs] == ["Daily Loving Message"]