From e593ad6c144c1c1c7d69e936210b5249b79d9e29 Mon Sep 17 00:00:00 2001 From: ming1523 Date: Sat, 15 Aug 2026 15:34:55 +0800 Subject: [PATCH] fix(scheduler): coerce serialized task timestamps (#4785) * fix(scheduler): coerce serialized task timestamps * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: xsfx20 <15558128926@qq.com> Co-authored-by: Willem Jiang Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../persistence/scheduled_tasks/sql.py | 22 +++++++++- .../tests/test_scheduled_task_repository.py | 40 +++++++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py index 46d7ac0e6..dcbdcce17 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py @@ -27,6 +27,24 @@ def _lease_is_alive(lease_expires_at: datetime | None, *, now: datetime, grace_s return lease_expires_at >= now - timedelta(seconds=grace_seconds) +def _coerce_datetime(value: datetime | str | None) -> datetime | None: + """Convert serialized task timestamps back before binding DateTime fields.""" + if value is None or isinstance(value, datetime): + return value + if isinstance(value, str): + try: + text = value + if text.endswith("Z"): + text = f"{text[:-1]}+00:00" + dt = datetime.fromisoformat(text) + except ValueError as exc: + raise ValueError(f"invalid scheduled task timestamp: {value!r}") from exc + if dt.tzinfo is None: + return dt.replace(tzinfo=UTC) + return dt.astimezone(UTC) + raise TypeError(f"scheduled task timestamp must be datetime, str, or None: {type(value).__name__}") + + class ScheduledTaskRepository: def __init__( self, @@ -206,7 +224,7 @@ class ScheduledTaskRepository: *, status: str, next_run_at: datetime | None, - last_run_at: datetime | None, + last_run_at: datetime | str | None, last_run_id: str | None, last_thread_id: str | None, last_error: str | None, @@ -237,7 +255,7 @@ class ScheduledTaskRepository: row.status = status row.last_error = last_error row.next_run_at = next_run_at - row.last_run_at = last_run_at + row.last_run_at = _coerce_datetime(last_run_at) row.last_run_id = last_run_id row.last_thread_id = last_thread_id if increment_run_count: diff --git a/backend/tests/test_scheduled_task_repository.py b/backend/tests/test_scheduled_task_repository.py index 92f11bd36..0566ea324 100644 --- a/backend/tests/test_scheduled_task_repository.py +++ b/backend/tests/test_scheduled_task_repository.py @@ -519,6 +519,46 @@ async def test_update_after_launch_protect_terminal_keeps_hook_result(tmp_path): await close_engine() +@pytest.mark.asyncio +async def test_update_after_launch_coerces_serialized_last_run_at(tmp_path): + """Task rows returned by the repository serialize timestamps as ISO strings.""" + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + sf = get_session_factory() + assert sf is not None + + repo = ScheduledTaskRepository(sf) + await repo.create( + task_id="task-serialized-timestamp", + user_id="user-1", + thread_id=None, + context_mode="fresh_thread_per_run", + assistant_id="lead_agent", + title="serialized timestamp", + prompt="p", + schedule_type="cron", + schedule_spec={"cron": "0 9 * * *"}, + timezone="UTC", + next_run_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC), + ) + + await repo.update_after_launch( + "task-serialized-timestamp", + status="enabled", + next_run_at=datetime(2026, 7, 3, 1, 0, tzinfo=UTC), + last_run_at="2026-07-02T01:00:00+00:00", + last_run_id="run-serialized", + last_thread_id="thread-serialized", + last_error=None, + increment_run_count=True, + ) + + task = await repo.get("task-serialized-timestamp", user_id="user-1") + assert task is not None + assert task["last_run_at"] == "2026-07-02T01:00:00+00:00" + + await close_engine() + + @pytest.mark.asyncio async def test_list_by_task_paginates(tmp_path): await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))