deer-flow/backend/docs/SCHEDULE_DESIGN_zh.md
rayhpeng b56e939c4b docs(schedule): cover the ports and the application service
The design doc stopped at the model layer while ports.py and service.py
had already landed, so it described the inner ring as one third of what
is actually there and still called the other two "to be built".

Two new chapters. Ports covers the three contracts that matter more than
the signatures — another user's task reads as absent rather than
forbidden, only ThreadBusyError or LaunchFailedError may escape a
launch, and RunOutcome keeps the run runtime out of the domain — plus
the two deliberate absences (no Clock, no claimer identity) and the line
between single-threaded semantics, which the contract owns, and
atomicity, which it does not. The service chapter walks dispatch_task's
four exits as a diagram, explains why the global budget is not a
per-poll batch size, and records why the context change travels packaged
rather than behind a sentinel.

The rest follows: the migration map now shows the inner ring complete
and only adapters outstanding, the overview gains the discipline each of
the three layers is held to, the dev guide gains "add a use case" and a
warning against writing rules in the service, and the pitfalls and
glossary pick up what the new layers introduce.
2026-07-28 14:20:36 +08:00

33 KiB
Raw Blame History

定时任务Schedule模块设计

面向想理解或扩展定时任务模块的人。读完你将能回答:一条定时规则从创建到执行经历了什么、它的每条业务规则住在哪个文件、以及你要改它时该动哪里。

配套文档:HEXAGONAL_ARCHITECTURE_zh.md(本文遵循的架构分层)、backend/AGENTS.md(编码规约与调度相关的运行时约定)。

本文覆盖整个内圈domain/schedule/:模型、端口、应用服务)。适配器与入口仍在旧位置,见 §2。


1. 这个模块做什么

一句话:

让用户注册一条「到点、或按 cron 用这段 prompt 起一次 agent run」的规则由后台轮询器按时把它派发进既有的 Gateway run 生命周期。

一条最简单的使用路径:

用户在 /workspace/scheduled-tasks 建一条任务
  "每天 09:00Asia/Shanghai帮我总结昨天的 GitHub issue"
       ↓
后台轮询器每 5 秒扫一次,发现它到点了
       ↓
起一个 agent run新 thread、非交互模式
       ↓
run 结束后回写执行记录,并算出下一次的时间

一条硬约束(写在 backend/AGENTS.md 里):调度器只决定 when,不得引入第二套执行栈。它最终只做一件事——在正确的时刻调用一次现有的 run 启动入口,然后记账。模块的全部复杂度都在"正确的时刻"和"记账"的并发与崩溃语义上,而不在执行本身。

两个核心概念,对应两张表、两个聚合:

概念 是什么 聚合
任务task 用户注册的规则,长期存在 ScheduledTask scheduled_tasks
执行run 规则的一次触发,只读历史 ScheduledRun scheduled_task_runs

2. 当前状态:新旧并存

这一节请先读,否则你会在代码库里迷路。

模块正处于六边形重构的中途。整个内圈已经落地,但生产代码路径还没有切过来:

flowchart LR
    subgraph NEW["✅ 已落地 · 内圈"]
        direction TB
        SV["service.py<br/>ScheduleService用例编排"]
        PO["ports.py<br/>4 个 Protocol + 2 个 DTO"]
        M["model/<br/>ScheduleSpec · ScheduledTask · ScheduledRun"]
        SV --> PO
        SV --> M
    end
    subgraph OLD["⏳ 仍是旧形态 · 生产路径"]
        R["gateway/routers/scheduled_tasks.py<br/>入参校验 + 业务判断混在一起"]
        S["app/scheduler/service.py<br/>轮询 + 派发编排 + 状态推导"]
        P["persistence/scheduled_task*/sql.py<br/>仓储返回裸 dict"]
        C["deerflow/scheduler/schedules.py<br/>时区 / cron 计算"]
    end
    subgraph TODO["🚧 待建 · 外圈"]
        AD["app/infra/persistence/<br/>app/infra/schedule/<br/>app/scheduler/poller.py"]
    end

    R -.->|尚未调用| SV
    S -.->|尚未调用| SV
    AD -.->|将实现| PO

    style NEW fill:#eef6ff
    style OLD fill:#f5f5f5

