github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-01 19:06:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(schedule): add the schedule domain model

Browse Source 
First step of the scheduled-task context's hexagonal migration: the
inner-ring model, with zero infrastructure dependencies.

- ScheduleSpec / SchedulePolicy value objects
- ScheduledTask aggregate root
- ScheduledRun aggregate
- 9 domain errors, 5 enums

Rules are migrated verbatim from their current homes, each method's
docstring citing the source line: timezone/cron/next-run calculation
from deerflow/scheduler/schedules.py, context-mode and re-arm rules from
routers/scheduled_tasks.py, and the four status-derivation rules from
app/scheduler/service.py.

Two things previously held by convention are now enforced by
construction. Validation and normalization live in __post_init__, so
building a ScheduleSpec field-by-field cannot bypass them. The skipped
tombstone is a separate factory, so it can never be written as the
transient queued row that would collide with uq_scheduled_task_run_active.

The domain does not serialize itself: mapping the stored schedule_spec
JSON in and out stays with the adapter layer, keeping Mapping[str, Any]
out of every domain signature.

Production code still runs through app/scheduler/service.py -- this
commit adds no call sites and changes no behavior.
This commit is contained in:
rayhpeng 2026-07-28 11:15:06 +08:00
parent 551865abcf
commit 9d0b09558e
9 changed files with 1669 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

468
backend/docs/SCHEDULE_DESIGN_zh.md Normal file
View File

