Harden Celery schedule parsing and startup safety

tests/test_celery_schedule_safety.py

@@ -0,0 +1,38 @@
+import os
+import subprocess
+import sys
+
+import pytest
+
+
+def _run_python_import(code: str, env_overrides: dict[str, str]) -> subprocess.CompletedProcess:
+    env = os.environ.copy()
+    env.update(env_overrides)
+    return subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+        env=env,
+        check=False,
+    )
+
+
+@pytest.mark.django_db
+def test_invalid_cron_does_not_crash_config_import_path():
+    result = _run_python_import(
+        (
+            "import config; "
+            "from config.celery import app; "
+            "print(f'beat_schedule_size={len(app.conf.beat_schedule or {})}')"
+        ),
+        {
+            "DJANGO_SETTINGS_MODULE": "config.settings.development",
+            "DJANGO_ENV": "development",
+            "DJANGO_DEBUG": "1",
+            "INGESTION_SCHEDULE_ENABLED": "1",
+            "INGESTION_SCHEDULE_CRON": "bad cron value",
+        },
+    )
+
+    assert result.returncode == 0
+    assert "beat_schedule_size=0" in result.stdout

tests/test_ingestion_tasks.py

@@ -31,6 +31,18 @@ def test_build_periodic_schedule_disabled(settings):
     assert build_periodic_schedule() == {}
 
 
+@pytest.mark.django_db
+def test_build_periodic_schedule_invalid_cron_disables_task_and_logs(settings, caplog):
+    settings.INGESTION_SCHEDULE_ENABLED = True
+    settings.INGESTION_SCHEDULE_CRON = "invalid-cron"
+
+    with caplog.at_level("ERROR"):
+        schedule = build_periodic_schedule()
+
+    assert schedule == {}
+    assert any("Invalid periodic ingestion schedule config. Task disabled." in message for message in caplog.messages)
+
+
 @pytest.mark.django_db
 def test_trigger_incremental_sync_skips_overlapping_runs(settings):
     settings.INGESTION_PREVENT_OVERLAP = True