定时任务:修复智能体执行 422、新增描述字段、公开页删除按钮等

- 定时任务执行时按 agent.id 解析智能体,老数据中存的中文显示名会回退按 name 查找,避免 422
- scheduled_tasks 表新增 description 字段(迁移 20260522_01),创建/编辑表单支持,公开页标题下显示
- 公开页(/public/scheduled-tasks/{id})新增删除单次执行记录入口,使用样式化 Dialog 确认
- 公开页左侧标题区精简:去掉 schedule/timezone/最近状态两行;执行列表标题剥离尾部时间戳,"计划"改为"执行时间"
- 工作区定时任务列表:超长标题改为按字符换行不再截断;按 created_at 倒序(新建任务在前)
- 启动 Alembic 升级加超时与单独线程,MySQL 缺库时自动 CREATE DATABASE
- vite build 增加 8GB 内存上限

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.7 committed 2026-05-22 10:55:59 +08:00
1 parent a892f55cb2
commit a258122e22
14 files changed
+451 -75

No files matched your search

+1 -1
View File
@@ -5,7 +5,7 @@
"type": "module",
"scripts": {
"dev": "vite",
"build": "vite build",
"build": "node --max-old-space-size=8192 ./node_modules/vite/bin/vite.js build",
"preview": "vite preview",
"typecheck": "tsc --noEmit"
},
@@ -46,6 +46,7 @@ export async function listScheduledTasks(params: ListScheduledTasksParams = {})
export async function createScheduledTask(data: {
name: string;
description?: string | null;
prompt: string;
schedule: string;
timezone?: string | null;
@@ -203,6 +204,7 @@ export async function updateScheduledTask(
taskId: string,
data: Partial<{
name: string;
description: string | null;
prompt: string;
schedule: string;
timezone: string;
@@ -329,6 +331,7 @@ export async function adminUpdateScheduledTask(
taskId: string,
data: Partial<{
name: string;
description: string | null;
prompt: string;
schedule: string;
timezone: string;
@@ -14,6 +14,7 @@ export type ScheduledTaskSubscription = {
export type ScheduledTask = {
task_id: string;
name: string;
description?: string | null;
prompt: string;
schedule_text: string;
cron_expr: string;
@@ -27,6 +28,8 @@ export type ScheduledTask = {
scheduler_thread_id: string;
published?: boolean;
published_at?: string | null;
created_at?: string | null;
updated_at?: string | null;
role?: ScheduledTaskRole;
_subscription?: ScheduledTaskSubscription | null;
creator_user_id?: string | null;
@@ -7,6 +7,7 @@ import {
FileText,
Loader2,
Pencil,
Trash2,
X,
} from "lucide-react";
import { useCallback, useEffect, useMemo, useState } from "react";
@@ -16,6 +17,14 @@ import { toast } from "sonner";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import {
Dialog,
DialogContent,
DialogDescription,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { ScrollArea } from "@/components/ui/scroll-area";
import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs";
import { Textarea } from "@/components/ui/textarea";
@@ -34,6 +43,7 @@ import { cn } from "@/lib/utils";
type PublicTask = {
task_id: string;
name: string;
description?: string | null;
schedule_text: string;
cron_expr?: string;
timezone: string;
@@ -142,6 +152,17 @@ async function updatePublicRunBody(runId: string, content: string): Promise<void
);
}
async function deletePublicRun(runId: string): Promise<void> {
const response = await fetch(
`${publicBaseUrl()}/runs/${encodeURIComponent(runId)}`,
{ method: "DELETE" },
);
if (!response.ok && response.status !== 204) {
const body = await response.json().catch(() => ({ detail: "Request failed" }));
throw new Error(typeof body.detail === "string" ? body.detail : "Request failed");
}
}
function publicRunFileDownloadUrl(runId: string, index: number): string {
return `${publicBaseUrl()}/runs/${encodeURIComponent(runId)}/files/${index}/download`;
}
@@ -196,6 +217,20 @@ function stripExtension(name: string): string {
return name.slice(0, dot);
}
/**
* Strip a trailing ` - YYYY-MM-DD HH:MM[:SS]` (or `_YYYY-MM-DD ...`) suffix
* that the agent appends to run titles. The scheduled-for column is already
* displayed separately below the title, so the duplicated date looks like noise.
*/
function stripTimestampSuffix(title: string): string {
if (!title) return title;
const cleaned = title.replace(
/[\s_]*[-_]?\s*\d{4}-\d{1,2}-\d{1,2}[\sT]+\d{1,2}:\d{2}(?::\d{2})?\s*$/,
"",
);
return cleaned.trim() || title;
}
const BODY_TAB_VALUE = "__body__";
/**
@@ -212,6 +247,8 @@ const BODY_TAB_VALUE = "__body__";
export default function PublicScheduledTaskPage() {
const { taskId } = useParams<{ taskId: string }>();
const [selectedRunId, setSelectedRunId] = useState<string | null>(null);
const [pendingDeleteRun, setPendingDeleteRun] = useState<{ id: string; title: string } | null>(null);
const queryClient = useQueryClient();
const { data, isLoading, isError, error } = useQuery({
queryKey: ["public-scheduled-task", taskId],
@@ -222,6 +259,21 @@ export default function PublicScheduledTaskPage() {
refetchInterval: 8000,
});
const deleteRunMutation = useMutation({
mutationFn: (runId: string) => deletePublicRun(runId),
onSuccess: async (_data, runId) => {
toast.success("已删除");
if (selectedRunId === runId) {
setSelectedRunId(null);
}
setPendingDeleteRun(null);
await queryClient.invalidateQueries({ queryKey: ["public-scheduled-task", taskId] });
},
onError: (err) => {
toast.error(err instanceof Error ? err.message : "删除失败");
},
});
// Public viewer only surfaces runs that have actual content to show:
// pending/running/failed are hidden, and succeeded runs with neither a body
// nor a Markdown attachment fall out too (nothing to render).
@@ -301,12 +353,11 @@ export default function PublicScheduledTaskPage() {
{task?.enabled ? "启用" : "暂停"}
</Badge>
</div>
<p className="text-muted-foreground truncate text-xs">
{task?.schedule_text} · {task?.timezone}
</p>
<p className="text-muted-foreground text-xs">
最近状态:{statusLabel(task?.last_status)} · 上次:{formatDate(task?.last_run_at)}
</p>
{task?.description && task.description.trim() && (
<p className="text-muted-foreground line-clamp-3 text-xs leading-relaxed break-all whitespace-pre-wrap">
{task.description.trim()}
</p>
)}
</CardHeader>
<CardContent className="min-h-0 flex-1 p-0">
<ScrollArea className="size-full">
@@ -319,15 +370,27 @@ export default function PublicScheduledTaskPage() {
{runs.map((run) => {
const active = run.id === selectedRunId;
const runRunning = run.status === "running" || run.status === "pending";
const runTitle = run.result?.title || formatDate(run.scheduled_for);
const runTitle = stripTimestampSuffix(
run.result?.title || formatDate(run.scheduled_for),
);
const isDeletingThis =
deleteRunMutation.isPending && deleteRunMutation.variables === run.id;
return (
<button
<div
key={run.id}
role="button"
tabIndex={0}
className={cn(
"bg-card hover:bg-accent/60 rounded-md border p-3 text-left transition-colors",
"bg-card hover:bg-accent/60 group rounded-md border p-3 text-left transition-colors cursor-pointer focus:outline-none focus-visible:ring-2 focus-visible:ring-ring",
active && "border-primary bg-accent",
)}
onClick={() => setSelectedRunId(run.id)}
onKeyDown={(event) => {
if (event.key === "Enter" || event.key === " ") {
event.preventDefault();
setSelectedRunId(run.id);
}
}}
>
<div className="flex items-start justify-between gap-2">
<div className="min-w-0 flex-1">
@@ -345,21 +408,40 @@ export default function PublicScheduledTaskPage() {
</TooltipContent>
</Tooltip>
<div className="text-muted-foreground mt-1 text-xs">
计划:{formatDate(run.scheduled_for)}
执行时间:{formatDate(run.scheduled_for)}
</div>
</div>
<Badge
variant={statusVariant(run.status)}
className={cn(
"shrink-0",
runRunning && "border-primary/30 bg-primary/10 text-primary",
)}
>
{runRunning && <Loader2 className="size-3 animate-spin" />}
{statusLabel(run.status)}
</Badge>
<div className="flex shrink-0 items-center gap-1">
<Badge
variant={statusVariant(run.status)}
className={cn(
"shrink-0",
runRunning && "border-primary/30 bg-primary/10 text-primary",
)}
>
{runRunning && <Loader2 className="size-3 animate-spin" />}
{statusLabel(run.status)}
</Badge>
<Button
size="icon"
variant="ghost"
className="text-muted-foreground hover:text-destructive size-7"
disabled={isDeletingThis}
onClick={(event) => {
event.stopPropagation();
setPendingDeleteRun({ id: run.id, title: runTitle });
}}
title="删除记录"
>
{isDeletingThis ? (
<Loader2 className="size-3.5 animate-spin" />
) : (
<Trash2 className="size-3.5" />
)}
</Button>
</div>
</div>
</button>
</div>
);
})}
</div>
@@ -377,6 +459,54 @@ export default function PublicScheduledTaskPage() {
)}
</Card>
</div>
<Dialog
open={pendingDeleteRun !== null}
onOpenChange={(open) => {
if (!open && !deleteRunMutation.isPending) {
setPendingDeleteRun(null);
}
}}
>
<DialogContent className="sm:max-w-[420px]">
<DialogHeader>
<DialogTitle>删除执行记录</DialogTitle>
<DialogDescription>
确认删除这条执行记录吗?此操作不可恢复。
</DialogDescription>
</DialogHeader>
{pendingDeleteRun && (
<div className="bg-muted/40 text-foreground rounded-md border px-3 py-2 text-sm break-all">
{pendingDeleteRun.title}
</div>
)}
<DialogFooter>
<Button
variant="outline"
onClick={() => setPendingDeleteRun(null)}
disabled={deleteRunMutation.isPending}
>
取消
</Button>
<Button
variant="destructive"
onClick={() => {
if (pendingDeleteRun) {
deleteRunMutation.mutate(pendingDeleteRun.id);
}
}}
disabled={deleteRunMutation.isPending}
>
{deleteRunMutation.isPending ? (
<Loader2 className="size-4 animate-spin" />
) : (
<Trash2 className="size-4" />
)}
删除
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
</TooltipProvider>
);
}
+26 -8
View File
@@ -78,6 +78,7 @@ function displayUsername(value?: string | null): string {
const emptyForm = {
name: "",
description: "",
schedule: "",
timezone: "Asia/Shanghai",
prompt: "",
@@ -116,9 +117,13 @@ export default function ScheduledTasksPage() {
const sortedTasks = useMemo(
() =>
[...tasks].sort((a, b) =>
String(a.next_run_at ?? "").localeCompare(String(b.next_run_at ?? "")),
),
[...tasks].sort((a, b) => {
// Newest task first. Fall back to next_run_at so tasks that lack a
// created_at (older rows / subscribed tasks) still order consistently.
const aKey = a.created_at ?? a.next_run_at ?? "";
const bKey = b.created_at ?? b.next_run_at ?? "";
return String(bKey).localeCompare(String(aKey));
}),
[tasks],
);
const selectedTask = sortedTasks.find((task) => task.task_id === selectedTaskId) ?? null;
@@ -162,6 +167,7 @@ export default function ScheduledTasksPage() {
setEditing(task);
setForm({
name: task.name,
description: task.description ?? "",
schedule: task.schedule_text,
timezone: task.timezone,
prompt: task.prompt,
@@ -190,6 +196,7 @@ export default function ScheduledTasksPage() {
if (creating) {
const task = await actions.create.mutateAsync({
name: form.name || "定时任务",
description: form.description.trim() || null,
schedule: cronSchedule,
timezone: form.timezone || "Asia/Shanghai",
prompt: form.prompt,
@@ -202,6 +209,7 @@ export default function ScheduledTasksPage() {
taskId: editing.task_id,
data: {
name: form.name,
description: form.description.trim() || null,
schedule: cronSchedule,
timezone: form.timezone,
prompt: form.prompt,
@@ -324,9 +332,11 @@ export default function ScheduledTasksPage() {
onClick={() => setSelectedTaskId(task.task_id)}
>
<div className="flex items-start justify-between gap-3">
<div className="min-w-0">
<div className="flex items-center gap-2">
<div className="truncate font-medium">{task.name}</div>
<div className="min-w-0 flex-1">
<div className="font-medium leading-snug break-all">
{task.name}
</div>
<div className="mt-1 flex flex-wrap items-center gap-1">
{isRunningNow && (
<Badge
variant="secondary"
@@ -351,7 +361,7 @@ export default function ScheduledTasksPage() {
{task.schedule_text} · {task.timezone}
</div>
</div>
<Badge variant={task.enabled ? "default" : "secondary"}>
<Badge variant={task.enabled ? "default" : "secondary"} className="shrink-0">
{task.enabled ? "启用" : "暂停"}
</Badge>
</div>
@@ -760,6 +770,14 @@ export default function ScheduledTasksPage() {
}
placeholder="任务名称"
/>
<Textarea
className="min-h-16"
value={form.description}
onChange={(event) =>
setForm((value) => ({ ...value, description: event.target.value }))
}
placeholder="任务描述(可选,将显示在公开页标题下方)"
/>
<div className="flex flex-col gap-1">
<p className="text-muted-foreground text-xs">执行频率</p>
<ScheduleBuilder value={scheduleConfig} onChange={setScheduleConfig} />
@@ -786,7 +804,7 @@ export default function ScheduledTasksPage() {
<SelectContent>
<SelectItem value="__default__">默认助手(不指定)</SelectItem>
{agents.map((agent) => (
<SelectItem key={agent.name} value={agent.name}>
<SelectItem key={agent.id} value={agent.id}>
{agent.name}
{agent.description ? ` — ${agent.description}` : ""}
</SelectItem>
@@ -1,8 +1,10 @@
import asyncio
import logging
import os
import sys
from collections.abc import AsyncGenerator
import asyncio
import logging
import os
import queue
import sys
import threading
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from dotenv import load_dotenv
@@ -245,8 +247,7 @@ async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
async def _run_db_migrations() -> None:
"""Run Alembic migrations at startup so schema changes are applied automatically."""
import asyncio
from pathlib import Path
from pathlib import Path
from alembic import command
from alembic.config import Config
@@ -259,18 +260,51 @@ async def _run_db_migrations() -> None:
logger.info("Persistence backend=memory — skipping Alembic migrations")
return
db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
def _run_sync() -> None:
cfg = Config()
cfg.set_main_option("script_location", migrations_dir)
cfg.set_main_option("sqlalchemy.url", db_url)
command.upgrade(cfg, "head")
try:
await asyncio.get_running_loop().run_in_executor(None, _run_sync)
logger.info("Database migrations applied (head)")
if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations")
return
db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1)
def _run_sync() -> None:
try:
cfg = Config()
cfg.set_main_option("script_location", migrations_dir)
cfg.set_main_option("sqlalchemy.url", db_url)
command.upgrade(cfg, "head")
except BaseException as exc:
result_queue.put(exc)
else:
result_queue.put(None)
try:
try:
timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
except ValueError:
timeout_s = 60.0
worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True)
worker.start()
try:
deadline = asyncio.get_running_loop().time() + timeout_s
while result_queue.empty():
if asyncio.get_running_loop().time() >= deadline:
raise asyncio.TimeoutError
await asyncio.sleep(0.2)
result = result_queue.get_nowait()
except asyncio.TimeoutError:
logger.error(
"Database migration did not finish within %.0fs; continuing startup on create_all schema. "
"Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 "
"to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.",
timeout_s,
)
return
if result is not None:
raise result
logger.info("Database migrations applied (head)")
except Exception:
logger.exception("Database migration failed — schema may be out of date")
@@ -15,6 +15,7 @@ Surface area:
- ``PUT /api/public/scheduled-tasks/runs/{run_id}/files/{i}``— overwrite an editable result file
- ``PUT /api/public/scheduled-tasks/runs/{run_id}/body`` — overwrite the inline Markdown body
- ``GET /api/public/scheduled-tasks/runs/{run_id}/download`` — zip / md download of result
- ``DELETE /api/public/scheduled-tasks/runs/{run_id}`` — remove a single run (no auth)
The PUT endpoints intentionally skip ownership checks so the anonymous public
viewer at ``#/public/scheduled-tasks/<task_id>`` can fix typos in result MD
@@ -355,6 +356,20 @@ async def public_update_run_file(
raise HTTPException(status_code=400, detail=str(exc))
@router.delete("/runs/{run_id}", status_code=204)
async def public_delete_run(run_id: str) -> Response:
"""Remove a single run regardless of owner.
Anonymous — no ownership check, mirroring the existing public PUT endpoints.
Tighten this if the deployment requires read-only public sharing.
"""
service = get_scheduled_task_service()
deleted = await service.admin_delete_run(run_id)
if not deleted:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
return Response(status_code=204)
@router.put("/runs/{run_id}/body")
async def public_update_run_body(run_id: str, body: PublicRunBodyUpdateRequest) -> dict[str, Any]:
"""Overwrite the inline Markdown body (``result.content``) of a run.
@@ -31,6 +31,7 @@ router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
class ScheduledTaskCreateRequest(BaseModel):
name: str = Field(default="Scheduled task")
description: str | None = None
prompt: str
schedule: str
timezone: str | None = None
@@ -40,6 +41,7 @@ class ScheduledTaskCreateRequest(BaseModel):
class ScheduledTaskUpdateRequest(BaseModel):
name: str | None = None
description: str | None = None
prompt: str | None = None
schedule: str | None = None
timezone: str | None = None
@@ -547,6 +549,7 @@ async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Reque
task = await service.create_task(
user_id=user_id,
name=body.name,
description=body.description,
prompt=body.prompt,
schedule=body.schedule,
timezone=body.timezone,
@@ -11,14 +11,16 @@ None and fall back to in-memory implementations.
from __future__ import annotations
import asyncio
import json
import logging
import os
import sys
from importlib.util import find_spec
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
import json
import logging
import os
import sys
from importlib.util import find_spec
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.engine import make_url
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
def _json_serializer(obj: object) -> str:
@@ -131,7 +133,7 @@ async def _run_alembic_upgrade(url: str) -> None:
logger.exception("Auto-applying Alembic migrations failed; continuing with create_all schema")
async def _auto_create_postgres_db(url: str) -> None:
async def _auto_create_postgres_db(url: str) -> None:
"""Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
The target database name is extracted from *url*. The connection is
@@ -154,11 +156,50 @@ async def _auto_create_postgres_db(url: str) -> None:
async with maint_engine.connect() as conn:
await conn.execute(text(f'CREATE DATABASE "{db_name}"'))
logger.info("Auto-created PostgreSQL database: %s", db_name)
finally:
await maint_engine.dispose()
async def init_engine(
finally:
await maint_engine.dispose()
async def _auto_create_mysql_db(url: str) -> None:
"""Create the target MySQL database when it is missing.
``create_all`` can create tables, but MySQL requires the database/schema to
exist before the application URL can connect. We connect to the built-in
``mysql`` maintenance database on the same server and issue a conservative
``CREATE DATABASE IF NOT EXISTS``.
"""
parsed = make_url(url)
db_name = parsed.database
if not db_name:
raise ValueError("Cannot auto-create MySQL database: no database name in URL")
maint_url = parsed.set(database="mysql")
maint_engine = create_async_engine(
maint_url.render_as_string(hide_password=False),
isolation_level="AUTOCOMMIT",
pool_pre_ping=True,
)
quoted = "`" + db_name.replace("`", "``") + "`"
try:
async with maint_engine.connect() as conn:
await conn.execute(
text(
f"CREATE DATABASE IF NOT EXISTS {quoted} "
"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
)
)
logger.info("Auto-created MySQL database if missing: %s", db_name)
finally:
await maint_engine.dispose()
def _is_mysql_unknown_database_error(exc: Exception) -> bool:
message = str(exc).lower()
return "unknown database" in message or "1049" in message
async def init_engine(
backend: str,
*,
url: str = "",
@@ -261,21 +302,37 @@ async def init_engine(
# This is expected during initial scaffolding or minimal installs.
logger.debug("deerflow.persistence.models not found; skipping auto-create tables")
try:
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
except Exception as exc:
if backend == "postgres" and "does not exist" in str(exc):
try:
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
except Exception as exc:
if backend == "postgres" and "does not exist" in str(exc):
# Database not yet created — attempt to auto-create it, then retry.
await _auto_create_postgres_db(url)
# Rebuild engine against the now-existing database
await _engine.dispose()
_engine = create_async_engine(url, echo=echo, pool_size=pool_size, pool_pre_ping=True, json_serializer=_json_serializer)
_session_factory = async_sessionmaker(_engine, expire_on_commit=False)
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
else:
raise
_session_factory = async_sessionmaker(_engine, expire_on_commit=False)
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
elif backend == "mysql" and _is_mysql_unknown_database_error(exc):
# Database not yet created. Create it, rebuild the app engine
# against the target DB, then let SQLAlchemy create all tables.
await _auto_create_mysql_db(url)
await _engine.dispose()
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
pool_recycle=1800,
json_serializer=_json_serializer,
)
_session_factory = async_sessionmaker(_engine, expire_on_commit=False)
async with _engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
else:
raise
# Apply in-place schema fixes (column widenings, etc.) that ``create_all``
# cannot do on its own. Failures don't abort startup — see the helper.
@@ -0,0 +1,24 @@
"""Merge agent pin/featured and user memory config migration heads.
Revision ID: 20260521_02
Revises: 20260520_03, 20260521_01
Create Date: 2026-05-21
"""
from __future__ import annotations
from collections.abc import Sequence
revision: str = "20260521_02"
down_revision: str | tuple[str, str] = ("20260520_03", "20260521_01")
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -0,0 +1,36 @@
"""Add description column to scheduled_tasks.
Revision ID: 20260522_01
Revises: 20260521_02
Create Date: 2026-05-22
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260522_01"
down_revision = "20260521_02"
branch_labels = None
depends_on = None
def _has_column(table_name: str, column_name: str) -> bool:
inspector = inspect(op.get_bind())
return any(column["name"] == column_name for column in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_column("scheduled_tasks", "description"):
op.add_column(
"scheduled_tasks",
sa.Column("description", sa.Text(), nullable=True),
)
def downgrade() -> None:
if _has_column("scheduled_tasks", "description"):
op.drop_column("scheduled_tasks", "description")
@@ -31,6 +31,7 @@ class ScheduledTaskRow(Base):
source_agent_name: Mapped[str | None] = mapped_column(String(128), index=True)
execution_agent_name: Mapped[str | None] = mapped_column(String(128), index=True)
name: Mapped[str] = mapped_column(String(127))
description: Mapped[str | None] = mapped_column(Text)
prompt: Mapped[str] = mapped_column(Text)
schedule_text: Mapped[str] = mapped_column(String(256))
cron_expr: Mapped[str] = mapped_column(String(128))
@@ -76,6 +76,7 @@ class ScheduledTaskRepository(ScheduledTaskStore):
source_agent_name=data.get("source_agent_name"),
execution_agent_name=data.get("execution_agent_name"),
name=data["name"],
description=data.get("description"),
prompt=data["prompt"],
schedule_text=data["schedule_text"],
cron_expr=data["cron_expr"],
@@ -18,6 +18,7 @@ from zoneinfo import ZoneInfo
from croniter import croniter
from langgraph.checkpoint.base import empty_checkpoint
from deerflow.config.agents_config import AGENT_ID_PATTERN
from deerflow.config.paths import get_paths
from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
from deerflow.runtime import serialize_channel_values
@@ -445,6 +446,7 @@ class ScheduledTaskService:
prompt: str,
schedule: str,
timezone: str | None = None,
description: str | None = None,
source_thread_id: str | None = None,
source_agent_name: str | None = None,
execution_agent_name: str | None = None,
@@ -459,6 +461,7 @@ class ScheduledTaskService:
context = dict(execution_context or {})
if execution_agent_name:
context["agent_name"] = execution_agent_name
normalized_description = description.strip() if isinstance(description, str) else None
task = await self.store.create_task(
{
"task_id": str(uuid.uuid4()),
@@ -468,6 +471,7 @@ class ScheduledTaskService:
"source_agent_name": source_agent_name,
"execution_agent_name": execution_agent_name,
"name": final_name,
"description": normalized_description or None,
"prompt": prompt.strip(),
"schedule_text": schedule.strip(),
"cron_expr": cron_expr,
@@ -490,6 +494,15 @@ class ScheduledTaskService:
for key in ("name", "prompt", "enabled", "execution_agent_name"):
if key in data and data[key] is not None:
update_data[key] = data[key]
# Description is nullable — accept ``None`` (clear) and empty strings
# (also clear) explicitly, so the field can be removed via PATCH.
if "description" in data:
raw = data["description"]
if isinstance(raw, str):
trimmed = raw.strip()
update_data["description"] = trimmed or None
else:
update_data["description"] = None
# Names are matched against the per-user uniqueness constraint, so keep
# the stored value trimmed to match create_task's normalization.
if isinstance(update_data.get("name"), str):
@@ -858,8 +871,17 @@ class ScheduledTaskService:
try:
request = SimpleNamespace(app=self.app)
execution_context = dict(task.get("execution_context") or {})
if task.get("execution_agent_name"):
execution_context["agent_name"] = task["execution_agent_name"]
stored_agent_ref = task.get("execution_agent_name")
if stored_agent_ref:
resolved_agent_id = await self._resolve_agent_id(stored_agent_ref, task["user_id"])
if resolved_agent_id:
execution_context["agent_name"] = resolved_agent_id
else:
logger.warning(
"Scheduled task %s references unknown agent %r; falling back to default lead agent",
task.get("task_id"),
stored_agent_ref,
)
execution_context.update(
{
"is_scheduled_run": True,
@@ -884,6 +906,35 @@ class ScheduledTaskService:
finally:
reset_current_user(token)
async def _resolve_agent_id(self, stored_ref: str, user_id: str) -> str | None:
"""Resolve a scheduled task's ``execution_agent_name`` to a stable agent id.
Older tasks (created before the frontend was fixed) stored the agent's
Chinese **display name** here; the runtime validator only accepts ids
matching ``^[A-Za-z0-9_-]+$``, so those rows would 422 forever. We
gracefully recover by looking the stored value up first as an id, then
as a display name. Returns ``None`` when nothing visible matches — the
caller then falls back to the default lead agent rather than crashing.
"""
store = getattr(self.app.state, "agent_store", None)
if store is None:
return stored_ref if AGENT_ID_PATTERN.fullmatch(stored_ref) else None
if AGENT_ID_PATTERN.fullmatch(stored_ref):
try:
row = await store.get_visible(stored_ref, user_id)
except Exception:
row = None
if row is not None:
return row["id"]
try:
agents = await store.list_visible(user_id)
except Exception:
return None
for agent in agents:
if agent.get("name") == stored_ref:
return agent.get("id")
return None
async def _extract_latest_ai_text(self, thread_id: str) -> str | None:
checkpointer = self.app.state.checkpointer
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}