@ -0,0 +1,468 @@
# 定时任务（Schedule）模块设计
> 面向想理解或扩展定时任务模块的人。读完你将能回答：一条定时规则从创建到执行经历了什么、它的每条业务规则住在哪个文件、以及你要改它时该动哪里。
>
> 配套文档：[`HEXAGONAL_ARCHITECTURE_zh.md`](HEXAGONAL_ARCHITECTURE_zh.md)（本文遵循的架构分层）、`backend/AGENTS.md`（编码规约与调度相关的运行时约定）。
>
> **本文当前只覆盖领域模型层**（`domain/schedule/model/`）——六边形迁移的第一步。端口、应用服务与适配器仍在旧位置，见 §2。
---
## 1. 这个模块做什么
一句话：
> 让用户注册一条「到点、或按 cron 用这段 prompt 起一次 agent run」的规则，由后台轮询器按时把它派发进**既有的** Gateway run 生命周期。
一条最简单的使用路径：
```
用户在 /workspace/scheduled-tasks 建一条任务
"每天 09:00（Asia/Shanghai）帮我总结昨天的 GitHub issue"
↓
后台轮询器每 5 秒扫一次，发现它到点了
↓
起一个 agent run（新 thread、非交互模式）
↓
run 结束后回写执行记录，并算出下一次的时间
```
**一条硬约束**（写在 `backend/AGENTS.md` 里）：调度器只决定 **when**，不得引入第二套执行栈。它最终只做一件事——在正确的时刻调用一次现有的 run 启动入口，然后记账。模块的全部复杂度都在"正确的时刻"和"记账"的并发与崩溃语义上，而不在执行本身。
两个核心概念，对应两张表、两个聚合：
| 概念 | 是什么 | 聚合 | 表 |
|---|---|---|---|
| **任务**（task） | 用户注册的**规则**，长期存在 | `ScheduledTask` | `scheduled_tasks` |
| **执行**（run） | 规则的**一次触发**，只读历史 | `ScheduledRun` | `scheduled_task_runs` |
---
## 2. 当前状态：新旧并存
**这一节请先读，否则你会在代码库里迷路。**
模块正处于六边形重构的中途。领域模型已经落地，但**生产代码路径还没有切过来**：
```mermaid
flowchart LR
subgraph NEW["✅ 已落地 · 内圈"]
M["deerflow/domain/schedule/model/<br/>ScheduleSpec · ScheduledTask · ScheduledRun"]
end
subgraph OLD["⏳ 仍是旧形态 · 生产路径"]
R["app/gateway/routers/scheduled_tasks.py<br/>入参校验 + 业务判断混在一起"]
S["app/scheduler/service.py<br/>轮询 + 派发编排 + 状态推导"]
P["deerflow/persistence/scheduled_task*/sql.py<br/>仓储返回裸 dict"]
C["deerflow/scheduler/schedules.py<br/>时区 / cron 计算"]
end
subgraph TODO["🚧 待建"]
PO["domain/schedule/ports.py"]
SV["domain/schedule/service.py"]
AD["app/infra/persistence/ + app/infra/schedule/"]
end
R -.->|尚未调用| M
S -.->|尚未调用| M
M --> PO --> SV --> AD
style M fill:#d8ecff
style OLD fill:#f5f5f5
```
含义很具体：
- **`domain/schedule/model/` 是唯一的真相声明处**，但目前只有域测试在用它。运行中的定时任务走的仍是 `app/scheduler/service.py` 那套。
- 两边**规则内容一致**（模型是逐条从旧代码搬迁的，每个方法的 docstring 都标了来源行号），但**代码是重复的**。这是迁移中间态的正常代价。
- 你现在改一条业务规则，要**两边都改**，直到迁移完成。旧位置见 §10 的索引。
- 新增业务规则请**只写在领域模型里**，然后在旧位置调用它——不要再往 router / service 里加新的判断。
迁移完成后 `app/scheduler/service.py`、`deerflow/scheduler/` 整包、两个 `sql.py` 都会消失，本文会补上端口与服务层章节。
---
## 3. 领域模型全景
`packages/harness/deerflow/domain/schedule/model/` 是一个包而非单文件，因为这个上下文有两个聚合加一个值对象：
```
model/
├── errors.py 9 个领域错误，零依赖
├── enums.py 5 个枚举，零依赖
├── spec.py ScheduleSpec（值对象）· SchedulePolicy（值对象）
├── task.py ScheduledTask（聚合根）· TERMINAL_TASK_STATUSES
└── run.py ScheduledRun（聚合）· ACTIVE/TERMINAL_RUN_STATUSES
```
**纪律**：这一层零基础设施依赖——没有 SQL、没有 HTTP、没有配置读取、不看时钟（`now` 一律由调用方显式传入）。CI 的 `tests/test_harness_domain_purity.py` 会 AST 扫描整个 `domain/` 目录执法。唯一的第三方依赖是 `croniter`，它是确定性纯计算库（无 IO、无全局状态），与标准库 `zoneinfo` 同性质——日历计算本身就是定时任务的领域知识。
```mermaid
classDiagram
class ScheduledTask {
<<聚合根 · frozen>>
+ScheduleSpec schedule
+TaskStatus status
+ContextMode context_mode
+create(...)$
+resolve_execution_thread() str
+ensure_mutable()
+status_after_launch(trigger)
+status_after_failure(trigger)
+status_after_skip()
+status_after_completion(outcome)
+with_schedule(...) ScheduledTask
+with_context(...) ScheduledTask
+paused() / resumed()
}
class ScheduleSpec {
<<值对象 · frozen>>
+ScheduleType schedule_type
+str timezone
+str|None cron
+datetime|None run_at
+next_after(now) datetime|None
+ensure_launchable(now, policy)
}
class SchedulePolicy {
<<值对象 · frozen>>
+int min_once_delay_seconds
}
class ScheduledRun {
<<聚合 · frozen>>
+RunStatus status
+TriggerKind trigger
+queued(...)$
+skipped_tombstone(...)$
+is_active bool
}
ScheduledTask *-- ScheduleSpec : 持有
ScheduledTask ..> SchedulePolicy : 方法入参
ScheduleSpec ..> SchedulePolicy : 方法入参
```
三个关系值得注意：
- **两个聚合之间只有 `task_id` 字符串引用，没有对象引用。** 一次派发要写两张表，且现状就不在同一个事务里——它们是各自独立的一致性边界。
- **`SchedulePolicy` 是"传入"而非"持有"。** 它承载运营可调阈值（当前只有 `min_once_delay_seconds`），由组合根从 `config.scheduler` 构造。聚合若持有它，同一个任务对象在不同部署配置下语义就不同了。它的领域默认值是 `0`（不施加约束），真正的业务阈值只能由外圈注入。
- **`ScheduledTask` 持有的是解析后的 `ScheduleSpec`，不是原始 dict。** dict ↔ 值对象的映射属于适配器层。
---
## 4. `ScheduleSpec`：调度在什么时候
它把三个存储字段（`schedule_type` / `schedule_spec` JSON / `timezone`）解析成一个校验过的值对象。
### 4.1 两种调度类型
| 类型 | 依据字段 | 语义 |
|---|---|---|
| `cron` | `cron`（5 段） | 周期性，永远有下一次 |
| `once` | `run_at` | 一次性，用完即止 |
### 4.2 创建即一致
所有校验与规范化都在 `__post_init__` 里，**不在工厂方法里**。原因很实际：frozen dataclass 仍然可以逐字段直接构造，把规则放在 `cron_schedule()` / `once_at()` 里等于留了一条绕过通道。
构造期做四件事：
1. 时区必须是合法 IANA 名（先做，因为下面本地化 `run_at` 要用它）
2. `cron` 必须存在且恰好 5 段，空白折叠后写回
3. `once` 必须带 `run_at`
4. naive 的 `run_at` 按**任务自己的时区**本地化——`2026-08-01T09:00` 配 `Asia/Shanghai` 意为"上海时间早上九点"，不是 UTC 九点
非法输入在**任何 IO 发生之前**就抛 `InvalidScheduleError`。
### 4.3 两个算时间的方法，别用反
这是本模块最容易写错的一处：
```mermaid
flowchart TD
subgraph CU["创建 / 更新路径（用户提交）"]
A1["ensure_launchable(now, policy)"] --> A2{"ONCE?"}
A2 -->|是| A3{"没有未来的一次?"}
A3 -->|是| A4["InvalidScheduleError<br/>必须是未来时间"]
A3 -->|否| A5{"距今 < min_once_delay?"}
A5 -->|是| A6["InvalidScheduleError<br/>至少提前 N 秒"]
A5 -->|否| A7["返回时间"]
A2 -->|否| A8["CRON 不受提前量约束"]
end
subgraph DP["派发路径（重排下一次）"]
B1["next_after(now)"] --> B2["纯计算，不做任何提交期校验"]
end
style A4 fill:#ffe0e0
style A6 fill:#ffe0e0
```
- `next_after(now)` —— 算下一次是什么时候。cron 在**任务时区**里算，返回 UTC；once 只在还没过期时返回。
- `ensure_launchable(now, policy)` —— 同样的计算，外加**只在用户提交时才成立**的约束：once 必须在未来，且至少提前 `min_once_delay_seconds`（默认配置 60 秒，防止用户建一个立刻就要跑的任务）。cron 从不受这个下限约束。
**用反的后果**：拿 `ensure_launchable` 做派发后重排，会让一个正常执行完的 cron 任务被"提前量不足"拒绝。
### 4.4 时区是真的按时区算
cron 表达式在任务声明的时区里求值，然后转成 UTC 存储。这意味着夏令时切换会被正确吸收：
```
America/New_York 的 "0 9 * * *"
2026-03-07（EST, UTC-5）→ 14:00 UTC
2026-03-09（EDT, UTC-4）→ 13:00 UTC
```
用户看到的始终是"每天早上九点"。
### 4.5 它不知道自己怎么被存储
`ScheduleSpec` 上**没有**序列化方法。数据库里那个 `schedule_spec` JSON 列（以及 HTTP 请求/响应里的同名字段）与值对象之间的双向映射，属于适配器层：
```
{"cron": "0 9 * * *"} ←──→ ScheduleSpec(CRON, "Asia/Shanghai", cron="0 9 * * *")
（JSON 列 / HTTP 字段） （值对象）
↑
app/infra/schedule/spec_mapping.py
```
这不是洁癖，是一条可执行的判据：**一旦领域方法的签名里出现 `Mapping[str, Any]`，就说明领域在处理持久化/传输格式了**。解析这件事天然可以切成两半——结构校验（键在不在？值是不是字符串？）属于边界，值校验（cron 是不是 5 段、时区认不认识、`run_at` 有没有）属于 `__post_init__`。切开之后领域完全不需要看见 dict，签名全部强类型。
同样的形状在 Feedback 上下文里也成立：`Feedback` 聚合对 ORM 行一无所知，转换全在 `app/infra/persistence/feedback.py`。
---
## 5. `ScheduledTask`：规则本体
聚合根，持有全部不变量。
### 5.1 字段分组
| 组 | 字段 | 说明 |
|---|---|---|
| 身份 | `task_id` `user_id` | 一切读写按 `user_id` 隔离 |
| 内容 | `title` `prompt` | |
| 调度 | `schedule`（`ScheduleSpec`） | |
| 执行上下文 | `context_mode` `thread_id` `assistant_id` | 见 5.2 |
| 状态 | `status` `overlap_policy` | 见 5.3、5.5 |
| 调度游标 | `next_run_at` | **认领的唯一依据**：为空或在未来 ⇒ 不会被派发 |
| 回执 | `last_run_at` `last_run_id` `last_thread_id` `last_error` `run_count` | 仅供展示 |
### 5.2 执行上下文：每次新会话，还是复用同一个
| `context_mode` | 语义 |
|---|---|
| `fresh_thread_per_run`（默认） | 每次派发新建一个 thread，各次执行互不干扰 |
| `reuse_thread` | 所有执行都落在同一个 thread 里，agent 能看到历史 |
不变量：`reuse_thread` **必须**带 `thread_id`，构造期强制。
`resolve_execution_thread()` 回答"这次派发用哪个 thread"。注意它**不是幂等的**——`fresh_thread_per_run` 每次调用都生成新 UUID。调用方必须每次派发只调一次，把结果存进局部变量，供 run 记录、启动调用、返回结果三处复用。
### 5.3 状态机
```mermaid
stateDiagram-v2
[*] --> enabled: create()
enabled --> paused: paused()
paused --> enabled: resumed()
enabled --> running: 被轮询器认领
running --> enabled: 派发完成（cron）
running --> running: 派发完成（once，等回调）
running --> completed: run 成功（once）
running --> failed: run 失败（once）
running --> cancelled: run 被中断（once）
completed --> enabled: with_schedule 改到未来
failed --> enabled: with_schedule 改到未来
cancelled --> enabled: with_schedule 改到未来
note right of running
RUNNING ≠ "agent 正在跑"
而是"这一轮的调度所有权被持有"
end note
```
三件反直觉的事，理解了它们就理解了这个状态机：
**① `running` 不表示 agent 在执行。** 轮询器认领任务的那一刻就写 `running`，此时 run 还没创建。它真正的含义是"某个轮询进程持有这一轮的调度所有权（租约）"。`ensure_mutable()` 因此在这个状态拒绝编辑——正在被派发的任务改不得。
**② cron 任务几乎从不停在 `running`。** 派发一完成立刻回 `enabled`。只有 `once` 会停在 `running` 等待完成回调——因为在启动那一刻宣布 `completed` 会在 run 失败或进程崩溃时永久说谎。
**③ 终态可以被重新武装。** 把一个 `completed` / `failed` / `cancelled` 的任务的调度改到未来时间，状态会被强制拉回 `enabled`（`TERMINAL_TASK_STATUSES`）。不这么做的话，接口会返回 200、`next_run_at` 有值、但**永远不触发**——静默死亡。
### 5.4 四条状态推导规则
派发流程的每个出口都要回答"任务接下来是什么状态"。这四个方法就是答案，**判定顺序即语义**（现状是 `if/elif/else`，写成并列的 `if` 会静默改变行为）：
```mermaid
flowchart TD
L0["status_after_launch(trigger)<br/>启动成功后"] --> L1{"ONCE?"}
L1 -->|是| LR["RUNNING"]
L1 -->|否| L2{"MANUAL 且当前 PAUSED?"}
L2 -->|是| LP["PAUSED"]
L2 -->|否| LE["ENABLED"]
F0["status_after_failure(trigger)<br/>启动失败后"] --> F1{"MANUAL?"}
F1 -->|是| FS["保持原状态"]
F1 -->|否| F2{"ONCE?"}
F2 -->|是| FF["FAILED"]
F2 -->|否| FE["ENABLED"]
K0["status_after_skip()<br/>因重叠被跳过"] --> K1{"ONCE?"}
K1 -->|是| KF["FAILED"]
K1 -->|否| KE["ENABLED"]
C0["status_after_completion(outcome)<br/>run 到达终态"] --> C1{"ONCE?"}
C1 -->|否| CN["None（不改）"]
C1 -->|是| C2{"outcome"}
C2 -->|SUCCESS| CC["COMPLETED"]
C2 -->|INTERRUPTED| CX["CANCELLED"]
C2 -->|FAILED| CF["FAILED"]
style L1 fill:#ffd9d9
style F1 fill:#ffd9d9
```
两个**判定顺序陷阱**（红色节点），都有专门的测试锁住：
- **`status_after_launch` 先判 ONCE**：一个 `paused` 的 once 任务被手动触发，结果是 `RUNNING` 而不是 `PAUSED`。
- **`status_after_failure` 先判 MANUAL**：一个 once 任务被手动触发且失败，保持原状态而不是 `FAILED`——失败的手动触发不能吃掉这个任务本来的调度未来。
另外三处设计意图：
- `status_after_skip` **没有 trigger 参数**。跳过只发生在自动调度路径上——手动触发遇到重叠是直接拒绝、不留记录的，所以那个分支根本不存在，加参数会暗示一个虚构的可能性。
- once 被跳过是 `FAILED` 而非 `COMPLETED`：唯一的那次机会丢了，说"完成"等于谎称执行过。
- 完成回调里 `INTERRUPTED` 映射到 `CANCELLED` 而非 `FAILED`：用户主动取消、或同 thread 被新 run 抢占，都不是执行失败。
### 5.5 重叠策略
`overlap_policy` 目前固定为 `"skip"`：一个任务同时最多有一个活跃执行，到点时若上一次还没结束，这一次就被跳过。
聚合上用 `skips_on_overlap` 属性封装这个判断，字符串比较只存在于这一处——将来加第二种策略（比如 `queue`）只改这里。之所以不做成枚举：目前只有一个取值，单值枚举是噪音。
---
## 6. `ScheduledRun`：一次执行的记账
字段几乎是 `scheduled_task_runs` 表的镜像，业务逻辑很少。它存在的核心理由是**两个具名工厂**：
| 工厂 | 产出状态 | 用在哪 |
|---|---|---|
| `ScheduledRun.queued(...)` | `QUEUED`（活跃） | 正常派发 |
| `ScheduledRun.skipped_tombstone(...)` | `SKIPPED`（**直接终态**） | 因重叠被跳过 |
**为什么墓碑必须是第二个工厂，而不是"先 queued 再改状态"**——这是全模块最容易写错的一处：
数据库上有一个部分唯一索引 `uq_scheduled_task_run_active`，谓词是 `status IN ('queued','running')`，保证一个任务最多一条活跃执行。跳过发生时，上一次执行**还占着**那个槽位。如果墓碑先建成 `queued`，它自己就会撞上这个索引。`skipped` 落在谓词之外，永不冲突。
用两个具名工厂而不是一个可变状态的构造器，就是用类型系统把这条规则焊死，让错误实现写不出来。
`ACTIVE_RUN_STATUSES` 这个常量必须与上述索引谓词保持逐字一致——它是"跳过"判断的快路径依据，而索引是并发下的最终仲裁者，两者漂移会让它们对不上。
执行状态共六个：`queued → running → success | failed | interrupted`，外加旁路的 `skipped`。
---
## 7. 二次开发指引
### 7.1 给任务加一个字段
例：加一个 `notify_on_failure: bool`。
1. `model/task.py` 的 `ScheduledTask` 加字段（带默认值）
2. 若有约束，写进 `__post_init__`
3. `persistence/scheduled_tasks/model.py` 的 ORM 行加列
4. 新增一个 alembic revision（`cd backend && make migrate-rev MSG="..."`），用 `_helpers.py` 的幂等 helper
5. router 的请求/响应模型加字段
6. 前端 `frontend/src/core/scheduled-tasks/types.ts` 同步
7. 域测试补一条
### 7.2 加一种调度类型
例：加 `interval`（每 N 分钟）。
1. `model/enums.py` 的 `ScheduleType` 加成员
2. `model/spec.py`：加承载参数的字段（如 `interval_seconds`）、在 `__post_init__` 加校验、在 `next_after` 加一个分支
3. 适配器的 `spec_mapping`（见 §4.5）：进出两个方向各加一个分支
4. **逐个检查 `task.py` 里四个 `status_after_*`**——它们目前都在问"是不是 ONCE"，新类型会落进 else 分支。确认那是你要的语义（大概率是：interval 与 cron 同属周期性）
5. 域测试：新类型在 §5.4 四张表里各补一行
不需要改数据库——`schedule_spec` 是 JSON 列。
### 7.3 改一条状态推导规则
只改 `model/task.py` 对应的那个方法，**顺便改 `app/scheduler/service.py` 里的旧副本**（见 §2）。域测试里对应的真值表用例必须同步更新——那张表就是规则的规格说明。
### 7.4 加一种重叠策略
例：加 `queue`（排队而非跳过）。
这是改动面最大的一种，因为它触及数据库不变量：
1. `model/task.py`：`skips_on_overlap` 拆成策略判断
2. `scheduled_task_runs` 的部分唯一索引**必须**改成条件化的（`... AND overlap_policy = 'skip'`）——现在它是纯状态谓词，会把排队的执行也挡掉。索引同时定义在 ORM `__table_args__` 和迁移文件里，**两处都要改**（空库 bootstrap 走 `create_all`，不执行迁移）
3. 跳过路径的墓碑逻辑要相应分叉
动手前先读 `backend/AGENTS.md` 里关于这个索引的整段说明。
### 7.5 不要做的事
- **不要在 router 或 service 里新增业务判断**——那是迁移前的旧形态，新规则一律进领域模型
- **不要在领域层读配置或时钟**——阈值通过 `SchedulePolicy` 注入，`now` 显式传参；CI 的纯度测试会拦截基础设施导入
- **不要为定时执行另建一套运行栈**——必须复用现有的 run 生命周期
---
## 8. 常见陷阱速查
| 陷阱 | 后果 |
|---|---|
| 用 `ensure_launchable` 做派发后重排 | 正常执行完的 cron 任务被"提前量不足"拒绝 |
| 把 `status_after_*` 的 `if/elif` 写成并列 `if` | 两处判定顺序失效，静默改变行为 |
| 多次调用 `resolve_execution_thread()` | 每次拿到不同的 thread，记录与实际执行对不上 |
| 墓碑先建 `queued` 再改 `skipped` | 撞上唯一索引，跳过流程直接报错 |
| 改了 `ACTIVE_RUN_STATUSES` 没改索引谓词 | 快路径与数据库仲裁者判断不一致 |
| 终态任务改了调度但没重新武装 | 接口返回 200，任务永不触发 |
| 只改领域模型，忘了 `app/scheduler/service.py` | 迁移完成前，生产行为不变 |
---
## 9. 术语表
| 术语 | 含义 |
|---|---|
| 任务（task） | 用户注册的定时**规则**，长期存在 |
| 执行（run） | 规则的**一次**触发，只读历史 |
| 派发（dispatch） | 把一个到期任务变成一次执行的动作 |
| 触发方式（trigger） | `scheduled`（轮询器）或 `manual`（用户点"立即执行"） |
| 认领（claim） | 轮询器取得某任务这一轮调度所有权 |
| 租约（lease） | 认领时盖的带过期时间的戳，用于崩溃恢复 |
| 重叠（overlap） | 到点时上一次执行还没结束 |
| 墓碑（tombstone） | 被跳过的那次执行留下的终态记录 |
| 重新武装（re-arm） | 把终态任务拉回 `enabled` 使其可再次被认领 |
---
## 10. 代码索引
**领域模型（已迁移，本文覆盖）**
| 文件 | 内容 |
|---|---|
| [`domain/schedule/model/enums.py`](../packages/harness/deerflow/domain/schedule/model/enums.py) | `TaskStatus` `RunStatus` `ScheduleType` `ContextMode` `TriggerKind` |
| [`domain/schedule/model/errors.py`](../packages/harness/deerflow/domain/schedule/model/errors.py) | 9 个领域错误 |
| [`domain/schedule/model/spec.py`](../packages/harness/deerflow/domain/schedule/model/spec.py) | `ScheduleSpec` `SchedulePolicy` |
| [`domain/schedule/model/task.py`](../packages/harness/deerflow/domain/schedule/model/task.py) | `ScheduledTask` `TERMINAL_TASK_STATUSES` |
| [`domain/schedule/model/run.py`](../packages/harness/deerflow/domain/schedule/model/run.py) | `ScheduledRun` `ACTIVE_RUN_STATUSES` `TERMINAL_RUN_STATUSES` |
| [`tests/test_schedule_domain.py`](../tests/test_schedule_domain.py) | 域测试，全同步零 IO；四张真值表逐格覆盖 |
**尚未迁移（生产路径，见 §2）**
| 文件 | 内容 |
|---|---|
| [`app/gateway/routers/scheduled_tasks.py`](../app/gateway/routers/scheduled_tasks.py) | 10 个 HTTP 端点，含入参校验与业务判断 |
| [`app/scheduler/service.py`](../app/scheduler/service.py) | 轮询循环、派发编排、状态推导、完成回调、启动清扫 |
| [`deerflow/scheduler/schedules.py`](../packages/harness/deerflow/scheduler/schedules.py) | 时区 / cron / 下次时间计算 |
| [`persistence/scheduled_tasks/`](../packages/harness/deerflow/persistence/scheduled_tasks/) | 任务表 ORM + 仓储 |
| [`persistence/scheduled_task_runs/`](../packages/harness/deerflow/persistence/scheduled_task_runs/) | 执行表 ORM + 仓储，含唯一索引定义 |
| [`config/scheduler_config.py`](../packages/harness/deerflow/config/scheduler_config.py) | `enabled` `poll_interval_seconds` `lease_seconds` `max_concurrent_runs` `min_once_delay_seconds` |
**前端**
`frontend/src/core/scheduled-tasks/`（类型、API、cron 解析、预设配方）与 `frontend/src/app/workspace/scheduled-tasks/page.tsx`。

