diff --git a/.gitignore b/.gitignore index cd3c1330..1b033763 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/frontend-web/vite.config.ts b/frontend-web/vite.config.ts index 76bb457e..b560c6d7 100644 --- a/frontend-web/vite.config.ts +++ b/frontend-web/vite.config.ts @@ -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: [ { diff --git a/offline-backend-20260512/.env b/offline-backend-20260512/.env new file mode 100644 index 00000000..f40e4270 --- /dev/null +++ b/offline-backend-20260512/.env @@ -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 diff --git a/offline-backend-20260512/MYSQL_SWITCH_RELEASE_NOTES_LINUX_DOCKER_ZH.md b/offline-backend-20260512/MYSQL_SWITCH_RELEASE_NOTES_LINUX_DOCKER_ZH.md new file mode 100644 index 00000000..2fc884b7 --- /dev/null +++ b/offline-backend-20260512/MYSQL_SWITCH_RELEASE_NOTES_LINUX_DOCKER_ZH.md @@ -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://: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 持久卷内,不需要改启动脚本。 diff --git a/offline-backend-20260512/MYSQL_SWITCH_RELEASE_NOTES_WINDOWS_LOCAL_ZH.md b/offline-backend-20260512/MYSQL_SWITCH_RELEASE_NOTES_WINDOWS_LOCAL_ZH.md new file mode 100644 index 00000000..be40df55 --- /dev/null +++ b/offline-backend-20260512/MYSQL_SWITCH_RELEASE_NOTES_WINDOWS_LOCAL_ZH.md @@ -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 天内会话使用。归档功能正式实现时,需要把“归档文件缺失”作为可预期异常处理。 diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index e5d75009..2247d38c 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -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` diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index bdd40a23..d37216ba 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -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: diff --git a/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py b/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py index 34c813be..9856ba1b 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py +++ b/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py @@ -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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index 17dbaf7e..e9b7bd0d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py b/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py index e11cfeea..c24ad83a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/community/aio_sandbox/remote_backend.py @@ -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() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py index 0b1de45e..bd78d38d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py @@ -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}") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/model.py index 4b7a0146..2e1a2419 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py index a1858894..49d8dc76 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py @@ -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 "", ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/feedback/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/feedback/model.py index 433e038f..bc8a2c48 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/feedback/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/feedback/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py index 89f5277c..9d16e1ba 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py @@ -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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/env.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/env.py index 78b04a6f..e715a51c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/env.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/env.py @@ -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() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260512_01_scheduled_task_run_result.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260512_01_scheduled_task_run_result.py index 0eaeaec2..2dadcf66 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260512_01_scheduled_task_run_result.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260512_01_scheduled_task_run_result.py @@ -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: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_01_agents_table.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_01_agents_table.py index 261dcb54..6447bb2e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_01_agents_table.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_01_agents_table.py @@ -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()), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260515_01_tags_tables.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260515_01_tags_tables.py index 33b7091f..1374460d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260515_01_tags_tables.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260515_01_tags_tables.py @@ -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=""), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/run_event.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/run_event.py index cf7c1657..c61c6d14 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/run_event.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/run_event.py @@ -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"), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py index eb6d9998..64d8c442 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py @@ -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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/recommended_questions/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/recommended_questions/model.py index 74f46345..a8c37f32 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/recommended_questions/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/recommended_questions/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py index 76157e2f..175e7ed7 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py @@ -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"),) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py index eeacf9e1..5b384ca9 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tags/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tags/model.py index eb65d690..d83bba99 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tags/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tags/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py index 72936070..6b0331cb 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_shares/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_shares/model.py index 00e442a0..1fa67101 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_shares/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_shares/model.py @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/types.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/types.py new file mode 100644 index 00000000..7f30abeb --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/types.py @@ -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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/user/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/user/model.py index 487cb5ba..c079bdfd 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/user/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/user/model.py @@ -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), ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/__init__.py index 530dd84e..733e0d61 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/__init__.py @@ -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"] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py index 61265318..032cbd3e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py @@ -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: diff --git a/offline-backend-20260512/backend/packages/harness/pyproject.toml b/offline-backend-20260512/backend/packages/harness/pyproject.toml index 33c3e6fa..c27248dc 100644 --- a/offline-backend-20260512/backend/packages/harness/pyproject.toml +++ b/offline-backend-20260512/backend/packages/harness/pyproject.toml @@ -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"] diff --git a/offline-backend-20260512/backend/pyproject.toml b/offline-backend-20260512/backend/pyproject.toml index d739191d..d4ea5346 100644 --- a/offline-backend-20260512/backend/pyproject.toml +++ b/offline-backend-20260512/backend/pyproject.toml @@ -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 = [ diff --git a/offline-backend-20260512/backend/scripts/migrate_business_sqlite_to_mysql.py b/offline-backend-20260512/backend/scripts/migrate_business_sqlite_to_mysql.py new file mode 100644 index 00000000..b233e9ab --- /dev/null +++ b/offline-backend-20260512/backend/scripts/migrate_business_sqlite_to_mysql.py @@ -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() diff --git a/offline-backend-20260512/backend/scripts/update_database_config.py b/offline-backend-20260512/backend/scripts/update_database_config.py new file mode 100644 index 00000000..52735cef --- /dev/null +++ b/offline-backend-20260512/backend/scripts/update_database_config.py @@ -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() diff --git a/offline-backend-20260512/backend/tests/test_scheduled_task_name_uniqueness.py b/offline-backend-20260512/backend/tests/test_scheduled_task_name_uniqueness.py new file mode 100644 index 00000000..ea4b21cd --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_scheduled_task_name_uniqueness.py @@ -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": "日报"}) diff --git a/offline-backend-20260512/backend/tests/test_scheduled_task_run_lookup.py b/offline-backend-20260512/backend/tests/test_scheduled_task_run_lookup.py new file mode 100644 index 00000000..796c9b95 --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_scheduled_task_run_lookup.py @@ -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 diff --git a/offline-backend-20260512/config.yaml b/offline-backend-20260512/config.yaml new file mode 100644 index 00000000..acc54823 --- /dev/null +++ b/offline-backend-20260512/config.yaml @@ -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 diff --git a/offline-backend-20260512/switch_database_to_mysql.ps1 b/offline-backend-20260512/switch_database_to_mysql.ps1 new file mode 100644 index 00000000..5599950f --- /dev/null +++ b/offline-backend-20260512/switch_database_to_mysql.ps1 @@ -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" diff --git a/offline-backend-20260512/switch_database_to_mysql.sh b/offline-backend-20260512/switch_database_to_mysql.sh new file mode 100644 index 00000000..f73058b1 --- /dev/null +++ b/offline-backend-20260512/switch_database_to_mysql.sh @@ -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"