含义很具体:

  • domain/schedule/ 是唯一的真相声明处,但目前只有测试在用它。运行中的定时任务走的仍是 app/scheduler/service.py 那套。
  • 两边规则内容一致(内圈是逐条从旧代码搬迁的,每个方法的 docstring 都标了来源行号),但代码是重复的。这是迁移中间态的正常代价。
  • 你现在改一条业务规则,要两边都改,直到迁移完成。旧位置见 §12 的索引。
  • 新增业务规则请只写在内圈,然后在旧位置调用它——不要再往 router / 旧 service 里加新的判断。

还差什么:四个端口都没有真实实现。适配器落地后,app/scheduler/service.pydeerflow/scheduler/ 整包、两个 sql.py 都会消失router 瘦身成协议转换。

3. 内圈全景

domain/schedule/
├── service.py       ScheduleService —— 用例编排input port
├── ports.py         4 个 Protocol + LaunchedRun / RunOutcomeoutput ports
└── model/           领域模型;是包而非单文件,因为这里有两个聚合加值对象
    ├── errors.py    9 个领域错误,零依赖
    ├── enums.py     6 个枚举,零依赖
    ├── spec.py      ScheduleSpec值对象· SchedulePolicy值对象
    ├── task.py      ScheduledTask聚合根· TERMINAL_TASK_STATUSES
    └── run.py       ScheduledRun聚合· ACTIVE/TERMINAL_RUN_STATUSES

三层的纪律各不相同:

文件 职责 纪律
model/ 业务事实与不变量 零依赖;不知道存储和 HTTP 存在;不变量在构造期校验
ports.py 领域声明的接口 技术中立——签名里不出现 SQL、表名、HTTP 状态码、运行时类型
service.py 用例编排 内圈唯一调用 output port 的地方;user_id 显式传参;自身不含业务规则

一个关键理解:service 调用 port 是合法的——port 是领域自己声明、自己拥有的接口,调用自己的抽象不构成对外圈的依赖。运行时注入的实现来自外圈,但 service 只见 Protocol 类型。

纪律:这一层零基础设施依赖——没有 SQL、没有 HTTP、没有配置读取、不看时钟now 一律由调用方显式传入。CI 的 tests/test_harness_domain_purity.py 会 AST 扫描整个 domain/ 目录执法。唯一的第三方依赖是 croniter,它是确定性纯计算库(无 IO、无全局状态与标准库 zoneinfo 同性质——日历计算本身就是定时任务的领域知识。

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 cron5 段) 周期性,永远有下一次
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:00Asia/Shanghai 意为"上海时间早上九点",不是 UTC 九点

非法输入在任何 IO 发生之前就抛 InvalidScheduleError

4.3 两个算时间的方法,别用反

这是本模块最容易写错的一处:

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 在任务时区里算,返回 UTConce 只在还没过期时返回。
  • 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-07EST, UTC-5→ 14:00 UTC
  2026-03-09EDT, 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
调度 scheduleScheduleSpec
执行上下文 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 状态机

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 的任务的调度改到未来时间,状态会被强制拉回 enabledTERMINAL_TASK_STATUSES)。不这么做的话,接口会返回 200、next_run_at 有值、但永远不触发——静默死亡。

5.4 四条状态推导规则

派发流程的每个出口都要回答"任务接下来是什么状态"。这四个方法就是答案,判定顺序即语义(现状是 if/elif/else,写成并列的 if 会静默改变行为):

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. 端口:领域对外的四个依赖

ports.py 声明领域需要外界做什么,由外圈实现。签名一律技术中立——出现 SQL、表名、HTTP 状态码就是越界了。

端口 回答什么问题
ScheduledTaskRepository 规则存在哪、怎么按用户隔离、怎么原子地认领到期任务
ScheduledRunRepository 执行记录存在哪、谁来仲裁"一个任务只能有一条活跃执行"
RunLauncher 怎么真正启动一次 agent run
ThreadLookup 这个 thread 存在吗、这个用户能用吗

外加两个 DTOLaunchedRun(启动成功后拿到的 run 身份)、RunOutcome(一次执行到达终态的领域表述)。

7.1 三条约定,比签名更重要

① 越权一律表现为"不存在"。 别人的任务在读取时返回 None / False / 从列表里消失,而不是抛权限错误——调用方不能借此判断"这个 id 到底存不存在"。