58
backend/packages/harness/deerflow/domain/schedule/__init__.py Normal file
View File

@ -0,0 +1,58 @@
"""Schedule bounded context: standing instructions to run a prompt on time.
Public API of the context. Import domain objects from here; ports will live in
`deerflow.domain.schedule.ports` -- they are contracts consumed by adapters and
tests, not everyday call-site symbols.
`ScheduleService` is not exported yet: this commit lands the model only.
"""
from deerflow.domain.schedule.model import (
ACTIVE_RUN_STATUSES,
CRON_FIELD_COUNT,
TERMINAL_RUN_STATUSES,
TERMINAL_TASK_STATUSES,
ActiveRunConflictError,
ContextMode,
InvalidContextModeError,
InvalidScheduleError,
LaunchFailedError,
RunStatus,
ScheduledRun,
ScheduledTask,
ScheduleError,
SchedulePolicy,
ScheduleSpec,
ScheduleType,
TaskNotFoundError,
TaskNotMutableError,
TaskStatus,
ThreadBusyError,
ThreadNotFoundError,
TriggerKind,
)
__all__ = [
"ACTIVE_RUN_STATUSES",
"CRON_FIELD_COUNT",
"TERMINAL_RUN_STATUSES",
"TERMINAL_TASK_STATUSES",
"ActiveRunConflictError",
"ContextMode",
"InvalidContextModeError",
"InvalidScheduleError",
"LaunchFailedError",
"RunStatus",
"ScheduleError",
"SchedulePolicy",
"ScheduleSpec",
"ScheduleType",
"ScheduledRun",
"ScheduledTask",
"TaskNotFoundError",
"TaskNotMutableError",
"TaskStatus",
"ThreadBusyError",
"ThreadNotFoundError",
"TriggerKind",
]

