rootandClaude Opus 5.5
9858b8494d
modules/{auth-settings,sources-identity,acquisition-organize,agent} placeholders listed in
SERVER_MODULES; ServerModule registers routes (under /api/v1, authenticated), tasks,
outbox routes, capabilities and background loops. bootstrap() builds ServerContext (db,
config, logger, events, engine, outbox writer, secrets placeholder, settings reader,
capability registry), fills unregistered tasks with stubs and rejects duplicate task or
outbox ownership. worker starts the OpenWorkflow worker, OutboxRelay and module start().
--role migrate also migrates the engine schema; api/worker connect without migrating.
tasks: TaskRegistry.has, migrateEngineSchema.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
@umt/tasks
持久任务引擎封装。依据 ADR-002:OpenWorkflow 0.10.1(精确锁定)+ postgres.js,引擎表固定在 openworkflow schema。业务代码只依赖本包导出的 TaskEngine 端口,不直接 import openworkflow。
引擎边界
- 引擎只管过程:推进步骤、定时唤醒、重试、租约。业务状态不放引擎,download_task / approval / asset_operation 等都写业务表。
- UI 读业务表。
TaskEngine.get/list只是只读投影,给 task.inspect 和管理页看状态、attempts、错误用。不要读openworkflow.step_attempts来拼时间线:被中断的 attempt 会一直停在running。 - 引擎用自己的 postgres.js 连接池,不能和业务写入放在同一个事务里入队。入队要走 outbox。
组成
| 导出 | 作用 |
|---|---|
OpenWorkflowEngine.connect({ databaseUrl, db, registry }) |
TaskEngine 实现:enqueue / signal / cancel / get / list / startWorker / stop |
defineTask / TaskRegistry / createDefaultRegistry() |
任务定义与注册表;默认注册表里有 §3.1-8 的 9 个任务(目前是空壳) |
OutboxRelay |
用 FOR UPDATE SKIP LOCKED 认领 outbox_event 中 status=pending 且 next_attempt_at 已到的行,按 type 路由后 enqueue(幂等键取行的 idempotency_key,缺省 outbox:<id>),成功标记 dispatched + published_at;失败 attempts+1 指数退避(上限 5 分钟),达 maxAttempts 置 dead;多进程并发时不会重复 |
migrateEngineSchema(databaseUrl, schema?) |
只跑引擎自己的迁移后断开(部署时 --role migrate) |
withAttemptToken(db, opKey, fn, { commit }) |
操作账本 fencing,表是 tasks_op_ledger |
| 表 | tasks_run_meta、tasks_fact、tasks_op_ledger 由 packages/db 迁移 0013_tasks_tables 建立(类型见 @umt/db 的 DB);本包不再自带迁移,只通过注入的 Kysely 访问 |
TaskContext 包含 input(已经过 zod 校验)、runId、traceId、priority、attemptToken(每次执行 handler 都会重新生成)、abortSignal、logger(自动脱敏)、step.run / step.sleep / step.waitForFact、throwIfCanceled()。
- 取消:
cancel(runId, reason)先写tasks_run_meta.cancel_requested_at,再取消引擎里的 run。handler 执行期间会轮询这个标志(默认 1s)并触发abortSignal;每个step.run/sleep开始前也会检查一次。进行中的步骤只能靠自己检查abortSignal来中止。已完成的外部动作不会撤销,补偿要写成显式步骤。 - 等待外部条件:
signal(runId, name, payload)先把事实写进tasks_fact,再发送引擎信号。step.waitForFact(name, { check?, timeout })每一轮都先查库,查不到才带超时等待信号。所以事实比等待方先到也不会丢。审批、下载完成这类业务事实,先写业务表,再用check去读业务表。
新增/填充任务(W2 规则)
- 在
src/definitions.ts里用defineTask({ name, version, inputSchema, retryPolicy, handler })定义任务。输入只放业务 id,handler 自己去读业务表。重试策略要显式写:外部网络类 5 次(1s→60s),文件类 3 次。 - 步骤体必须幂等。worker 被 kill 后,正在执行的步骤会在新 worker 上从头重跑(最长要等约 30s 租约到期)。已完成的步骤不会重跑。
- 有外部副作用的步骤都走
withAttemptToken,opKey 按语义生成,例如submit:<planId>:- 已经有结果时直接返回;
info.resumed === true表示上一次 attempt 没留下结果。这时先对账外部状态(qB 按 hash 或管理标签查,文件按目标路径 + 大小/hash 查),再决定是收尾还是重做;- 业务状态放在
commit(trx, result)里提交,和 fencing 校验在同一个事务。租约丢失的旧 attempt 提交时会抛FencingError。
- 等授权、等下载完成,用
step.waitForFact,不要直接等信号。同一个 run 里,fact 名不要重复使用。 - 长步骤(比如复制大文件)要分块检查
ctx.abortSignal,被中止时throw ctx.abortSignal.reason。 - 步骤名在同一版本内保持稳定。重放是按步骤名匹配的。
- 日志只用
ctx.logger。passkey、token 这类字段和 URL 参数会被脱敏,但也不要主动把凭证传给 logger。
版本升级规则
- 不兼容的改动(改步骤名或顺序、改输入结构、改已有步骤的语义)必须把
version+1,同时保留旧版本定义并注册。worker 用startWorker({ versions: ["submitDownload@1", "submitDownload@2"] })或者省略参数(表示全部版本)。等list({ taskName, status })查不到旧版本的 run 了,再删除旧定义。 - 新 run 一律用最高版本。旧版本 run 如果找不到对应的 worker,会无限退避等待,不会失败。
- 同一版本里,只允许在末尾追加步骤,或插入幂等步骤。
已知缺口(OpenWorkflow 0.10.1)
- 引擎信号不持久,同名信号同一时间只能有一个等待者。已经用
tasks_fact+waitForFact补上。 - 进行中的步骤拿不到引擎的 AbortSignal 和租约状态。已经用 PG 取消标志和操作账本 fencing 补上。
- 租约固定 30s,不可配置。崩溃后最长约 30s 才会续跑。
- 引擎幂等键只在 24h 内有效,而且不比较输入。长期去重靠业务表的唯一约束。
priority只记录在tasks_run_meta和 run 输入里,引擎仍按available_at先后认领,不支持真正的优先级调度。- 取消只阻止后续步骤,不撤销已完成的外部动作。在已取消的 run 上,handler 抛出的错误会被引擎拒收,只在 worker 日志里出现。
- 引擎表(openworkflow schema)由
migrateEngineSchema(databaseUrl)迁移:server 的--role migrate调用它;--role all单进程连接时顺带迁移,api/worker分进程以runMigrations:false连接。
测试
deno task test(或本包 deno task test)会连接开发库,每个测试 schema 先用 @umt/db 的 migrateToLatest 建全部业务表(DATABASE_URL,默认 postgres://umt:umt@localhost:55433/umt)。每个测试文件使用独立的 tasks_test_<名>_<pid+随机> schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。