diff --git a/README.md b/README.md index 7578325fe3..da9ae0eece 100644 --- a/README.md +++ b/README.md @@ -3432,6 +3432,7 @@ schedule tools keep their existing inputs. See the [group delivery guide](backen ### Lifecycle, safety caps and stop conditions - The tasks page and the REST API (`POST` / `PATCH /api/scheduled-tasks`) accept the same per-run goal (`goal_objective`), safety cap (`max_runs`, `end_at`) and stop condition (`stop_condition`) as a conversation. Sending `null` in a PATCH clears any of these four; an `end_at` without a UTC offset is wall-clock time in the task's timezone. +- New or changed one-time `run_at` values and submitted `end_at` values without an offset must exist in the chosen timezone. For these fields, a time skipped by a daylight-saving transition is rejected; choose a valid local time or supply an explicit instant with a UTC offset. Repeated local times keep the earlier occurrence unless an offset selects the later one. Recurring cron schedules keep their existing runtime behavior: a fixed-time occurrence in a skipped local time fires at the shifted time that day. - A stop condition is the user's "stop when …" rule. It is stored in its own field (migration `0031`), never inside the task instructions. Only when a run starts does DeerFlow append it to that run's message and ask the run to call `stop_scheduled_task` when the rule holds. While `scheduler.tool_enabled` is on, every scheduled run can stop its own schedule, whether a chat or the tasks page created the task; with it off, the run is asked to report a met rule instead of calling a tool it does not have. - Goal tasks created on the tasks page are now evaluated like chat-created ones, and their runs also receive the saved notes and the previous-run reference. - Resume computes the next run from now, so a long pause never causes a catch-up run. A one-time task whose time has passed returns `422 once_time_passed` and needs a new time. Resuming an active task changes nothing; pausing a finished task returns `409 task_finished`. diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md index 7c9e9a094b..d25829db40 100644 --- a/backend/app/gateway/routers/AGENTS.md +++ b/backend/app/gateway/routers/AGENTS.md @@ -4,6 +4,12 @@ Shared scheduled-task update validation preserves an unchanged future `once` instant on a live task inside `scheduler.min_once_delay_seconds`; new/changed instants and terminal reactivation still enforce the minimum delay. +Scheduled-task admission rejects nonexistent naive `run_at` and `end_at` +wall times through the shared Gateway datetime resolver. Keep this check out +of runtime `next_run_at`: legacy tasks retain their accepted execution time, +including unchanged live once definitions. Explicit offsets retain their instant; +ambiguous naive times keep the existing fold choice. + The MCP task list accepts optional `TaskStatus` and `active_only` query filters. Pass both to the repository before its limit; preserve `threads:read`, owner checks, current-thread incarnation scoping, and the unfiltered array response. diff --git a/backend/app/gateway/scheduled_task_validation.py b/backend/app/gateway/scheduled_task_validation.py index 291dc346a5..5488ae1d14 100644 --- a/backend/app/gateway/scheduled_task_validation.py +++ b/backend/app/gateway/scheduled_task_validation.py @@ -59,6 +59,26 @@ def _as_datetime(value: datetime | str | None) -> datetime | None: return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) +def _resolve_local_datetime(value: datetime, timezone: str) -> datetime: + """Resolve a submitted wall time without silently moving it across a clock gap.""" + if value.tzinfo is not None: + return value.astimezone(UTC) + zone = ZoneInfo(timezone) + resolved = value.replace(tzinfo=zone).astimezone(UTC) + if resolved.astimezone(zone).replace(tzinfo=None) != value: + raise ValueError(f"Local time {value.isoformat()} does not exist in timezone {timezone}; choose a valid local time or an explicit instant with a UTC offset") + return resolved + + +def _validate_once_wall_time(schedule_spec: dict[str, Any], timezone: str) -> None: + run_at = schedule_spec.get("run_at") + if isinstance(run_at, str): + try: + _resolve_local_datetime(datetime.fromisoformat(run_at), timezone) + except ValueError as exc: + raise ValueError(f"run_at: {exc}") from exc + + def resolve_end_at(value: datetime | str | None, timezone: str) -> datetime | None: """Return ``end_at`` as aware UTC; a naive value is wall-clock time in the task's timezone. @@ -73,9 +93,10 @@ def resolve_end_at(value: datetime | str | None, timezone: str) -> datetime | No raise scheduler_error(422, "invalid_request", "end_at must be an ISO 8601 date-time") from exc if not isinstance(value, datetime): raise scheduler_error(422, "invalid_request", "end_at must be an ISO 8601 date-time") - if value.tzinfo is None: - value = value.replace(tzinfo=ZoneInfo(timezone)) - return value.astimezone(UTC) + try: + return _resolve_local_datetime(value, timezone) + except ValueError as exc: + raise scheduler_error(422, "invalid_request", f"end_at: {exc}") from exc def validate_task_bounds( @@ -156,6 +177,8 @@ def _validate_schedule(schedule_type: str, schedule_spec: dict[str, Any], timezo except ValueError as exc: raise scheduler_error(422, "invalid_timezone", str(exc)) from exc try: + if schedule_type == "once": + _validate_once_wall_time(schedule_spec, timezone) if schedule_type == "cron": raw_cron = schedule_spec.get("cron") if not isinstance(raw_cron, str): @@ -237,6 +260,18 @@ def _recompute_schedule(existing: dict[str, Any], updates: dict[str, Any], *, sc schedule_spec = dict(updates["schedule_spec"]) if isinstance(updates.get("schedule_spec"), dict) else dict(existing["schedule_spec"]) timezone = str(updates.get("timezone", existing["timezone"])) try: + if schedule_type == "once": + # Preserve already admitted legacy wall times on no-op edits. New + # 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 timezone == existing["timezone"] + and existing["status"] not in TERMINAL_TASK_STATUSES + and existing.get("next_run_at") is not None + ) + if not unchanged_definition: + _validate_once_wall_time(schedule_spec, timezone) if schedule_type == "cron": raw_cron = schedule_spec.get("cron") if not isinstance(raw_cron, str): diff --git a/backend/docs/API.md b/backend/docs/API.md index 361601273e..db80fbff12 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -1673,6 +1673,16 @@ POST /api/scheduled-tasks/preview-cron | `stop_condition` | string \| null | The user's "stop when …" rule, stored in its own field. Whitespace collapses to single spaces; at most 500 characters; blank means none. It is never part of `prompt`: each launch appends it as an instruction to call `stop_scheduled_task` (or, with `scheduler.tool_enabled` off, to report a met rule). | | `notification_group_connection_id` | string \| null | One additional connected Feishu group belonging to the task owner. Select a group binding returned by `GET /api/channels/connections`; it must record `chat_type=group` and a matching chat/workspace ID. Omitted in PATCH preserves the target; null clears it. Invalid choices return `invalid_notification_group` (422), lookup outages return `scheduler_unavailable` (503). | +For one-time `schedule_spec.run_at` and `end_at` without an offset, a local time +skipped by a timezone clock change is rejected: `422 invalid_schedule` for +`run_at`, or `422 invalid_request` for `end_at`, including an end-time renewal +on Resume. This applies to REST and conversation scheduling. Ambiguous local +times retain the earlier occurrence (`fold=0`); explicit offsets preserve the +specified instant. Unchanged live once definitions keep their previously +accepted execution time; a changed definition or a PATCH that rearms a terminal +task is validated again. Ordinary Resume retains the admitted stored schedule +without revalidating its wall time; a renewed `end_at` is still checked. + In a PATCH, sending `null` clears `goal_objective`, `max_runs`, `end_at` or `stop_condition` or `notification_group_connection_id`; for every other field `null` means unchanged (`assistant_id: null` restores `lead_agent`). PATCH may change `schedule_type` together with diff --git a/backend/tests/test_scheduled_task_router_behavior.py b/backend/tests/test_scheduled_task_router_behavior.py index 717e486441..5f3cf60d58 100644 --- a/backend/tests/test_scheduled_task_router_behavior.py +++ b/backend/tests/test_scheduled_task_router_behavior.py @@ -1747,6 +1747,37 @@ async def test_naive_end_at_is_wall_clock_time_in_the_task_timezone(http): assert datetime.fromisoformat(patched.json()["end_at"]) == datetime(2030, 6, 1, 10, 30, tzinfo=UTC) +@pytest.mark.asyncio +@pytest.mark.parametrize(("operation", "field"), [("create", "run_at"), ("update", "run_at"), ("create", "end_at"), ("update", "end_at"), ("resume", "end_at")]) +async def test_http_rejects_nonexistent_local_time_without_writing(http, monkeypatch, operation, field): + class FixedDatetime(datetime): + @classmethod + def now(cls, tz=None): + return cls(2026, 10, 22, tzinfo=UTC) + + monkeypatch.setattr(scheduled_tasks, "datetime", FixedDatetime) + wall_time = "2027-03-14T02:30:00" + fields = {"schedule_type": "once", "schedule_spec": {"run_at": wall_time}} if field == "run_at" else {"end_at": wall_time} + if operation == "create": + response = await http.client.post("/api/scheduled-tasks", json={**_CREATE, "timezone": "America/New_York", **fields}) + else: + await _seed(http.repo, timezone="America/New_York", next_run_at=datetime(2026, 10, 23, tzinfo=UTC)) + if operation == "resume": + await http.repo.update("task-l", user_id=OWNER, updates={"status": "paused"}) + before = await http.repo.get("task-l", user_id=OWNER) + if operation == "update": + response = await http.client.patch("/api/scheduled-tasks/task-l", json=fields) + else: + response = await http.client.post("/api/scheduled-tasks/task-l/resume", json=fields) + detail = coded(response, 422, "invalid_schedule" if field == "run_at" else "invalid_request") + assert "does not exist" in detail["message"] + if operation == "create": + assert await http.repo.list_by_user(OWNER) == [] + else: + assert await http.repo.get("task-l", user_id=OWNER) == before + assert http.launches == [] + + @pytest.mark.asyncio async def test_create_keeps_the_prompt_and_stores_the_stop_condition_apart(http): response = await http.client.post( diff --git a/backend/tests/test_scheduled_task_wall_times.py b/backend/tests/test_scheduled_task_wall_times.py new file mode 100644 index 0000000000..cbc631d55d --- /dev/null +++ b/backend/tests/test_scheduled_task_wall_times.py @@ -0,0 +1,156 @@ +"""HTTP and conversation admission must not silently move nonexistent local times.""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock +from zoneinfo import ZoneInfo + +import pytest +from fastapi import HTTPException + +from app.gateway.routers.scheduled_tasks import ScheduledTaskCreateRequest, ScheduledTaskUpdateRequest +from app.gateway.scheduled_task_validation import ( + prepare_resume_updates, + resolve_end_at, + validate_scheduled_task_create, + validate_scheduled_task_update, +) +from deerflow.config.scheduler_config import SchedulerConfig +from deerflow.scheduler.schedules import next_run_at + +NOW = datetime(2026, 10, 22, tzinfo=UTC) +GAPS = [ + ("America/New_York", "2027-03-14T02:30:00"), + ("Europe/Berlin", "2027-03-28T02:30:00"), + ("Australia/Sydney", "2027-10-03T02:30:00"), + ("Australia/Lord_Howe", "2027-10-03T02:15:00"), +] + + +def _existing(timezone, **updates): + return { + "schedule_type": "once", + "schedule_spec": {"run_at": "2027-01-01T09:00:00"}, + "timezone": timezone, + "status": "enabled", + "next_run_at": datetime(2027, 1, 1, tzinfo=UTC), + "context_mode": "fresh_thread_per_run", + "thread_id": None, + "goal_objective": None, + "max_runs": None, + "end_at": None, + **updates, + } + + +def _time_fields(field, value): + return {"schedule_spec": {"run_at": value}} if field == "run_at" else {"end_at": value} + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("timezone", "wall_time"), GAPS) +@pytest.mark.parametrize("field", ["run_at", "end_at"]) +@pytest.mark.parametrize("operation", ["create", "update"]) +async def test_shared_admission_rejects_nonexistent_wall_time(timezone, wall_time, field, operation): + fields = _time_fields(field, wall_time) + with pytest.raises(HTTPException) as caught: + if operation == "create": + body = ScheduledTaskCreateRequest( + title="Report", + prompt="Write the report", + schedule_type="once", + timezone=timezone, + **({"schedule_spec": {"run_at": "2027-01-01T09:00:00"}} | fields), + ) + await validate_scheduled_task_create(body, user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(return_value="lead_agent"), now=NOW) + else: + body = ScheduledTaskUpdateRequest(**fields) + await validate_scheduled_task_update(_existing(timezone), body.model_dump(exclude_unset=True), user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(), used=0, now=NOW) + assert caught.value.status_code == 422 + assert caught.value.detail["code"] == ("invalid_schedule" if field == "run_at" else "invalid_request") + assert caught.value.detail["message"].startswith(f"{field}: ") + assert "does not exist" in caught.value.detail["message"] + assert timezone in caught.value.detail["message"] + + +@pytest.mark.parametrize(("timezone", "wall_time"), GAPS) +def test_resume_rejects_nonexistent_end_at(timezone, wall_time): + with pytest.raises(HTTPException) as caught: + prepare_resume_updates(_existing(timezone, status="paused"), {"end_at": wall_time}, used=0, now=NOW) + assert caught.value.status_code == 422 + assert caught.value.detail["code"] == "invalid_request" + assert "does not exist" in caught.value.detail["message"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("timezone", "wall_time", "expected"), + [ + ("America/New_York", "2027-03-14T01:30:00", "2027-03-14T06:30:00+00:00"), + ("America/New_York", "2027-03-14T03:30:00", "2027-03-14T07:30:00+00:00"), + ("Europe/Berlin", "2027-03-28T01:30:00", "2027-03-28T00:30:00+00:00"), + ("Europe/Berlin", "2027-03-28T03:30:00", "2027-03-28T01:30:00+00:00"), + ("Australia/Sydney", "2027-10-03T01:30:00", "2027-10-02T15:30:00+00:00"), + ("Australia/Sydney", "2027-10-03T03:30:00", "2027-10-02T16:30:00+00:00"), + ("Australia/Lord_Howe", "2027-10-03T01:45:00", "2027-10-02T15:15:00+00:00"), + ("Australia/Lord_Howe", "2027-10-03T02:45:00", "2027-10-02T15:45:00+00:00"), + ("America/New_York", "2027-11-07T01:30:00", "2027-11-07T05:30:00+00:00"), + ("America/New_York", "2027-11-07T01:30:00-05:00", "2027-11-07T06:30:00+00:00"), + ("America/New_York", "2027-03-14T02:30:00-05:00", "2027-03-14T07:30:00+00:00"), + ("Asia/Shanghai", "2027-03-14T02:30:00", "2027-03-13T18:30:00+00:00"), + ], +) +async def test_valid_wall_times_and_explicit_instants_keep_their_meaning(timezone, wall_time, expected): + body = ScheduledTaskCreateRequest(title="Report", prompt="Write the report", schedule_type="once", schedule_spec={"run_at": wall_time}, timezone=timezone) + definition = await validate_scheduled_task_create(body, user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(return_value="lead_agent"), now=NOW) + assert definition["next_run_at"] == datetime.fromisoformat(expected) + assert resolve_end_at(wall_time, timezone) == datetime.fromisoformat(expected) + + +def test_end_at_preserves_an_aware_second_fold(): + end_at = datetime(2027, 11, 7, 1, 30, tzinfo=ZoneInfo("America/New_York"), fold=1) + assert resolve_end_at(end_at, "America/New_York") == datetime(2027, 11, 7, 6, 30, tzinfo=UTC) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["enabled", "paused"]) +async def test_unchanged_legacy_gap_schedule_keeps_its_already_accepted_instant(status): + timezone, wall_time = GAPS[0] + spec = {"run_at": wall_time} + accepted = next_run_at("once", spec, timezone, now=NOW) + existing = _existing(timezone, status=status, schedule_spec=spec, next_run_at=accepted) + updates = await validate_scheduled_task_update( + existing, {"schedule_spec": spec, "timezone": timezone, "title": "Renamed"}, user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(), used=0, now=NOW + ) + assert updates["next_run_at"] == accepted + assert next_run_at("once", spec, timezone, now=accepted) is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("change", ["timezone", "terminal"]) +async def test_gap_schedule_is_revalidated_when_reinterpreted_or_rearmed(change): + timezone, wall_time = GAPS[0] + spec = {"run_at": wall_time} + existing = _existing("UTC" if change == "timezone" else timezone, schedule_spec=spec, status="completed" if change == "terminal" else "enabled") + fields = {"timezone": timezone} if change == "timezone" else {"schedule_spec": spec} + with pytest.raises(HTTPException) as caught: + await validate_scheduled_task_update(existing, fields, user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(), used=0, now=NOW) + assert caught.value.status_code == 422 + assert caught.value.detail["code"] == "invalid_schedule" + assert caught.value.detail["message"].startswith("run_at: ") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["create", "update", "reactivate"]) +@pytest.mark.parametrize("run_at", ["not-a-date", "2027-02-30T09:00:00"]) +async def test_invalid_once_date_names_run_at_in_the_admission_error(operation, run_at): + spec = {"run_at": run_at} + with pytest.raises(HTTPException) as caught: + if operation == "create": + body = ScheduledTaskCreateRequest(title="Report", prompt="Write the report", schedule_type="once", schedule_spec=spec, timezone="UTC") + await validate_scheduled_task_create(body, user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(), now=NOW) + else: + existing = _existing("UTC", status="completed" if operation == "reactivate" else "enabled") + await validate_scheduled_task_update(existing, {"schedule_spec": spec}, user_id="owner", thread_store=None, scheduler_config=SchedulerConfig(), assistant_resolver=AsyncMock(), used=0, now=NOW) + assert caught.value.status_code == 422 + assert caught.value.detail["code"] == "invalid_schedule" + assert caught.value.detail["message"].startswith("run_at: ") diff --git a/backend/tests/test_scheduler_capability_access.py b/backend/tests/test_scheduler_capability_access.py index 94896f1647..7fcedd32a0 100644 --- a/backend/tests/test_scheduler_capability_access.py +++ b/backend/tests/test_scheduler_capability_access.py @@ -236,6 +236,31 @@ async def test_create_uses_bound_origin_and_echoes_goal_and_stop_details(monkeyp assert "stop_instructions" not in task +@pytest.mark.asyncio +@pytest.mark.parametrize(("action", "field"), [("create", "run_at"), ("update", "run_at"), ("create", "end_at"), ("update", "end_at"), ("resume", "end_at")]) +async def test_conversation_rejects_nonexistent_local_time_before_persistence(monkeypatch, action, field): + from app.gateway import scheduled_task_access as access + + class FixedDatetime(datetime): + @classmethod + def now(cls, tz=None): + return cls(2026, 10, 22, tzinfo=UTC) + + monkeypatch.setattr(access, "datetime", FixedDatetime) + task = _task(timezone="America/New_York", status="paused" if action == "resume" else "enabled", next_run_at="2026-10-23T00:00:00+00:00") + prepare, repo, *_ = _setup(monkeypatch, task=task) + capability = await prepare(client_timezone="America/New_York") + wall_time = "2027-03-14T02:30:00" + fields = {"schedule_type": "once", "schedule_spec": {"run_at": wall_time}} if field == "run_at" else {"end_at": wall_time} + payload = {"title": "Report", "prompt": "Write the report", "schedule_type": "interval", "schedule_spec": {"every_seconds": 3600}} if action == "create" else {"task_id": "task-1"} + result = await capability.manage(action=action, request={**payload, **fields}) + assert result.get("status_code") == 422, result + assert result["code"] == ("invalid_schedule" if field == "run_at" else "invalid_request") + assert "does not exist" in result["error"] + repo.create.assert_not_awaited() + repo.update.assert_not_awaited() + + @pytest.mark.asyncio @pytest.mark.parametrize("objective", ["", " \t\n ", "g" * 4001]) async def test_create_rejects_invalid_goal_before_persisting_schedule(monkeypatch, objective):