mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-28 08:56:13 +00:00
4fc08b4f15
* feat: add scheduled tasks MVP
* fix: harden scheduled task execution semantics
* feat(scheduled-tasks): preset-driven schedule form with timezone and live preview
Replace the raw cron input with a preset Select (hourly/daily/weekly/monthly/custom)
plus structured inputs (time picker, weekday toggles, day-of-month), datetime-local
for one-time tasks, a timezone selector defaulting to the browser timezone, and a
live human-readable preview. Reuses one ScheduledTaskScheduleInput for create and
edit; backend contract unchanged; zero new deps (pure Intl + DST-safe offset helpers).
* feat(scheduled-tasks): full-page i18n + recipe templates + E2E locale pin
Localize the rest of the scheduled-tasks page (filters, detail pane, actions,
edit form, run list, enum values) via t.scheduledTasks.* in en/zh. Add four
built-in recipe templates (GitHub Trending, news digest, issue triage, weekly
report) exposed as a chip row that pre-fills title + prompt + schedule. Pin
Playwright locale to en-US so E2E selectors stay stable against i18n. No backend
change, no new deps.
* fix(scheduled-tasks): idempotent 0003 migration, update head constants, future-date once test
Merge with main surfaced three CI failures:
- 0003_scheduled_tasks create_table collided with legacy test seeds that
build from full metadata; guard with inspector.has_table so the revision
no-ops when the table already exists (0004/0005 are already idempotent via
_helpers.py).
- persistence bootstrap concurrency/regression tests pinned HEAD to main's
0002_runs_token_usage; bump to the new head 0005_scheduled_task_thread_nullable.
- once-task router test used a fixed past run_at and tripped the
must-be-in-the-future validation; use a future date.
* address review: ok-check, 502 for trigger failure, mock fields, migration filename, doc fences
- fetchThreadScheduledTasks now checks response.ok like the other fetchers.
- trigger endpoint returns 502 (not 409) when dispatch fails outright, so
clients can distinguish a real conflict from a server-side failure.
- E2E mock normalizes scheduled-task objects with context_mode/last_thread_id
and nullable thread_id, matching the backend contract the UI renders against.
- Rename 0002_scheduled_tasks.py -> 0003_scheduled_tasks.py to match its
revision id (file was renamed in spirit already; filename now follows).
- CONFIGURATION.md: close the Tool Groups yaml fence and drop the stray fence
after the Scheduler notes so the sections render correctly.
* fix(scheduled-tasks): harden lease, poller, config, and frontend UX after review
* fix(scheduled-tasks): harden run lifecycle, overlap skip, non_interactive gating, and DST conversion after review
- defer a once task's terminal status to the run-completion hook; the task
stays running until the real outcome, and a startup sweep cancels once
tasks orphaned by a crash (launch-time 'completed' could stick forever)
- record interrupted runs as a distinct 'interrupted' run status with a
readable message; an interrupted once task ends 'cancelled', not 'failed'
- enforce overlap_policy=skip for fresh_thread_per_run via an active-run
pre-check (same-thread ConflictError can never fire across fresh threads)
- protect terminal run statuses from the late launch-path 'running' write
- honor context.non_interactive only for internally-authenticated callers;
arbitrary clients can no longer strip ask_clarification
- fix DST-stale timezone offset in zonedLocalToUtcIso by re-deriving the
offset at the resolved instant (once tasks fired an hour late around
spring-forward and the create->edit round-trip diverged)
- drop dead ScheduledTaskRunRepository.update_by_run_id; share one Gateway
API error helper between channels and scheduled-tasks frontends
* fix(scheduled-tasks): close review round-3 gaps in guards, concurrency, and API ergonomics
- scrub internal-only context keys (non_interactive) from the assembled run
config for non-internal callers: gating body.context alone left the same
key smuggle-able through the free-form body.config copied verbatim by
build_run_config
- guard update_after_launch with protect_terminal so the launch bookkeeping
write cannot clobber a once task already finalized by a fast-failing run's
completion hook (parent-row sibling of the run-row guard)
- reject a manual trigger while the task has an active run (409) instead of
launching a duplicate concurrent run on fresh_thread_per_run
- re-arm a terminal once task to enabled when PATCH pushes run_at into the
future; previously the endpoint returned 200 with a next_run_at that could
never be claimed
- make max_concurrent_runs a real global cap: each poll claims only into the
remaining budget of active (queued/running) scheduled runs
- paginate GET /scheduled-tasks/{id}/runs (limit<=200, offset) and push the
thread filter of /threads/{id}/scheduled-tasks into SQL
- stamp context.user_id on scheduler-launched runs, matching IM channels, so
user-scoped guardrail providers see the owning user
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
233 lines
8.6 KiB
Python
233 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy import and_, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
|
|
from deerflow.utils.time import coerce_iso
|
|
|
|
TERMINAL_TASK_STATUSES: frozenset[str] = frozenset({"completed", "failed", "cancelled"})
|
|
|
|
|
|
class ScheduledTaskRepository:
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
|
self._sf = session_factory
|
|
|
|
@staticmethod
|
|
def _row_to_dict(row: ScheduledTaskRow) -> dict[str, Any]:
|
|
data = row.to_dict()
|
|
for key in (
|
|
"created_at",
|
|
"updated_at",
|
|
"next_run_at",
|
|
"last_run_at",
|
|
"lease_expires_at",
|
|
):
|
|
if data.get(key) is not None:
|
|
data[key] = coerce_iso(data[key])
|
|
return data
|
|
|
|
async def create(
|
|
self,
|
|
*,
|
|
task_id: str,
|
|
user_id: str,
|
|
thread_id: str | None,
|
|
context_mode: str,
|
|
assistant_id: str | None,
|
|
title: str,
|
|
prompt: str,
|
|
schedule_type: str,
|
|
schedule_spec: dict[str, Any],
|
|
timezone: str,
|
|
next_run_at: datetime | None,
|
|
) -> dict[str, Any]:
|
|
now = datetime.now(UTC)
|
|
row = ScheduledTaskRow(
|
|
id=task_id,
|
|
user_id=user_id,
|
|
thread_id=thread_id,
|
|
context_mode=context_mode,
|
|
assistant_id=assistant_id,
|
|
title=title,
|
|
prompt=prompt,
|
|
schedule_type=schedule_type,
|
|
schedule_spec=schedule_spec,
|
|
timezone=timezone,
|
|
next_run_at=next_run_at,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
async with self._sf() as session:
|
|
session.add(row)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return self._row_to_dict(row)
|
|
|
|
async def get(self, task_id: str, *, user_id: str) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(ScheduledTaskRow, task_id)
|
|
if row is None or row.user_id != user_id:
|
|
return None
|
|
return self._row_to_dict(row)
|
|
|
|
async def list_by_user(self, user_id: str) -> list[dict[str, Any]]:
|
|
stmt = select(ScheduledTaskRow).where(ScheduledTaskRow.user_id == user_id).order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc())
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
return [self._row_to_dict(row) for row in result.scalars()]
|
|
|
|
async def update(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
user_id: str,
|
|
updates: dict[str, Any],
|
|
) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
row = await session.get(ScheduledTaskRow, task_id)
|
|
if row is None or row.user_id != user_id:
|
|
return None
|
|
for key, value in updates.items():
|
|
if hasattr(row, key):
|
|
setattr(row, key, value)
|
|
row.updated_at = datetime.now(UTC)
|
|
await session.commit()
|
|
await session.refresh(row)
|
|
return self._row_to_dict(row)
|
|
|
|
async def delete(self, task_id: str, *, user_id: str) -> bool:
|
|
async with self._sf() as session:
|
|
row = await session.get(ScheduledTaskRow, task_id)
|
|
if row is None or row.user_id != user_id:
|
|
return False
|
|
await session.delete(row)
|
|
await session.commit()
|
|
return True
|
|
|
|
async def claim_due_tasks(
|
|
self,
|
|
*,
|
|
now: datetime,
|
|
lease_owner: str,
|
|
lease_seconds: int,
|
|
limit: int,
|
|
) -> list[dict[str, Any]]:
|
|
lease_expires_at = now + timedelta(seconds=lease_seconds)
|
|
stmt = (
|
|
select(ScheduledTaskRow)
|
|
.where(
|
|
ScheduledTaskRow.next_run_at.is_not(None),
|
|
ScheduledTaskRow.next_run_at <= now,
|
|
or_(
|
|
and_(
|
|
ScheduledTaskRow.status == "enabled",
|
|
or_(
|
|
ScheduledTaskRow.lease_expires_at.is_(None),
|
|
ScheduledTaskRow.lease_expires_at < now,
|
|
),
|
|
),
|
|
# A task stuck in "running" with an expired lease means the
|
|
# claiming process died between claim and dispatch; it must
|
|
# stay reclaimable or the task is dead forever.
|
|
and_(
|
|
ScheduledTaskRow.status == "running",
|
|
ScheduledTaskRow.lease_expires_at.is_not(None),
|
|
ScheduledTaskRow.lease_expires_at < now,
|
|
),
|
|
),
|
|
)
|
|
.order_by(ScheduledTaskRow.next_run_at.asc(), ScheduledTaskRow.id.asc())
|
|
.limit(limit)
|
|
.with_for_update(skip_locked=True)
|
|
)
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
rows = list(result.scalars())
|
|
for row in rows:
|
|
row.lease_owner = lease_owner
|
|
row.lease_expires_at = lease_expires_at
|
|
row.status = "running"
|
|
row.updated_at = datetime.now(UTC)
|
|
await session.commit()
|
|
return [self._row_to_dict(row) for row in rows]
|
|
|
|
async def update_after_launch(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
status: str,
|
|
next_run_at: datetime | None,
|
|
last_run_at: datetime | None,
|
|
last_run_id: str | None,
|
|
last_thread_id: str | None,
|
|
last_error: str | None,
|
|
increment_run_count: bool,
|
|
protect_terminal: bool = False,
|
|
) -> None:
|
|
async with self._sf() as session:
|
|
row = await session.get(ScheduledTaskRow, task_id)
|
|
if row is None:
|
|
return
|
|
if protect_terminal and row.status in TERMINAL_TASK_STATUSES:
|
|
# A fast-failing run can reach handle_run_completion (which
|
|
# finalizes a `once` task) before this launch-path write
|
|
# commits; keep the hook's status/error and only record the
|
|
# launch bookkeeping.
|
|
pass
|
|
else:
|
|
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_id = last_run_id
|
|
row.last_thread_id = last_thread_id
|
|
if increment_run_count:
|
|
row.run_count += 1
|
|
row.lease_owner = None
|
|
row.lease_expires_at = None
|
|
row.updated_at = datetime.now(UTC)
|
|
await session.commit()
|
|
|
|
async def list_by_user_and_thread(self, user_id: str, thread_id: str) -> list[dict[str, Any]]:
|
|
stmt = (
|
|
select(ScheduledTaskRow)
|
|
.where(
|
|
ScheduledTaskRow.user_id == user_id,
|
|
ScheduledTaskRow.thread_id == thread_id,
|
|
)
|
|
.order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc())
|
|
)
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
return [self._row_to_dict(row) for row in result.scalars()]
|
|
|
|
async def cancel_stuck_once_tasks(self, *, error: str) -> int:
|
|
"""Reconcile ``once`` tasks orphaned in ``running`` by a process crash.
|
|
|
|
A launched ``once`` task stays ``running`` until the in-process
|
|
completion hook moves it to a terminal status; its lease was cleared at
|
|
launch, so the claim query's expired-lease reclaim branch never sees
|
|
it. After a crash the hook is gone and the task would be stuck forever.
|
|
Tasks still holding a lease are left alone — they were claimed but not
|
|
launched, and expired-lease reclaim recovers them safely.
|
|
"""
|
|
stmt = select(ScheduledTaskRow).where(
|
|
ScheduledTaskRow.schedule_type == "once",
|
|
ScheduledTaskRow.status == "running",
|
|
ScheduledTaskRow.lease_expires_at.is_(None),
|
|
)
|
|
async with self._sf() as session:
|
|
result = await session.execute(stmt)
|
|
rows = list(result.scalars())
|
|
now = datetime.now(UTC)
|
|
for row in rows:
|
|
row.status = "cancelled"
|
|
row.last_error = error
|
|
row.updated_at = now
|
|
await session.commit()
|
|
return len(rows)
|