mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(scheduler): stop interval-task metadata edits from failing with 500 on the reused next_run_at (#5964)
* fix(scheduler): coerce serialized timestamps in ScheduledTaskRepository.update()
Editing an interval task's title or prompt returned 500. The edit dialog
always sends schedule_spec beside the changed field, and when the cadence
is unchanged the PATCH route keeps existing["next_run_at"] — a value the
repository had handed back serialized as an ISO string. update() assigned
it to the DateTime column untouched, so SQLite raised StatementError
("only accepts Python datetime") and Postgres a DataError; only a cadence
change worked, because that path computes a fresh datetime.
Coerce every serialized timestamp update() accepts through the existing
_coerce_datetime, sharing one _TIMESTAMP_KEYS definition with the row
serializer, as update_after_launch() already does (#4785). The router test
now drives the real SQLite repository for the same-cadence rename and the
cadence change; the in-memory stand-in the other router tests use never
saw the type mismatch.
* docs(changelog): reference #5964 in the scheduled-task PATCH entries
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
34a105c084
commit
1e26f00664
11
CHANGELOG.md
11
CHANGELOG.md
@ -323,6 +323,16 @@ This release closes that milestone with **181 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **scheduler:** Editing an interval task's title or prompt no longer fails with
|
||||
a 500. The edit dialog always sends `schedule_spec` beside the changed field,
|
||||
and when the cadence is unchanged `PATCH /api/scheduled-tasks/{id}` keeps
|
||||
the task's existing `next_run_at` — a value the repository had handed back
|
||||
serialized as an ISO string. `ScheduledTaskRepository.update()` assigned it
|
||||
to the `DateTime` column untouched, so SQLite raised `StatementError`
|
||||
("only accepts Python datetime") and Postgres a `DataError`; only changing
|
||||
the cadence worked, because that path computes a fresh datetime. The
|
||||
repository now coerces every serialized timestamp it accepts in `update()`,
|
||||
as `update_after_launch()` already did. ([#5964])
|
||||
- **gateway:** `GET /api/skills`, `GET /api/skills/custom` and
|
||||
`GET /api/skills/{name}` no longer walk the skill directories on the event
|
||||
loop. Each called `load_skills()` inline, which resolves the caller's
|
||||
@ -6178,4 +6188,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#5934]: https://github.com/bytedance/deer-flow/pull/5934
|
||||
[#5935]: https://github.com/bytedance/deer-flow/pull/5935
|
||||
[#5945]: https://github.com/bytedance/deer-flow/pull/5945
|
||||
[#5964]: https://github.com/bytedance/deer-flow/pull/5964
|
||||
|
||||
|
||||
@ -328,6 +328,12 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **调度器:** 修改间隔任务的标题或 prompt 不再返回 500。编辑对话框总是把 `schedule_spec` 与被修改的
|
||||
字段一起发送;当节奏没有变化时,`PATCH /api/scheduled-tasks/{id}` 会沿用任务原有的 `next_run_at`——
|
||||
而这个值是仓储以 ISO 字符串序列化后返回的。`ScheduledTaskRepository.update()` 原样把它赋给
|
||||
`DateTime` 列,于是 SQLite 抛出 `StatementError`("only accepts Python datetime"),Postgres 抛出
|
||||
`DataError`;只有改变节奏才能成功,因为那条路径会重新计算出 datetime。现在仓储在 `update()` 中
|
||||
会像 `update_after_launch()` 一样,把接收到的所有序列化时间戳转换回来。([#5964])
|
||||
- **网关:** `GET /api/skills`、`GET /api/skills/custom` 与 `GET /api/skills/{name}` 不再在事件循环上
|
||||
遍历技能目录。三者此前都内联调用 `load_skills()`:它会解析调用者的存储、扫描所有公共与自定义
|
||||
技能目录并解析每个 `SKILL.md`,工作量随已安装技能数增长。#5747 已经为自定义技能内容路由把同一
|
||||
@ -5246,3 +5252,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子
|
||||
[#5934]: https://github.com/bytedance/deer-flow/pull/5934
|
||||
[#5935]: https://github.com/bytedance/deer-flow/pull/5935
|
||||
[#5945]: https://github.com/bytedance/deer-flow/pull/5945
|
||||
[#5964]: https://github.com/bytedance/deer-flow/pull/5964
|
||||
|
||||
@ -36,6 +36,11 @@ def _lease_is_alive(lease_expires_at: datetime | None, *, now: datetime, grace_s
|
||||
return lease_expires_at >= now - timedelta(seconds=grace_seconds)
|
||||
|
||||
|
||||
# Task columns that ``_row_to_dict`` serializes as ISO strings and that every
|
||||
# write path must coerce back (``_coerce_datetime``) before binding.
|
||||
_TIMESTAMP_KEYS = frozenset({"created_at", "updated_at", "next_run_at", "last_run_at", "lease_expires_at"})
|
||||
|
||||
|
||||
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):
|
||||
@ -67,13 +72,7 @@ class ScheduledTaskRepository:
|
||||
@staticmethod
|
||||
def _row_to_dict(row: ScheduledTaskRow) -> dict[str, Any]:
|
||||
data = row.to_dict(exclude={"last_occurrence_seq"})
|
||||
for key in (
|
||||
"created_at",
|
||||
"updated_at",
|
||||
"next_run_at",
|
||||
"last_run_at",
|
||||
"lease_expires_at",
|
||||
):
|
||||
for key in _TIMESTAMP_KEYS:
|
||||
if data.get(key) is not None:
|
||||
data[key] = coerce_iso(data[key])
|
||||
return data
|
||||
@ -270,7 +269,10 @@ class ScheduledTaskRepository:
|
||||
raise ActiveScheduledTaskMutationConflict(active_status)
|
||||
for key, value in updates.items():
|
||||
if hasattr(row, key):
|
||||
setattr(row, key, value)
|
||||
# Callers pass timestamps back in the serialized form
|
||||
# ``_row_to_dict`` returned (the PATCH route reuses an
|
||||
# interval task's ``next_run_at`` unchanged); bind datetimes.
|
||||
setattr(row, key, _coerce_datetime(value) if key in _TIMESTAMP_KEYS else value)
|
||||
row.updated_at = datetime.now(UTC)
|
||||
await session.commit()
|
||||
await session.refresh(row)
|
||||
|
||||
@ -854,6 +854,51 @@ async def test_update_after_launch_protect_terminal_keeps_hook_result(tmp_path):
|
||||
await close_engine()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_coerces_serialized_timestamps_it_previously_returned(tmp_path):
|
||||
"""``update()`` is fed the dict it handed out: the PATCH route reuses an interval task's
|
||||
``next_run_at`` unchanged, which ``_row_to_dict`` had serialized to an ISO string. Every
|
||||
timestamp column must be coerced back on the way in, as ``update_after_launch`` does."""
|
||||
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)
|
||||
created = await repo.create(
|
||||
task_id="task-update-serialized",
|
||||
user_id="user-1",
|
||||
thread_id=None,
|
||||
context_mode="fresh_thread_per_run",
|
||||
assistant_id="lead_agent",
|
||||
title="serialized update",
|
||||
prompt="p",
|
||||
schedule_type="interval",
|
||||
schedule_spec={"every_seconds": 3600},
|
||||
timezone="UTC",
|
||||
next_run_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
|
||||
)
|
||||
assert created["next_run_at"] == "2026-07-02T01:00:00+00:00"
|
||||
|
||||
updated = await repo.update(
|
||||
"task-update-serialized",
|
||||
user_id="user-1",
|
||||
updates={
|
||||
"title": "renamed",
|
||||
"next_run_at": created["next_run_at"], # the string form the router passes back
|
||||
"last_run_at": "2026-07-01T01:00:00Z", # the other accepted spelling
|
||||
"lease_expires_at": "2026-07-02T01:05:00+00:00",
|
||||
},
|
||||
)
|
||||
|
||||
assert updated is not None
|
||||
assert updated["title"] == "renamed"
|
||||
assert updated["next_run_at"] == "2026-07-02T01:00:00+00:00"
|
||||
assert updated["last_run_at"] == "2026-07-01T01:00:00+00:00"
|
||||
assert updated["lease_expires_at"] == "2026-07-02T01:05:00+00:00"
|
||||
|
||||
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."""
|
||||
|
||||
@ -460,6 +460,69 @@ async def test_update_rechecks_atomic_mutability_after_router_precheck(tmp_path)
|
||||
await close_engine()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("cadence_changed", [False, True], ids=["same-cadence", "new-cadence"])
|
||||
async def test_update_interval_task_metadata_through_the_sql_repository(tmp_path, cadence_changed):
|
||||
"""The edit dialog always sends ``schedule_spec`` beside the changed field. For an interval
|
||||
task whose cadence did not change the router keeps ``existing["next_run_at"]`` — which the
|
||||
repository hands back serialized as an ISO string — and passes it straight to
|
||||
``repo.update``. The SQL store assigned it to a ``DateTime`` column untouched, so renaming an
|
||||
interval task failed with a 500 on SQLite and Postgres alike; the in-memory stand-in the other
|
||||
router tests use never noticed. Drive the real router against the real SQLite repository."""
|
||||
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
|
||||
old_repo = scheduled_tasks.get_scheduled_task_repo
|
||||
old_thread_store = scheduled_tasks.get_thread_store
|
||||
old_config = scheduled_tasks.get_config
|
||||
old_user = scheduled_tasks.get_optional_user_from_request
|
||||
try:
|
||||
sf = get_session_factory()
|
||||
assert sf is not None
|
||||
repo = ScheduledTaskRepository(sf)
|
||||
original_next_run_at = datetime(2030, 1, 1, 9, 0, tzinfo=UTC)
|
||||
task = await repo.create(
|
||||
task_id="task-router-interval-rename",
|
||||
user_id="user-1",
|
||||
thread_id=None,
|
||||
context_mode="fresh_thread_per_run",
|
||||
assistant_id="lead_agent",
|
||||
title="Hourly digest",
|
||||
prompt="summarize",
|
||||
schedule_type="interval",
|
||||
schedule_spec={"every_seconds": 3600},
|
||||
timezone="UTC",
|
||||
next_run_at=original_next_run_at,
|
||||
)
|
||||
assert isinstance(task["next_run_at"], str) # the serialized form the router reuses
|
||||
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
||||
scheduled_tasks.get_thread_store = lambda _request: SimpleNamespace(check_access=AsyncMock(return_value=True))
|
||||
scheduled_tasks.get_config = lambda: _Config()
|
||||
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=SimpleNamespace(id="user-1"))
|
||||
|
||||
every_seconds = 7200 if cadence_changed else 3600
|
||||
updated = await call_unwrapped(
|
||||
scheduled_tasks.update_scheduled_task,
|
||||
task_id=task["id"],
|
||||
request=SimpleNamespace(),
|
||||
body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Hourly digest (renamed)", schedule_spec={"every_seconds": every_seconds}),
|
||||
)
|
||||
|
||||
assert updated["title"] == "Hourly digest (renamed)"
|
||||
assert updated["schedule_spec"] == {"every_seconds": every_seconds}
|
||||
stored = await repo.get(task["id"], user_id="user-1")
|
||||
assert stored is not None and stored["title"] == "Hourly digest (renamed)"
|
||||
stored_next_run_at = datetime.fromisoformat(stored["next_run_at"].replace("Z", "+00:00"))
|
||||
if cadence_changed:
|
||||
assert stored_next_run_at != original_next_run_at # recomputed from the new cadence
|
||||
else:
|
||||
assert stored_next_run_at == original_next_run_at # a pure rename keeps the next occurrence
|
||||
finally:
|
||||
scheduled_tasks.get_scheduled_task_repo = old_repo
|
||||
scheduled_tasks.get_thread_store = old_thread_store
|
||||
scheduled_tasks.get_config = old_config
|
||||
scheduled_tasks.get_optional_user_from_request = old_user
|
||||
await close_engine()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_scheduled_task_deletes_repo_row():
|
||||
repo = _Repo()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user