67
backend/packages/harness/deerflow/domain/schedule/model/__init__.py Normal file
View File

@ -0,0 +1,67 @@
"""Domain model of the schedule context.
A package rather than a single module because this context has two
aggregates plus a value object (Feedback, the first slice, needed only
92 lines and stayed a module). Re-exports below make the split
invisible to callers: `deerflow.domain.schedule.model` exposes exactly
what a single `model.py` would have.
The re-exports below are ordered alphabetically because ruff's isort rule
owns that block; alphabetical happens to satisfy the real dependency order
too (enums and errors depend on nothing, run depends on enums, spec depends
on enums and errors, task depends on all three), so it needs no override.
What actually keeps the package acyclic is a rule isort cannot express:
**a submodule imports its siblings directly, never this package.** Reaching
back through `deerflow.domain.schedule.model` makes a submodule depend on a
partially initialized package — it only works while the symbol happens to
sit above that submodule's own line here, and reordering this block silently
breaks it. Keep it that way.
"""
from deerflow.domain.schedule.model.enums import (
ContextMode,
RunStatus,
ScheduleType,
TaskStatus,
TriggerKind,
)
from deerflow.domain.schedule.model.errors import (
ActiveRunConflictError,
InvalidContextModeError,
InvalidScheduleError,
LaunchFailedError,
ScheduleError,
TaskNotFoundError,
TaskNotMutableError,
ThreadBusyError,
ThreadNotFoundError,
)
from deerflow.domain.schedule.model.run import ACTIVE_RUN_STATUSES, TERMINAL_RUN_STATUSES, ScheduledRun
from deerflow.domain.schedule.model.spec import CRON_FIELD_COUNT, SchedulePolicy, ScheduleSpec
from deerflow.domain.schedule.model.task import TERMINAL_TASK_STATUSES, ScheduledTask
__all__ = [
"ACTIVE_RUN_STATUSES",
"CRON_FIELD_COUNT",
"TERMINAL_RUN_STATUSES",
"TERMINAL_TASK_STATUSES",
"ActiveRunConflictError",
"ContextMode",
"InvalidContextModeError",
"InvalidScheduleError",
"LaunchFailedError",
"RunStatus",
"ScheduleError",
"SchedulePolicy",
"ScheduleSpec",
"ScheduleType",
"ScheduledRun",
"ScheduledTask",
"TaskNotFoundError",
"TaskNotMutableError",
"TaskStatus",
"ThreadBusyError",
"ThreadNotFoundError",
"TriggerKind",
]

44
backend/packages/harness/deerflow/domain/schedule/model/enums.py Normal file
View File

@ -0,0 +1,44 @@
from __future__ import annotations
from enum import StrEnum
class TaskStatus(StrEnum):
ENABLED = "enabled"
PAUSED = "paused"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class ContextMode(StrEnum):
FRESH_THREAD_PER_RUN = "fresh_thread_per_run"
REUSE_THREAD = "reuse_thread"
class ScheduleType(StrEnum):
ONCE = "once"
CRON = "cron"
class RunStatus(StrEnum):
QUEUED = "queued"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
INTERRUPTED = "interrupted"
SKIPPED = "skipped"
class TriggerKind(StrEnum):
"""What caused a dispatch.
The two kinds diverge in almost every decision the domain makes — how an
overlap is handled, what status survives a failed launch, whether a paused
task stays paused — so this is a first-class concept rather than the raw
string the old service compared in four places.
"""
SCHEDULED = "scheduled"
MANUAL = "manual"

48
backend/packages/harness/deerflow/domain/schedule/model/errors.py Normal file
View File

@ -0,0 +1,48 @@
from __future__ import annotations
class ScheduleError(Exception):
"""Base error for the schedule domain."""
class InvalidScheduleError(ScheduleError):
"""Timezone, cron expression, or run_at is not usable."""
class InvalidContextModeError(ScheduleError):
"""context_mode is unknown, or reuse_thread is missing its thread_id."""
class TaskNotFoundError(ScheduleError):
"""The task does not exist or does not belong to the user."""
class TaskNotMutableError(ScheduleError):
"""The task is currently running and cannot be edited."""
class ThreadNotFoundError(ScheduleError):
"""reuse_thread points at a thread the user cannot access."""
class ActiveRunConflictError(ScheduleError):
"""The task already holds its single active run slot.
Raised by the run repository when the partial unique index
``uq_scheduled_task_run_active`` rejects a second active row. Moved here
from ``persistence/scheduled_task_runs/sql.py`` (was
``ActiveScheduledRunConflict``) so the domain owns its own vocabulary.
"""
class ThreadBusyError(ScheduleError):
"""The execution thread already has an in-flight run.
Translated by the RunLauncher adapter from ConflictError / HTTP 409.
This is what removes `from fastapi import HTTPException` from the
orchestration layer.
"""
class LaunchFailedError(ScheduleError):
"""The run could not be launched for any non-conflict reason."""

97
backend/packages/harness/deerflow/domain/schedule/model/run.py Normal file
View File

@ -0,0 +1,97 @@
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from deerflow.domain.schedule.model.enums import RunStatus
ACTIVE_RUN_STATUSES: tuple[RunStatus, ...] = (RunStatus.QUEUED, RunStatus.RUNNING)
"""The statuses that occupy a task's single active-run slot.
Must stay in lockstep with the predicate of the partial unique index
``uq_scheduled_task_run_active`` (``status IN ('queued','running')``, declared
in ``persistence/scheduled_task_runs/model.py``). Drift here silently
decouples the overlap fast path from its atomic arbiter — the consistency
assertion lives in a separate test module rather than the domain tests, which
stay dependency-free.
"""
TERMINAL_RUN_STATUSES: frozenset[RunStatus] = frozenset(
{
RunStatus.SUCCESS,
RunStatus.FAILED,
RunStatus.SKIPPED,
RunStatus.INTERRUPTED,
}
)
"""Statuses a run row can no longer leave.
Used by the repository's ``protect_terminal`` compare-and-set: a fast-failing
run can reach the completion hook before the launch path's own write lands.
"""
@dataclass(frozen=True)
class ScheduledRun:
"""One execution record of a scheduled task — the history row.
A separate aggregate from ``ScheduledTask``: the two are written in
independent transactions and reference each other only by ``task_id``.
"""
record_id: str
task_id: str
thread_id: str
scheduled_for: datetime
trigger: str
status: RunStatus
run_id: str | None = None
error: str | None = None
started_at: datetime | None = None
finished_at: datetime | None = None
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
@classmethod
def queued(cls, *, task_id: str, thread_id: str, scheduled_for: datetime, trigger: str) -> ScheduledRun:
"""The active row of a normal dispatch.
Inserting this is what the unique index arbitrates: the loser of a
concurrent insert surfaces as ``ActiveRunConflictError``.
The ``task-run-{hex}`` id shape is depended on by existing rows and by
the run metadata that links a Gateway run back to this record — do not
change it.
"""
return cls(
record_id=f"task-run-{uuid.uuid4().hex}",
task_id=task_id,
thread_id=thread_id,
scheduled_for=scheduled_for,
trigger=trigger,
status=RunStatus.QUEUED,
)
@classmethod
def skipped_tombstone(cls, *, task_id: str, thread_id: str, scheduled_for: datetime, trigger: str) -> ScheduledRun:
"""A dropped occurrence, created directly as terminal ``SKIPPED``.
Deliberately a second factory rather than ``queued()`` followed by a
status change: ``QUEUED`` falls inside ``uq_scheduled_task_run_active``'s
predicate and would collide with the pre-existing run that still holds
the task's single active slot. ``SKIPPED`` is outside the predicate and
can never conflict (service.py:249-256).
"""
return cls(
record_id=f"task-run-{uuid.uuid4().hex}",
task_id=task_id,
thread_id=thread_id,
scheduled_for=scheduled_for,
trigger=trigger,
status=RunStatus.SKIPPED,
)
@property
def is_active(self) -> bool:
"""Whether this row occupies the task's single active-run slot."""
return self.status in ACTIVE_RUN_STATUSES

