feat: 业务库切换 MySQL + 修复定时任务运行详情递归与重名 500

- 业务 ORM 数据支持 MySQL 后端(asyncmy),新增 PortableJSON/
  BeijingDateTime 可移植列类型、迁移脚本与切库脚本
- 修复 GET /api/scheduled-tasks/runs/{id} 的 _load_run 无限递归
  (误调用自身,应委托 service.get_run)导致的 RecursionError 500
- 定时任务名按用户唯一:create/update/admin_update 主动校验,
  冲突抛 DuplicateTaskNameError,路由转 409 而非裸 IntegrityError 500
- 补充回归测试 test_scheduled_task_run_lookup / _name_uniqueness

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.7 committed 2026-05-18 19:34:14 +08:00
1 parent d6f018e581
commit 358d5c72a6
40 files changed
+2128 -190

No files matched your search

-2
View File
@@ -47,11 +47,9 @@ offline-backend-20260505/bundle/start_backend.sh
offline-backend-20260505/bundle/stop_backend.sh
# offline-backend-20260512 runtime artifacts (large files, local config, data)
offline-backend-20260512/.env
offline-backend-20260512/.env.example
offline-backend-20260512/DEPLOY_BACKEND_OFFLINE_ZH.md
offline-backend-20260512/SHA256SUMS
offline-backend-20260512/config.yaml
offline-backend-20260512/config.yaml.example
offline-backend-20260512/data/
offline-backend-20260512/deerflow-backend-offline-image-20260505.tar
+7 -8
View File
@@ -1,14 +1,13 @@
import path from "node:path";
import { fileURLToPath } from "node:url";
import react from "@vitejs/plugin-react";
import sourceIdentifierPlugin from "vite-plugin-source-identifier";
import { defineConfig } from "vite";
import path from "node:path";
import { fileURLToPath } from "node:url";
import react from "@vitejs/plugin-react";
import { defineConfig } from "vite";
const rootDir = fileURLToPath(new URL(".", import.meta.url));
export default defineConfig({
plugins: [react(), sourceIdentifierPlugin()],
export default defineConfig({
plugins: [react()],
resolve: {
alias: [
{
+61
View File
@@ -0,0 +1,61 @@
# ========================
# EDIT BEFORE DEPLOYMENT
# ========================
# This file is loaded by start_backend.scheduler.sh through docker --env-file.
# Put server-specific values here. Do not configure Python mirrors only on the
# host shell; values in this file are what the container will receive.
# AI provider keys. Replace placeholders and rotate any test key before release.
DEEPSEEK_API_KEY=sk-cea9133f82e24e9f939136746b05d5f8
OPENAI_API_KEY=your-openai-api-key
ANTHROPIC_API_KEY=your-anthropic-api-key
TAVILY_API_KEY=your-tavily-api-key
SERPER_API_KEY=your-serper-api-key
# Auth and frontend access.
# AUTH_JWT_SECRET must stay stable across restarts, or existing logins expire.
# GATEWAY_CORS_ORIGINS must include the exact frontend origins, separated by commas.
AUTH_JWT_SECRET=i3_BiBK44sYHiFIkChhDs46y57ZewJKO3diQDw01anI
GATEWAY_CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
GATEWAY_CORS_ALLOW_ALL=0
GATEWAY_ENABLE_DOCS=true
GATEWAY_WORKERS=2
DEER_FLOW_AUTH_DISABLED=false
AUTH_TRUSTED_PROXIES=
# Scheduler dependency behavior.
# Keep SCHEDULER_INSTALL_DEPS=1 if croniter is not already baked into the image.
# Set the Python mirror below if the server cannot access public PyPI.
SCHEDULER_INSTALL_DEPS=1
SCHEDULER_PIP_PACKAGES=croniter==6.2.2
SCHEDULED_TASKS_ENABLED=true
# Element / Matrix delivery for scheduled task results.
# If disabled or not configured, scheduled tasks still run but no Element DM is sent.
ELEMENT_ENABLED=false
ELEMENT_HOMESERVER_URL=
ELEMENT_ACCESS_TOKEN=
ELEMENT_BOT_USERNAME=scheduled-bot
ELEMENT_BOT_PASSWORD=
ELEMENT_AUTO_REGISTER=false
ELEMENT_STATE_FILE=/app/backend/.deer-flow/element-bot-token
# Python private package index used by uv/pip inside the backend container and
# propagated into sandbox containers for agent-installed dependencies.
# Uncomment and set these to your internal mirror when public PyPI is blocked.
# UV_INDEX_URL=http://your-python-mirror/simple
# PIP_INDEX_URL=http://your-python-mirror/simple
# UV_TRUSTED_HOST=your-python-mirror
# PIP_TRUSTED_HOST=your-python-mirror
# PIP_EXTRA_INDEX_URL=
# UV_EXTRA_INDEX_URL=
# HTTP_PROXY=
# HTTPS_PROXY=
# NO_PROXY=
# Alternative generic names also supported by the startup script.
# PYTHON_PACKAGE_INDEX_URL=http://your-python-mirror/simple
# PYTHON_EXTRA_INDEX_URL=
# PYTHON_TRUSTED_HOST=your-python-mirror
MYSQL_DATABASE_URL=mysql+asyncmy://kland:kland%4088124932@192.168.0.123:3306/deerflow?charset=utf8mb4
@@ -0,0 +1,255 @@
# Linux Docker 版:业务库切换 MySQL 发版说明
适用场景:Linux 服务器上使用 `offline-backend-20260512/start_backend.sh` 启动 Docker 后端。
原启动脚本保持不变:
```bash
bash start_backend.sh
```
单独新增切换脚本:
```bash
bash switch_database_to_mysql.sh
```
## 一、切换后的数据布局
迁移到 MySQL 的业务表:
```text
users
threads_meta
runs
run_events
feedback
agents
skills
scheduled_tasks
scheduled_task_runs
scheduled_task_subscriptions
scheduled_task_delivery_profiles
scheduler_threads
llm_call_metrics
notifications
recommended_questions
tags
tag_assignments
thread_shares
```
继续留在 SQLite:
```text
checkpoints
writes
```
SQLite 路径:
```text
宿主机: offline-backend-20260512/data/data/deerflow.db
容器内: /app/backend/.deer-flow/data/deerflow.db
```
## 二、前置条件
MySQL 建议提前建库:
```sql
CREATE DATABASE deerflow
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;
```
本次 Linux 内网库连接信息按以下方式配置:
```bash
MYSQL_DATABASE_URL='mysql+asyncmy://root:1qaz%40WSX@172.16.4.xxx:3306/deerflow?charset=utf8mb4'
```
注意:密码里的 `@` 在 URL 中必须写成 `%40`,所以 `1qaz@WSX` 要写成 `1qaz%40WSX`。
离线包 `wheelhouse` 中需要有:
```text
asyncmy-0.2.10-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl
```
当前包已经放入该 wheel。
如果使用内网 pip 私服,在 `.env` 中配置:
```env
PIP_INDEX_URL=http://your-pip-mirror/simple
PIP_TRUSTED_HOST=your-pip-mirror
UV_INDEX_URL=http://your-pip-mirror/simple
UV_TRUSTED_HOST=your-pip-mirror
```
这些变量会传给后端容器和 AIO 沙箱容器。
## 三、发版步骤
进入部署目录:
```bash
cd offline-backend-20260512
```
先启动一次后端,让 SQLite schema 自动升级到当前代码版本:
```bash
bash start_backend.sh
```
确认服务正常后停止:
```bash
bash stop_backend.sh
```
执行一键切换:
```bash
MYSQL_DATABASE_URL='mysql+asyncmy://root:1qaz%40WSX@172.16.4.xxx:3306/deerflow?charset=utf8mb4' \
bash switch_database_to_mysql.sh
```
如果数据库 `deerflow` 已经由 DBA 提前建好,或者当前账号不能访问 MySQL 系统库 `mysql`,可以跳过自动建库:
```bash
CREATE_DATABASE=0 \
MYSQL_DATABASE_URL='mysql+asyncmy://root:1qaz%40WSX@172.16.4.xxx:3306/deerflow?charset=utf8mb4' \
bash switch_database_to_mysql.sh
```
如果目标 MySQL 表已有旧数据,并确认要清空后重迁:
```bash
TRUNCATE_TARGET=1 \
MYSQL_DATABASE_URL='mysql+asyncmy://root:1qaz%40WSX@172.16.4.xxx:3306/deerflow?charset=utf8mb4' \
bash switch_database_to_mysql.sh
```
切换完成后,继续用原启动脚本:
```bash
bash start_backend.sh
```
## 四、脚本会自动做什么
`switch_database_to_mysql.sh` 会自动:
1. 检查 Docker 是否可用。
2. 检查后端容器是否已停止。
3. 备份:
```text
data/backups/mysql-switch-YYYYMMDD_HHMMSS/config.yaml.bak
data/backups/mysql-switch-YYYYMMDD_HHMMSS/env.bak
data/backups/mysql-switch-YYYYMMDD_HHMMSS/deerflow.db.bak
```
4. 启动一次临时迁移容器。
5. 从 `/app/wheelhouse` 安装 `asyncmy` 到持久目录:
```text
容器内: /app/backend/.deer-flow/python-vendor
宿主机: offline-backend-20260512/data/python-vendor
```
6. 创建/升级 MySQL schema。
7. 默认清理失败重跑留下的空表:
```text
DROP_EMPTY_TARGET_SCHEMA=1
```
该清理只会删除已知业务表中“全部为空”的半成品表;如果任意表已有数据,脚本不会删除,会继续交给迁移校验保护。
8. 从 SQLite 迁移业务表到 MySQL。
9. 校验每张业务表行数。
10. 修改 `config.yaml`:
```yaml
database:
backend: mysql
mysql_url: $MYSQL_DATABASE_URL
pool_size: 10
echo_sql: false
checkpointer:
type: sqlite
connection_string: /app/backend/.deer-flow/data/deerflow.db
```
11. 将 `MYSQL_DATABASE_URL` 写入 `.env`。
## 五、验证
健康检查:
```bash
curl http://<server-ip>:8001/health
```
功能验证:
```text
登录/退出
历史会话列表
打开旧会话
新建对话
定时任务
分享链接
后端日志无 MySQL driver 或连接错误
```
## 六、回滚
停止后端:
```bash
bash stop_backend.sh
```
恢复备份:
```bash
cp data/backups/mysql-switch-YYYYMMDD_HHMMSS/config.yaml.bak config.yaml
cp data/backups/mysql-switch-YYYYMMDD_HHMMSS/env.bak .env
cp data/backups/mysql-switch-YYYYMMDD_HHMMSS/deerflow.db.bak data/data/deerflow.db
```
重新启动:
```bash
bash start_backend.sh
```
## 七、注意事项
- `MYSQL_DATABASE_URL` 不要写 `127.0.0.1` 指向宿主机 MySQL。容器内的 `127.0.0.1` 是容器自己。
- 如果 MySQL 在宿主机,使用宿主机内网 IP 或 Docker 可访问主机名。
- 当前 Linux 内网 MySQL 示例使用 `172.16.4.xxx:3306`。
- 密码含 `@`、`#`、`?`、`/` 等 URL 特殊字符时必须 URL 编码;本次 `1qaz@WSX` 写成 `1qaz%40WSX`。
- 已同步 Windows 实测问题修复:
```text
老 MySQL 767 bytes 索引长度限制:已收短被索引/唯一约束覆盖的字符串列。
老 MySQL 不支持 JSON 类型:MySQL 下 JSON 字段自动使用 TEXT 存储,应用层仍读写 dict/list。
失败重跑的半成品空表:switch_database_to_mysql.sh 默认传入 --drop-empty-target-schema 自动清理。
```
- `data/data/deerflow.db` 不能删除,因为 `checkpoints/writes` 仍在里面。
- 后续归档文件建议放:
```text
容器内: /app/backend/.deer-flow/archives/conversations
宿主机: offline-backend-20260512/data/archives/conversations
```
该路径已在现有 Docker 持久卷内,不需要改启动脚本。
@@ -0,0 +1,230 @@
# Windows 本地版:业务库切换 MySQL 发版说明
适用场景:Windows 机器上直接在 `offline-backend-20260512/backend` 目录本地启动后端,不使用 Docker。
原本地启动方式保持不变:
```powershell
cd offline-backend-20260512\backend
$env:PYTHONPATH='.'
uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001
```
单独新增切换脚本:
```powershell
cd offline-backend-20260512
.\switch_database_to_mysql.ps1
```
## 一、切换后的数据布局
迁移到 MySQL 的业务表:
```text
users
threads_meta
runs
run_events
feedback
agents
skills
scheduled_tasks
scheduled_task_runs
scheduled_task_subscriptions
scheduled_task_delivery_profiles
scheduler_threads
llm_call_metrics
notifications
recommended_questions
tags
tag_assignments
thread_shares
```
继续留在 SQLite:
```text
checkpoints
writes
```
Windows 本地默认 SQLite 路径:
```text
offline-backend-20260512\backend\.deer-flow\data\deerflow.db
```
如果你的本地 `DEER_FLOW_HOME` 或 SQLite 路径不是默认值,执行切换脚本时用 `-SqliteDb` 指定真实文件。
## 二、前置条件
本机需要:
```text
Python 3.12+
uv
Windows 能访问的 MySQL
```
MySQL 建议提前建库:
```sql
CREATE DATABASE deerflow
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;
```
如果 MySQL 就在 Windows 本机,连接串可以使用 `127.0.0.1`:
```text
mysql+asyncmy://user:password@127.0.0.1:3306/deerflow?charset=utf8mb4
```
切换脚本会在本地 `backend` 的 uv 环境中安装:
```text
asyncmy==0.2.10
```
如果本机不能访问公网 pip,请先在当前 PowerShell 配置内网 pip 私服:
```powershell
$env:PIP_INDEX_URL='http://your-pip-mirror/simple'
$env:PIP_TRUSTED_HOST='your-pip-mirror'
$env:UV_INDEX_URL='http://your-pip-mirror/simple'
$env:UV_TRUSTED_HOST='your-pip-mirror'
```
## 三、发版步骤
先启动一次后端,让 SQLite schema 自动升级到当前代码版本:
```powershell
cd offline-backend-20260512\backend
$env:PYTHONPATH='.'
uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001
```
确认服务正常后按 `Ctrl + C` 停止后端。
回到发版目录,执行一键切换:
```powershell
cd offline-backend-20260512
$env:MYSQL_DATABASE_URL='mysql+asyncmy://user:password@127.0.0.1:3306/deerflow?charset=utf8mb4'
.\switch_database_to_mysql.ps1
```
也可以直接用参数传入:
```powershell
.\switch_database_to_mysql.ps1 -MysqlDatabaseUrl 'mysql+asyncmy://user:password@127.0.0.1:3306/deerflow?charset=utf8mb4'
```
如果 SQLite 不在默认路径:
```powershell
.\switch_database_to_mysql.ps1 `
-SqliteDb 'D:\deerflow-data\data\deerflow.db' `
-MysqlDatabaseUrl 'mysql+asyncmy://user:password@127.0.0.1:3306/deerflow?charset=utf8mb4'
```
如果目标 MySQL 表已有旧数据,并确认要清空后重迁:
```powershell
.\switch_database_to_mysql.ps1 `
-TruncateTarget 1 `
-MysqlDatabaseUrl 'mysql+asyncmy://user:password@127.0.0.1:3306/deerflow?charset=utf8mb4'
```
切换完成后,继续用原本地启动方式:
```powershell
cd backend
$env:PYTHONPATH='.'
uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001
```
## 四、脚本会自动做什么
`switch_database_to_mysql.ps1` 会自动:
1. 检查 `uv`、配置文件、迁移脚本、SQLite 文件是否存在。
2. 备份:
```text
backend\.deer-flow\backups\mysql-switch-YYYYMMDD_HHMMSS\config.yaml.bak
backend\.deer-flow\backups\mysql-switch-YYYYMMDD_HHMMSS\env.bak
backend\.deer-flow\backups\mysql-switch-YYYYMMDD_HHMMSS\deerflow.db.bak
```
3. 将 `MYSQL_DATABASE_URL` 写入 `.env`。
4. 如果 `backend\.venv` 不存在,自动执行 `uv venv`。
5. 执行 `uv pip install asyncmy==0.2.10`。
6. 创建/升级 MySQL schema。
7. 从 SQLite 迁移业务表到 MySQL。
8. 校验每张业务表行数。
9. 修改 `config.yaml`:
```yaml
database:
backend: mysql
mysql_url: $MYSQL_DATABASE_URL
pool_size: 10
echo_sql: false
checkpointer:
type: sqlite
connection_string: D:\...\offline-backend-20260512\backend\.deer-flow\data\deerflow.db
```
## 五、验证
启动后端后检查:
```powershell
(Invoke-WebRequest -UseBasicParsing http://127.0.0.1:8001/health).StatusCode
```
功能验证:
```text
登录/退出
历史会话列表
打开旧会话
新建对话
定时任务
分享链接
后端日志无 MySQL driver 或连接错误
```
## 六、回滚
停止本地后端进程后恢复备份:
```powershell
Copy-Item backend\.deer-flow\backups\mysql-switch-YYYYMMDD_HHMMSS\config.yaml.bak config.yaml -Force
Copy-Item backend\.deer-flow\backups\mysql-switch-YYYYMMDD_HHMMSS\env.bak .env -Force
Copy-Item backend\.deer-flow\backups\mysql-switch-YYYYMMDD_HHMMSS\deerflow.db.bak backend\.deer-flow\data\deerflow.db -Force
```
重新启动本地后端即可。
## 七、归档文件存储
后续 90 天前对话归档建议放在:
```text
offline-backend-20260512\backend\.deer-flow\archives\conversations
```
如果你设置了自定义 `DEER_FLOW_HOME`,则放在:
```text
%DEER_FLOW_HOME%\archives\conversations
```
Windows 本地版不涉及 Docker 挂载。这个目录只需要跟随你的本地数据目录一起备份即可。
归档文件被手动删除一部分时,正确行为应是该历史会话无法恢复并返回明确提示,不应该影响系统启动或 90 天内会话使用。归档功能正式实现时,需要把“归档文件缺失”作为可预期异常处理。
@@ -237,10 +237,19 @@ Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` →
**Stuck-run recovery**: each execution attempt is bounded by `_RUN_ATTEMPT_TIMEOUT_SECONDS` (30 min) via `asyncio.wait_for`. On timeout the run is retried up to `_MAX_RUN_RETRIES` (3) times, then marked `failed`. A watchdog (`_reconcile_stuck_runs`, run every poll) recovers runs left in `running` with no in-process owner — e.g. orphaned by a crash or restart — re-driving them through the same retry budget so a task can never stay perpetually "executing". The `scheduled_task_runs.retry_count` column tracks consumed retries.
**Unique task names**: task names are unique per user (`uq_scheduled_tasks_user_name`). `create_task`/`update_task`/`admin_update_task` proactively check via `_assert_name_available()` and raise `DuplicateTaskNameError` on a collision; the router translates that into HTTP `409` instead of letting the raw DB `IntegrityError` surface as a `500`.
### Custom Agent Persistence
Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
### Portable Column Types (`deerflow/persistence/types.py`)
Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` / `mysql` backends. **All ORM timestamp/JSON columns must use these, not the raw SQLAlchemy types.**
- **`PortableJSON`** — native `JSON` on SQLite/Postgres; `TEXT` (with manual `json.dumps`/`loads`) on MySQL, where older servers lack JSON support.
- **`BeijingDateTime`** — stores timestamps as Beijing (UTC+8) wall-clock time. The app records time with `datetime.now(UTC)`, but MySQL `DATETIME` / SQLite have no timezone and persist whatever wall clock the driver serializes — so a raw `DateTime` column would store UTC and read 8 hours behind Beijing. `BeijingDateTime` converts UTC→Beijing on bind and re-attaches `+08:00` on result, so the database holds Beijing time while Python still gets timezone-aware datetimes (instant comparisons stay correct, including in WHERE clauses). PostgreSQL `timestamptz` stores the instant unambiguously, so values pass through unchanged there. Uses a fixed UTC+8 offset (China has no DST), so no `tzdata` dependency.
### Sandbox System (`packages/harness/deerflow/sandbox/`)
**Interface**: Abstract `Sandbox` with `execute_command`, `read_file`, `write_file`, `list_dir`
@@ -254,7 +254,7 @@ async def _run_db_migrations() -> None:
logger.info("Persistence backend=memory — skipping Alembic migrations")
return
db_url = str(engine.url)
db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
def _run_sync() -> None:
@@ -15,7 +15,7 @@ from fastapi.responses import FileResponse, Response, StreamingResponse
from pydantic import BaseModel, Field
from app.gateway.deps import get_current_user, get_local_provider, get_optional_user_from_request, get_tag_store
from deerflow.runtime.scheduler import get_scheduled_task_service
from deerflow.runtime.scheduler import DuplicateTaskNameError, get_scheduled_task_service
router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
@@ -75,7 +75,7 @@ async def _load_run(service, request: Request, run_id: str, user_id: str) -> dic
Lets admins open run results for tasks they do not own (the admin
management view links into the shared run-detail page).
"""
run = await _load_run(service, request, run_id, user_id)
run = await service.get_run(user_id, run_id)
if run is None and await _is_admin(request):
run = await service.admin_get_run(run_id)
return run
@@ -381,15 +381,18 @@ async def update_delivery_profile(body: ScheduledTaskDeliveryProfileUpdateReques
async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Request) -> dict:
service = get_scheduled_task_service()
user_id = await _user_id(request)
task = await service.create_task(
user_id=user_id,
name=body.name,
prompt=body.prompt,
schedule=body.schedule,
timezone=body.timezone,
execution_agent_name=body.execution_agent_name,
execution_context=body.execution_context,
)
try:
task = await service.create_task(
user_id=user_id,
name=body.name,
prompt=body.prompt,
schedule=body.schedule,
timezone=body.timezone,
execution_agent_name=body.execution_agent_name,
execution_context=body.execution_context,
)
except DuplicateTaskNameError:
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
await _enrich_with_creator(task)
return task
@@ -398,7 +401,10 @@ async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Reque
async def update_scheduled_task(task_id: str, body: ScheduledTaskUpdateRequest, request: Request) -> dict:
service = get_scheduled_task_service()
user_id = await _user_id(request)
task = await service.update_task(user_id, task_id, body.model_dump(exclude_unset=True))
try:
task = await service.update_task(user_id, task_id, body.model_dump(exclude_unset=True))
except DuplicateTaskNameError:
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
if task is None:
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
await _enrich_with_creator(task)
@@ -565,7 +571,10 @@ async def admin_update_scheduled_task(task_id: str, body: ScheduledTaskUpdateReq
"""Update any task regardless of owner. Admin-only."""
await _require_admin(request)
service = get_scheduled_task_service()
task = await service.admin_update_task(task_id, body.model_dump(exclude_unset=True))
try:
task = await service.admin_update_task(task_id, body.model_dump(exclude_unset=True))
except DuplicateTaskNameError:
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
if task is None:
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
await _enrich_with_creator(task)
@@ -43,9 +43,27 @@ logger = logging.getLogger(__name__)
DEFAULT_IMAGE = "enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest"
DEFAULT_PORT = 8080
DEFAULT_CONTAINER_PREFIX = "deer-flow-sandbox"
DEFAULT_IDLE_TIMEOUT = 600 # 10 minutes in seconds
DEFAULT_REPLICAS = 3 # Maximum concurrent sandbox containers
IDLE_CHECK_INTERVAL = 60 # Check every 60 seconds
DEFAULT_IDLE_TIMEOUT = 600 # 10 minutes in seconds
DEFAULT_REPLICAS = 3 # Maximum concurrent sandbox containers
IDLE_CHECK_INTERVAL = 60 # Check every 60 seconds
PACKAGE_MANAGER_ENV_KEYS = (
"PIP_INDEX_URL",
"PIP_EXTRA_INDEX_URL",
"PIP_TRUSTED_HOST",
"PIP_CERT",
"PIP_CLIENT_CERT",
"UV_INDEX_URL",
"UV_EXTRA_INDEX_URL",
"UV_DEFAULT_INDEX",
"UV_INDEX",
"UV_TRUSTED_HOST",
"HTTP_PROXY",
"HTTPS_PROXY",
"NO_PROXY",
"http_proxy",
"https_proxy",
"no_proxy",
)
def _lock_file_exclusive(lock_file) -> None:
@@ -142,9 +160,9 @@ class AioSandboxProvider(SandboxProvider):
Local provider manages container lifecycle directly (start/stop).
"""
provisioner_url = self._config.get("provisioner_url")
if provisioner_url:
logger.info(f"Using remote sandbox backend with provisioner at {provisioner_url}")
return RemoteSandboxBackend(provisioner_url=provisioner_url)
if provisioner_url:
logger.info(f"Using remote sandbox backend with provisioner at {provisioner_url}")
return RemoteSandboxBackend(provisioner_url=provisioner_url, environment=self._config["environment"])
logger.info("Using local container sandbox backend")
return LocalContainerBackend(
@@ -172,20 +190,24 @@ class AioSandboxProvider(SandboxProvider):
"idle_timeout": idle_timeout if idle_timeout is not None else DEFAULT_IDLE_TIMEOUT,
"replicas": replicas if replicas is not None else DEFAULT_REPLICAS,
"mounts": sandbox_config.mounts or [],
"environment": self._resolve_env_vars(sandbox_config.environment or {}),
"environment": self._resolve_env_vars(sandbox_config.environment or {}),
# provisioner URL for dynamic pod management (e.g. http://provisioner:8002)
"provisioner_url": getattr(sandbox_config, "provisioner_url", None) or "",
}
@staticmethod
def _resolve_env_vars(env_config: dict[str, str]) -> dict[str, str]:
"""Resolve environment variable references (values starting with $)."""
resolved = {}
for key, value in env_config.items():
if isinstance(value, str) and value.startswith("$"):
env_name = value[1:]
resolved[key] = os.environ.get(env_name, "")
else:
def _resolve_env_vars(env_config: dict[str, str]) -> dict[str, str]:
"""Resolve environment variable references (values starting with $)."""
resolved = {}
for key in PACKAGE_MANAGER_ENV_KEYS:
value = os.environ.get(key)
if value:
resolved[key] = value
for key, value in env_config.items():
if isinstance(value, str) and value.startswith("$"):
env_name = value[1:]
resolved[key] = os.environ.get(env_name, "")
else:
resolved[key] = str(value)
return resolved
@@ -40,14 +40,17 @@ class RemoteSandboxBackend(SandboxBackend):
provisioner_url: http://provisioner:8002
"""
def __init__(self, provisioner_url: str):
"""Initialize with the provisioner service URL.
Args:
provisioner_url: URL of the provisioner service
(e.g., ``http://provisioner:8002``).
"""
self._provisioner_url = provisioner_url.rstrip("/")
def __init__(self, provisioner_url: str, environment: dict[str, str] | None = None):
"""Initialize with the provisioner service URL.
Args:
provisioner_url: URL of the provisioner service
(e.g., ``http://provisioner:8002``).
environment: Environment variables for sandbox pods. The remote
provisioner may ignore this field if it does not support it.
"""
self._provisioner_url = provisioner_url.rstrip("/")
self._environment = environment or {}
@property
def provisioner_url(self) -> str:
@@ -91,10 +94,11 @@ class RemoteSandboxBackend(SandboxBackend):
try:
resp = requests.post(
f"{self._provisioner_url}/api/sandboxes",
json={
"sandbox_id": sandbox_id,
"thread_id": thread_id,
},
json={
"sandbox_id": sandbox_id,
"thread_id": thread_id,
"environment": self._environment,
},
timeout=30,
)
resp.raise_for_status()
@@ -11,8 +11,13 @@ blocking, making a unified file safe for both workloads. Writers
that contend for the lock wait via the default 5-second sqlite3
busy timeout rather than failing immediately.
Postgres mode: both use the same database URL but maintain independent
connection pools with different lifecycles.
Postgres mode: both use the same database URL but maintain independent
connection pools with different lifecycles.
MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does
not have a built-in MySQL backend in this project, so deployments that choose
MySQL for business tables should configure the legacy ``checkpointer:``
section separately (usually sqlite) to preserve conversation checkpoints.
Memory mode: checkpointer uses MemorySaver, app uses in-memory stores.
No database is initialized.
@@ -21,12 +26,12 @@ Sensitive values (postgres_url) should use $VAR syntax in config.yaml
to reference environment variables from .env:
database:
backend: postgres
postgres_url: $DATABASE_URL
The $VAR resolution is handled by AppConfig.resolve_env_variables()
before this config is instantiated -- DatabaseConfig itself does not
need to do any environment variable processing.
backend: postgres
postgres_url: $DATABASE_URL
The $VAR resolution is handled by AppConfig.resolve_env_variables()
before this config is instantiated -- DatabaseConfig itself does not
need to do any environment variable processing.
"""
from __future__ import annotations
@@ -37,31 +42,40 @@ from typing import Literal
from pydantic import BaseModel, Field
class DatabaseConfig(BaseModel):
backend: Literal["memory", "sqlite", "postgres"] = Field(
default="memory",
description=("Storage backend for both checkpointer and application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment."),
)
class DatabaseConfig(BaseModel):
backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field(
default="memory",
description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."),
)
sqlite_dir: str = Field(
default=".deer-flow/data",
description=("Directory for the SQLite database file. Both checkpointer and application data share {sqlite_dir}/deerflow.db."),
)
postgres_url: str = Field(
default="",
description=(
"PostgreSQL connection URL, shared by checkpointer and app. "
postgres_url: str = Field(
default="",
description=(
"PostgreSQL connection URL, shared by checkpointer and app. "
"Use $DATABASE_URL in config.yaml to reference .env. "
"Example: postgresql://user:pass@host:5432/deerflow "
"(the +asyncpg driver suffix is added automatically where needed)."
),
)
),
)
mysql_url: str = Field(
default="",
description=(
"MySQL connection URL for the application ORM. "
"Use $MYSQL_DATABASE_URL in config.yaml to reference .env. "
"Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. "
"If mysql:// is provided, the +asyncmy driver suffix is added automatically."
),
)
echo_sql: bool = Field(
default=False,
description="Echo all SQL statements to log (debug only).",
)
pool_size: int = Field(
default=5,
description="Connection pool size for the app ORM engine (postgres only).",
description="Connection pool size for the app ORM engine (postgres/mysql only).",
)
# -- Derived helpers (not user-configured) --
@@ -94,9 +108,16 @@ class DatabaseConfig(BaseModel):
"""SQLAlchemy async URL for the application ORM engine."""
if self.backend == "sqlite":
return f"sqlite+aiosqlite:///{self.sqlite_path}"
if self.backend == "postgres":
url = self.postgres_url
if url.startswith("postgresql://"):
url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
return url
raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")
if self.backend == "postgres":
url = self.postgres_url
if url.startswith("postgresql://"):
url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
return url
if self.backend == "mysql":
url = self.mysql_url
if not url:
raise ValueError("database.mysql_url is required for backend='mysql'")
if url.startswith("mysql://"):
url = url.replace("mysql://", "mysql+asyncmy://", 1)
return url
raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, DateTime, String
from sqlalchemy import Boolean, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class AgentRow(Base):
@@ -16,9 +17,9 @@ class AgentRow(Base):
# Stored as a plain string so the same id can be used as the filesystem
# directory name under {base_dir}/agents/{id}/.
id: Mapped[str] = mapped_column(String(64), primary_key=True)
name: Mapped[str] = mapped_column(String(256), nullable=False, index=True)
name: Mapped[str] = mapped_column(String(191), nullable=False, index=True)
description: Mapped[str] = mapped_column(String(1024), nullable=False, default="")
user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -10,10 +10,13 @@ None and fall back to in-memory implementations.
from __future__ import annotations
import json
import logging
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
import json
import logging
import sys
from importlib.util import find_spec
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
def _json_serializer(obj: object) -> str:
@@ -23,11 +26,37 @@ def _json_serializer(obj: object) -> str:
logger = logging.getLogger(__name__)
_engine: AsyncEngine | None = None
_session_factory: async_sessionmaker[AsyncSession] | None = None
async def _auto_create_postgres_db(url: str) -> None:
_engine: AsyncEngine | None = None
_session_factory: async_sessionmaker[AsyncSession] | None = None
def _add_runtime_vendor_path() -> None:
"""Allow offline deployments to place optional DB drivers in DEER_FLOW_HOME.
The offline Docker image keeps the Python environment inside the image, while
``/app/backend/.deer-flow`` is a persistent bind mount. The MySQL switch
script installs ``asyncmy`` into ``{DEER_FLOW_HOME}/python-vendor`` so the
unchanged startup script can still import it after container recreation.
"""
try:
from deerflow.config.runtime_paths import runtime_home
vendor_dir = runtime_home() / "python-vendor"
except Exception:
vendor_dir = Path(".deer-flow/python-vendor").resolve()
vendor = str(vendor_dir)
if vendor_dir.exists() and vendor not in sys.path:
sys.path.insert(0, vendor)
def _mysql_driver_name(url: str) -> str:
if url.startswith("mysql+aiomysql://"):
return "aiomysql"
return "asyncmy"
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
@@ -65,8 +94,8 @@ async def init_engine(
"""Create the async engine and session factory, then auto-create tables.
Args:
backend: "memory", "sqlite", or "postgres".
url: SQLAlchemy async URL (for sqlite/postgres).
backend: "memory", "sqlite", "postgres", or "mysql".
url: SQLAlchemy async URL (for sqlite/postgres/mysql).
echo: Echo SQL to log.
pool_size: Postgres connection pool size.
sqlite_dir: Directory to create for SQLite (ensured to exist).
@@ -77,11 +106,20 @@ async def init_engine(
logger.info("Persistence backend=memory -- ORM engine not initialized")
return
if backend == "postgres":
try:
import asyncpg # noqa: F401
except ImportError:
raise ImportError("database.backend is set to 'postgres' but asyncpg is not installed.\nInstall it with:\n uv sync --extra postgres\nOr switch to backend: sqlite in config.yaml for single-node deployment.") from None
if backend == "postgres":
try:
import asyncpg # noqa: F401
except ImportError:
raise ImportError("database.backend is set to 'postgres' but asyncpg is not installed.\nInstall it with:\n uv sync --extra postgres\nOr switch to backend: sqlite in config.yaml for single-node deployment.") from None
if backend == "mysql":
_add_runtime_vendor_path()
driver = _mysql_driver_name(url)
if find_spec(driver) is None:
raise ImportError(
"database.backend is set to 'mysql' but the MySQL async driver is not installed.\n"
"Install asyncmy, or run switch_database_to_mysql.sh so it can install asyncmy into "
f"{Path('.deer-flow/python-vendor')} from the offline wheelhouse."
)
if backend == "sqlite":
import os
@@ -112,16 +150,25 @@ async def init_engine(
cursor.execute("PRAGMA foreign_keys=ON;")
finally:
cursor.close()
elif backend == "postgres":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
json_serializer=_json_serializer,
)
else:
raise ValueError(f"Unknown persistence backend: {backend!r}")
elif backend == "postgres":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
json_serializer=_json_serializer,
)
elif backend == "mysql":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
pool_recycle=1800,
json_serializer=_json_serializer,
)
else:
raise ValueError(f"Unknown persistence backend: {backend!r}")
_session_factory = async_sessionmaker(_engine, expire_on_commit=False)
@@ -162,10 +209,10 @@ async def init_engine_from_config(config) -> None:
await init_engine("memory")
return
await init_engine(
backend=config.backend,
url=config.app_sqlalchemy_url,
echo=config.echo_sql,
pool_size=config.pool_size,
backend=config.backend,
url=config.app_sqlalchemy_url,
echo=config.echo_sql,
pool_size=config.pool_size,
sqlite_dir=config.sqlite_dir if config.backend == "sqlite" else "",
)
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import DateTime, String, Text, UniqueConstraint
from sqlalchemy import String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class FeedbackRow(Base):
@@ -29,4 +30,4 @@ class FeedbackRow(Base):
comment: Mapped[str | None] = mapped_column(Text)
# Optional text feedback from the user
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import DateTime, Float, Integer, String, Text
from sqlalchemy import Float, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class LlmCallMetricRow(Base):
@@ -18,7 +19,7 @@ class LlmCallMetricRow(Base):
id: Mapped[str] = mapped_column(String(64), primary_key=True)
# Wall-clock time the LLM call ended. Indexed because admin views always
# sort/filter by recency.
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True)
user_id: Mapped[str | None] = mapped_column(String(64), index=True)
thread_id: Mapped[str | None] = mapped_column(String(64), index=True)
@@ -44,12 +44,13 @@ def run_migrations_offline() -> None:
context.run_migrations()
def do_run_migrations(connection):
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=True, # Required for SQLite ALTER TABLE support
)
def do_run_migrations(connection):
render_as_batch = connection.dialect.name == "sqlite"
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=render_as_batch, # Required for SQLite ALTER TABLE support
)
with context.begin_transaction():
context.run_migrations()
@@ -7,9 +7,11 @@ Create Date: 2026-05-12
from __future__ import annotations
from alembic import op
from sqlalchemy import inspect
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
import sqlalchemy as sa
from deerflow.persistence.types import PortableJSON
revision = "20260512_01"
@@ -23,9 +25,9 @@ def _has_column(table_name: str, column_name: str) -> bool:
return any(column["name"] == column_name for column in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_column("scheduled_task_runs", "result_json"):
op.add_column("scheduled_task_runs", sa.Column("result_json", sa.JSON(), nullable=True))
def upgrade() -> None:
if not _has_column("scheduled_task_runs", "result_json"):
op.add_column("scheduled_task_runs", sa.Column("result_json", PortableJSON(), nullable=True))
def downgrade() -> None:
@@ -30,7 +30,7 @@ def upgrade() -> None:
op.create_table(
"agents",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("name", sa.String(256), nullable=False),
sa.Column("name", sa.String(191), nullable=False),
sa.Column("description", sa.String(1024), nullable=False, server_default=""),
sa.Column("user_id", sa.String(64), nullable=True),
sa.Column("published", sa.Boolean(), nullable=False, server_default=sa.false()),
@@ -28,7 +28,7 @@ def upgrade() -> None:
op.create_table(
"tags",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("name", sa.String(256), nullable=False),
sa.Column("name", sa.String(159), nullable=False),
sa.Column("tag_type", sa.String(32), nullable=False),
sa.Column("color", sa.String(16), nullable=False, server_default="gray"),
sa.Column("description", sa.String(1024), nullable=False, server_default=""),
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import JSON, DateTime, Index, String, Text, UniqueConstraint
from sqlalchemy import Index, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class RunEventRow(Base):
@@ -24,9 +25,9 @@ class RunEventRow(Base):
category: Mapped[str] = mapped_column(String(16), nullable=False)
# "message" | "trace" | "lifecycle"
content: Mapped[str] = mapped_column(Text, default="")
event_metadata: Mapped[dict] = mapped_column(JSON, default=dict)
event_metadata: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
seq: Mapped[int] = mapped_column(nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
__table_args__ = (
UniqueConstraint("thread_id", "seq", name="uq_events_thread_seq"),
@@ -5,10 +5,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import JSON, Boolean, DateTime, String, Text
from sqlalchemy import Boolean, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class NotificationRow(Base):
@@ -19,6 +20,6 @@ class NotificationRow(Base):
type: Mapped[str] = mapped_column(String(64), nullable=False)
title: Mapped[str] = mapped_column(String(256), nullable=False, default="")
body: Mapped[str] = mapped_column(Text, nullable=False, default="")
payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict)
payload: Mapped[dict[str, Any]] = mapped_column(PortableJSON(), nullable=False, default=dict)
read: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), index=True)
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import DateTime, Integer, String, Text
from sqlalchemy import Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class RecommendedQuestionRow(Base):
@@ -17,5 +18,5 @@ class RecommendedQuestionRow(Base):
title: Mapped[str] = mapped_column(String(256), nullable=False)
content: Mapped[str] = mapped_column(Text, nullable=False)
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import JSON, DateTime, Index, String, Text
from sqlalchemy import Index, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class RunRow(Base):
@@ -22,8 +23,8 @@ class RunRow(Base):
model_name: Mapped[str | None] = mapped_column(String(128))
multitask_strategy: Mapped[str] = mapped_column(String(20), default="reject")
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
kwargs_json: Mapped[dict] = mapped_column(JSON, default=dict)
metadata_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
kwargs_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
error: Mapped[str | None] = mapped_column(Text)
# Convenience fields (for listing pages without querying RunEventStore)
@@ -43,7 +44,7 @@ class RunRow(Base):
# Follow-up association
follow_up_to_run_id: Mapped[str | None] = mapped_column(String(64))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
__table_args__ = (Index("ix_runs_thread_status", "thread_id", "status"),)
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import JSON, Boolean, DateTime, Integer, String, Text, UniqueConstraint
from sqlalchemy import Boolean, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class SchedulerThreadRow(Base):
@@ -15,8 +16,8 @@ class SchedulerThreadRow(Base):
user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
thread_id: Mapped[str] = mapped_column(String(64), unique=True, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class ScheduledTaskRow(Base):
@@ -29,20 +30,20 @@ class ScheduledTaskRow(Base):
source_thread_id: Mapped[str | None] = mapped_column(String(64), index=True)
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(256))
name: Mapped[str] = mapped_column(String(127))
prompt: Mapped[str] = mapped_column(Text)
schedule_text: Mapped[str] = mapped_column(String(256))
cron_expr: Mapped[str] = mapped_column(String(128))
timezone: Mapped[str] = mapped_column(String(64), default="UTC")
enabled: Mapped[bool] = mapped_column(Boolean, default=True)
next_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True)
last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
next_run_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), index=True)
last_run_at: Mapped[datetime | None] = mapped_column(BeijingDateTime())
last_status: Mapped[str | None] = mapped_column(String(32))
execution_context_json: Mapped[dict] = mapped_column(JSON, default=dict)
execution_context_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
published: Mapped[bool] = mapped_column(Boolean, default=False, index=True)
published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
published_at: Mapped[datetime | None] = mapped_column(BeijingDateTime())
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class ScheduledTaskSubscriptionRow(Base):
@@ -53,8 +54,8 @@ class ScheduledTaskSubscriptionRow(Base):
task_id: Mapped[str] = mapped_column(String(64), index=True)
user_id: Mapped[str] = mapped_column(String(64), index=True)
notify_element: Mapped[bool] = mapped_column(Boolean, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class ScheduledTaskRunRow(Base):
@@ -66,25 +67,25 @@ class ScheduledTaskRunRow(Base):
user_id: Mapped[str] = mapped_column(String(64), index=True)
scheduler_thread_id: Mapped[str] = mapped_column(String(64), index=True)
agent_run_id: Mapped[str | None] = mapped_column(String(64), index=True)
scheduled_for: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
scheduled_for: Mapped[datetime] = mapped_column(BeijingDateTime(), index=True)
started_at: Mapped[datetime | None] = mapped_column(BeijingDateTime())
finished_at: Mapped[datetime | None] = mapped_column(BeijingDateTime())
status: Mapped[str] = mapped_column(String(32), default="pending")
error: Mapped[str | None] = mapped_column(Text)
result_json: Mapped[dict | None] = mapped_column(JSON, default=dict)
result_json: Mapped[dict | None] = mapped_column(PortableJSON(), default=dict)
# Number of recovery attempts made after the first try. A run is retried
# when it overruns the per-attempt timeout or is found orphaned in the
# "running" state (e.g. after a service restart). See ScheduledTaskService.
retry_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class ScheduledTaskDeliveryProfileRow(Base):
__tablename__ = "scheduled_task_delivery_profiles"
user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
element_user_id: Mapped[str | None] = mapped_column(String(256), index=True)
element_user_id: Mapped[str | None] = mapped_column(String(191), index=True)
element_room_id: Mapped[str | None] = mapped_column(String(256), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -14,10 +14,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import DateTime, String, UniqueConstraint
from sqlalchemy import String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
# Allowed tag/target types. Kept here so persistence + API share one source.
TAG_TYPES: tuple[str, ...] = ("agent", "skill", "scheduled_task")
@@ -28,12 +29,12 @@ class TagRow(Base):
__table_args__ = (UniqueConstraint("tag_type", "name", name="uq_tags_type_name"),)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
name: Mapped[str] = mapped_column(String(256), nullable=False)
name: Mapped[str] = mapped_column(String(159), nullable=False)
tag_type: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
color: Mapped[str] = mapped_column(String(16), nullable=False, default="gray")
description: Mapped[str] = mapped_column(String(1024), nullable=False, default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class TagAssignmentRow(Base):
@@ -44,4 +45,4 @@ class TagAssignmentRow(Base):
tag_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
target_type: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
target_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
@@ -4,10 +4,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import JSON, DateTime, String
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON
class ThreadMetaRow(Base):
@@ -18,6 +19,6 @@ class ThreadMetaRow(Base):
user_id: Mapped[str | None] = mapped_column(String(64), index=True)
display_name: Mapped[str | None] = mapped_column(String(256))
status: Mapped[str] = mapped_column(String(20), default="idle")
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
metadata_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -10,10 +10,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, DateTime, String
from sqlalchemy import Boolean, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class ThreadShareRow(Base):
@@ -33,4 +34,4 @@ class ThreadShareRow(Base):
# "view" — a public, read-only static page anyone can open without
# logging in; viewers cannot send new messages.
mode: Mapped[str] = mapped_column(String(16), nullable=False, default="import", server_default="import")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
@@ -0,0 +1,85 @@
"""Portable SQLAlchemy column types."""
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from typing import Any
from sqlalchemy import JSON, DateTime, Text
from sqlalchemy.types import TypeDecorator
# China Standard Time. China has observed UTC+8 with no DST since 1991, so a
# fixed offset is exact and avoids depending on the `tzdata` package (which the
# stdlib `zoneinfo` needs on Windows / minimal containers).
BEIJING_TZ = timezone(timedelta(hours=8))
class PortableJSON(TypeDecorator):
"""Use native JSON where available, and TEXT on older MySQL servers."""
impl = JSON
cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == "mysql":
return dialect.type_descriptor(Text())
return dialect.type_descriptor(JSON())
def process_bind_param(self, value: Any, dialect) -> Any:
if dialect.name != "mysql" or value is None:
return value
return json.dumps(value, ensure_ascii=False)
def process_result_value(self, value: Any, dialect) -> Any:
if dialect.name != "mysql" or value is None or not isinstance(value, str):
return value
if not value:
return None
return json.loads(value)
class BeijingDateTime(TypeDecorator):
"""Store timestamps as Beijing (UTC+8) wall-clock time.
The application works in UTC internally (``datetime.now(UTC)`` everywhere).
MySQL ``DATETIME`` and SQLite have no timezone, so the driver serializes the
wall-clock fields of whatever datetime it is given and drops ``tzinfo``.
Without conversion a row stores the UTC wall clock and an operator browsing
the table sees a time 8 hours behind Beijing.
This decorator converts UTC-aware values to Beijing on the way in (stored as
naive Beijing wall-clock) and re-attaches the Beijing zone on the way out, so
the database holds Beijing time while Python keeps receiving correct
timezone-aware datetimes — instant comparisons stay correct either way.
PostgreSQL has a real ``TIMESTAMP WITH TIME ZONE`` type and stores the
instant unambiguously, so values pass through unchanged there.
"""
impl = DateTime
cache_ok = True
def load_dialect_impl(self, dialect):
# timezone=True keeps PostgreSQL on timestamptz; MySQL/SQLite ignore it.
return dialect.type_descriptor(DateTime(timezone=True))
def process_bind_param(self, value: Any, dialect) -> Any:
if not isinstance(value, datetime):
return value
if dialect.name == "postgresql":
return value
if value.tzinfo is None:
# Naive values follow the codebase convention of being UTC.
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(BEIJING_TZ).replace(tzinfo=None)
def process_result_value(self, value: Any, dialect) -> Any:
if not isinstance(value, datetime):
return value
if dialect.name == "postgresql":
return value
if value.tzinfo is None:
# Stored as naive Beijing wall-clock; re-attach the zone.
return value.replace(tzinfo=BEIJING_TZ)
return value.astimezone(BEIJING_TZ)
@@ -13,10 +13,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, DateTime, Index, String, text
from sqlalchemy import Boolean, Index, String, text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class UserRow(Base):
@@ -25,7 +26,7 @@ class UserRow(Base):
# UUIDs are stored as 36-char strings for cross-backend portability.
id: Mapped[str] = mapped_column(String(36), primary_key=True)
email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False, index=True)
email: Mapped[str] = mapped_column(String(191), unique=True, nullable=False, index=True)
password_hash: Mapped[str | None] = mapped_column(String(128), nullable=True)
# "admin" | "user" — kept as plain string to avoid ALTER TABLE pain
@@ -33,7 +34,7 @@ class UserRow(Base):
system_role: Mapped[str] = mapped_column(String(16), nullable=False, default="user")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
BeijingDateTime(),
nullable=False,
default=lambda: datetime.now(UTC),
)
@@ -1,4 +1,4 @@
from .registry import get_scheduled_task_service, set_scheduled_task_service
from .service import ScheduledTaskService
from .service import DuplicateTaskNameError, ScheduledTaskService
__all__ = ["ScheduledTaskService", "get_scheduled_task_service", "set_scheduled_task_service"]
__all__ = ["DuplicateTaskNameError", "ScheduledTaskService", "get_scheduled_task_service", "set_scheduled_task_service"]
@@ -36,6 +36,19 @@ _RUN_ATTEMPT_TIMEOUT_SECONDS = 30 * 60 # 30 minutes
_MAX_RUN_RETRIES = 3
class DuplicateTaskNameError(Exception):
"""A user already owns a scheduled task with the requested name.
Task names are unique per user (``uq_scheduled_tasks_user_name``). The
service raises this instead of letting the raw DB ``IntegrityError`` bubble
up so the API layer can return a friendly 409 rather than a 500.
"""
def __init__(self, name: str) -> None:
self.name = name
super().__init__(f"A scheduled task named {name!r} already exists")
@dataclass
class _RuntimeUser:
id: str
@@ -366,6 +379,20 @@ class ScheduledTaskService:
async def update_delivery_profile(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
return await self.store.update_delivery_profile(user_id, data)
async def _assert_name_available(self, user_id: str, name: str, *, exclude_task_id: str | None = None) -> None:
"""Raise :class:`DuplicateTaskNameError` if ``user_id`` already owns ``name``.
Enforces ``uq_scheduled_tasks_user_name`` proactively so callers get a
clean conflict error instead of a raw DB ``IntegrityError``. The check
compares the trimmed name; ``exclude_task_id`` skips the task being
renamed so an unchanged name on update is not flagged as a collision.
"""
for existing in await self.store.list_tasks(user_id):
if exclude_task_id is not None and existing.get("task_id") == exclude_task_id:
continue
if (existing.get("name") or "").strip() == name:
raise DuplicateTaskNameError(name)
async def create_task(
self,
*,
@@ -382,6 +409,8 @@ class ScheduledTaskService:
tz = _normalize_timezone(timezone)
cron_expr = _schedule_to_cron(schedule)
next_run_at = _next_run_at(cron_expr, tz)
final_name = name.strip() or "Scheduled task"
await self._assert_name_available(user_id, final_name)
scheduler_thread = await self.ensure_scheduler_thread(user_id)
context = dict(execution_context or {})
if execution_agent_name:
@@ -394,7 +423,7 @@ class ScheduledTaskService:
"source_thread_id": source_thread_id,
"source_agent_name": source_agent_name,
"execution_agent_name": execution_agent_name,
"name": name.strip() or "Scheduled task",
"name": final_name,
"prompt": prompt.strip(),
"schedule_text": schedule.strip(),
"cron_expr": cron_expr,
@@ -417,6 +446,10 @@ 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]
# 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):
update_data["name"] = update_data["name"].strip() or "Scheduled task"
timezone = data.get("timezone")
if timezone is not None:
update_data["timezone"] = _normalize_timezone(timezone)
@@ -443,6 +476,8 @@ class ScheduledTaskService:
if existing is None:
return None
update_data = self._build_task_update(data, existing)
if "name" in update_data:
await self._assert_name_available(user_id, update_data["name"], exclude_task_id=task_id)
return await self.store.update_task(task_id, user_id, update_data)
async def delete_task(self, user_id: str, task_id: str) -> bool:
@@ -463,6 +498,8 @@ class ScheduledTaskService:
if existing is None:
return None
update_data = self._build_task_update(data, existing)
if "name" in update_data:
await self._assert_name_available(existing["user_id"], update_data["name"], exclude_task_id=task_id)
return await self.store.update_task_any(task_id, update_data)
async def admin_delete_task(self, task_id: str) -> bool:
@@ -40,13 +40,14 @@ dependencies = [
[project.optional-dependencies]
ollama = ["langchain-ollama>=0.3.0"]
postgres = [
"asyncpg>=0.29",
"langgraph-checkpoint-postgres>=3.0.5",
"psycopg[binary]>=3.3.3",
"psycopg-pool>=3.3.0",
]
pymupdf = ["pymupdf4llm>=0.0.17"]
postgres = [
"asyncpg>=0.29",
"langgraph-checkpoint-postgres>=3.0.5",
"psycopg[binary]>=3.3.3",
"psycopg-pool>=3.3.0",
]
mysql = ["asyncmy>=0.2.10"]
pymupdf = ["pymupdf4llm>=0.0.17"]
[build-system]
requires = ["hatchling"]
@@ -24,8 +24,9 @@ dependencies = [
"croniter>=6.2.2",
]
[project.optional-dependencies]
postgres = ["deerflow-harness[postgres]"]
[project.optional-dependencies]
postgres = ["deerflow-harness[postgres]"]
mysql = ["deerflow-harness[mysql]"]
[dependency-groups]
dev = [
@@ -0,0 +1,433 @@
"""Migrate DeerFlow business tables from SQLite to MySQL.
This intentionally excludes LangGraph checkpoint tables (``checkpoints`` and
``writes``). Keep those on SQLite unless/until a MySQL checkpointer is added.
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import sqlite3
import sys
from collections.abc import Iterable
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import sqlalchemy as sa
from sqlalchemy import MetaData, text
from sqlalchemy.engine import make_url
from sqlalchemy.ext.asyncio import create_async_engine
BUSINESS_TABLES = [
"users",
"threads_meta",
"runs",
"run_events",
"feedback",
"agents",
"skills",
"scheduled_tasks",
"scheduled_task_subscriptions",
"scheduled_task_runs",
"scheduled_task_delivery_profiles",
"scheduler_threads",
"llm_call_metrics",
"notifications",
"recommended_questions",
"tags",
"tag_assignments",
"thread_shares",
]
JSON_COLUMNS = {
("threads_meta", "metadata_json"),
("runs", "metadata_json"),
("runs", "kwargs_json"),
("run_events", "event_metadata"),
("scheduled_tasks", "execution_context_json"),
("scheduled_task_runs", "result_json"),
("notifications", "payload"),
}
def _json_serializer(obj: object) -> str:
return json.dumps(obj, ensure_ascii=False)
def _add_vendor_path(vendor_dir: str | None) -> None:
candidates: list[Path] = []
if vendor_dir:
candidates.append(Path(vendor_dir))
if env_home := os.getenv("DEER_FLOW_HOME"):
candidates.append(Path(env_home) / "python-vendor")
candidates.append(Path(".deer-flow/python-vendor"))
for candidate in candidates:
resolved = candidate.resolve()
if resolved.exists():
value = str(resolved)
if value not in sys.path:
sys.path.insert(0, value)
def _normalize_mysql_url(raw: str) -> str:
if raw.startswith("mysql://"):
return raw.replace("mysql://", "mysql+asyncmy://", 1)
return raw
def _quote_sqlite_identifier(name: str) -> str:
return '"' + name.replace('"', '""') + '"'
def _sqlite_tables(conn: sqlite3.Connection) -> set[str]:
rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
return {str(row[0]) for row in rows}
def _sqlite_columns(conn: sqlite3.Connection, table_name: str) -> list[str]:
rows = conn.execute(f"PRAGMA table_info({_quote_sqlite_identifier(table_name)})").fetchall()
return [str(row[1]) for row in rows]
def _sqlite_count(conn: sqlite3.Connection, table_name: str) -> int:
return int(conn.execute(f"SELECT COUNT(*) FROM {_quote_sqlite_identifier(table_name)}").fetchone()[0])
def _parse_datetime(value: Any) -> Any:
if value is None or isinstance(value, datetime):
return value
if isinstance(value, (int, float)):
return datetime.fromtimestamp(value, tz=timezone.utc).replace(tzinfo=None)
if not isinstance(value, str):
return value
raw = value.strip()
if not raw:
return None
if raw.endswith("Z"):
raw = raw[:-1] + "+00:00"
parsed = datetime.fromisoformat(raw)
if parsed.tzinfo is not None:
parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
return parsed
def _parse_json(value: Any) -> Any:
if value is None or isinstance(value, (dict, list, int, float, bool)):
return value
if isinstance(value, bytes):
value = value.decode("utf-8")
if isinstance(value, str):
raw = value.strip()
if not raw:
return None
return json.loads(raw)
return value
def _convert_value(table_name: str, column: sa.Column, value: Any) -> Any:
if value is None:
return None
if (table_name, column.name) in JSON_COLUMNS or isinstance(column.type, sa.JSON):
parsed = _parse_json(value)
if not isinstance(column.type, sa.JSON):
return json.dumps(parsed, ensure_ascii=False)
return parsed
if isinstance(column.type, sa.DateTime):
return _parse_datetime(value)
if isinstance(column.type, sa.Boolean):
return bool(value)
return value
def _build_rows(
*,
table_name: str,
table: sa.Table,
rows: Iterable[sqlite3.Row],
columns: list[str],
) -> list[dict[str, Any]]:
result: list[dict[str, Any]] = []
for row in rows:
item: dict[str, Any] = {}
for column_name in columns:
column = table.c[column_name]
item[column_name] = _convert_value(table_name, column, row[column_name])
result.append(item)
return result
async def _create_database_if_needed(mysql_url: str) -> None:
parsed = make_url(mysql_url)
db_name = parsed.database
if not db_name:
raise ValueError("MySQL URL must include a database name")
maintenance = parsed.set(database="mysql")
engine = create_async_engine(
maintenance.render_as_string(hide_password=False),
isolation_level="AUTOCOMMIT",
pool_pre_ping=True,
)
quoted = "`" + db_name.replace("`", "``") + "`"
try:
async with engine.connect() as conn:
await conn.execute(text(f"CREATE DATABASE IF NOT EXISTS {quoted} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"))
finally:
await engine.dispose()
async def _create_schema(mysql_url: str) -> None:
from deerflow.persistence.base import Base
import deerflow.persistence.models # noqa: F401
engine = create_async_engine(mysql_url, pool_pre_ping=True, json_serializer=_json_serializer)
try:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
finally:
await engine.dispose()
def _quote_mysql_identifier(name: str) -> str:
return "`" + name.replace("`", "``") + "`"
async def _drop_empty_target_schema(mysql_url: str, table_names: list[str]) -> None:
"""Drop known target tables only when all existing known tables are empty.
This is intentionally conservative. It lets a failed first attempt clean up
half-created MySQL tables before re-running with fixed schema definitions,
while refusing to touch a database that already contains migrated rows.
"""
engine = create_async_engine(mysql_url, pool_pre_ping=True)
try:
async with engine.begin() as conn:
result = await conn.execute(
text(
"SELECT TABLE_NAME FROM information_schema.TABLES "
"WHERE TABLE_SCHEMA = DATABASE()"
)
)
existing = {str(row[0]) for row in result}
candidates = [name for name in table_names if name in existing]
if not candidates:
return
non_empty: list[str] = []
for table_name in candidates:
count = int(
(
await conn.execute(
text(f"SELECT COUNT(*) FROM {_quote_mysql_identifier(table_name)}")
)
).scalar_one()
)
if count > 0:
non_empty.append(table_name)
if non_empty:
joined = ", ".join(non_empty)
print(f"Target MySQL has non-empty tables; skip empty-schema cleanup: {joined}")
return
await conn.execute(text("SET FOREIGN_KEY_CHECKS=0"))
try:
for table_name in reversed(candidates):
await conn.execute(text(f"DROP TABLE {_quote_mysql_identifier(table_name)}"))
print(f"dropped empty target table: {table_name}")
finally:
await conn.execute(text("SET FOREIGN_KEY_CHECKS=1"))
finally:
await engine.dispose()
def _run_alembic(mysql_url: str) -> None:
from alembic import command
from alembic.config import Config
import deerflow.persistence as persistence_pkg
migrations_dir = str(Path(persistence_pkg.__file__).parent / "migrations")
cfg = Config()
cfg.set_main_option("script_location", migrations_dir)
cfg.set_main_option("sqlalchemy.url", mysql_url.replace("%", "%%"))
command.upgrade(cfg, "head")
async def _reflect_target(mysql_url: str) -> tuple[sa.ext.asyncio.AsyncEngine, MetaData]:
engine = create_async_engine(
mysql_url,
pool_pre_ping=True,
pool_recycle=1800,
json_serializer=_json_serializer,
)
metadata = MetaData()
async with engine.begin() as conn:
await conn.run_sync(metadata.reflect)
return engine, metadata
async def _target_count(conn, table: sa.Table) -> int:
return int((await conn.execute(sa.select(sa.func.count()).select_from(table))).scalar_one())
async def _ensure_target_empty(engine, metadata: MetaData, table_names: list[str]) -> None:
non_empty: list[str] = []
async with engine.connect() as conn:
for table_name in table_names:
table = metadata.tables.get(table_name)
if table is not None and await _target_count(conn, table) > 0:
non_empty.append(table_name)
if non_empty:
joined = ", ".join(non_empty)
raise RuntimeError(f"Target MySQL tables are not empty: {joined}. Re-run with --truncate-target if this is intentional.")
async def _truncate_target(engine, metadata: MetaData, table_names: list[str]) -> None:
async with engine.begin() as conn:
await conn.execute(text("SET FOREIGN_KEY_CHECKS=0"))
try:
for table_name in reversed(table_names):
table = metadata.tables.get(table_name)
if table is not None:
await conn.execute(table.delete())
finally:
await conn.execute(text("SET FOREIGN_KEY_CHECKS=1"))
async def _copy_table(
*,
source: sqlite3.Connection,
engine,
metadata: MetaData,
table_name: str,
batch_size: int,
) -> tuple[int, int]:
table = metadata.tables.get(table_name)
if table is None:
print(f"skip {table_name}: not present in MySQL schema")
return 0, 0
source_columns = _sqlite_columns(source, table_name)
columns = [name for name in source_columns if name in table.c]
if not columns:
print(f"skip {table_name}: no shared columns")
return 0, 0
selected = ", ".join(_quote_sqlite_identifier(name) for name in columns)
query = f"SELECT {selected} FROM {_quote_sqlite_identifier(table_name)}"
cursor = source.execute(query)
copied = 0
async with engine.begin() as conn:
while True:
batch = cursor.fetchmany(batch_size)
if not batch:
break
payload = _build_rows(table_name=table_name, table=table, rows=batch, columns=columns)
if payload:
await conn.execute(table.insert(), payload)
copied += len(payload)
expected = _sqlite_count(source, table_name)
return expected, copied
async def _verify_counts(source: sqlite3.Connection, engine, metadata: MetaData, table_names: list[str]) -> None:
failures: list[str] = []
async with engine.connect() as conn:
for table_name in table_names:
table = metadata.tables.get(table_name)
if table is None:
continue
source_count = _sqlite_count(source, table_name)
target_count = await _target_count(conn, table)
if source_count != target_count:
failures.append(f"{table_name}: sqlite={source_count}, mysql={target_count}")
if failures:
raise RuntimeError("Row count verification failed:\n " + "\n ".join(failures))
async def _migrate_data(args: argparse.Namespace, mysql_url: str) -> None:
source_path = Path(args.sqlite).resolve()
if not source_path.is_file():
raise FileNotFoundError(f"SQLite database not found: {source_path}")
source = sqlite3.connect(str(source_path))
source.row_factory = sqlite3.Row
try:
source_tables = _sqlite_tables(source)
table_names = [name for name in BUSINESS_TABLES if name in source_tables]
missing = [name for name in BUSINESS_TABLES if name not in source_tables]
if missing:
print("SQLite tables not present, skipped: " + ", ".join(missing))
engine, metadata = await _reflect_target(mysql_url)
try:
if args.truncate_target:
await _truncate_target(engine, metadata, table_names)
else:
await _ensure_target_empty(engine, metadata, table_names)
for table_name in table_names:
expected, copied = await _copy_table(
source=source,
engine=engine,
metadata=metadata,
table_name=table_name,
batch_size=args.batch_size,
)
print(f"{table_name}: copied {copied}/{expected}")
await _verify_counts(source, engine, metadata, table_names)
finally:
await engine.dispose()
finally:
source.close()
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Migrate DeerFlow business tables from SQLite to MySQL")
parser.add_argument("--sqlite", required=True, help="Source SQLite deerflow.db path")
parser.add_argument("--mysql-url", default=os.getenv("MYSQL_DATABASE_URL", ""), help="Target MySQL SQLAlchemy URL")
parser.add_argument("--vendor-dir", default="", help="Optional Python vendor directory containing asyncmy")
parser.add_argument("--batch-size", type=int, default=1000)
parser.add_argument("--create-database", action="store_true", help="Create the target MySQL database if it does not exist")
parser.add_argument("--drop-empty-target-schema", action="store_true", help="Drop known target tables first, but only when all are empty")
parser.add_argument("--skip-schema", action="store_true", help="Do not create/update target schema before copying")
parser.add_argument("--truncate-target", action="store_true", help="Delete existing target rows before copying")
return parser.parse_args()
def main() -> None:
args = _parse_args()
_add_vendor_path(args.vendor_dir)
mysql_url = _normalize_mysql_url(args.mysql_url.strip())
if not mysql_url:
raise SystemExit("Missing --mysql-url or MYSQL_DATABASE_URL")
if args.create_database:
asyncio.run(_create_database_if_needed(mysql_url))
if not args.skip_schema:
if args.drop_empty_target_schema:
asyncio.run(_drop_empty_target_schema(mysql_url, BUSINESS_TABLES + ["alembic_version"]))
asyncio.run(_create_schema(mysql_url))
_run_alembic(mysql_url)
asyncio.run(_migrate_data(args, mysql_url))
print("Business table migration completed successfully.")
if __name__ == "__main__":
main()
@@ -0,0 +1,47 @@
"""Patch config.yaml for the MySQL business DB + SQLite checkpointer split."""
from __future__ import annotations
import argparse
from pathlib import Path
import yaml
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Switch DeerFlow config.yaml to MySQL business storage")
parser.add_argument("--config", required=True, help="Path to config.yaml")
parser.add_argument("--mysql-url-env", default="MYSQL_DATABASE_URL", help="Environment variable referenced by database.mysql_url")
parser.add_argument("--checkpointer-sqlite", required=True, help="SQLite deerflow.db path used by LangGraph checkpointer")
parser.add_argument("--pool-size", type=int, default=10)
return parser.parse_args()
def main() -> None:
args = _parse_args()
config_path = Path(args.config)
data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
if not isinstance(data, dict):
raise SystemExit(f"Expected a mapping in {config_path}")
database = dict(data.get("database") or {})
database["backend"] = "mysql"
database["mysql_url"] = f"${args.mysql_url_env}"
database.setdefault("pool_size", args.pool_size)
database.setdefault("echo_sql", False)
data["database"] = database
data["checkpointer"] = {
"type": "sqlite",
"connection_string": args.checkpointer_sqlite,
}
config_path.write_text(
yaml.safe_dump(data, allow_unicode=True, sort_keys=False),
encoding="utf-8",
)
print(f"Updated {config_path}: database.backend=mysql, checkpointer.type=sqlite")
if __name__ == "__main__":
main()
@@ -0,0 +1,116 @@
"""Tests for per-user scheduled-task name uniqueness.
Task names are unique per user (``uq_scheduled_tasks_user_name``). The service
proactively rejects collisions with :class:`DuplicateTaskNameError` so the API
returns a 409 instead of letting a raw DB ``IntegrityError`` surface as a 500.
"""
from __future__ import annotations
from datetime import UTC, datetime
import pytest
from deerflow.persistence.scheduled_tasks import MemoryScheduledTaskStore
from deerflow.runtime.scheduler import DuplicateTaskNameError
from deerflow.runtime.scheduler.service import ScheduledTaskService
@pytest.fixture
def store() -> MemoryScheduledTaskStore:
return MemoryScheduledTaskStore()
def _build_service(store: MemoryScheduledTaskStore) -> ScheduledTaskService:
service = ScheduledTaskService(app=None, store=store, poll_interval_seconds=60.0)
service._element_client = None # noqa: SLF001
return service
def _seed_task(store: MemoryScheduledTaskStore, *, task_id: str, user_id: str, name: str) -> dict:
task = {
"task_id": task_id,
"user_id": user_id,
"scheduler_thread_id": f"thread-{user_id}",
"name": name,
"prompt": "hello",
"schedule_text": "0 9 * * *",
"cron_expr": "0 9 * * *",
"timezone": "UTC",
"enabled": True,
"next_run_at": datetime.now(UTC),
"execution_context": {},
}
store._tasks[task_id] = task # noqa: SLF001
return task
@pytest.mark.asyncio
async def test_create_task_rejects_duplicate_name(store):
"""A second task with the same (user_id, name) is refused before any insert."""
_seed_task(store, task_id="t1", user_id="alice", name="日报")
service = _build_service(store)
with pytest.raises(DuplicateTaskNameError):
await service.create_task(user_id="alice", name="日报", prompt="p", schedule="0 9 * * *")
@pytest.mark.asyncio
async def test_create_task_treats_whitespace_padded_name_as_duplicate(store):
"""The name is trimmed before the uniqueness check, matching what is stored."""
_seed_task(store, task_id="t1", user_id="alice", name="日报")
service = _build_service(store)
with pytest.raises(DuplicateTaskNameError):
await service.create_task(user_id="alice", name=" 日报 ", prompt="p", schedule="0 9 * * *")
@pytest.mark.asyncio
async def test_create_task_allows_same_name_for_different_user(store):
"""The constraint is per user — another user may reuse the name.
The create proceeds past the name check and only fails later because the
test service has no app wired for scheduler-thread setup; the point here is
that no DuplicateTaskNameError is raised.
"""
_seed_task(store, task_id="t1", user_id="alice", name="日报")
service = _build_service(store)
with pytest.raises(Exception) as exc_info:
await service.create_task(user_id="bob", name="日报", prompt="p", schedule="0 9 * * *")
assert not isinstance(exc_info.value, DuplicateTaskNameError)
@pytest.mark.asyncio
async def test_update_task_rejects_rename_to_existing_name(store):
"""Renaming a task onto a sibling's name is refused."""
_seed_task(store, task_id="t1", user_id="alice", name="日报")
_seed_task(store, task_id="t2", user_id="alice", name="周报")
service = _build_service(store)
with pytest.raises(DuplicateTaskNameError):
await service.update_task("alice", "t2", {"name": "日报"})
@pytest.mark.asyncio
async def test_update_task_allows_keeping_own_name(store):
"""Re-saving a task without changing its name is not a self-collision."""
_seed_task(store, task_id="t1", user_id="alice", name="日报")
service = _build_service(store)
updated = await service.update_task("alice", "t1", {"name": "日报", "prompt": "new"})
assert updated is not None
assert updated["name"] == "日报"
assert updated["prompt"] == "new"
@pytest.mark.asyncio
async def test_admin_update_task_rejects_rename_to_existing_name(store):
"""Admin cross-user rename still honours the owner's name uniqueness."""
_seed_task(store, task_id="t1", user_id="alice", name="日报")
_seed_task(store, task_id="t2", user_id="alice", name="周报")
service = _build_service(store)
with pytest.raises(DuplicateTaskNameError):
await service.admin_update_task("t2", {"name": "日报"})
@@ -0,0 +1,80 @@
"""Regression tests for the scheduled-task run lookup helper.
Guards against a recursion bug where ``_load_run`` called itself instead of
delegating to ``service.get_run`` — every ``GET /api/scheduled-tasks/runs/{id}``
blew the stack with ``RecursionError`` and returned 500.
"""
from __future__ import annotations
import pytest
from app.gateway.routers import scheduled_tasks as st
class _FakeService:
"""Records delegation so the test can assert no self-recursion happened."""
def __init__(self, *, owner_run=None, admin_run=None) -> None:
self._owner_run = owner_run
self._admin_run = admin_run
self.get_run_calls: list[tuple[str, str]] = []
self.admin_get_run_calls: list[str] = []
async def get_run(self, user_id: str, run_id: str):
self.get_run_calls.append((user_id, run_id))
return self._owner_run
async def admin_get_run(self, run_id: str):
self.admin_get_run_calls.append(run_id)
return self._admin_run
class _User:
def __init__(self, system_role: str | None) -> None:
self.system_role = system_role
def _patch_user(monkeypatch, user) -> None:
async def _fake(_request):
return user
monkeypatch.setattr(st, "get_optional_user_from_request", _fake)
@pytest.mark.asyncio
async def test_load_run_returns_owner_run_without_recursing(monkeypatch):
"""Owner lookup hits service.get_run exactly once — no self-recursion."""
_patch_user(monkeypatch, _User("user"))
service = _FakeService(owner_run={"id": "r1"})
run = await st._load_run(service, object(), "r1", "alice")
assert run == {"id": "r1"}
assert service.get_run_calls == [("alice", "r1")]
assert service.admin_get_run_calls == [] # admin fallback skipped when found
@pytest.mark.asyncio
async def test_load_run_admin_fallback_when_owner_miss(monkeypatch):
"""An admin who is not the owner falls back to admin_get_run."""
_patch_user(monkeypatch, _User("admin"))
service = _FakeService(owner_run=None, admin_run={"id": "r2"})
run = await st._load_run(service, object(), "r2", "bob")
assert run == {"id": "r2"}
assert service.get_run_calls == [("bob", "r2")]
assert service.admin_get_run_calls == ["r2"]
@pytest.mark.asyncio
async def test_load_run_non_admin_miss_returns_none(monkeypatch):
"""A non-admin who is not the owner gets None (router turns it into 404)."""
_patch_user(monkeypatch, _User("user"))
service = _FakeService(owner_run=None, admin_run={"id": "r3"})
run = await st._load_run(service, object(), "r3", "carol")
assert run is None
assert service.admin_get_run_calls == [] # no admin fallback for non-admins
+87
View File
@@ -0,0 +1,87 @@
config_version: 8
log_level: info
models:
- name: deepseek-chat
display_name: DeepSeek Chat
use: langchain_openai:ChatOpenAI
model: deepseek-chat
api_key: sk-e1c22b7931dc446f8b67e57878edd105
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: deepseek-v4-flash
display_name: DeepSeek V4 Flash
use: langchain_openai:ChatOpenAI
model: deepseek-v4-flash
api_key: sk-e1c22b7931dc446f8b67e57878edd105
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
- name: deepseek-v4-pro
display_name: DeepSeek V4 Pro
use: langchain_openai:ChatOpenAI
model: deepseek-v4-pro
api_key: sk-e1c22b7931dc446f8b67e57878edd105
base_url: https://api.deepseek.com
request_timeout: 600.0
max_retries: 2
max_tokens: 4096
temperature: 1.0
supports_vision: true
supports_thinking: true
tool_groups:
- name: web
- name: file:read
- name: file:write
- name: bash
tools:
- name: web_search
group: web
use: deerflow.community.ddg_search.tools:web_search_tool
max_results: 5
- name: ls
group: file:read
use: deerflow.sandbox.tools:ls_tool
- name: read_file
group: file:read
use: deerflow.sandbox.tools:read_file_tool
- name: glob
group: file:read
use: deerflow.sandbox.tools:glob_tool
max_results: 200
- name: grep
group: file:read
use: deerflow.sandbox.tools:grep_tool
max_results: 100
- name: write_file
group: file:write
use: deerflow.sandbox.tools:write_file_tool
- name: str_replace
group: file:write
use: deerflow.sandbox.tools:str_replace_tool
- name: bash
group: bash
use: deerflow.sandbox.tools:bash_tool
sandbox:
use: deerflow.sandbox.local:LocalSandboxProvider
allow_host_bash: true
skills:
path: /app/skills
agents_api:
enabled: true
database:
backend: mysql
mysql_url: $MYSQL_DATABASE_URL
pool_size: 10
echo_sql: false
checkpointer:
type: sqlite
connection_string: D:\cursor-work\offline-backend-20260512\backend\.deer-flow\data\deerflow.db
@@ -0,0 +1,215 @@
param(
[string]$MysqlDatabaseUrl = $env:MYSQL_DATABASE_URL,
[string]$MysqlPipPackages = $(if ($env:MYSQL_PIP_PACKAGES) { $env:MYSQL_PIP_PACKAGES } else { "asyncmy==0.2.10" }),
[string]$CreateDatabase = $(if ($env:CREATE_DATABASE) { $env:CREATE_DATABASE } else { "1" }),
[string]$DropEmptyTargetSchema = $(if ($env:DROP_EMPTY_TARGET_SCHEMA) { $env:DROP_EMPTY_TARGET_SCHEMA } else { "1" }),
[string]$TruncateTarget = $(if ($env:TRUNCATE_TARGET) { $env:TRUNCATE_TARGET } else { "0" }),
[string]$SqliteDb = $(if ($env:SQLITE_DB) { $env:SQLITE_DB } else { "" }),
[string]$ConfigFile = $(if ($env:DEER_FLOW_CONFIG_PATH) { $env:DEER_FLOW_CONFIG_PATH } else { "" }),
[string]$EnvFile = $(if ($env:ENV_FILE) { $env:ENV_FILE } else { "" }),
[string]$PythonExe = $(if ($env:PYTHON_EXE) { $env:PYTHON_EXE } else { "" })
)
$ErrorActionPreference = "Stop"
function Die([string]$Message) {
Write-Error $Message
exit 1
}
function Require-Command([string]$Name) {
if (-not (Get-Command $Name -ErrorAction SilentlyContinue)) {
Die "command not found: $Name"
}
}
function Require-File([string]$Path) {
if (-not (Test-Path -LiteralPath $Path -PathType Leaf)) {
Die "required file not found: $Path"
}
}
function Require-Dir([string]$Path) {
if (-not (Test-Path -LiteralPath $Path -PathType Container)) {
Die "required directory not found: $Path"
}
}
function Resolve-ExistingFile([string]$Path) {
return (Resolve-Path -LiteralPath $Path).Path
}
function Get-EnvFileValue([string]$Path, [string]$Key) {
if (-not (Test-Path -LiteralPath $Path -PathType Leaf)) {
return ""
}
$line = Get-Content -LiteralPath $Path |
Where-Object { $_ -match "^\s*$([regex]::Escape($Key))=" } |
Select-Object -Last 1
if (-not $line) {
return ""
}
return ($line -replace "^\s*$([regex]::Escape($Key))=", "")
}
function Upsert-EnvFileValue([string]$Path, [string]$Key, [string]$Value) {
$existing = @()
if (Test-Path -LiteralPath $Path -PathType Leaf) {
$existing = Get-Content -LiteralPath $Path | Where-Object { $_ -notmatch "^\s*$([regex]::Escape($Key))=" }
}
$existing + "$Key=$Value" | Set-Content -LiteralPath $Path -Encoding UTF8
}
function Is-Enabled([string]$Value) {
return @("1", "true", "yes", "on") -contains $Value.ToLowerInvariant()
}
function Run-Native([string]$Command, [string[]]$Arguments) {
& $Command @Arguments
if ($LASTEXITCODE -ne 0) {
Die "$Command failed with exit code $LASTEXITCODE"
}
}
$ScriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$BackendDir = Join-Path $ScriptDir "backend"
$DefaultDeerFlowHome = Join-Path $BackendDir ".deer-flow"
if (-not $ConfigFile) {
$ConfigFile = Join-Path $ScriptDir "config.yaml"
}
if (-not $EnvFile) {
$EnvFile = Join-Path $ScriptDir ".env"
}
if (-not $SqliteDb) {
$SqliteDb = Join-Path $DefaultDeerFlowHome "data\deerflow.db"
}
if ($PythonExe) {
Require-File $PythonExe
$PythonExe = Resolve-ExistingFile $PythonExe
} else {
Require-Command uv
}
Require-File $EnvFile
Require-File $ConfigFile
Require-File $SqliteDb
Require-File (Join-Path $BackendDir "pyproject.toml")
Require-File (Join-Path $BackendDir "scripts\migrate_business_sqlite_to_mysql.py")
Require-File (Join-Path $BackendDir "scripts\update_database_config.py")
Require-File (Join-Path $ScriptDir "extensions_config.json")
Require-File (Join-Path $ScriptDir "skill_evolution.json")
Require-Dir (Join-Path $ScriptDir "skills")
if (-not $MysqlDatabaseUrl) {
$MysqlDatabaseUrl = Get-EnvFileValue $EnvFile "MYSQL_DATABASE_URL"
}
if (-not $MysqlDatabaseUrl) {
Die "MYSQL_DATABASE_URL is required. Example: .\switch_database_to_mysql.ps1 -MysqlDatabaseUrl 'mysql+asyncmy://user:pass@127.0.0.1:3306/deerflow?charset=utf8mb4'"
}
$ConfigFile = Resolve-ExistingFile $ConfigFile
$EnvFile = Resolve-ExistingFile $EnvFile
$SqliteDb = Resolve-ExistingFile $SqliteDb
$DeerFlowHome = if ($env:DEER_FLOW_HOME) { $env:DEER_FLOW_HOME } else { $DefaultDeerFlowHome }
New-Item -ItemType Directory -Force -Path $DeerFlowHome | Out-Null
$DeerFlowHome = (Resolve-Path -LiteralPath $DeerFlowHome).Path
$BackupRoot = Join-Path $DeerFlowHome "backups"
New-Item -ItemType Directory -Force -Path $BackupRoot | Out-Null
$Timestamp = Get-Date -Format "yyyyMMdd_HHmmss"
$BackupDir = Join-Path $BackupRoot "mysql-switch-$Timestamp"
New-Item -ItemType Directory -Force -Path $BackupDir | Out-Null
Copy-Item -LiteralPath $ConfigFile -Destination (Join-Path $BackupDir "config.yaml.bak") -Force
Copy-Item -LiteralPath $EnvFile -Destination (Join-Path $BackupDir "env.bak") -Force
Copy-Item -LiteralPath $SqliteDb -Destination (Join-Path $BackupDir "deerflow.db.bak") -Force
$env:MYSQL_DATABASE_URL = $MysqlDatabaseUrl
$env:DEER_FLOW_PROJECT_ROOT = $ScriptDir
$env:DEER_FLOW_HOME = $DeerFlowHome
$env:DEER_FLOW_CONFIG_PATH = $ConfigFile
$env:DEER_FLOW_EXTENSIONS_CONFIG_PATH = Join-Path $ScriptDir "extensions_config.json"
$env:DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH = Join-Path $ScriptDir "skill_evolution.json"
$env:DEER_FLOW_SKILLS_PATH = Join-Path $ScriptDir "skills"
$PythonPathParts = @(
$BackendDir,
(Join-Path $BackendDir "packages\harness")
)
if ($env:PYTHONPATH) {
$PythonPathParts += $env:PYTHONPATH
}
$env:PYTHONPATH = ($PythonPathParts -join [IO.Path]::PathSeparator)
Push-Location $BackendDir
try {
$packages = @($MysqlPipPackages -split "\s+" | Where-Object { $_ })
if ($PythonExe) {
if ($packages.Count -gt 0) {
Write-Host "Installing MySQL driver into selected Python environment: $MysqlPipPackages"
Run-Native $PythonExe (@("-m", "pip", "install") + $packages)
}
Run-Native $PythonExe @("-c", "import asyncmy")
} else {
if (-not (Test-Path -LiteralPath (Join-Path $BackendDir ".venv") -PathType Container)) {
Write-Host "Creating local uv virtual environment..."
Run-Native "uv" @("venv")
}
if ($packages.Count -gt 0) {
Write-Host "Installing MySQL driver into local backend environment: $MysqlPipPackages"
Run-Native "uv" (@("pip", "install") + $packages)
}
Run-Native "uv" @("run", "--no-sync", "python", "-c", "import asyncmy")
}
$migrationArgs = @(
"scripts\migrate_business_sqlite_to_mysql.py",
"--sqlite", $SqliteDb,
"--mysql-url", $MysqlDatabaseUrl
)
if (Is-Enabled $CreateDatabase) {
$migrationArgs += "--create-database"
}
if (Is-Enabled $DropEmptyTargetSchema) {
$migrationArgs += "--drop-empty-target-schema"
}
if (Is-Enabled $TruncateTarget) {
$migrationArgs += "--truncate-target"
}
Write-Host "Backups written to: $BackupDir"
Write-Host "Migrating business tables to MySQL..."
if ($PythonExe) {
Run-Native $PythonExe $migrationArgs
} else {
Run-Native "uv" (@("run", "--no-sync", "python") + $migrationArgs)
}
$configArgs = @(
"scripts\update_database_config.py",
"--config", $ConfigFile,
"--mysql-url-env", "MYSQL_DATABASE_URL",
"--checkpointer-sqlite", $SqliteDb
)
if ($PythonExe) {
Run-Native $PythonExe $configArgs
} else {
Run-Native "uv" (@("run", "--no-sync", "python") + $configArgs)
}
Upsert-EnvFileValue $EnvFile "MYSQL_DATABASE_URL" $MysqlDatabaseUrl
}
finally {
Pop-Location
}
Write-Host ""
Write-Host "MySQL switch completed."
Write-Host " Config updated: $ConfigFile"
Write-Host " Env updated: $EnvFile"
Write-Host " Backup dir: $BackupDir"
Write-Host ""
Write-Host "Start the local backend from backend directory, for example:"
Write-Host " cd backend"
Write-Host " `$env:PYTHONPATH='.'"
Write-Host " uv run uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001"
@@ -0,0 +1,166 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
IMAGE_TAG="${IMAGE_TAG:-deerflow-backend-offline:20260505}"
CONTAINER_NAME="${CONTAINER_NAME:-deerflow-backend}"
IMAGE_TAR="${IMAGE_TAR:-${SCRIPT_DIR}/deerflow-backend-offline-image-20260505.tar}"
MYSQL_DRIVER_IMPORT="${MYSQL_DRIVER_IMPORT:-asyncmy}"
MYSQL_PIP_PACKAGES="${MYSQL_PIP_PACKAGES:-asyncmy==0.2.10}"
CREATE_DATABASE="${CREATE_DATABASE:-1}"
DROP_EMPTY_TARGET_SCHEMA="${DROP_EMPTY_TARGET_SCHEMA:-1}"
TRUNCATE_TARGET="${TRUNCATE_TARGET:-0}"
SQLITE_DB_HOST="${SQLITE_DB_HOST:-${SCRIPT_DIR}/data/data/deerflow.db}"
SQLITE_DB_CONTAINER="${SQLITE_DB_CONTAINER:-/app/backend/.deer-flow/data/deerflow.db}"
VENDOR_DIR_CONTAINER="${VENDOR_DIR_CONTAINER:-/app/backend/.deer-flow/python-vendor}"
die() {
echo "ERROR: $*" >&2
exit 1
}
require_cmd() {
command -v "$1" >/dev/null 2>&1 || die "command not found: $1"
}
require_file() {
[ -f "$1" ] || die "required file not found: $1"
}
read_env_value() {
local key="$1"
[ -f "${SCRIPT_DIR}/.env" ] || return 0
grep -E "^[[:space:]]*${key}=" "${SCRIPT_DIR}/.env" 2>/dev/null | tail -n 1 | sed 's/^[^=]*=//'
}
upsert_env_key() {
local key="$1"
local value="$2"
local tmp
tmp="$(mktemp "${SCRIPT_DIR}/.env.mysql-switch.XXXXXX")"
grep -Ev "^[[:space:]]*${key}=" "${SCRIPT_DIR}/.env" >"${tmp}" || true
printf '%s=%s\n' "${key}" "${value}" >>"${tmp}"
mv "${tmp}" "${SCRIPT_DIR}/.env"
}
require_cmd docker
require_file "${SCRIPT_DIR}/.env"
require_file "${SCRIPT_DIR}/config.yaml"
require_file "${SCRIPT_DIR}/backend/pyproject.toml"
require_file "${SCRIPT_DIR}/backend/app/gateway/app.py"
require_file "${SCRIPT_DIR}/backend/scripts/migrate_business_sqlite_to_mysql.py"
require_file "${SCRIPT_DIR}/backend/scripts/update_database_config.py"
require_file "${SCRIPT_DIR}/extensions_config.json"
require_file "${SCRIPT_DIR}/skill_evolution.json"
[ -d "${SCRIPT_DIR}/skills" ] || die "required directory not found: ${SCRIPT_DIR}/skills"
[ -d "${SCRIPT_DIR}/wheelhouse" ] || die "required directory not found: ${SCRIPT_DIR}/wheelhouse"
MYSQL_DATABASE_URL="${MYSQL_DATABASE_URL:-$(read_env_value MYSQL_DATABASE_URL)}"
[ -n "${MYSQL_DATABASE_URL}" ] || die "MYSQL_DATABASE_URL is required. Example: MYSQL_DATABASE_URL='mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4' bash switch_database_to_mysql.sh"
if docker ps --format '{{.Names}}' | awk -v name="${CONTAINER_NAME}" '$0 == name { found=1 } END { exit(found ? 0 : 1) }'; then
die "container ${CONTAINER_NAME} is running. Stop the backend first: bash stop_backend.sh"
fi
require_file "${SQLITE_DB_HOST}"
mkdir -p "${SCRIPT_DIR}/data/backups"
timestamp="$(date +%Y%m%d_%H%M%S)"
backup_dir="${SCRIPT_DIR}/data/backups/mysql-switch-${timestamp}"
mkdir -p "${backup_dir}"
cp "${SCRIPT_DIR}/config.yaml" "${backup_dir}/config.yaml.bak"
cp "${SCRIPT_DIR}/.env" "${backup_dir}/env.bak"
cp "${SQLITE_DB_HOST}" "${backup_dir}/deerflow.db.bak"
upsert_env_key "MYSQL_DATABASE_URL" "${MYSQL_DATABASE_URL}"
if ! docker image inspect "${IMAGE_TAG}" >/dev/null 2>&1; then
require_file "${IMAGE_TAR}"
echo "Loading Docker image from ${IMAGE_TAR}"
docker load -i "${IMAGE_TAR}"
fi
if docker ps -a --format '{{.Names}}' | awk -v name="${CONTAINER_NAME}-mysql-switch" '$0 == name { found=1 } END { exit(found ? 0 : 1) }'; then
docker rm -f "${CONTAINER_NAME}-mysql-switch" >/dev/null
fi
echo "Backups written to: ${backup_dir}"
echo "Migrating business tables to MySQL..."
docker run --rm \
--name "${CONTAINER_NAME}-mysql-switch" \
--env-file "${SCRIPT_DIR}/.env" \
-e MYSQL_DATABASE_URL="${MYSQL_DATABASE_URL}" \
-e MYSQL_DRIVER_IMPORT="${MYSQL_DRIVER_IMPORT}" \
-e MYSQL_PIP_PACKAGES="${MYSQL_PIP_PACKAGES}" \
-e CREATE_DATABASE="${CREATE_DATABASE}" \
-e DROP_EMPTY_TARGET_SCHEMA="${DROP_EMPTY_TARGET_SCHEMA}" \
-e TRUNCATE_TARGET="${TRUNCATE_TARGET}" \
-e VENDOR_DIR_CONTAINER="${VENDOR_DIR_CONTAINER}" \
-e DEER_FLOW_PROJECT_ROOT=/app \
-e DEER_FLOW_HOME=/app/backend/.deer-flow \
-e DEER_FLOW_CONFIG_PATH=/app/config.yaml \
-e DEER_FLOW_EXTENSIONS_CONFIG_PATH=/app/extensions_config.json \
-e DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH=/app/skill_evolution.json \
-e DEER_FLOW_SKILLS_PATH=/app/skills \
-e UV_LINK_MODE=copy \
-v "${SCRIPT_DIR}/backend/app:/app/backend/app" \
-v "${SCRIPT_DIR}/backend/packages:/app/backend/packages" \
-v "${SCRIPT_DIR}/backend/scripts:/app/backend/scripts" \
-v "${SCRIPT_DIR}/backend/pyproject.toml:/app/backend/pyproject.toml:ro" \
-v "${SCRIPT_DIR}/backend/uv.lock:/app/backend/uv.lock:ro" \
-v "${SCRIPT_DIR}/backend/langgraph.json:/app/backend/langgraph.json:ro" \
-v "${SCRIPT_DIR}/config.yaml:/app/config.yaml" \
-v "${SCRIPT_DIR}/extensions_config.json:/app/extensions_config.json:ro" \
-v "${SCRIPT_DIR}/skill_evolution.json:/app/skill_evolution.json:ro" \
-v "${SCRIPT_DIR}/skills:/app/skills:ro" \
-v "${SCRIPT_DIR}/wheelhouse:/app/wheelhouse:ro" \
-v "${SCRIPT_DIR}/data:/app/backend/.deer-flow" \
"${IMAGE_TAG}" \
sh -c '
set -eu
cd /app/backend
mkdir -p "${VENDOR_DIR_CONTAINER:-/app/backend/.deer-flow/python-vendor}"
export PYTHONPATH=".:${VENDOR_DIR_CONTAINER:-/app/backend/.deer-flow/python-vendor}"
if ! uv run --no-sync python -c "import ${MYSQL_DRIVER_IMPORT:-asyncmy}" >/dev/null 2>&1; then
echo "Installing ${MYSQL_PIP_PACKAGES:-asyncmy==0.2.10} into ${VENDOR_DIR_CONTAINER:-/app/backend/.deer-flow/python-vendor}"
uv pip install \
--target "${VENDOR_DIR_CONTAINER:-/app/backend/.deer-flow/python-vendor}" \
--no-index \
--find-links /app/wheelhouse \
${MYSQL_PIP_PACKAGES:-asyncmy==0.2.10}
fi
MIGRATION_FLAGS=""
if [ "${CREATE_DATABASE:-1}" = "1" ]; then
MIGRATION_FLAGS="${MIGRATION_FLAGS} --create-database"
fi
if [ "${DROP_EMPTY_TARGET_SCHEMA:-1}" = "1" ]; then
MIGRATION_FLAGS="${MIGRATION_FLAGS} --drop-empty-target-schema"
fi
if [ "${TRUNCATE_TARGET:-0}" = "1" ]; then
MIGRATION_FLAGS="${MIGRATION_FLAGS} --truncate-target"
fi
uv run --no-sync python scripts/migrate_business_sqlite_to_mysql.py \
--sqlite "'"${SQLITE_DB_CONTAINER}"'" \
--mysql-url "${MYSQL_DATABASE_URL}" \
--vendor-dir "${VENDOR_DIR_CONTAINER:-/app/backend/.deer-flow/python-vendor}" \
${MIGRATION_FLAGS}
uv run --no-sync python scripts/update_database_config.py \
--config /app/config.yaml \
--mysql-url-env MYSQL_DATABASE_URL \
--checkpointer-sqlite "'"${SQLITE_DB_CONTAINER}"'"
'
echo
echo "MySQL switch completed."
echo " Config updated: ${SCRIPT_DIR}/config.yaml"
echo " Env updated: ${SCRIPT_DIR}/.env"
echo " Backup dir: ${backup_dir}"
echo
echo "Start backend with the unchanged script:"
echo " bash start_backend.sh"