diff --git a/backend/docs/SCHEDULE_DESIGN_zh.md b/backend/docs/SCHEDULE_DESIGN_zh.md new file mode 100644 index 000000000..3c7308fed --- /dev/null +++ b/backend/docs/SCHEDULE_DESIGN_zh.md @@ -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/
ScheduleSpec · ScheduledTask · ScheduledRun"] + end + subgraph OLD["⏳ 仍是旧形态 · 生产路径"] + R["app/gateway/routers/scheduled_tasks.py
入参校验 + 业务判断混在一起"] + S["app/scheduler/service.py
轮询 + 派发编排 + 状态推导"] + P["deerflow/persistence/scheduled_task*/sql.py
仓储返回裸 dict"] + C["deerflow/scheduler/schedules.py
时区 / 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
必须是未来时间"] + A3 -->|否| A5{"距今 < min_once_delay?"} + A5 -->|是| A6["InvalidScheduleError
至少提前 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)
启动成功后"] --> L1{"ONCE?"} + L1 -->|是| LR["RUNNING"] + L1 -->|否| L2{"MANUAL 且当前 PAUSED?"} + L2 -->|是| LP["PAUSED"] + L2 -->|否| LE["ENABLED"] + + F0["status_after_failure(trigger)
启动失败后"] --> F1{"MANUAL?"} + F1 -->|是| FS["保持原状态"] + F1 -->|否| F2{"ONCE?"} + F2 -->|是| FF["FAILED"] + F2 -->|否| FE["ENABLED"] + + K0["status_after_skip()
因重叠被跳过"] --> K1{"ONCE?"} + K1 -->|是| KF["FAILED"] + K1 -->|否| KE["ENABLED"] + + C0["status_after_completion(outcome)
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`。 diff --git a/backend/packages/harness/deerflow/domain/schedule/__init__.py b/backend/packages/harness/deerflow/domain/schedule/__init__.py new file mode 100644 index 000000000..38497879a --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/__init__.py @@ -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", +] diff --git a/backend/packages/harness/deerflow/domain/schedule/model/__init__.py b/backend/packages/harness/deerflow/domain/schedule/model/__init__.py new file mode 100644 index 000000000..fbc9edbcc --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/__init__.py @@ -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", +] diff --git a/backend/packages/harness/deerflow/domain/schedule/model/enums.py b/backend/packages/harness/deerflow/domain/schedule/model/enums.py new file mode 100644 index 000000000..f112fafed --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/enums.py @@ -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" diff --git a/backend/packages/harness/deerflow/domain/schedule/model/errors.py b/backend/packages/harness/deerflow/domain/schedule/model/errors.py new file mode 100644 index 000000000..d222e5bae --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/errors.py @@ -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.""" diff --git a/backend/packages/harness/deerflow/domain/schedule/model/run.py b/backend/packages/harness/deerflow/domain/schedule/model/run.py new file mode 100644 index 000000000..373387f6e --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/run.py @@ -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 diff --git a/backend/packages/harness/deerflow/domain/schedule/model/spec.py b/backend/packages/harness/deerflow/domain/schedule/model/spec.py new file mode 100644 index 000000000..3176d2dcb --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/spec.py @@ -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 diff --git a/backend/packages/harness/deerflow/domain/schedule/model/task.py b/backend/packages/harness/deerflow/domain/schedule/model/task.py new file mode 100644 index 000000000..0f72bc919 --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/model/task.py @@ -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) diff --git a/backend/tests/test_schedule_domain.py b/backend/tests/test_schedule_domain.py new file mode 100644 index 000000000..96c808d1e --- /dev/null +++ b/backend/tests/test_schedule_domain.py @@ -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)