125
backend/packages/harness/deerflow/domain/schedule/model/spec.py Normal file
View File

@ -0,0 +1,125 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from croniter import croniter
from deerflow.domain.schedule.model.enums import ScheduleType
from deerflow.domain.schedule.model.errors import InvalidScheduleError
CRON_FIELD_COUNT = 5
@dataclass(frozen=True)
class SchedulePolicy:
"""Operator-tunable thresholds the domain needs but must not read itself."""
min_once_delay_seconds: int = 0
@dataclass(frozen=True)
class ScheduleSpec:
"""Parsed, validated view of (schedule_type, schedule_spec, timezone).
The stored JSON spec is mapped in and out by the adapter layer, never here:
a `Mapping[str, Any]` in a domain signature would mean the domain is
handling a persistence/transport format. The two halves of that parsing
split cleanly — structural checks (is the key present? is it a str?) belong
to the boundary, value rules (5-field cron, resolvable timezone, run_at
present) belong to __post_init__ below. Storage keeps the same raw JSON, so
this needs no migration.
Normalization happens in __post_init__ rather than in the factories below,
so direct construction cannot bypass it: a frozen dataclass is still
constructible field-by-field, and "valid on construction" has to hold for
that path too.
"""
schedule_type: ScheduleType
timezone: str
cron: str | None = None
run_at: datetime | None = None
def __post_init__(self) -> None:
# The timezone is checked first because normalizing a naive run_at
# below needs it to already be known-good.
try:
zone = ZoneInfo(self.timezone)
except ZoneInfoNotFoundError as exc:
raise InvalidScheduleError(f"Unknown timezone: {self.timezone}") from exc
if self.schedule_type is ScheduleType.CRON:
if not self.cron:
raise InvalidScheduleError("cron schedule requires schedule_spec.cron")
fields = [part for part in self.cron.split() if part]
if len(fields) != CRON_FIELD_COUNT:
raise InvalidScheduleError(f"Cron expression must contain exactly {CRON_FIELD_COUNT} fields")
object.__setattr__(self, "cron", " ".join(fields))
if self.schedule_type is ScheduleType.ONCE:
if self.run_at is None:
raise InvalidScheduleError("once schedule requires run_at")
if self.run_at.tzinfo is None:
# A naive run_at means wall-clock time in this schedule's own
# timezone (schedules.py:40-43). Localizing here rather than at
# every read site keeps the rest of this class tz-aware only.
object.__setattr__(self, "run_at", self.run_at.replace(tzinfo=zone))
@classmethod
def cron_schedule(cls, expr: str, timezone: str) -> ScheduleSpec:
"""Readability sugar — all validation lives in __post_init__."""
return cls(ScheduleType.CRON, timezone, cron=expr)
@classmethod
def once_at(cls, run_at: datetime, timezone: str) -> ScheduleSpec:
"""Readability sugar — all validation lives in __post_init__."""
return cls(ScheduleType.ONCE, timezone, run_at=run_at)
def next_after(self, now: datetime) -> datetime | None:
"""Next fire time in UTC, or None when there is no future occurrence.
The dispatch-path calculation (was `next_run_at` in schedules.py:24-55).
It applies no submission-time policy — see ensure_launchable for that,
and do not swap the two: re-arming a cron task through the stricter one
would reject it right after a perfectly normal launch.
ONCE returns run_at while it is still ahead of `now`, else None (the
single occurrence is in the past). CRON is evaluated in this schedule's
timezone and returned as UTC. A naive `now` is read as UTC
(schedules.py:32-33).
"""
if now.tzinfo is None:
now = now.replace(tzinfo=UTC)
if self.schedule_type is ScheduleType.ONCE:
return self.run_at if self.run_at > now else None
zone = ZoneInfo(self.timezone)
next_local = croniter(self.cron, now.astimezone(zone)).get_next(datetime)
if next_local.tzinfo is None:
next_local = next_local.replace(tzinfo=zone)
return next_local.astimezone(UTC)
def ensure_launchable(self, now: datetime, policy: SchedulePolicy) -> datetime | None:
"""Next fire time, with the constraints that only apply at submission.
Used by create/update; the dispatch path must use next_after instead.
Raises:
InvalidScheduleError: a ONCE schedule with no future occurrence, or
one closer than policy.min_once_delay_seconds. CRON is never
subject to the delay floor (router:105, router:196).
"""
if now.tzinfo is None:
now = now.replace(tzinfo=UTC)
next_at = self.next_after(now)
if self.schedule_type is not ScheduleType.ONCE:
return next_at
if next_at is None:
raise InvalidScheduleError("once schedule must be in the future")
if (next_at - now).total_seconds() < policy.min_once_delay_seconds:
raise InvalidScheduleError(f"once schedule must be at least {policy.min_once_delay_seconds} seconds in the future")
return next_at

278
backend/packages/harness/deerflow/domain/schedule/model/task.py Normal file
View File

