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]>

deno.json

@@ -65,7 +65,7 @@
     "types": "mkdir -p types && deno types > types/deno.d.ts",
     "typecheck": "deno task types && deno task typecheck:tsc && deno task typecheck:deno",
     "typecheck:tsc": "deno task --members -j 1 typecheck",
-    "typecheck:deno": "deno check apps/server/src/main.ts packages/contracts/src/index.ts packages/db/src/index.ts packages/db/scripts/migrate.ts packages/db/scripts/gen-types.ts packages/db/migrations/*.ts packages/domain/src/index.ts packages/tasks/src/index.ts packages/tasks/migrations/*.ts packages/agent/src/index.ts packages/adapters/src/index.ts packages/testkit/src/index.ts",
+    "typecheck:deno": "deno check apps/server/src/main.ts packages/contracts/src/index.ts packages/db/src/index.ts packages/db/scripts/migrate.ts packages/db/scripts/gen-types.ts packages/db/migrations/*.ts packages/domain/src/index.ts packages/tasks/src/index.ts packages/agent/src/index.ts packages/adapters/src/index.ts packages/testkit/src/index.ts",
     "lint": "deno run -A npm:@biomejs/biome check .",
     "fmt": "deno run -A npm:@biomejs/biome check --write .",
     "test": "deno task --members -j 1 test",

packages/db/README.md

@@ -27,8 +27,10 @@ PostgreSQL 访问层:Kysely 客户端与连接池、表类型(`src/types.ts`)、
 | `0010_agent` | agent_run、artifact、decision、approval、automation_grant |
 | `0011_ops` | reconcile_run、server_mapping |
 | `0012_organization` | organization_record(change_seq 序列 + 触发器)、organization_file_result;回填 asset_operation.organization_record_id |
+| `0013_tasks_tables` | 并入 @umt/tasks 自有表:tasks_run_meta、tasks_fact、tasks_op_ledger(if not exists,可接管 W1 旧库) |
+| `0014_agent_run_event` | agent_run_event:Agent 运行事件,主键 (run_id, seq),删 run 级联 |
 
-## 表清单(41 张)
+## 表清单(45 张)
 
 ### 底座 / 运维
 | 表 | 用途与关键不变量 |
@@ -100,6 +102,14 @@ PostgreSQL 访问层:Kysely 客户端与连接池、表类型(`src/types.ts`)、
 | `decision` | 待处理中心条目;同一 `dedupe_key` 同时只有一个未关闭条目 |
 | `approval` | 单次审批,绑定 plan + payload_hash + expected_revision |
 | `automation_grant` | 持续授权(动作 + 范围 + 预算 + 策略版本),可撤销 |
+| `agent_run_event` | Agent 运行事件(9 种,见 contracts `AgentRunEventType`);**(run_id, seq) 主键**,先落库再推 SSE;payload = 事件主体去掉 type |
+
+### 任务引擎(@umt/tasks 内部使用,不经 API 暴露)
+| 表 | 用途与关键不变量 |
+|---|---|
+| `tasks_run_meta` | 引擎 run 的业务侧元数据(traceId/优先级/取消标志) |
+| `tasks_fact` | signal 前先落库的事实,waitForFact 以它为准 |
+| `tasks_op_ledger` | 操作账本 fencing:attempt_token,旧 attempt 提交被拒 |
 
 ### 整理
 | 表 | 用途与关键不变量 |

packages/db/migrations/0013_tasks_tables.ts