RunLauncher 只允许两种异常逃逸。

执行 thread 已经忙  ->  ThreadBusyError
其他任何失败        ->  LaunchFailedError

这条约定是整个重构的支点。旧代码的编排层要靠 isinstance(exc, HTTPException) and exc.status_code == 409 去嗅探"线程忙",于是业务逻辑依赖了 Web 框架。翻译交给适配器之后,领域只认自己的两个错误。两者必须区分,因为结果不同:自动调度遇到线程忙是一次跳过的机会,真正的失败则要记为失败。

RunOutcome 挡住运行时类型。 旧的完成回调直接吃 deerflow.runtime.RunRecord——一个基础设施类型,纯度测试会拦下它。现在由外圈先转换成 RunOutcome,顺带承担了旧代码内联做的过滤:一次不带定时任务元数据、或还没到终态的 run压根产生不出 RunOutcomeservice 也就不会被调用。

7.2 两个刻意的缺席

没有 Clock 端口。 now 一律由调用方显式传参(run_once(now=...)dispatch_task(now=...)),领域从不读时钟,测试天然确定。再加一个 Clock 只会制造"到底该用参数还是 self._clock.now()"的第二个真相源。

claim_due 不接收"谁在认领"。 那是进程身份,不是规则。适配器可以记一个(诊断用),但没有任何代码读回来——决定认领能否被接管的只有过期时间。相比之下 lease_seconds(多久算过期)留在了 SchedulePolicy 里,因为它直接决定崩溃后多快能恢复,是领域关心的策略。

7.3 原子性归实现,不归契约

claim_due 的 docstring 明说:单线程语义(选哪些行、写什么状态)是契约的一部分,原子性不是。一个内存实现可以满足全部单线程规则却毫无并发保证。

这条边界很重要——tests/schedule_fakes.py 的模块 docstring 和 test_schedule_fakes.py 都重复了它:契约测试全绿不代表可以跑多个调度器实例。并发由真数据库上的 test_scheduled_task_dispatch_race.py 单独负责。


8. 应用服务:用例编排

ScheduleService 是这个上下文的 input port。主适配器HTTP router、轮询器、完成回调调它然后把返回值翻译成自己的协议。它自身不含业务规则——每个判断都委托给聚合。

8.1 用例清单

分类 方法
list_tasks · list_tasks_by_thread · get_task · list_task_runs
create_task · update_task · pause_task · resume_task · delete_task
派发 trigger_task(手动) · run_once(轮询一轮) · dispatch_task(单次派发)
生命周期 handle_run_completion · reconcile_on_startup

8.2 dispatch_task:四条出口

整个模块的风险中心。它把一个到期任务变成一次执行,有且只有四种结局:

flowchart TD
    A["dispatch_task(task, now, trigger)"] --> B["resolve_execution_thread()<br/>只调一次"]
    B --> C{"skips_on_overlap<br/>且已有活跃执行?"}
    C -->|是, 手动| D["CONFLICT<br/>不留任何记录"]
    C -->|是, 自动| E["SKIPPED<br/>写终态墓碑"]
    C -->|否| F["创建 queued 记录"]
    F -->|"活跃槽位被抢"| G{"trigger?"}
    G -->|手动| D
    G -->|自动| E
    F -->|插入成功| H["launcher.launch(...)"]
    H -->|ThreadBusyError + 自动 + skip| E
    H -->|ThreadBusyError + 其他| I["CONFLICT<br/>留失败记录"]
    H -->|LaunchFailedError| J["FAILED"]
    H -->|成功| K["LAUNCHED<br/>回写两张表"]

    style D fill:#fff0e0
    style E fill:#e8f0ff
    style J fill:#ffe0e0
    style K fill:#e2f7e2

快路径与槽位仲裁必须产生相同结果。 has_active 是非原子的快路径——两个并发派发可以都通过它。真正的仲裁者是仓储:第二条活跃记录会被拒绝,抛 ActiveRunConflictError。调用方不能分辨自己被哪一种机制拦下,否则重试行为就会分叉。test_schedule_service.py 里那条断言逐字段比较两条路径的 DispatchResult,就是钉死这一点。