@ -0,0 +1,278 @@
from __future__ import annotations
import uuid
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime
from deerflow.domain.schedule.model.enums import ContextMode, RunStatus, ScheduleType, TaskStatus, TriggerKind
from deerflow.domain.schedule.model.errors import InvalidContextModeError, TaskNotMutableError
from deerflow.domain.schedule.model.spec import SchedulePolicy, ScheduleSpec
TERMINAL_TASK_STATUSES = frozenset({TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED})
"""The statuses in which a task's current schedule is done.
Lives here rather than in enums.py: it is a business subset of TaskStatus, so
it belongs beside the aggregate that reasons about it.
Two rules read it, and they share one constant rather than drifting into two
same-valued ones:
- `with_schedule` re-arms them to ENABLED once the schedule moves into the
future — claiming only admits ENABLED rows, so leaving a terminal status
there would hand back a next_run_at that silently never fires.
- The repository's `protect_terminal` compare-and-set refuses to overwrite
them, because a fast-failing run's completion hook can land before the launch
path's own write (scheduled_tasks/sql.py:12).
The two coincide because terminal *is* the definition of re-armable. The name
states what the statuses are; what each rule does with them belongs to that
rule.
"""
def _parse_context_mode(value: str | ContextMode) -> ContextMode:
"""Coerce a wire-level context_mode into the enum.
Reported as InvalidContextModeError rather than letting ValueError
escape, so the router can map the whole ScheduleError family uniformly
— an unknown mode was a 422 at router:76-77.
"""
try:
return ContextMode(value)
except ValueError as exc:
raise InvalidContextModeError(f"Unsupported context_mode: {value}") from exc
@dataclass(frozen=True)
class ScheduledTask:
"""A user's standing instruction to run one prompt on a schedule.
Aggregate root of the schedule context. Holds every invariant the old
router/service pair scattered across three files; nothing here knows about
HTTP, SQL, or the run runtime.
The `with_*` / `paused` / `resumed` transitions return a new aggregate and
deliberately leave `updated_at` alone: the repository stamps it on write
(scheduled_tasks/sql.py:97), and a second clock read here would make the
domain both impure and a competing source of truth for the same column.
"""
task_id: str
user_id: str
title: str
prompt: str
schedule: ScheduleSpec
context_mode: ContextMode = ContextMode.FRESH_THREAD_PER_RUN
thread_id: str | None = None
assistant_id: str | None = "lead_agent"
status: TaskStatus = TaskStatus.ENABLED
# The MVP fixes this to "skip"; a single-valued enum would be noise. The
# comparison is confined to `skips_on_overlap` so a second policy only has
# to change one place.
overlap_policy: str = "skip"
next_run_at: datetime | None = None
last_run_at: datetime | None = None
last_run_id: str | None = None
last_thread_id: str | None = None
last_error: str | None = None
run_count: int = 0
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
def __post_init__(self) -> None:
if self.context_mode is ContextMode.REUSE_THREAD and not self.thread_id:
raise InvalidContextModeError("reuse_thread requires thread_id")
@classmethod
def create(
cls,
*,
user_id: str,
title: str,
prompt: str,
schedule: ScheduleSpec,
context_mode: str | ContextMode,
thread_id: str | None,
now: datetime,
policy: SchedulePolicy,
) -> ScheduledTask:
"""Factory: generate identity, normalize context, validate invariants.
FRESH_THREAD_PER_RUN drops any supplied thread_id (router:164-165 does
the same on update) — carrying a thread the task will never use would
make `resolve_execution_thread` ambiguous.
Thread existence/ownership is NOT checked here: that needs the
ThreadLookup port and belongs to the service.
"""
mode = _parse_context_mode(context_mode)
effective_thread = thread_id if mode is ContextMode.REUSE_THREAD else None
return cls(
task_id=f"task-{uuid.uuid4().hex}",
user_id=user_id,
title=title,
prompt=prompt,
schedule=schedule,
context_mode=mode,
thread_id=effective_thread,
next_run_at=schedule.ensure_launchable(now, policy),
)
@property
def skips_on_overlap(self) -> bool:
"""Whether an overlapping dispatch is dropped rather than queued."""
return self.overlap_policy == "skip"
def resolve_execution_thread(self) -> str:
"""Pick the thread this dispatch executes in (service.py:94-96).
NOT idempotent: FRESH_THREAD_PER_RUN mints a new uuid4 on every call,
so a dispatch must call this exactly once and reuse the value for the
run row, the launch, and the result.
The empty-thread_id fallback is kept even though __post_init__ makes it
unreachable for new aggregates — rows predating that invariant can
still carry REUSE_THREAD with no thread.
"""
if self.context_mode is ContextMode.FRESH_THREAD_PER_RUN or not self.thread_id:
return str(uuid.uuid4())
return self.thread_id
def ensure_mutable(self) -> None:
"""Reject edits while this dispatch round's lease is held.
Was `_ensure_task_mutable` in the router (409). Called by every
state-changing method here rather than by the caller, so the rule
cannot be forgotten at a new call site. The router historically applied
it to update/pause/resume but not trigger/delete; that asymmetry is
intentional and is preserved by keeping those two off this path.
Raises:
TaskNotMutableError: the task is currently RUNNING.
"""
if self.status is TaskStatus.RUNNING:
raise TaskNotMutableError("Scheduled task is currently running; retry after the active execution finishes")
def status_after_launch(self, *, trigger: TriggerKind) -> TaskStatus:
"""Status to persist after a successful launch (service.py:152-161).
Precondition on `self.status`: for a SCHEDULED trigger it is already
RUNNING (written by claim_due); for a MANUAL trigger it is the
user-facing status. The PAUSED rule below depends on that.
Decision order is load-bearing — ONCE is tested first, so manually
triggering a paused ONCE task yields RUNNING, not PAUSED:
ONCE -> RUNNING (await the completion
hook; declaring COMPLETED at
launch would stick if the run
fails or the process dies)
MANUAL and self.status PAUSED -> PAUSED (a manual run must not
silently resume the schedule)
otherwise -> ENABLED
"""
if self.schedule.schedule_type is ScheduleType.ONCE:
return TaskStatus.RUNNING
if trigger is TriggerKind.MANUAL and self.status is TaskStatus.PAUSED:
return TaskStatus.PAUSED
return TaskStatus.ENABLED
def status_after_failure(self, *, trigger: TriggerKind) -> TaskStatus:
"""Status to persist after a failed launch (was _task_status_for_failure).
Decision order is load-bearing — MANUAL is tested first, so a failed
manual trigger of a ONCE task keeps its status instead of burning the
occurrence:
MANUAL -> self.status (unchanged; the old dict-based code
needed an `or "enabled"` fallback, which the
field default makes unnecessary here)
SCHEDULED, ONCE -> FAILED
SCHEDULED, CRON -> ENABLED (the next occurrence still stands)
"""
if trigger is TriggerKind.MANUAL:
return self.status
if self.schedule.schedule_type is ScheduleType.ONCE:
return TaskStatus.FAILED
return TaskStatus.ENABLED
def status_after_skip(self) -> TaskStatus:
"""Status to persist after an overlap skip (was _task_status_for_skip).
No `trigger` parameter: a skip only happens on the SCHEDULED path — a
manual trigger that overlaps is rejected outright and records no run
row at all, so a parameter here would imply a branch that cannot exist.
ONCE -> FAILED (the single occurrence was lost; COMPLETED would
claim an execution that never happened)
CRON -> ENABLED
"""
if self.schedule.schedule_type is ScheduleType.ONCE:
return TaskStatus.FAILED
return TaskStatus.ENABLED
def status_after_completion(self, outcome: RunStatus) -> TaskStatus | None:
"""Status to persist when a launched run reaches a terminal state.
Returns None when the status must not change — every CRON task, whose
schedule outlives any single run.
CRON -> None
ONCE, SUCCESS -> COMPLETED
ONCE, INTERRUPTED -> CANCELLED (a cancel or same-thread takeover is
not an execution failure)
ONCE, otherwise -> FAILED
The occurrence is consumed either way: the run did launch, so re-arming
would risk duplicate side effects. Note the caller writes `last_error`
unconditionally, whether or not this returns None (service.py:346).
"""
if self.schedule.schedule_type is not ScheduleType.ONCE:
return None
if outcome is RunStatus.SUCCESS:
return TaskStatus.COMPLETED
if outcome is RunStatus.INTERRUPTED:
return TaskStatus.CANCELLED
return TaskStatus.FAILED
def with_schedule(self, schedule: ScheduleSpec, *, now: datetime, policy: SchedulePolicy) -> ScheduledTask:
"""Replace the schedule and recompute next_run_at (router:172-208).
A terminal task whose schedule just moved into the future must be
re-armed: claim_due only admits ENABLED rows, so leaving the terminal
status would return 200 with a next_run_at that silently never fires.
Raises:
TaskNotMutableError: via ensure_mutable.
InvalidScheduleError: via ensure_launchable.
"""
self.ensure_mutable()
next_at = schedule.ensure_launchable(now, policy)
status = TaskStatus.ENABLED if next_at is not None and self.status in TERMINAL_TASK_STATUSES else self.status
return replace(self, schedule=schedule, next_run_at=next_at, status=status)
def with_context(self, context_mode: str | ContextMode, thread_id: str | None) -> ScheduledTask:
"""Replace execution context (router:153-165).
FRESH_THREAD_PER_RUN forces thread_id to None. Thread existence is the
service's job (ThreadLookup port), not this aggregate's.
Raises:
TaskNotMutableError: via ensure_mutable.
InvalidContextModeError: unknown mode, or REUSE_THREAD with no
thread (raised by __post_init__ on the replacement).
"""
self.ensure_mutable()
mode = _parse_context_mode(context_mode)
effective_thread = thread_id if mode is ContextMode.REUSE_THREAD else None
return replace(self, context_mode=mode, thread_id=effective_thread)
def paused(self) -> ScheduledTask:
"""Stop claiming this task until resumed."""
self.ensure_mutable()
return replace(self, status=TaskStatus.PAUSED)
def resumed(self) -> ScheduledTask:
"""Re-admit this task to claim_due."""
self.ensure_mutable()
return replace(self, status=TaskStatus.ENABLED)

484
backend/tests/test_schedule_domain.py Normal file
View File