@@ -0,0 +1,60 @@
+import type { Kysely } from "kysely";
+import { dropTables, execDdl, oneOf } from "../src/migration-helpers.ts";
+
+// @umt/tasks 的三张自有表并入统一迁移(W1 审查 S2;原 packages/tasks/migrations/0001_tasks_tables)。
+// 表结构与原迁移一致,tasks 包通过注入的 Kysely 访问:
+// - tasks_run_meta:引擎 run 的业务侧元数据(traceId/优先级/取消标志),取消标志是 abortSignal 的来源
+// - tasks_fact:signal 之前先落库的事实;waitForFact 以它为准,信号只用来加速唤醒
+// - tasks_op_ledger:操作账本(fencing),记录 attempt_token,旧 attempt 的提交会被拒绝
+// 用 if not exists:W1 期间用 migrateTasksTables 建过表的库可以直接接管;顺手删掉旧的独立迁移记录表。
+
+// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
+export async function up(db: Kysely<any>): Promise<void> {
+  await execDdl(
+    db,
+    `
+create table if not exists tasks_run_meta (
+  run_id text primary key,
+  task_name text not null,
+  task_version integer not null,
+  idempotency_key text not null,
+  trace_id text,
+  priority integer not null default 0,
+  cancel_requested_at timestamptz,
+  cancel_reason text,
+  created_at timestamptz not null default now()
+);
+create index if not exists tasks_run_meta_task_idx on tasks_run_meta (task_name, created_at);
+
+create table if not exists tasks_fact (
+  id bigserial primary key,
+  run_id text not null,
+  name text not null,
+  payload jsonb,
+  created_at timestamptz not null default now()
+);
+create index if not exists tasks_fact_run_name_idx on tasks_fact (run_id, name, id);
+
+create table if not exists tasks_op_ledger (
+  op_key text primary key,
+  run_id text,
+  attempt_token uuid not null,
+  attempts integer not null default 1,
+  status text not null ${oneOf("status", ["started", "succeeded"])},
+  result jsonb,
+  last_error text,
+  started_at timestamptz not null default now(),
+  completed_at timestamptz
+);
+
+do $$ begin
+  execute format('drop table if exists %I.tasks_migration, %I.tasks_migration_lock', current_schema(), current_schema());
+end $$;
+`,
+  );
+}
+
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+export async function down(db: Kysely<any>): Promise<void> {
+  await execDdl(db, dropTables(["tasks_op_ledger", "tasks_fact", "tasks_run_meta"]));
+}

packages/db/migrations/0014_agent_run_event.ts

@@ -0,0 +1,40 @@
+import type { Kysely } from "kysely";
+import { dropTables, execDdl, oneOf } from "../src/migration-helpers.ts";
+
+// Agent 运行事件持久化(W1 审查 B9;REQ §10.6):先落库再推 SSE,断线/跨进程按 (run_id, seq) 回放。
+// type 取值与 contracts AgentRunEventType 一致;payload = 事件主体去掉 type(contracts agentRunEventToRow)。
+
+const TYPES = [
+  "run_started",
+  "text_delta",
+  "thinking_delta",
+  "tool_call_start",
+  "tool_call_end",
+  "tool_call_rejected",
+  "compaction",
+  "run_completed",
+  "run_failed",
+];
+
+// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
+export async function up(db: Kysely<any>): Promise<void> {
+  await execDdl(
+    db,
+    `
+create table agent_run_event (
+  run_id uuid not null references agent_run (id) on delete cascade,
+  seq bigint not null check (seq >= 0),
+  type text not null ${oneOf("type", TYPES)},
+  at timestamptz not null default now(),
+  payload jsonb not null default '{}'::jsonb,
+  constraint agent_run_event_pkey primary key (run_id, seq)
+);
+create index agent_run_event_at_idx on agent_run_event (at);
+`,
+  );
+}
+
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+export async function down(db: Kysely<any>): Promise<void> {
+  await execDdl(db, dropTables(["agent_run_event"]));
+}

packages/db/src/types.ts

@@ -37,6 +37,16 @@ export type AgentRunStatus =
   | "failed"
   | "cancelled"
   | "budget_exceeded";
+export type AgentRunEventType =
+  | "run_started"
+  | "text_delta"
+  | "thinking_delta"
+  | "tool_call_start"
+  | "tool_call_end"
+  | "tool_call_rejected"
+  | "compaction"
+  | "run_completed"
+  | "run_failed";
 export type ApprovalStatus = "pending" | "approved" | "rejected" | "revoked" | "expired" | "consumed";
 export type ArtifactKind = "evidence" | "plan" | "proposal" | "report" | "snapshot" | "log";
 export type AssetLocationLocationKind = "library" | "download_cache" | "hardlink" | "external";
