github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-27 00:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity
deer-flow/backend/app/gateway/routers/scheduled_tasks.py
Xinmin Zeng 4fc08b4f15
feat: add scheduled tasks MVP (#3898) 
* 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>
2026-07-04 21:51:57 +08:00

311 lines
13 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import uuid
from datetime import UTC, datetime
from typing import Any
from fastapi import APIRouter, HTTPException, Query, Request
from pydantic import BaseModel, Field
from app.gateway.authz import require_permission
from app.gateway.deps import (
get_config,
get_optional_user_from_request,
get_scheduled_task_repo,
get_scheduled_task_run_repo,
get_scheduled_task_service,
get_thread_store,
)
from deerflow.scheduler.schedules import (
next_run_at as compute_next_run_at,
)
from deerflow.scheduler.schedules import (
normalize_cron_expression,
validate_timezone,
)
router = APIRouter(prefix="/api", tags=["scheduled-tasks"])
def _ensure_task_mutable(task: dict[str, Any]) -> None:
if task.get("status") == "running":
raise HTTPException(
status_code=409,
detail="Scheduled task is currently running; retry after the active execution finishes",
)
class ScheduledTaskCreateRequest(BaseModel):
thread_id: str | None = None
context_mode: str = "fresh_thread_per_run"
title: str = Field(min_length=1)
prompt: str = Field(min_length=1)
schedule_type: str
schedule_spec: dict[str, Any]
timezone: str
class ScheduledTaskUpdateRequest(BaseModel):
context_mode: str | None = None
thread_id: str | None = None
title: str | None = Field(default=None, min_length=1)
prompt: str | None = Field(default=None, min_length=1)
schedule_spec: dict[str, Any] | None = None
timezone: str | None = None
@router.get("/scheduled-tasks")
@require_permission("threads", "read")
async def list_scheduled_tasks(request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
return []
return await repo.list_by_user(str(user.id))
@router.post("/scheduled-tasks")
@require_permission("threads", "write")
async def create_scheduled_task(request: Request, body: ScheduledTaskCreateRequest):
config = get_config()
repo = get_scheduled_task_repo(request)
thread_store = get_thread_store(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
if body.context_mode not in {"fresh_thread_per_run", "reuse_thread"}:
raise HTTPException(status_code=422, detail="Unsupported context_mode")
if body.context_mode == "reuse_thread":
if not body.thread_id:
raise HTTPException(status_code=422, detail="reuse_thread requires thread_id")
if not await thread_store.check_access(body.thread_id, str(user.id), require_existing=True):
raise HTTPException(status_code=404, detail="Thread not found")
if body.schedule_type not in {"once", "cron"}:
raise HTTPException(status_code=422, detail="Unsupported schedule_type")
schedule_spec = dict(body.schedule_spec)
try:
validate_timezone(body.timezone)
if body.schedule_type == "cron":
raw_cron = schedule_spec.get("cron")
if not isinstance(raw_cron, str):
raise HTTPException(status_code=422, detail="cron schedule requires schedule_spec.cron")
schedule_spec["cron"] = normalize_cron_expression(raw_cron)
next_run_at = compute_next_run_at(
body.schedule_type,
schedule_spec,
body.timezone,
now=datetime.now(UTC),
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if body.schedule_type == "once" and next_run_at is None:
raise HTTPException(status_code=422, detail="once schedule must be in the future")
if body.schedule_type == "once" and next_run_at is not None and (next_run_at - datetime.now(UTC)).total_seconds() < config.scheduler.min_once_delay_seconds:
raise HTTPException(
status_code=422,
detail=(f"once schedule must be at least {config.scheduler.min_once_delay_seconds} seconds in the future"),
)
return await repo.create(
task_id=f"task-{uuid.uuid4().hex}",
user_id=str(user.id),
thread_id=body.thread_id,
context_mode=body.context_mode,
assistant_id="lead_agent",
title=body.title,
prompt=body.prompt,
schedule_type=body.schedule_type,
schedule_spec=schedule_spec,
timezone=body.timezone,
next_run_at=next_run_at,
)
@router.get("/scheduled-tasks/{task_id}")
@require_permission("threads", "read")
async def get_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
task = await repo.get(task_id, user_id=str(user.id))
if task is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return task
@router.patch("/scheduled-tasks/{task_id}")
@require_permission("threads", "write")
async def update_scheduled_task(task_id: str, request: Request, body: ScheduledTaskUpdateRequest):
config = get_config()
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
existing = await repo.get(task_id, user_id=str(user.id))
if existing is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
_ensure_task_mutable(existing)
updates = body.model_dump(exclude_none=True)
if "context_mode" in updates:
if updates["context_mode"] not in {"fresh_thread_per_run", "reuse_thread"}:
raise HTTPException(status_code=422, detail="Unsupported context_mode")
effective_context_mode = str(updates.get("context_mode", existing["context_mode"]))
effective_thread_id = updates.get("thread_id", existing.get("thread_id"))
if effective_context_mode == "reuse_thread":
if not effective_thread_id:
raise HTTPException(status_code=422, detail="reuse_thread requires thread_id")
thread_store = get_thread_store(request)
if not await thread_store.check_access(str(effective_thread_id), str(user.id), require_existing=True):
raise HTTPException(status_code=404, detail="Thread not found")
elif effective_context_mode == "fresh_thread_per_run":
updates["thread_id"] = None
if "timezone" in updates:
try:
validate_timezone(str(updates["timezone"]))
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if "schedule_spec" in updates or "timezone" in updates:
schedule_spec = dict(existing["schedule_spec"])
if "schedule_spec" in updates and isinstance(updates["schedule_spec"], dict):
schedule_spec = dict(updates["schedule_spec"])
timezone = str(updates.get("timezone", existing["timezone"]))
try:
if existing["schedule_type"] == "cron":
raw_cron = schedule_spec.get("cron")
if not isinstance(raw_cron, str):
raise HTTPException(
status_code=422,
detail="cron schedule requires schedule_spec.cron",
)
schedule_spec["cron"] = normalize_cron_expression(raw_cron)
next_run_at = compute_next_run_at(
existing["schedule_type"],
schedule_spec,
timezone,
now=datetime.now(UTC),
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if existing["schedule_type"] == "once" and next_run_at is None:
raise HTTPException(status_code=422, detail="once schedule must be in the future")
if existing["schedule_type"] == "once" and next_run_at is not None and (next_run_at - datetime.now(UTC)).total_seconds() < config.scheduler.min_once_delay_seconds:
raise HTTPException(
status_code=422,
detail=(f"once schedule must be at least {config.scheduler.min_once_delay_seconds} seconds in the future"),
)
updates["schedule_spec"] = schedule_spec
updates["next_run_at"] = next_run_at
# A terminal task (completed/failed/cancelled) whose schedule was just
# pushed into the future must be re-armed: claim_due_tasks only admits
# "enabled" rows, so leaving the terminal status would return 200 with
# a next_run_at that silently never fires.
if next_run_at is not None and existing["status"] in {"completed", "failed", "cancelled"}:
updates["status"] = "enabled"
updated = await repo.update(
task_id,
user_id=str(user.id),
updates=updates,
)
return updated
@router.post("/scheduled-tasks/{task_id}/pause")
@require_permission("threads", "write")
async def pause_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
existing = await repo.get(task_id, user_id=str(user.id))
if existing is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
_ensure_task_mutable(existing)
updated = await repo.update(task_id, user_id=str(user.id), updates={"status": "paused"})
if updated is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return updated
@router.post("/scheduled-tasks/{task_id}/resume")
@require_permission("threads", "write")
async def resume_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
existing = await repo.get(task_id, user_id=str(user.id))
if existing is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
_ensure_task_mutable(existing)
updated = await repo.update(task_id, user_id=str(user.id), updates={"status": "enabled"})
if updated is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return updated
@router.post("/scheduled-tasks/{task_id}/trigger")
@require_permission("threads", "write")
async def trigger_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
service = get_scheduled_task_service(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
task = await repo.get(task_id, user_id=str(user.id))
if task is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
result = await service.dispatch_task(task, now=datetime.now(UTC), trigger="manual")
if result["outcome"] == "conflict":
raise HTTPException(status_code=409, detail=result["error"] or "Scheduled task trigger conflicted with an active run")
if result["outcome"] == "failed":
raise HTTPException(status_code=502, detail=result["error"] or "Scheduled task trigger failed")
return {"id": task_id, "triggered": True}
@router.delete("/scheduled-tasks/{task_id}")
@require_permission("threads", "write")
async def delete_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
deleted = await repo.delete(task_id, user_id=str(user.id))
if not deleted:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return {"id": task_id, "deleted": deleted}
@router.get("/scheduled-tasks/{task_id}/runs")
@require_permission("threads", "read")
async def list_scheduled_task_runs(
task_id: str,
request: Request,
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
):
task_repo = get_scheduled_task_repo(request)
run_repo = get_scheduled_task_run_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
task = await task_repo.get(task_id, user_id=str(user.id))
if task is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return await run_repo.list_by_task(task_id, limit=limit, offset=offset)
@router.get("/threads/{thread_id}/scheduled-tasks")
@require_permission("threads", "read", owner_check=True)
async def list_thread_scheduled_tasks(thread_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
return await repo.list_by_user_and_thread(str(user.id), thread_id)
Reference in New Issue View Git Blame Copy Permalink