feat(db): 0013 tasks tables and 0014 agent_run_event; tasks uses db migrations

tasks_run_meta/tasks_fact/tasks_op_ledger move into packages/db (if not exists, old
tasks_migration tables dropped); packages/tasks/migrations and migrateTasksTables
removed, tests migrate their schema with @umt/db. agent_run_event(run_id, seq, type,
at, payload) with (run_id, seq) primary key; types regenerated, db-sync covers it.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

packages/tasks/README.md

@@ -16,7 +16,7 @@
 | `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`;多进程并发时不会重复 |
 | `withAttemptToken(db, opKey, fn, { commit })` | 操作账本 fencing,表是 `tasks_op_ledger` |
-| `migrateTasksTables(db)` | 本包迁移(`migrations/`):`tasks_run_meta`、`tasks_fact`、`tasks_op_ledger`。W2 并入 `packages/db` |
+| 表 | `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()`。
 
@@ -54,4 +54,4 @@
 
 ## 测试
 
-`deno task test`(或本包 `deno task test`)会连接开发库(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_<名>_<pid+随机>` schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。
+`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。

packages/tasks/src/index.ts

@@ -26,7 +26,6 @@ export {
   silentSink,
   type TaskLogger,
 } from "./logger.ts";
-export { migrateTasksTables, TASKS_MIGRATIONS_DIR } from "./migrate.ts";
 export {
   type AttemptInfo,
   assertAttemptToken,

packages/tasks/test/crash-recovery.test.ts

@@ -6,6 +6,7 @@ import {
   connectDb,
   createSideEffectTable,
   dropSchemas,
+  migrateBizSchema,
   pgAvailable,
   resetSchemas,
   setupEngine,
@@ -46,6 +47,7 @@ describe.skipIf(!hasPg)("崩溃恢复(worker 子进程 kill -9)", () => {
   beforeAll(async () => {
     await resetSchemas(schemas);
     db = connectDb(schemas.biz);
+    await migrateBizSchema(db, schemas.biz);
     await createSideEffectTable(db);
     engine = await setupEngine(db, schemas, createTestRegistry(db)); // 只做客户端,不起 worker
   }, 30_000);

packages/tasks/test/engine.test.ts

@@ -6,6 +6,7 @@ import {
   connectDb,
   createSideEffectTable,
   dropSchemas,
+  migrateBizSchema,
   pgAvailable,
   resetSchemas,
   setupEngine,
@@ -25,6 +26,7 @@ describe.skipIf(!hasPg)("OpenWorkflowEngine(进程内 worker)", () => {
   beforeAll(async () => {
     await resetSchemas(schemas);
     db = connectDb(schemas.biz);
+    await migrateBizSchema(db, schemas.biz);
     await createSideEffectTable(db);
     engine = await setupEngine(db, schemas, createTestRegistry(db));
     await engine.startWorker({ concurrency: 4 });

packages/tasks/test/helpers.ts

@@ -1,6 +1,7 @@
+import { migrateToLatest } from "@umt/db";
 import { Kysely, PostgresDialect, sql } from "kysely";
 import pg from "pg";
-import { migrateTasksTables, OpenWorkflowEngine, silentSink, type TaskRegistry } from "../src/index.ts";
+import { OpenWorkflowEngine, silentSink, type TaskRegistry } from "../src/index.ts";
 
 export const PG_URL =
   process.env.TASKS_TEST_DATABASE_URL ?? process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
@@ -32,11 +33,17 @@ export function testSchemas(suffix: string) {
   return { biz, ow: `${biz}_ow` };
 }
 
+/** search_path 带上 public:db 迁移用到的 pg_trgm 操作符类装在 public */
 export function connectDb(schema: string): TestDb {
-  const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema}` });
+  const pool = new pg.Pool({ connectionString: PG_URL, max: 6, options: `-c search_path=${schema},public` });
   return new Kysely({ dialect: new PostgresDialect({ pool }) });
 }
 
+/** 用 @umt/db 的统一迁移在测试 schema 里建全部业务表(含 tasks_* 与 outbox_event) */
+export async function migrateBizSchema(db: TestDb, schema: string) {
+  await migrateToLatest(db, { migrationTableSchema: schema });
+}
+
 export async function resetSchemas(schemas: { biz: string; ow: string }) {
   const admin = connectDb("public");
   try {
@@ -58,21 +65,6 @@ export async function dropSchemas(schemas: { biz: string; ow: string }) {
   }
 }
 
-/** 与 packages/db 0001_init + 0002_foundation 的 outbox_event 同构 */
-export async function createOutboxTable(db: TestDb) {
-  await sql`CREATE TABLE outbox_event (
-    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
-    at timestamptz NOT NULL DEFAULT now(),
-    type text NOT NULL,
-    payload jsonb NOT NULL DEFAULT '{}'::jsonb,
-    published_at timestamptz,
-    idempotency_key text UNIQUE,
-    status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'dispatched', 'failed', 'dead')),
-    attempts integer NOT NULL DEFAULT 0,
-    next_attempt_at timestamptz NOT NULL DEFAULT now(),
-    last_error jsonb)`.execute(db);
-}
-
 export async function createSideEffectTable(db: TestDb) {
   await sql`CREATE TABLE IF NOT EXISTS side_effect (
     id bigserial PRIMARY KEY, run_id text NOT NULL, step text NOT NULL, note text, pid int,
@@ -85,8 +77,8 @@ export async function sideEffects(db: TestDb, runId: string) {
   return r.rows;
 }
 
+/** 只连接引擎;业务表由调用方先 migrateBizSchema 建好 */
 export async function setupEngine(db: TestDb, schemas: { ow: string }, registry: TaskRegistry) {
-  await migrateTasksTables(db);
   return OpenWorkflowEngine.connect({
     databaseUrl: PG_URL,
     db,

packages/tasks/test/op-ledger.test.ts

@@ -1,7 +1,15 @@
 import { sql } from "kysely";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { FencingError, getOperation, migrateTasksTables, withAttemptToken } from "../src/index.ts";
-import { connectDb, dropSchemas, pgAvailable, resetSchemas, type TestDb, testSchemas } from "./helpers.ts";
+import { FencingError, getOperation, withAttemptToken } from "../src/index.ts";
+import {
+  connectDb,
+  dropSchemas,
+  migrateBizSchema,
+  pgAvailable,
+  resetSchemas,
+  type TestDb,
+  testSchemas,
+} from "./helpers.ts";
 
 const schemas = testSchemas("ledger");
 const hasPg = await pgAvailable();
@@ -12,8 +20,8 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => {
   beforeAll(async () => {
     await resetSchemas(schemas);
     db = connectDb(schemas.biz);
-    await migrateTasksTables(db);
-    await sql`CREATE TABLE download_task (id text PRIMARY KEY, hash text)`.execute(db);
+    await migrateBizSchema(db, schemas.biz);
+    await sql`CREATE TABLE ledger_probe (id text PRIMARY KEY, hash text)`.execute(db);
   });
 
   afterAll(async () => {
@@ -34,7 +42,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => {
         {
           runId: "r1",
           commit: async (trx, result) => {
-            await sql`INSERT INTO download_task (id, hash) VALUES ('d1', ${result.hash})`.execute(trx);
+            await sql`INSERT INTO ledger_probe (id, hash) VALUES ('d1', ${result.hash})`.execute(trx);
           },
         },
       );
@@ -65,7 +73,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => {
         return "old";
       },
       {
-        commit: async (trx) => void (await sql`INSERT INTO download_task (id, hash) VALUES ('d2', 'old')`.execute(trx)),
+        commit: async (trx) => void (await sql`INSERT INTO ledger_probe (id, hash) VALUES ('d2', 'old')`.execute(trx)),
       },
     );
     await oldStartedP;
@@ -80,7 +88,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => {
         return "new";
       },
       {
-        commit: async (trx) => void (await sql`INSERT INTO download_task (id, hash) VALUES ('d2', 'new')`.execute(trx)),
+        commit: async (trx) => void (await sql`INSERT INTO ledger_probe (id, hash) VALUES ('d2', 'new')`.execute(trx)),
       },
     );
     expect(fresh).toBe("new");
@@ -88,7 +96,7 @@ describe.skipIf(!hasPg)("withAttemptToken(操作账本 fencing)", () => {
 
     releaseOld();
     await expect(oldAttempt).rejects.toBeInstanceOf(FencingError);
-    const r = await sql<{ hash: string }>`SELECT hash FROM download_task WHERE id = 'd2'`.execute(db);
+    const r = await sql<{ hash: string }>`SELECT hash FROM ledger_probe WHERE id = 'd2'`.execute(db);
     expect(r.rows).toEqual([{ hash: "new" }]);
     expect(await getOperation(db, "op-2")).toMatchObject({ status: "succeeded", attempts: 2, result: "new" });
   });

packages/tasks/test/outbox.test.ts

@@ -4,9 +4,9 @@ import { type EnqueueOptions, OutboxRelay, type TaskEngine } from "../src/index.
 import { createTestRegistry } from "./fixtures/tasks.ts";
 import {
   connectDb,
-  createOutboxTable,
   createSideEffectTable,
   dropSchemas,
+  migrateBizSchema,
   pgAvailable,
   resetSchemas,
   setupEngine,
@@ -24,7 +24,7 @@ describe.skipIf(!hasPg)("OutboxRelay", () => {
   beforeAll(async () => {
     await resetSchemas(schemas);
     db = connectDb(schemas.biz);
-    await createOutboxTable(db);
+    await migrateBizSchema(db, schemas.biz);
     await createSideEffectTable(db);
     engine = await setupEngine(db, schemas, createTestRegistry(db));
   }, 30_000);
@@ -80,11 +80,12 @@ describe.skipIf(!hasPg)("OutboxRelay", () => {
       SELECT count(*) n FROM outbox_event WHERE status = 'dispatched' AND published_at IS NOT NULL AND attempts = 1`.execute(
       db,
     );
-    expect(dispatched.rows[0]!.n).toBe("60");
+    // @umt/db 把 int8 解析为 number(全局 pg 类型解析器),这里统一转成数字比较
+    expect(Number(dispatched.rows[0]!.n)).toBe(60);
 
     const runs = await sql<{ n: string; keys: string }>`
       SELECT count(*) n, count(DISTINCT idempotency_key) keys FROM tasks_run_meta WHERE task_name = 'echo'`.execute(db);
-    expect(runs.rows[0]).toEqual({ n: "60", keys: "60" });
+    expect([Number(runs.rows[0]!.n), Number(runs.rows[0]!.keys)]).toEqual([60, 60]);
     const traced = await sql<{ trace_id: string }>`
       SELECT m.trace_id FROM tasks_run_meta m
       JOIN outbox_event o ON m.idempotency_key = 'outbox:' || o.id::text