只有 launch 被 try 包住。 端口契约保证它只逃逸两种异常,所以后续记账写入的失败是真故障,会如实冒泡。旧代码把 launch 和记账一起包在 except Exception 里,结果是一次已经成功启动的执行会因为记账失败被标记成 failed。

8.3 run_once:预算是全局的

active = await runs.count_active()          # 跨所有任务
budget = policy.max_concurrent_runs - active
if budget <= 0: return []
claimed = await tasks.claim_due(now=..., lease_seconds=..., limit=budget)

max_concurrent_runs 限制的是同时活跃的执行总数,不是每轮的批量大小。长时间运行的任务会跨轮次累积,所以每一轮只能认领进剩余的额度。把它当成"每轮最多认领 N 个"会让长任务把系统压垮。

8.4 更新任务:为什么 context 是打包的

await service.update_task(
    task_id, user_id=..., now=...,
    title=None,            # None = 不改
    prompt=None,
    schedule=None,
    context=ContextChange(ContextMode.REUSE_THREAD, "thread-1"),
)

None 表示"没传",不需要哨兵值。这能成立的唯一原因是:所有可选字段里,只有 thread_idNone 本身有含义(解绑),而它和 context_mode 本来就一起变化——with_context 同时接收两者,切换到 fresh 模式就意味着清空绑定。打包成 ContextChange 之后歧义消失,其余字段就能用最朴素的 None

这也正好对上 HTTP 层router 的 PATCH 本来就是 exclude_none 语义。

8.5 完成回调与启动清扫

handle_run_completion(outcome, now) 先写执行记录的终态,再看任务:once 推向终态成功→completed / 中断→cancelled / 失败→failedcron 保持状态不变。last_error 无条件写入——cron 任务保住了调度,仍然要报告上次出了什么问题。任务在 run 飞行途中被删掉不算错误,静默返回。

reconcile_on_startup(error) 依次跑两个清扫并返回修复计数。它不吞异常——部分清扫失败要不要阻塞启动,是调用方的策略,不是领域的。


9. 二次开发指引

9.1 给任务加一个字段

例:加一个 notify_on_failure: bool

  1. model/task.pyScheduledTask 加字段(带默认值)
  2. 若有约束,写进 __post_init__
  3. 若用户要能设置它:service.pycreate_task / update_task 加参数
  4. persistence/scheduled_tasks/model.py 的 ORM 行加列
  5. 新增一个 alembic revisioncd backend && make migrate-rev MSG="..."),用 _helpers.py 的幂等 helper
  6. router 的请求/响应模型加字段
  7. 前端 frontend/src/core/scheduled-tasks/types.ts 同步
  8. 域测试补一条;若走了第 3 步service 测试也补一条

端口通常不用动——add / save 交换的是整个聚合,多一个字段不改变签名。

9.2 加一种调度类型

例:加 interval(每 N 分钟)。

  1. model/enums.pyScheduleType 加成员
  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 列。

9.3 改一条状态推导规则