@@ -175,6 +185,7 @@ export type SubscriptionOrigin = "user" | "agent" | "policy" | "import";
 export type SubscriptionPriority = "immediate" | "tracking" | "new_work" | "upgrade";
 export type SubscriptionStatus = "active" | "paused" | "completed" | "cancelled";
 export type TaskStatus = "pending" | "running" | "waiting" | "succeeded" | "failed" | "cancelled" | "dead";
+export type TasksOpLedgerStatus = "started" | "succeeded";
 export type TitleAliasKind = "original" | "translated" | "short" | "site_title" | "romanized" | "other";
 export type TitleAliasSource = "tmdb" | "user" | "site" | "agent" | "nfo" | "import";
 
@@ -250,6 +261,14 @@ export interface AgentRunTable {
   updated_at: GeneratedTimestamptz;
 }
 
+export interface AgentRunEventTable {
+  run_id: string;
+  seq: number;
+  type: AgentRunEventType;
+  at: GeneratedTimestamptz;
+  payload: Generated<Json>;
+}
+
 export interface ApiKeyTable {
   id: Generated<string>;
   name: string;
@@ -951,6 +970,38 @@ export interface TaskTable {
   updated_at: GeneratedTimestamptz;
 }
 
+export interface TasksFactTable {
+  id: Generated<number>;
+  run_id: string;
+  name: string;
+  payload: Json | null;
+  created_at: GeneratedTimestamptz;
+}
+
+export interface TasksOpLedgerTable {
+  op_key: string;
+  run_id: string | null;
+  attempt_token: string;
+  attempts: Generated<number>;
+  status: TasksOpLedgerStatus;
+  result: Json | null;
+  last_error: string | null;
+  started_at: GeneratedTimestamptz;
+  completed_at: Timestamptz | null;
+}
+
+export interface TasksRunMetaTable {
+  run_id: string;
+  task_name: string;
+  task_version: number;
+  idempotency_key: string;
+  trace_id: string | null;
+  priority: Generated<number>;
+  cancel_requested_at: Timestamptz | null;
+  cancel_reason: string | null;
+  created_at: GeneratedTimestamptz;
+}
+
 export interface TitleAliasTable {
   id: Generated<string>;
   media_entity_id: string;
@@ -969,6 +1020,7 @@ export interface DB {
   acquisition_plan: AcquisitionPlanTable;
   acquisition_slot: AcquisitionSlotTable;
   agent_run: AgentRunTable;
+  agent_run_event: AgentRunEventTable;
   api_key: ApiKeyTable;
   approval: ApprovalTable;
   artifact: ArtifactTable;
@@ -1006,6 +1058,9 @@ export interface DB {
   source_release: SourceReleaseTable;
   subscription: SubscriptionTable;
   task: TaskTable;
+  tasks_fact: TasksFactTable;
+  tasks_op_ledger: TasksOpLedgerTable;
+  tasks_run_meta: TasksRunMetaTable;
   title_alias: TitleAliasTable;
 }
 
@@ -1018,6 +1073,9 @@ export type AcquisitionSlotUpdate = Updateable<AcquisitionSlotTable>;
 export type AgentRunRow = Selectable<AgentRunTable>;
 export type NewAgentRun = Insertable<AgentRunTable>;
 export type AgentRunUpdate = Updateable<AgentRunTable>;
+export type AgentRunEventRow = Selectable<AgentRunEventTable>;
+export type NewAgentRunEvent = Insertable<AgentRunEventTable>;
+export type AgentRunEventUpdate = Updateable<AgentRunEventTable>;
 export type ApiKeyRow = Selectable<ApiKeyTable>;
 export type NewApiKey = Insertable<ApiKeyTable>;
 export type ApiKeyUpdate = Updateable<ApiKeyTable>;
@@ -1129,6 +1187,15 @@ export type SubscriptionUpdate = Updateable<SubscriptionTable>;
 export type TaskRow = Selectable<TaskTable>;
 export type NewTask = Insertable<TaskTable>;
 export type TaskUpdate = Updateable<TaskTable>;
+export type TasksFactRow = Selectable<TasksFactTable>;
+export type NewTasksFact = Insertable<TasksFactTable>;
+export type TasksFactUpdate = Updateable<TasksFactTable>;
+export type TasksOpLedgerRow = Selectable<TasksOpLedgerTable>;
+export type NewTasksOpLedger = Insertable<TasksOpLedgerTable>;
+export type TasksOpLedgerUpdate = Updateable<TasksOpLedgerTable>;
+export type TasksRunMetaRow = Selectable<TasksRunMetaTable>;
+export type NewTasksRunMeta = Insertable<TasksRunMetaTable>;
+export type TasksRunMetaUpdate = Updateable<TasksRunMetaTable>;
 export type TitleAliasRow = Selectable<TitleAliasTable>;
 export type NewTitleAlias = Insertable<TitleAliasTable>;
 export type TitleAliasUpdate = Updateable<TitleAliasTable>;

packages/db/test/schema.test.ts

@@ -63,8 +63,24 @@ const EXPECTED_TABLES = [
   "server_mapping",
   "organization_record",
   "organization_file_result",
+  // tasks 包自有表(0013)与 agent 事件(0014)
+  "tasks_run_meta",
+  "tasks_fact",
+  "tasks_op_ledger",
+  "agent_run_event",
 ].sort();
 
+/** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */
+const NON_ENTITY_TABLES = [
+  "settings",
+  "audit_event",
+  "outbox_event",
+  "tasks_run_meta",
+  "tasks_fact",
+  "tasks_op_ledger",
+  "agent_run_event",
+];
+
 describe.skipIf(skip)("schema migrations (real PG)", () => {
   let admin: Kysely<unknown>;
   let db: Kysely<DB>;
@@ -108,7 +124,7 @@ describe.skipIf(skip)("schema migrations (real PG)", () => {
         select table_name, column_name, column_default from information_schema.columns
          where table_schema = current_schema() and column_name in ('id', 'created_at', 'updated_at')`.execute(db)
     ).rows;
-    for (const t of EXPECTED_TABLES.filter((x) => !["settings", "audit_event", "outbox_event"].includes(x))) {
+    for (const t of EXPECTED_TABLES.filter((x) => !NON_ENTITY_TABLES.includes(x))) {
       const cols = rows.filter((r) => r.table_name === t);
       expect(cols.map((c) => c.column_name).sort(), t).toEqual(["created_at", "id", "updated_at"]);
       expect(cols.find((c) => c.column_name === "id")?.column_default, t).toBe("gen_random_uuid()");
@@ -291,6 +307,23 @@ describe.skipIf(skip)("schema migrations (real PG)", () => {
     expect(hit.rows.map((r) => r.id)).toContain(ent.id);
   });
 
+  it("agent_run_event:(run_id, seq) 唯一,删 run 级联删事件", async () => {
+    const run = await db
+      .insertInto("agent_run")
+      .values({ trigger: "user", goal: "g" })
+      .returning("id")
+      .executeTakeFirstOrThrow();
+    const ev = { run_id: run.id, seq: 0, type: "run_started" as const, payload: JSON.stringify({ goal: "g" }) };
+    await db.insertInto("agent_run_event").values(ev).execute();
+    await expect(db.insertInto("agent_run_event").values(ev).execute()).rejects.toThrow(/agent_run_event_pkey/);
+    await expect(
+      sql`insert into agent_run_event (run_id, seq, type) values (${run.id}, 1, 'bogus')`.execute(db),
+    ).rejects.toThrow(/check constraint/);
+    await db.deleteFrom("agent_run").where("id", "=", run.id).execute();
+    const left = await db.selectFrom("agent_run_event").select("seq").where("run_id", "=", run.id).execute();
+    expect(left).toEqual([]);
+  });
+
   it("迁移可完整 down 再 up(往返)", async () => {
     const migrator = createMigrator(db, { migrationTableSchema: schema });
     const down = await migrator.migrateTo(NO_MIGRATIONS);

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 { FencingError, getOperation, withAttemptToken } from "../src/index.ts";
-import { connectDb, dropSchemas, pgAvailable, resetSchemas, type TestDb, testSchemas } from "./helpers.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 migrateBizSchema(db, schemas.biz);
-    await sql`CREATE TABLE download_task (id text PRIMARY KEY, hash text)`.execute(db);
+    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