fix: include scheduled tasks in backups (#10644)

Author: wcqqq1214

astrbot/core/backup/constants.py

@@ -11,6 +11,7 @@
     CommandConfig,
     CommandConflict,
     ConversationV2,
+    CronJob,
     Persona,
     PersonaFolder,
     PlatformMessageHistory,
@@ -43,6 +44,7 @@
 MAIN_DB_MODELS: dict[str, type[SQLModel]] = {
     "platform_stats": PlatformStat,
     "conversations": ConversationV2,
+    "cron_jobs": CronJob,
     "personas": Persona,
     "persona_folders": PersonaFolder,
     "preferences": Preference,

tests/test_backup.py

@@ -28,6 +28,7 @@
 from astrbot.core.db.po import (
     ConversationV2,
 )
+from astrbot.core.db.sqlite import SQLiteDatabase
 from astrbot.core.utils.upload import UploadTooLargeError
 from astrbot.core.utils.version_comparator import VersionComparator
 from astrbot.dashboard.services.backup_service import (
@@ -1048,6 +1049,36 @@ def test_kb_metadata_models_contain_expected_tables(self):
 class TestBackupIntegration:
     """备份集成测试"""
 
+    @pytest.mark.asyncio
+    async def test_cron_jobs_roundtrip(self, tmp_path):
+        """Preserve a scheduled task through backup export and import."""
+        source = SQLiteDatabase(str(tmp_path / "source.db"))
+        target = SQLiteDatabase(str(tmp_path / "target.db"))
+        try:
+            job = await source.create_cron_job(
+                name="Daily reminder",
+                job_type="active_agent",
+                cron_expression="0 9 * * *",
+                timezone="Asia/Taipei",
+                payload={"message": "Reminder"},
+            )
+            exporter = AstrBotExporter(
+                main_db=source, config_path=str(tmp_path / "missing-config.json")
+            )
+            with patch(
+                "astrbot.core.backup.exporter.get_backup_directories", return_value={}
+            ):
+                zip_path = await exporter.export_all(output_dir=str(tmp_path / "backups"))
+
+            result = await AstrBotImporter(main_db=target).import_all(zip_path)
+            assert result.success, result.errors
+            restored = await target.list_cron_jobs()
+            assert len(restored) == 1
+            assert restored[0].model_dump() == job.model_dump()
+        finally:
+            await source.engine.dispose()
+            await target.engine.dispose()
+
     @pytest.mark.asyncio
     async def test_export_import_roundtrip(self, tmp_path):
         """测试导出-导入往返"""