只改 model/task.py 对应的那个方法,顺便改 app/scheduler/service.py 里的旧副本(见 §2domain/schedule/service.py 一般不用动——它只调用聚合,不复制规则。

域测试里对应的真值表用例必须同步更新——那张表就是规则的规格说明。

9.4 加一个用例

例:加"克隆一个任务"。

  1. service.py 加方法:读出聚合 → 用聚合方法或 replace 造出新的 → 经 add 持久化
  2. 不要在 service 里写规则。如果发现自己在写 if task.status is ...,那条判断属于聚合
  3. 只有当现有端口回答不了你的问题时才加端口方法——先问"这是新的存储能力,还是我把编排写复杂了"
  4. service 测试补一组(全 fake零 IO
  5. router 加端点,把领域错误映射成 HTTP 码

9.5 加一种重叠策略

例:加 queue(排队而非跳过)。

这是改动面最大的一种,因为它触及数据库不变量:

  1. model/task.pyskips_on_overlap 拆成策略判断
  2. scheduled_task_runs 的部分唯一索引必须改成条件化的(... AND overlap_policy = 'skip')——现在它是纯状态谓词,会把排队的执行也挡掉。索引同时定义在 ORM __table_args__ 和迁移文件里,两处都要改(空库 bootstrap 走 create_all,不执行迁移)
  3. 跳过路径的墓碑逻辑要相应分叉

动手前先读 backend/AGENTS.md 里关于这个索引的整段说明。

9.6 不要做的事

  • 不要在 router 或旧 service 里新增业务判断——新规则一律进聚合
  • 不要在 domain/schedule/service.py 里写规则——它只编排;出现 if task.status is ... 就说明放错地方了
  • 不要在领域层读配置或时钟——阈值通过 SchedulePolicy 注入,now 显式传参CI 的纯度测试会拦截基础设施导入
  • 不要让端口签名沾上技术词汇——Mapping[str, Any]、HTTP 状态码、表名出现在 ports.py 里都是信号
  • 不要为定时执行另建一套运行栈——必须复用现有的 run 生命周期

10. 常见陷阱速查

陷阱 后果
ensure_launchable 做派发后重排 正常执行完的 cron 任务被"提前量不足"拒绝
status_after_*if/elif 写成并列 if 两处判定顺序失效,静默改变行为
多次调用 resolve_execution_thread() 每次拿到不同的 thread记录与实际执行对不上
墓碑先建 queued 再改 skipped 撞上唯一索引,跳过流程直接报错
改了 ACTIVE_RUN_STATUSES 没改索引谓词 快路径与数据库仲裁者判断不一致
终态任务改了调度但没重新武装 接口返回 200任务永不触发
只改领域模型,忘了 app/scheduler/service.py 迁移完成前,生产行为不变
适配器让 launch 逃逸出第三种异常 领域收到不认识的错误;线程忙被记成失败而不是跳过
快路径与槽位仲裁产生不同结果 调用方能分辨被哪种机制拦下,重试行为分叉
把契约测试全绿当成可以多实例 fake 没有任何原子性;并发由真数据库的 dispatch_race 测试负责

11. 术语表

术语 含义
任务task 用户注册的定时规则,长期存在
执行run 规则的一次触发,只读历史
派发dispatch 把一个到期任务变成一次执行的动作
触发方式trigger scheduled(轮询器)或 manual(用户点"立即执行"
认领claim 轮询器取得某任务这一轮调度所有权
租约lease 认领时盖的带过期时间的戳,用于崩溃恢复
重叠overlap 到点时上一次执行还没结束
墓碑tombstone 被跳过的那次执行留下的终态记录
重新武装re-arm 把终态任务拉回 enabled 使其可再次被认领
端口port 领域声明、外圈实现的技术中立接口
input port 用例接口,被入口调用——这里就是 ScheduleService
output port 领域对外的依赖,被适配器实现——这里是四个 Protocol
聚合aggregate 一致性边界,不变量在构造期成立
活跃槽位active slot 一个任务同时最多持有一条 queued/running 执行

12. 代码索引

内圈(已迁移,本文覆盖)

文件 内容
domain/schedule/service.py ScheduleService · DispatchResult · ContextChange
domain/schedule/ports.py 4 个 Protocol · LaunchedRun · RunOutcome
domain/schedule/model/enums.py TaskStatus RunStatus ScheduleType ContextMode TriggerKind DispatchOutcome
domain/schedule/model/errors.py 9 个领域错误
domain/schedule/model/spec.py ScheduleSpec SchedulePolicy
domain/schedule/model/task.py ScheduledTask TERMINAL_TASK_STATUSES
domain/schedule/model/run.py ScheduledRun ACTIVE_RUN_STATUSES TERMINAL_RUN_STATUSES
tests/test_schedule_domain.py 域测试,全同步零 IO四张真值表逐格覆盖
tests/test_schedule_service.py 用例测试;完整生命周期跑在 fake 上,是迁移的验收标准
tests/schedule_fakes.py 四个端口的内存实现
tests/test_schedule_fakes.py 端口语义;适配器落地后升级成契约测试

尚未迁移(生产路径,见 §2

文件 内容
app/gateway/routers/scheduled_tasks.py 10 个 HTTP 端点,含入参校验与业务判断
app/scheduler/service.py 轮询循环、派发编排、状态推导、完成回调、启动清扫
deerflow/scheduler/schedules.py 时区 / cron / 下次时间计算
persistence/scheduled_tasks/ 任务表 ORM + 仓储
persistence/scheduled_task_runs/ 执行表 ORM + 仓储,含唯一索引定义
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