diff --git a/backend/app/gateway/scheduled_task_validation.py b/backend/app/gateway/scheduled_task_validation.py index 5488ae1d14..ec15bbeaf0 100644 --- a/backend/app/gateway/scheduled_task_validation.py +++ b/backend/app/gateway/scheduled_task_validation.py @@ -147,8 +147,15 @@ def validate_task_bounds( upcoming = _as_datetime(next_run_at) if end > now and upcoming is not None and end <= upcoming: raise scheduler_error(422, "end_at_before_first_run", "end_at must allow at least one scheduled occurrence") - if require_limit_for_frequent and max_runs is None and end_at is None and is_frequent_schedule(schedule_type, schedule_spec): - raise scheduler_error(422, "frequent_requires_limit", "Schedules more frequent than hourly require max_runs or end_at") + if require_limit_for_frequent and max_runs is None and end_at is None: + if is_frequent_schedule(schedule_type, _require_schedule_object(schedule_spec)): + raise scheduler_error(422, "frequent_requires_limit", "Schedules more frequent than hourly require max_runs or end_at") + + +def _require_schedule_object(spec: Any) -> dict[str, Any]: + if not isinstance(spec, dict): + raise scheduler_error(422, "invalid_schedule", "schedule_spec must be an object") + return spec def _normalized_stop_condition(value: Any) -> str | None: @@ -257,7 +264,8 @@ def _recompute_schedule(existing: dict[str, Any], updates: dict[str, Any], *, sc schedule_type = str(updates.get("schedule_type", existing["schedule_type"])) if schedule_type != existing["schedule_type"] and not isinstance(updates.get("schedule_spec"), dict): raise scheduler_error(422, "invalid_schedule", "changing schedule_type requires schedule_spec") - schedule_spec = dict(updates["schedule_spec"]) if isinstance(updates.get("schedule_spec"), dict) else dict(existing["schedule_spec"]) + previous_spec = existing["schedule_spec"] + schedule_spec = dict(_require_schedule_object(updates.get("schedule_spec", previous_spec))) timezone = str(updates.get("timezone", existing["timezone"])) try: if schedule_type == "once": @@ -265,7 +273,8 @@ def _recompute_schedule(existing: dict[str, Any], updates: dict[str, Any], *, sc # definitions and terminal reactivation must pass today's validation. unchanged_definition = ( existing["schedule_type"] == "once" - and schedule_spec.get("run_at") == existing["schedule_spec"].get("run_at") + and isinstance(previous_spec, dict) + and schedule_spec.get("run_at") == previous_spec.get("run_at") and timezone == existing["timezone"] and existing["status"] not in TERMINAL_TASK_STATUSES and existing.get("next_run_at") is not None @@ -279,10 +288,12 @@ def _recompute_schedule(existing: dict[str, Any], updates: dict[str, Any], *, sc schedule_spec["cron"] = normalize_cron_expression(raw_cron) if schedule_type == "interval": every_seconds = validate_interval_seconds(schedule_spec, scheduler_config.min_once_delay_seconds) - try: - previous_seconds = parse_interval_seconds(dict(existing["schedule_spec"])) if existing["schedule_type"] == "interval" else None - except ValueError: - previous_seconds = None + previous_seconds = None + if existing["schedule_type"] == "interval" and isinstance(previous_spec, dict): + try: + previous_seconds = parse_interval_seconds(previous_spec) + except ValueError: + pass # Re-saving an unchanged cadence keeps the next occurrence. if previous_seconds == every_seconds and existing.get("next_run_at") is not None: upcoming = existing["next_run_at"] @@ -379,10 +390,11 @@ def prepare_resume_updates(existing: dict[str, Any], renewal: dict[str, Any], *, if existing.get("status") == "enabled": return {} renewal = {key: value for key, value in renewal.items() if key in {"max_runs", "end_at"}} + schedule_spec = _require_schedule_object(existing["schedule_spec"]) upcoming = _as_datetime(existing.get("next_run_at")) if upcoming is None or upcoming <= now: try: - upcoming = next_run_at(existing["schedule_type"], existing["schedule_spec"], existing["timezone"], now=now) + upcoming = next_run_at(existing["schedule_type"], schedule_spec, existing["timezone"], now=now) except (ValueError, OverflowError) as exc: raise scheduler_error(422, "invalid_schedule", str(exc)) from exc if upcoming is None: diff --git a/backend/tests/test_scheduler_unschedulable_task.py b/backend/tests/test_scheduler_unschedulable_task.py index f11db6386f..a51e91c5e2 100644 --- a/backend/tests/test_scheduler_unschedulable_task.py +++ b/backend/tests/test_scheduler_unschedulable_task.py @@ -17,9 +17,10 @@ import pytest_asyncio from fastapi import HTTPException from sqlalchemy import update -from app.gateway.scheduled_task_validation import prepare_resume_updates +from app.gateway.scheduled_task_validation import prepare_resume_updates, validate_scheduled_task_update from app.scheduler.service import ScheduledTaskService from deerflow.config.database_config import DatabaseConfig +from deerflow.config.scheduler_config import SchedulerConfig from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config from deerflow.persistence.run import RunRepository from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository @@ -179,6 +180,47 @@ async def test_failed_recovery_pass_does_not_block_admission(repos, monkeypatch) assert launched == ["healthy"] +@pytest.mark.parametrize("invalid_spec", [None, 42, [42]]) +async def test_paused_interval_can_replace_a_malformed_stored_spec(repos, invalid_spec): + tasks, _, _, factory = repos + now = datetime.now(UTC) + await _create_task(tasks, "broken", next_run_at=now + timedelta(minutes=10), context_mode="fresh_thread_per_run") + async with factory() as session: + await session.execute(update(ScheduledTaskRow).where(ScheduledTaskRow.id == "broken").values(status="paused", schedule_spec=invalid_spec, next_run_at=None, origin_thread_id="thread-broken", max_runs=None, end_at=None)) + await session.commit() + existing = await tasks.get("broken", user_id="user-1") + assert existing is not None + + with pytest.raises(HTTPException) as resume_error: + prepare_resume_updates(existing, {}, used=0, now=now) + assert resume_error.value.status_code == 422 + assert resume_error.value.detail["code"] == "invalid_schedule" + + for fields in ({"timezone": "America/New_York"}, {"title": "Renamed"}): + with pytest.raises(HTTPException) as update_error: + await validate_scheduled_task_update(existing, fields, user_id="user-1", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=None, used=0, now=now) + assert update_error.value.status_code == 422 + assert update_error.value.detail["code"] == "invalid_schedule" + + updates = await validate_scheduled_task_update( + existing, + {"schedule_spec": {"every_seconds": 900}, "max_runs": 5}, + user_id="user-1", + thread_store=None, + scheduler_config=SchedulerConfig(), + assistant_resolver=None, + used=0, + now=now, + ) + repaired = await tasks.update("broken", user_id="user-1", updates=updates, require_mutable=True) + assert repaired is not None + assert repaired["status"] == "paused" + assert repaired["schedule_spec"] == {"every_seconds": 900} + resume = prepare_resume_updates(repaired, {}, used=0, now=now) + assert resume["status"] == "enabled" + assert resume["next_run_at"] == now + timedelta(seconds=900) + + async def test_resuming_reports_the_schedule_until_it_is_fixed(repos): """The pause is not a dead end: resume re-validates and names the problem.""" tasks, runs, run_store, factory = repos