@ -0,0 +1,484 @@
"""Domain tests for the schedule bounded context.
Deliberately synchronous and dependency-free: no pytest-asyncio, no fakes, no
IO. Everything under `deerflow.domain.schedule.model` is pure, and this file is
the proof — if a rule here ever needs a stub, the rule has leaked out of the
inner ring.
The three `status_after_*` truth tables (TestStatusAfter*) are the reason this
commit exists: those rules used to be static methods on
`app/scheduler/service.py` with no direct coverage at all, reachable only
through the service's integration tests.
"""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import pytest
from deerflow.domain.schedule.model import (
ACTIVE_RUN_STATUSES,
ContextMode,
InvalidContextModeError,
InvalidScheduleError,
RunStatus,
ScheduledRun,
ScheduledTask,
SchedulePolicy,
ScheduleSpec,
ScheduleType,
TaskNotMutableError,
TaskStatus,
TriggerKind,
)
NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
NO_DELAY = SchedulePolicy(min_once_delay_seconds=0)
MIN_60 = SchedulePolicy(min_once_delay_seconds=60)
def cron_spec(expr: str = "0 9 * * *", timezone: str = "UTC") -> ScheduleSpec:
return ScheduleSpec.cron_schedule(expr, timezone)
def once_spec(*, after_seconds: int = 3600, timezone: str = "UTC") -> ScheduleSpec:
return ScheduleSpec.once_at(NOW + timedelta(seconds=after_seconds), timezone)
def make_task(
*,
schedule: ScheduleSpec | None = None,
status: TaskStatus = TaskStatus.ENABLED,
context_mode: ContextMode = ContextMode.FRESH_THREAD_PER_RUN,
thread_id: str | None = None,
) -> ScheduledTask:
return ScheduledTask(
task_id="task-1",
user_id="user-1",
title="Daily summary",
prompt="Summarize",
schedule=schedule if schedule is not None else cron_spec(),
status=status,
context_mode=context_mode,
thread_id=thread_id,
)
# ---------------------------------------------------------------- A. ScheduleSpec invariants
class TestScheduleSpecInvariants:
def test_unknown_timezone_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="Unknown timezone"):
ScheduleSpec.cron_schedule("0 9 * * *", "Mars/Olympus_Mons")
def test_cron_without_expression_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="requires schedule_spec.cron"):
ScheduleSpec(ScheduleType.CRON, "UTC")
@pytest.mark.parametrize("expr", ["0 9 * *", "0 9 * * * *", "0"])
def test_cron_must_have_exactly_five_fields(self, expr):
with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
ScheduleSpec.cron_schedule(expr, "UTC")
def test_once_without_run_at_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="requires run_at"):
ScheduleSpec(ScheduleType.ONCE, "UTC")
def test_direct_construction_cannot_bypass_validation(self):
"""A frozen dataclass is still constructible field-by-field, so the
rules have to live in __post_init__ rather than in the factories."""
with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
ScheduleSpec(ScheduleType.CRON, "UTC", cron="not-a-cron")
def test_cron_whitespace_is_normalized_on_construction(self):
assert ScheduleSpec(ScheduleType.CRON, "UTC", cron=" 0 9 * * * ").cron == "0 9 * * *"
def test_naive_run_at_is_localized_to_the_schedule_timezone(self):
spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai")
assert spec.run_at.utcoffset() == timedelta(hours=8)
assert spec.run_at.astimezone(UTC) == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
def test_aware_run_at_is_left_alone(self):
run_at = datetime(2026, 8, 1, 9, 0, tzinfo=UTC)
assert ScheduleSpec.once_at(run_at, "Asia/Shanghai").run_at == run_at
def test_zulu_iso_input_is_accepted_as_the_same_instant(self):
"""The shape the frontend actually submits: `zonedLocalToUtcIso` output,
which carries a trailing Z. It is already aware, so localization must
leave it alone rather than reinterpreting it in the task's timezone."""
spec = ScheduleSpec.once_at(datetime.fromisoformat("2026-08-01T01:00:00Z"), "Asia/Shanghai")
assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
# ---------------------------------------------------------------- B. next_after
class TestNextAfter:
def test_once_in_the_future_returns_run_at(self):
spec = once_spec(after_seconds=3600)
assert spec.next_after(NOW) == NOW + timedelta(seconds=3600)
def test_once_in_the_past_returns_none(self):
spec = ScheduleSpec.once_at(NOW - timedelta(seconds=1), "UTC")
assert spec.next_after(NOW) is None
def test_once_exactly_now_returns_none(self):
"""`run_at > now`, not `>=` (schedules.py:44)."""
assert ScheduleSpec.once_at(NOW, "UTC").next_after(NOW) is None
def test_cron_returns_the_next_occurrence_in_utc(self):
# 09:00 UTC daily; NOW is 12:00 UTC, so the next one is tomorrow.
assert cron_spec("0 9 * * *", "UTC").next_after(NOW) == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)
def test_cron_is_evaluated_in_the_schedule_timezone(self):
# 09:00 Shanghai == 01:00 UTC. NOW (12:00 UTC) is past today's, so the
# answer is tomorrow 01:00 UTC -- not 09:00 UTC.
assert cron_spec("0 9 * * *", "Asia/Shanghai").next_after(NOW) == datetime(2026, 7, 28, 1, 0, tzinfo=UTC)
def test_cron_absorbs_a_dst_transition(self):
"""US DST starts 2026-03-08, so the same wall-clock cron maps to a
different UTC instant on either side of it. This is what evaluating in
the schedule's timezone buys."""
spec = cron_spec("0 9 * * *", "America/New_York")
before = spec.next_after(datetime(2026, 3, 7, 0, 0, tzinfo=UTC))
after = spec.next_after(datetime(2026, 3, 9, 0, 0, tzinfo=UTC))
assert before == datetime(2026, 3, 7, 14, 0, tzinfo=UTC) # EST, UTC-5
assert after == datetime(2026, 3, 9, 13, 0, tzinfo=UTC) # EDT, UTC-4
def test_naive_now_is_read_as_utc(self):
spec = cron_spec("0 9 * * *", "UTC")
assert spec.next_after(NOW.replace(tzinfo=None)) == spec.next_after(NOW)
# ---------------------------------------------------------------- C. ensure_launchable
class TestEnsureLaunchable:
def test_once_in_the_past_is_rejected(self):
spec = ScheduleSpec.once_at(NOW - timedelta(hours=1), "UTC")
with pytest.raises(InvalidScheduleError, match="must be in the future"):
spec.ensure_launchable(NOW, NO_DELAY)
def test_once_closer_than_the_minimum_delay_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="at least 60 seconds"):
once_spec(after_seconds=59).ensure_launchable(NOW, MIN_60)
def test_once_exactly_at_the_minimum_delay_is_accepted(self):
assert once_spec(after_seconds=60).ensure_launchable(NOW, MIN_60) == NOW + timedelta(seconds=60)
def test_cron_is_never_subject_to_the_delay_floor(self):
"""router:105 / router:196 apply the floor only to `once`; a cron
schedule whose next fire is seconds away is perfectly legal."""
spec = cron_spec("* * * * *", "UTC")
assert spec.ensure_launchable(NOW, MIN_60) is not None
def test_matches_next_after_for_a_valid_once_schedule(self):
spec = once_spec(after_seconds=3600)
assert spec.ensure_launchable(NOW, MIN_60) == spec.next_after(NOW)
# ---------------------------------------------------------------- D. value semantics
class TestScheduleSpecEquality:
"""A value object is its values -- two specs built from equivalent inputs
must compare equal, because the repository relies on that to decide whether
a schedule actually changed.
Serialization shape (the JSON written to `scheduled_tasks.schedule_spec`)
is deliberately NOT tested here: that mapping lives in the adapter layer,
and so do its tests.
"""
def test_equivalent_cron_inputs_converge(self):
assert ScheduleSpec.cron_schedule(" 0 9 * * * ", "UTC") == ScheduleSpec.cron_schedule("0 9 * * *", "UTC")
def test_equivalent_once_inputs_converge(self):
naive = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai")
aware = ScheduleSpec.once_at(datetime.fromisoformat("2026-08-01T01:00:00Z"), "Asia/Shanghai")
assert naive == aware
def test_timezone_is_part_of_identity(self):
at = datetime(2026, 8, 1, 9, 0, tzinfo=UTC)
assert ScheduleSpec.once_at(at, "UTC") != ScheduleSpec.once_at(at, "Asia/Shanghai")
# ---------------------------------------------------------------- E. ScheduledTask invariants
class TestScheduledTaskInvariants:
def test_reuse_thread_requires_a_thread(self):
with pytest.raises(InvalidContextModeError, match="reuse_thread requires thread_id"):
make_task(context_mode=ContextMode.REUSE_THREAD, thread_id=None)
def test_unknown_context_mode_is_a_domain_error(self):
"""Not a bare ValueError -- the router maps the ScheduleError family
uniformly (router:76-77 was a 422)."""
with pytest.raises(InvalidContextModeError, match="Unsupported context_mode"):
ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec(),
context_mode="teleport",
thread_id=None,
now=NOW,
policy=NO_DELAY,
)
def test_create_drops_thread_id_for_fresh_thread_mode(self):
task = ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec(),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id="thread-9",
now=NOW,
policy=NO_DELAY,
)
assert task.thread_id is None
def test_create_computes_next_run_at_and_defaults(self):
task = ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=cron_spec("0 9 * * *", "UTC"),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
now=NOW,
policy=NO_DELAY,
)
assert task.task_id.startswith("task-")
assert task.status is TaskStatus.ENABLED
assert task.run_count == 0
assert task.assistant_id == "lead_agent"
assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)
def test_create_rejects_a_once_schedule_inside_the_delay_floor(self):
with pytest.raises(InvalidScheduleError, match="at least 60 seconds"):
ScheduledTask.create(
user_id="user-1",
title="t",
prompt="p",
schedule=once_spec(after_seconds=10),
context_mode=ContextMode.FRESH_THREAD_PER_RUN,
thread_id=None,
now=NOW,
policy=MIN_60,
)
def test_skips_on_overlap_reflects_the_policy(self):
assert make_task().skips_on_overlap is True
# ---------------------------------------------------------------- F. the three truth tables
class TestStatusAfterLaunch:
@pytest.mark.parametrize(
("schedule", "trigger", "status", "expected"),
[
(once_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.RUNNING),
(once_spec(), TriggerKind.MANUAL, TaskStatus.ENABLED, TaskStatus.RUNNING),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.PAUSED, TaskStatus.PAUSED),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.ENABLED, TaskStatus.ENABLED),
(cron_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.ENABLED),
],
)
def test_truth_table(self, schedule, trigger, status, expected):
assert make_task(schedule=schedule, status=status).status_after_launch(trigger=trigger) is expected
def test_manual_trigger_of_paused_once_task_yields_running(self):
"""Decision-order trap: ONCE is tested before the MANUAL+PAUSED rule,
so the paused branch never sees a once task."""
task = make_task(schedule=once_spec(), status=TaskStatus.PAUSED)
assert task.status_after_launch(trigger=TriggerKind.MANUAL) is TaskStatus.RUNNING
class TestStatusAfterFailure:
@pytest.mark.parametrize(
("schedule", "trigger", "status", "expected"),
[
(once_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.FAILED),
(cron_spec(), TriggerKind.SCHEDULED, TaskStatus.RUNNING, TaskStatus.ENABLED),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.ENABLED, TaskStatus.ENABLED),
(cron_spec(), TriggerKind.MANUAL, TaskStatus.PAUSED, TaskStatus.PAUSED),
],
)
def test_truth_table(self, schedule, trigger, status, expected):
assert make_task(schedule=schedule, status=status).status_after_failure(trigger=trigger) is expected
@pytest.mark.parametrize("status", [TaskStatus.ENABLED, TaskStatus.PAUSED])
def test_failed_manual_trigger_of_once_task_keeps_status(self, status):
"""Decision-order trap: MANUAL is tested before ONCE, so a failed
manual trigger cannot burn the task's single scheduled occurrence."""
task = make_task(schedule=once_spec(), status=status)
assert task.status_after_failure(trigger=TriggerKind.MANUAL) is status
class TestStatusAfterSkip:
def test_once_is_failed_not_completed(self):
assert make_task(schedule=once_spec()).status_after_skip() is TaskStatus.FAILED
def test_cron_stays_enabled(self):
assert make_task(schedule=cron_spec()).status_after_skip() is TaskStatus.ENABLED
class TestStatusAfterCompletion:
@pytest.mark.parametrize("outcome", list(RunStatus))
def test_cron_never_changes_status(self, outcome):
assert make_task(schedule=cron_spec()).status_after_completion(outcome) is None
@pytest.mark.parametrize(
("outcome", "expected"),
[
(RunStatus.SUCCESS, TaskStatus.COMPLETED),
(RunStatus.INTERRUPTED, TaskStatus.CANCELLED),
(RunStatus.FAILED, TaskStatus.FAILED),
],
)
def test_once_maps_each_terminal_outcome(self, outcome, expected):
assert make_task(schedule=once_spec()).status_after_completion(outcome) is expected
def test_interrupted_is_cancelled_not_failed(self):
"""A user cancel or same-thread takeover carries no execution failure."""
task = make_task(schedule=once_spec())
assert task.status_after_completion(RunStatus.INTERRUPTED) is TaskStatus.CANCELLED
# ---------------------------------------------------------------- G. mutability gate
class TestMutabilityGate:
def test_ensure_mutable_rejects_a_running_task(self):
with pytest.raises(TaskNotMutableError, match="currently running"):
make_task(status=TaskStatus.RUNNING).ensure_mutable()
@pytest.mark.parametrize("status", [s for s in TaskStatus if s is not TaskStatus.RUNNING])
def test_ensure_mutable_allows_every_other_status(self, status):
make_task(status=status).ensure_mutable()
def test_pause_and_resume_are_gated(self):
running = make_task(status=TaskStatus.RUNNING)
with pytest.raises(TaskNotMutableError):
running.paused()
with pytest.raises(TaskNotMutableError):
running.resumed()
def test_pause_then_resume_round_trips(self):
task = make_task(status=TaskStatus.ENABLED)
assert task.paused().status is TaskStatus.PAUSED
assert task.paused().resumed().status is TaskStatus.ENABLED
def test_transitions_do_not_mutate_the_original(self):
task = make_task(status=TaskStatus.ENABLED)
task.paused()
assert task.status is TaskStatus.ENABLED
# ---------------------------------------------------------------- H. with_schedule
class TestWithSchedule:
@pytest.mark.parametrize("status", [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED])
def test_terminal_task_is_rearmed_by_a_future_schedule(self, status):
"""claim_due only admits ENABLED rows, so leaving the terminal status
would return a next_run_at that silently never fires (router:203-208)."""
task = make_task(schedule=once_spec(), status=status)
updated = task.with_schedule(once_spec(after_seconds=7200), now=NOW, policy=NO_DELAY)
assert updated.status is TaskStatus.ENABLED
assert updated.next_run_at == NOW + timedelta(seconds=7200)
@pytest.mark.parametrize("status", [TaskStatus.ENABLED, TaskStatus.PAUSED])
def test_non_terminal_status_is_preserved(self, status):
updated = make_task(status=status).with_schedule(cron_spec("0 10 * * *"), now=NOW, policy=NO_DELAY)
assert updated.status is status
def test_running_task_cannot_be_rescheduled(self):
task = make_task(status=TaskStatus.RUNNING)
with pytest.raises(TaskNotMutableError):
task.with_schedule(cron_spec("0 10 * * *"), now=NOW, policy=NO_DELAY)
def test_invalid_once_schedule_propagates(self):
task = make_task(schedule=once_spec(), status=TaskStatus.COMPLETED)
with pytest.raises(InvalidScheduleError, match="must be in the future"):
task.with_schedule(ScheduleSpec.once_at(NOW - timedelta(hours=1), "UTC"), now=NOW, policy=NO_DELAY)
# ---------------------------------------------------------------- I. with_context
class TestWithContext:
def test_switching_to_fresh_thread_clears_the_thread(self):
task = make_task(context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1")
updated = task.with_context(ContextMode.FRESH_THREAD_PER_RUN, "thread-1")
assert updated.context_mode is ContextMode.FRESH_THREAD_PER_RUN
assert updated.thread_id is None
def test_switching_to_reuse_thread_without_a_thread_is_rejected(self):
with pytest.raises(InvalidContextModeError, match="reuse_thread requires thread_id"):
make_task().with_context(ContextMode.REUSE_THREAD, None)
def test_switching_to_reuse_thread_keeps_the_thread(self):
updated = make_task().with_context("reuse_thread", "thread-7")
assert updated.context_mode is ContextMode.REUSE_THREAD
assert updated.thread_id == "thread-7"
def test_running_task_cannot_change_context(self):
with pytest.raises(TaskNotMutableError):
make_task(status=TaskStatus.RUNNING).with_context(ContextMode.FRESH_THREAD_PER_RUN, None)
# ---------------------------------------------------------------- J. execution thread
class TestResolveExecutionThread:
def test_fresh_thread_mode_mints_a_new_thread_every_call(self):
"""Non-idempotent by design -- a dispatch must call this once and reuse
the value for the run row, the launch, and the result."""
task = make_task(context_mode=ContextMode.FRESH_THREAD_PER_RUN)
assert task.resolve_execution_thread() != task.resolve_execution_thread()
def test_reuse_thread_mode_returns_the_bound_thread(self):
task = make_task(context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1")
assert task.resolve_execution_thread() == "thread-1"
assert task.resolve_execution_thread() == "thread-1"
def test_reuse_thread_with_an_empty_thread_falls_back_to_a_fresh_one(self):
"""Unreachable for new aggregates (__post_init__ forbids it) but rows
predating that invariant can still carry this shape."""
task = make_task()
legacy = ScheduledTask.__new__(ScheduledTask)
object.__setattr__(legacy, "context_mode", ContextMode.REUSE_THREAD)
object.__setattr__(legacy, "thread_id", None)
assert ScheduledTask.resolve_execution_thread(legacy) != task.thread_id
# ---------------------------------------------------------------- K. ScheduledRun
class TestScheduledRun:
def test_queued_run_occupies_the_active_slot(self):
run = ScheduledRun.queued(task_id="task-1", thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)
assert run.status is RunStatus.QUEUED
assert run.is_active is True
assert run.record_id.startswith("task-run-")
def test_skipped_tombstone_is_terminal_and_never_queued(self):
"""QUEUED falls inside uq_scheduled_task_run_active's predicate and
would collide with the pre-existing run still holding the slot."""
run = ScheduledRun.skipped_tombstone(task_id="task-1", thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)
assert run.status is RunStatus.SKIPPED
assert run.is_active is False
def test_each_run_gets_a_distinct_identity(self):
first = ScheduledRun.queued(task_id="task-1", thread_id="t", scheduled_for=NOW, trigger=TriggerKind.MANUAL)
second = ScheduledRun.queued(task_id="task-1", thread_id="t", scheduled_for=NOW, trigger=TriggerKind.MANUAL)
assert first.record_id != second.record_id
def test_active_statuses_are_exactly_queued_and_running(self):
assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING)