feat(storage): 数据保留补齐引擎运行与流水表,新增每天低峰的流水数据清理和 db-retention 脚本

- 保留规则改为声明式(retention.ts 每个任务的 rules),定时任务、立即清理、脚本 dry-run 共用一份条件;
  删得多(≥ 1000 行)的表删完 ANALYZE
- 引擎 run 单独保留期 engineRun(缺省 14 天):只删终态、过期、没被保留中业务任务引用、父 run 已结束的;
  step_attempts 级联,运行元数据 / 事实 / 信号跟随 run(原来按时间直接删,长时间运行的 run 会丢元数据)
- 业务任务被 agent_run / decision 等引用时不删(外键 set null,删了会断开引用)
- 新任务 storage.retention.pipeline(每天 04:40):已投递 outbox 事件(只删无幂等键或一次性幂等键)、
  识别尝试(每条资源留最近 3 次)、已识别且资源仍绑定作品的待识别条目(忽略 / 归档不删)、
  站点快照(每站最近一次成功的列表快照保留)
- settings.storage.retentionDays 追加 engineRun / outboxEvent / resolutionAttempt / identityResolved /
  siteSnapshot(旧库按缺省补),设置页「系统 → 数据保留」显示
- scripts/db-retention.ts:缺省 dry-run 输出每张表将删行数,--apply 执行,--only 选目标
- 测试:各表按规则删、被引用 / 未结束的不删、dry-run 与删除一致、分批;EXPLAIN 确认改写后的查询走索引

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

apps/server/src/modules/system/index.ts

@@ -202,7 +202,7 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule
               name: j.name,
               description: j.description,
               group: "maintain",
-              schedule: { type: "interval", everyMs: j.everyMs },
+              schedule: j.schedule ?? { type: "interval", everyMs: j.everyMs },
               firstRun: "after_interval",
             },
             run: async ({ now }) => ({

apps/server/src/modules/system/retention.ts

@@ -1,4 +1,4 @@
-import type { StorageCleanupTarget, StorageSettings } from "@umt/contracts";
+import type { ScheduleSpec, StorageCleanupTarget, StorageSettings } from "@umt/contracts";
 import { type DB, type Kysely, type RawBuilder, sql } from "@umt/db";
 
 // CACHE-2 保留策略清理(docs/runbook/storage.md):按表按时间列分批删除流水数据。
@@ -32,7 +32,7 @@ export interface BatchResult {
 const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
 
 /**
- * 分批删除 `table` 中满足 `where` 的行:每批 `delete … where ctid = any(array(select ctid … limit n))`,
+ * 分批删除 `table` 中满足 `where` 的行(`where` 里本表别名为 x):每批 `delete … where ctid = any(array(select ctid … limit n))`,
  * 单独事务 + 语句 / 锁超时。某批超时(57014)或等锁超时(55P03)时停止本次运行,不抛错。
  */
 export async function deleteInBatches(
@@ -53,7 +53,7 @@ export async function deleteInBatches(
         await sql`select set_config('statement_timeout', ${String(o.statementTimeoutMs ?? 15_000)}, true),
                          set_config('lock_timeout', ${String(o.lockTimeoutMs ?? 3000)}, true)`.execute(trx);
         const r = await sql`delete from ${table} where ctid = any(array(
-            select ctid from ${table} where ${where} limit ${size}))`.execute(trx);
+            select x.ctid from ${table} as x where ${where} limit ${size}))`.execute(trx);
         return Number(r.numAffectedRows ?? 0);
       });
     } catch (e) {
@@ -69,33 +69,237 @@ export async function deleteInBatches(
   }
 }
 
+/**
+ * 一条保留规则:`table` 里满足 `where`(本表别名 x)的行到期删除。
+ * cascade = 不单独删,随上一张表级联删除(只在 dry-run 里统计,如引擎 step_attempts)。
+ */
+export interface RetentionRule {
+  label: string;
+  /** 已加引号的表名(引擎表带 schema) */
+  table: string;
+  where: RawBuilder<unknown>;
+  cascade?: boolean;
+}
+
 export interface RetentionJob {
   target: StorageCleanupTarget;
   /** scheduled_job key */
   jobId: string;
   name: string;
   description: string;
+  /** 缺省按 everyMs 间隔;给了 schedule 以它为准(如每天低峰的 cron) */
   everyMs: number;
+  schedule?: ScheduleSpec;
   /** 保留期说明(存储面板) */
   retention(s: StorageSettings): string | null;
+  /** 删除类任务的规则(dry-run 与执行共用);归档 / ANALYZE 类没有 */
+  rules?(c: RetentionRuleContext): RetentionRule[];
+  /** dry-run 时的额外统计(非删除类,如归档条数) */
+  preview?(c: RetentionRuleContext): Promise<RetentionCount[]>;
   run(c: RetentionRunContext): Promise<string>;
 }
 
-export interface RetentionRunContext {
+export interface RetentionRuleContext {
   db: Kysely<DB>;
   engineSchema: string;
   settings: StorageSettings;
   now: Date;
+}
+
+export interface RetentionRunContext extends RetentionRuleContext {
   batch?: BatchOptions;
 }
 
+export interface RetentionCount {
+  label: string;
+  table: string;
+  rows: number;
+  cascade?: boolean;
+}
+
 const DAY_MS = 86_400_000;
+const HOUR = 3600_000;
+/** 一次删掉这么多行以上的表,删完顺带 ANALYZE(统计信息跟上,不等 autovacuum) */
+export const ANALYZE_AFTER_ROWS = 1000;
 const cutoffOf = (now: Date, days: number) => new Date(now.getTime() - days * DAY_MS);
-const t = (name: string) => sql.raw(`"${name}"`);
+const q = (name: string) => `"${name}"`;
+const truncNote = (rs: BatchResult[]) => (rs.some((r) => r.truncated) ? "，未删完，下次继续" : "");
 const summaryOf = (label: string, r: BatchResult) =>
   `${label}删除 ${r.deleted} 行（${r.batches} 批）${r.truncated ? "，未删完，下次继续" : ""}`;
 
-const HOUR = 3600_000;
+/** 按顺序执行规则(cascade 的跳过),删得多的表随后 ANALYZE */
+export async function applyRules(c: RetentionRunContext, rules: RetentionRule[]): Promise<BatchResult[]> {
+  const out: BatchResult[] = [];
+  const deletedByTable = new Map<string, number>();
+  for (const r of rules) {
+    if (r.cascade) continue;
+    const res = await deleteInBatches(c.db, sql.raw(r.table), r.where, c.batch);
+    out.push(res);
+    deletedByTable.set(r.table, (deletedByTable.get(r.table) ?? 0) + res.deleted);
+  }
+  for (const [table, n] of deletedByTable) {
+    if (n >= ANALYZE_AFTER_ROWS) await sql`analyze ${sql.raw(table)}`.execute(c.db);
+  }
+  return out;
+}
+
+/** dry-run:每条规则将删的行数(不改库) */
+export async function countRules(db: Kysely<DB>, rules: RetentionRule[]): Promise<RetentionCount[]> {
+  const out: RetentionCount[] = [];
+  for (const r of rules) {
+    const n = await sql<{
+      n: string;
+    }>`select count(*)::text as n from ${sql.raw(r.table)} as x where ${r.where}`.execute(db);
+    out.push({
+      label: r.label,
+      table: r.table,
+      rows: Number(n.rows[0]?.n ?? 0),
+      ...(r.cascade ? { cascade: true } : {}),
+    });
+  }
+  return out;
+}
+
+/** 一个保留任务的 dry-run 统计 */
+export async function previewJob(job: RetentionJob, c: RetentionRuleContext): Promise<RetentionCount[]> {
+  return [...(job.rules ? await countRules(c.db, job.rules(c)) : []), ...((await job.preview?.(c)) ?? [])];
+}
+
+// ── 业务任务与引擎 run(task_run)──
+// 引擎(openworkflow)语义:幂等键只在 24 小时内去重(DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS),恢复只看
+// pending / running / sleeping 的 run;父 run 在 step 里等待子 run 时会读子 run 的状态。所以只删:
+// 终态、结束早于保留期、没有被保留中的业务任务引用、父 run 也已结束的 run。step_attempts 随 run 级联删除。
+
+const TASK_DONE = sql.raw(`('succeeded', 'failed', 'cancelled', 'dead')`);
+const RUN_DONE = sql.raw(`('completed', 'succeeded', 'failed', 'canceled')`);
+/** 引用 task 的外键(都是 on delete set null):被引用的业务任务保留 */
+const TASK_REFS: [table: string, column: string][] = [
+  ["task", "parent_task_id"],
+  ["acquisition_plan", "task_id"],
+  ["asset_operation", "task_id"],
+  ["agent_run", "task_id"],
+  ["artifact", "task_id"],
+  ["decision", "task_id"],
+  ["reconcile_run", "task_id"],
+  ["organization_record", "task_id"],
+];
+
+/** 业务任务可删(别名 t) */
+function taskDeletable(alias: string, cutoff: Date): RawBuilder<unknown> {
+  const a = sql.raw(alias);
+  const refs = TASK_REFS.map(
+    ([tbl, col]) => sql`not exists (select 1 from ${sql.raw(q(tbl))} ref where ref.${sql.raw(q(col))} = ${a}.id)`,
+  );
+  return sql`(${a}.created_at < ${cutoff} and ${a}.status in ${TASK_DONE} and ${sql.join(refs, sql` and `)})`;
+}
+
+/** 引擎 run 可删(别名 r) */
+function runDeletable(c: RetentionRuleContext, alias: string): RawBuilder<unknown> {
+  const s = c.settings.retentionDays;
+  const cutoff = cutoffOf(c.now, s.engineRun);
+  const taskCutoff = cutoffOf(c.now, s.taskRun);
+  const ow = (n: string) => sql.raw(`${q(c.engineSchema)}.${q(n)}`);
+  const r = sql.raw(alias);
+  return sql`(${r}.status in ${RUN_DONE} and ${r}.created_at < ${cutoff}
+    and coalesce(${r}.finished_at, ${r}.updated_at) < ${cutoff}
+    and not exists (select 1 from task t where t.engine_run_id = ${r}.id and not ${taskDeletable("t", taskCutoff)})
+    and not exists (select 1 from ${ow("step_attempts")} psa
+                      join ${ow("workflow_runs")} p on p.namespace_id = psa.namespace_id and p.id = psa.workflow_run_id
+                     where psa.namespace_id = ${r}.parent_step_attempt_namespace_id and psa.id = ${r}.parent_step_attempt_id
+                       and p.status not in ${RUN_DONE}))`;
+}
+
+function taskRunRules(c: RetentionRuleContext): RetentionRule[] {
+  const s = c.settings.retentionDays;
+  const ow = (n: string) => `${q(c.engineSchema)}.${q(n)}`;
+  const owRaw = (n: string) => sql.raw(ow(n));
+  const engineCutoff = cutoffOf(c.now, s.engineRun);
+  /** 关联 run 已不存在或本次可删 */
+  const runGoneOr = (col: string) =>
+    sql`not exists (select 1 from ${owRaw("workflow_runs")} r where r.id = x.${sql.raw(col)} and not ${runDeletable(c, "r")})`;
+  return [
+    { label: "业务任务", table: q("task"), where: taskDeletable("x", cutoffOf(c.now, s.taskRun)) },
+    { label: "引擎 run", table: ow("workflow_runs"), where: runDeletable(c, "x") },
+    {
+      label: "引擎步骤（随 run 级联）",
+      table: ow("step_attempts"),
+      cascade: true,
+      where: sql`exists (select 1 from ${owRaw("workflow_runs")} r
+        where r.namespace_id = x.namespace_id and r.id = x.workflow_run_id and ${runDeletable(c, "r")})`,
+    },
+    {
+      label: "运行元数据",
+      table: q("tasks_run_meta"),
+      where: sql`x.created_at < ${engineCutoff} and ${runGoneOr("run_id")}`,
+    },
+    {
+      label: "运行事实",
+      table: q("tasks_fact"),
+      where: sql`x.created_at < ${engineCutoff} and ${runGoneOr("run_id")}`,
+    },
+    {
+      label: "引擎信号",
+      table: ow("workflow_signals"),
+      where: sql`x.created_at < ${engineCutoff} and ${runGoneOr("workflow_run_id")}`,
+    },
+    {
+      label: "操作账本",
+      table: q("tasks_op_ledger"),
+      where: sql`x.status = 'succeeded' and x.started_at < ${cutoffOf(c.now, s.taskRun)}`,
+    },
+  ];
+}
+
+// ── 流水数据(pipeline,每天低峰)──
+
+/**
+ * 可以删的已投递 outbox 事件:无幂等键,或幂等键只在创建时发一次 / 带时间槽的几类。
+ * 其它幂等键(如生命周期事件 `media.watched/…`、`decision.updated:<id>`)依赖「全表唯一」做永久去重,删了会重发,保留。
+ */
+export const OUTBOX_PURGEABLE_KEYS = ["release.observed:%", "feed.due:%", "decision.created:%", "download.submitted:%"];
+/** 每条资源至少保留最近几次识别尝试 */
+export const RESOLUTION_ATTEMPT_KEEP = 3;
+
+function pipelineRules(c: RetentionRuleContext): RetentionRule[] {
+  const s = c.settings.retentionDays;
+  const keys = sql.join(OUTBOX_PURGEABLE_KEYS.map((k) => sql`${k}`));
+  return [
+    {
+      label: "已投递事件",
+      table: q("outbox_event"),
+      where: sql`x.status = 'dispatched' and x.at < ${cutoffOf(c.now, s.outboxEvent)}
+        and (x.idempotency_key is null or x.idempotency_key like any(array[${keys}]::text[]))`,
+    },
+    {
+      label: "识别尝试",
+      table: q("resolution_attempt"),
+      where: sql`x.created_at < ${cutoffOf(c.now, s.resolutionAttempt)}
+        and exists (select 1 from resolution_attempt n
+                     where n.source_release_id = x.source_release_id and (n.created_at, n.id) > (x.created_at, x.id)
+                     offset ${RESOLUTION_ATTEMPT_KEEP - 1})`,
+    },
+    {
+      // 只删已识别且资源仍绑定作品的;忽略(dismissed)是「不再自动识别」的标记,归档(archived)仍可手动指定,都不删
+      label: "已识别的待识别条目",
+      table: q("identity_candidate"),
+      where: sql`x.status = 'resolved' and x.updated_at < ${cutoffOf(c.now, s.identityResolved)}
+        and exists (select 1 from source_release r where r.id = x.source_release_id and r.media_entity_id is not null)
+        and not exists (select 1 from metadata_enrichment m where m.identity_candidate_id = x.id)
+        and not exists (select 1 from decision d where d.id = x.decision_id
+                          and d.status in ('open', 'investigating', 'awaiting_user'))`,
+    },
+    {
+      // 每个站点最近一次成功的列表快照留着(站点刷新间隔按它判断)
+      label: "站点快照",
+      table: q("site_snapshot"),
+      where: sql`x.started_at < ${cutoffOf(c.now, s.siteSnapshot)} and x.status <> 'running'
+        and x.id is distinct from (select s2.id from site_snapshot s2
+                                    where s2.site_id = x.site_id and s2.kind in ('periodic', 'manual', 'backfill')
+                                      and s2.status in ('ok', 'partial')
+                                    order by s2.started_at desc limit 1)`,
+    },
+  ];
+}
 
 /**
  * 保留清理任务(维护组;heartbeat 目标由 system.heartbeat_cleanup 承担,images / kv_cache 由 cache 模块承担)。
@@ -109,10 +313,16 @@ export const RETENTION_JOBS: RetentionJob[] = [
     description: "删除过期的 RSS 抓取记录",
     everyMs: 6 * HOUR,
     retention: (s) => `${s.retentionDays.observation} 天`,
+    rules: (c) => [
+      {
+        label: "观察记录",
+        table: q("observation"),
+        where: sql`x.created_at < ${cutoffOf(c.now, c.settings.retentionDays.observation)}`,
+      },
+    ],
     async run(c) {
-      const cutoff = cutoffOf(c.now, c.settings.retentionDays.observation);
+      const [r] = await applyRules(c, this.rules!(c));
-      const r = await deleteInBatches(c.db, t("observation"), sql`created_at < ${cutoff}`, c.batch);
+      return summaryOf("观察记录", r!);
-      return summaryOf("观察记录", r);
     },
   },
   {
@@ -122,46 +332,30 @@ export const RETENTION_JOBS: RetentionJob[] = [
     description: "删除过期的页面实时推送记录",
     everyMs: HOUR,
     retention: (s) => `${s.retentionDays.sseEvent} 天`,
+    rules: (c) => [
+      {
+        label: "SSE 事件",
+        table: q("sse_event"),
+        where: sql`x.at < ${cutoffOf(c.now, c.settings.retentionDays.sseEvent)}`,
+      },
+    ],
     async run(c) {
-      const cutoff = cutoffOf(c.now, c.settings.retentionDays.sseEvent);
+      const [r] = await applyRules(c, this.rules!(c));
-      const r = await deleteInBatches(c.db, t("sse_event"), sql`at < ${cutoff}`, c.batch);
+      return summaryOf("SSE 事件", r!);
-      return summaryOf("SSE 事件", r);
     },
   },
   {
     target: "task_run",
     jobId: "storage.retention.task_run",
     name: "任务运行清理",
-    description: "删除过期的已结束任务",
+    description: "删除过期的已结束任务与引擎运行记录",
     everyMs: 6 * HOUR,
-    retention: (s) => `${s.retentionDays.taskRun} 天`,
+    retention: (s) => `任务 ${s.retentionDays.taskRun} 天，引擎运行 ${s.retentionDays.engineRun} 天`,
+    rules: taskRunRules,
     async run(c) {
-      const cutoff = cutoffOf(c.now, c.settings.retentionDays.taskRun);
+      const [task, runs, meta, facts, signals, ledger] = await applyRules(c, taskRunRules(c));
-      const e = (n: string) => sql.raw(`"${c.engineSchema}"."${n}"`);
+      const all = [task!, runs!, meta!, facts!, signals!, ledger!];
-      const task = await deleteInBatches(
+      return `业务任务 ${task!.deleted}、引擎 run ${runs!.deleted}、运行元数据 ${meta!.deleted + facts!.deleted + signals!.deleted + ledger!.deleted} 行已删除${truncNote(all)}`;
-        c.db,
-        t("task"),
-        sql`created_at < ${cutoff} and status in ('succeeded', 'failed', 'cancelled', 'dead')`,
-        c.batch,
-      );
-      // 引擎 run:step_attempts 随 run 级联删除
-      const runs = await deleteInBatches(
-        c.db,
-        e("workflow_runs"),
-        sql`created_at < ${cutoff} and status in ('completed', 'succeeded', 'failed', 'canceled')`,
-        c.batch,
-      );
-      // 运行元数据与 run 同时写入:早于保留期的 run 已结束(上一步删掉了),元数据按时间直接删
-      const meta = await deleteInBatches(c.db, t("tasks_run_meta"), sql`created_at < ${cutoff}`, c.batch);
-      const facts = await deleteInBatches(c.db, t("tasks_fact"), sql`created_at < ${cutoff}`, c.batch);
-      const ledger = await deleteInBatches(
-        c.db,
-        t("tasks_op_ledger"),
-        sql`status = 'succeeded' and started_at < ${cutoff}`,
-        c.batch,
-      );
-      const truncated = [task, runs, meta, facts, ledger].some((x) => x.truncated);
-      return `业务任务 ${task.deleted}、引擎 run ${runs.deleted}、运行元数据 ${meta.deleted + facts.deleted + ledger.deleted} 行已删除${truncated ? "，未删完，下次继续" : ""}`;
     },
   },
   {
@@ -171,18 +365,22 @@ export const RETENTION_JOBS: RetentionJob[] = [
     description: "删除已结束会话的过期消息，会话本身保留",
     everyMs: 6 * HOUR,
     retention: (s) => `${s.retentionDays.agentEvent} 天`,
-    async run(c) {
+    rules: (c) => {
       const cutoff = cutoffOf(c.now, c.settings.retentionDays.agentEvent);
-      const ended = sql`exists (select 1 from agent_run r where r.id = run_id
+      const ended = sql`exists (select 1 from agent_run r where r.id = x.run_id
         and r.status in ('succeeded', 'failed', 'cancelled', 'budget_exceeded'))`;
-      const ev = await deleteInBatches(c.db, t("agent_run_event"), sql`at < ${cutoff} and ${ended}`, c.batch);
+      return [
-      const msg = await deleteInBatches(
+        { label: "智能体事件", table: q("agent_run_event"), where: sql`x.at < ${cutoff} and ${ended}` },
-        c.db,
+        {
-        t("agent_run_message"),
+          label: "智能体消息",
-        sql`created_at < ${cutoff} and status = 'consumed' and ${ended}`,
+          table: q("agent_run_message"),
-        c.batch,
+          where: sql`x.created_at < ${cutoff} and x.status = 'consumed' and ${ended}`,
-      );
+        },
-      return `智能体事件 ${ev.deleted}、消息 ${msg.deleted} 行已删除${ev.truncated || msg.truncated ? "，未删完，下次继续" : ""}`;
+      ];
+    },
+    async run(c) {
+      const [ev, msg] = await applyRules(c, this.rules!(c));
+      return `智能体事件 ${ev!.deleted}、消息 ${msg!.deleted} 行已删除${truncNote([ev!, msg!])}`;
     },
   },
   {
@@ -192,15 +390,16 @@ export const RETENTION_JOBS: RetentionJob[] = [
     description: "删除过期的审计日志",
     everyMs: 24 * HOUR,
     retention: (s) => `${s.retentionDays.audit} 天`,
+    rules: (c) => [
+      {
+        label: "审计事件",
+        table: q("audit_event"),
+        where: sql`x.at < ${cutoffOf(c.now, c.settings.retentionDays.audit)} and x.target_type is distinct from 'capability'`,
+      },
+    ],
     async run(c) {
-      const cutoff = cutoffOf(c.now, c.settings.retentionDays.audit);
+      const [r] = await applyRules(c, this.rules!(c));
-      const r = await deleteInBatches(
+      return summaryOf("审计事件", r!);
-        c.db,
-        t("audit_event"),
-        sql`at < ${cutoff} and target_type is distinct from 'capability'`,
-        c.batch,
-      );
-      return summaryOf("审计事件", r);
     },
   },
   {
@@ -210,21 +409,36 @@ export const RETENTION_JOBS: RetentionJob[] = [
     description: "删除过期的能力调用记录",
     everyMs: 24 * HOUR,
     retention: (s) => `${s.retentionDays.capabilityStats} 天`,
-    async run(c) {
+    rules: (c) => {
       const cutoff = cutoffOf(c.now, c.settings.retentionDays.capabilityStats);
-      const a = await deleteInBatches(
+      return [
-        c.db,
+        { label: "能力审计", table: q("audit_event"), where: sql`x.at < ${cutoff} and x.target_type = 'capability'` },
-        t("audit_event"),
+        {
-        sql`at < ${cutoff} and target_type = 'capability'`,
+          label: "能力调用记录",
-        c.batch,
+          table: q("capability_invocation"),
-      );
+          where: sql`x.created_at < ${cutoff} and x.status <> 'running'`,
-      const inv = await deleteInBatches(
+        },
-        c.db,
+      ];
-        t("capability_invocation"),
+    },
-        sql`created_at < ${cutoff} and status <> 'running'`,
+    async run(c) {
-        c.batch,
+      const [a, inv] = await applyRules(c, this.rules!(c));
-      );
+      return `能力审计 ${a!.deleted}、调用记录 ${inv!.deleted} 行已删除${truncNote([a!, inv!])}`;
-      return `能力审计 ${a.deleted}、调用记录 ${inv.deleted} 行已删除${a.truncated || inv.truncated ? "，未删完，下次继续" : ""}`;
+    },
+  },
+  {
+    target: "pipeline",
+    jobId: "storage.retention.pipeline",
+    name: "流水数据清理",
+    description: "每天低峰删除过期的已投递事件、识别尝试、已识别的待识别条目与站点快照",
+    everyMs: 24 * HOUR,
+    schedule: { type: "cron", cron: "40 4 * * *" },
+    retention: (s) =>
+      `事件 ${s.retentionDays.outboxEvent} 天，识别尝试 ${s.retentionDays.resolutionAttempt} 天，已识别条目 ${s.retentionDays.identityResolved} 天，站点快照 ${s.retentionDays.siteSnapshot} 天`,
+    rules: pipelineRules,
+    async run(c) {
+      const rules = pipelineRules(c);
+      const rs = await applyRules(c, rules);
+      return `${rules.map((r, i) => `${r.label} ${rs[i]!.deleted}`).join("、")} 行已删除${truncNote(rs)}`;
     },
   },
   {
@@ -234,6 +448,14 @@ export const RETENTION_JOBS: RetentionJob[] = [
     description: "长期没有新线索的未识别资源归档，仍可手动指定",
     everyMs: 6 * HOUR,
     retention: (s) => `${s.retentionDays.identityPoolArchive} 天`,
+    async preview(c) {
+      const cutoff = cutoffOf(c.now, c.settings.retentionDays.identityPoolArchive);
+      const r = await sql<{ n: string }>`select count(*)::text as n from identity_candidate
+        where status = 'pending' and reason = 'unmatched' and coalesce(last_attempt_at, created_at) < ${cutoff}`.execute(
+        c.db,
+      );
+      return [{ label: "未识别条目（归档，不删除）", table: q("identity_candidate"), rows: Number(r.rows[0]?.n ?? 0) }];
+    },
     async run(c) {
       const cutoff = cutoffOf(c.now, c.settings.retentionDays.identityPoolArchive);
       let archived = 0;

apps/server/test/system/db-retention.test.ts

@@ -0,0 +1,281 @@
+import { DEFAULT_STORAGE_SETTINGS } from "@umt/contracts";
+import { type RawBuilder, sql } from "@umt/db";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import {
+  countRules,
+  OUTBOX_PURGEABLE_KEYS,
+  previewJob,
+  RETENTION_JOBS,
+  type RetentionRuleContext,
+} from "../../src/modules/system/retention.ts";
+import { createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
+
+// 数据库清理(docs/runbook/storage.md「保留规则」):引擎 run 只删终态且无业务引用 / 父 run 已结束的,
+// 元数据跟随 run;流水表各自的保留条件;被引用 / 未结束的不删;dry-run 统计与实际删除一致;分批。
+// 另外用 EXPLAIN 确认改写后的查询能走索引(别名 trigram、巡检回填部分索引)。
+
+const hasPg = await pgAvailable();
+const DAY = 86_400_000;
+
+describe.skipIf(!hasPg)("数据库清理(真实 PG)", () => {
+  let h: Harness;
+  const now = new Date();
+  const ago = (days: number) => new Date(now.getTime() - days * DAY);
+
+  beforeAll(async () => {
+    h = await createHarness({ engine: true });
+  }, 60_000);
+  afterAll(async () => {
+    await h?.close();
+  });
+
+  const job = (target: string) => RETENTION_JOBS.find((x) => x.target === target)!;
+  const ctx = (): RetentionRuleContext => ({
+    db: h.db,
+    engineSchema: h.engineSchema,
+    settings: DEFAULT_STORAGE_SETTINGS,
+    now,
+  });
+  const run = (target: string) => job(target).run({ ...ctx(), batch: { batchSize: 2, pauseMs: 0 } });
+  const ids = async (q: RawBuilder<{ id: string }>) => (await q.execute(h.db)).rows.map((r) => r.id).sort();
+  const ow = (n: string) => sql.raw(`"${h.engineSchema}"."${n}"`);
+
+  it("引擎 run:只删终态、过期、无保留中业务任务引用、父 run 已结束的;步骤级联,元数据跟随 run", async () => {
+    const wr = (id: string, status: string, at: Date, parentStep: string | null = null) =>
+      sql`insert into ${ow("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, attempts,
+            parent_step_attempt_namespace_id, parent_step_attempt_id, created_at, updated_at, finished_at)
+          values ('default', ${id}, 'w', 'v1', ${status}, '{}'::jsonb, 1, ${parentStep ? "default" : null}, ${parentStep},
+                  ${at}, ${at}, ${status === "running" ? null : at})`.execute(h.db);
+    const step = (id: string, runId: string, status: string) =>
+      sql`insert into ${ow("step_attempts")} (namespace_id, id, workflow_run_id, step_name, kind, status, config, created_at, updated_at)
+          values ('default', ${id}, ${runId}, 's', 'function', ${status}, '{}'::jsonb, ${ago(20)}, ${ago(20)})`.execute(
+        h.db,
+      );
+    const meta = (runId: string, at: Date) =>
+      sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key, created_at)
+          values (${runId}, 'w', 1, ${runId}, ${at})`.execute(h.db);
+
+    await wr("old-done", "completed", ago(20));
+    await step("old-done-s1", "old-done", "completed");
+    await step("old-done-s2", "old-done", "completed");
+    await meta("old-done", ago(20));
+    await wr("old-failed", "failed", ago(15));
+    await wr("fresh-done", "completed", ago(3));
+    await meta("fresh-done", ago(3));
+    await wr("old-running", "running", ago(20));
+    await meta("old-running", ago(20));
+    // 被保留中的业务任务引用(任务 60 天保留期内)
+    await wr("old-task-ref", "completed", ago(20));
+    await sql`insert into task (kind, idempotency_key, status, engine_run_id, created_at)
+      values ('k', 'ref', 'succeeded', 'old-task-ref', ${ago(20)})`.execute(h.db);
+    // 父 run 还在跑:子 run 已结束也保留(父 run 等子 run 时会读它的状态)
+    await wr("parent-running", "running", ago(20));
+    await step("parent-step", "parent-running", "running");
+    await wr("child-of-running", "completed", ago(20), "parent-step");
+    // 父 run 已结束的子 run 照常删
+    await wr("parent-done", "completed", ago(20));
+    await step("parent-done-step", "parent-done", "completed");
+    await wr("child-of-done", "completed", ago(20), "parent-done-step");
+    // 元数据的 run 早已不在
+    await meta("gone", ago(30));
+
+    const preview = await previewJob(job("task_run"), ctx());
+    const rows = (label: string) => preview.find((p) => p.label === label)?.rows;
+    expect(rows("引擎 run")).toBe(4);
+    expect(rows("引擎步骤（随 run 级联）")).toBe(3);
+    expect(rows("运行元数据")).toBe(2);
+
+    expect(await run("task_run")).toMatch(/业务任务 0、引擎 run 4、运行元数据 2 行已删除/);
+    expect(await ids(sql<{ id: string }>`select id from ${ow("workflow_runs")}`)).toEqual(
+      ["child-of-running", "fresh-done", "old-running", "old-task-ref", "parent-running"].sort(),
+    );
+    expect(await ids(sql<{ id: string }>`select id from ${ow("step_attempts")}`)).toEqual(["parent-step"]);
+    expect(await ids(sql<{ id: string }>`select run_id as id from tasks_run_meta`)).toEqual(
+      ["fresh-done", "old-running"].sort(),
+    );
+    // 再看一次:没有可删的了
+    expect((await previewJob(job("task_run"), ctx())).filter((p) => p.rows > 0)).toEqual([]);
+  });
+
+  it("业务任务:被其它表引用的不删(外键是 set null,删了会断开引用)", async () => {
+    const t = await sql<{
+      id: string;
+      idempotency_key: string;
+    }>`insert into task (kind, idempotency_key, status, created_at)
+      values ('k', 'old-free', 'succeeded', ${ago(61)}), ('k', 'old-ref', 'failed', ${ago(61)})
+      returning id, idempotency_key`.execute(h.db);
+    const ref = t.rows.find((r) => r.idempotency_key === "old-ref")!.id;
+    await sql`insert into agent_run (trigger, goal, status, task_id) values ('user', 'g', 'succeeded', ${ref})`.execute(
+      h.db,
+    );
+    expect(await run("task_run")).toMatch(/业务任务 1、/);
+    const left = await sql<{ k: string }>`select idempotency_key as k from task where kind = 'k' order by 1`.execute(
+      h.db,
+    );
+    expect(left.rows.map((r) => r.k)).toEqual(["old-ref", "ref"]);
+  });
+
+  it("流水数据:outbox / 识别尝试 / 已识别条目 / 站点快照各按规则删,dry-run 统计与删除一致,分批", async () => {
+    // outbox:已投递 + 过期 + (无键或一次性键)才删
+    const ob = (type: string, status: string, days: number, key: string | null) =>
+      sql`insert into outbox_event (type, status, at, idempotency_key) values (${type}, ${status}, ${ago(days)}, ${key})`.execute(
+        h.db,
+      );
+    await ob("del-null", "dispatched", 31, null);
+    await ob("del-release", "dispatched", 40, "release.observed:r1");
+    await ob("del-feed", "dispatched", 40, "feed.due:f1:123");
+    await ob("keep-lifecycle", "dispatched", 40, "media.watched/e1/subscribed/s1");
+    await ob("keep-decision-updated", "dispatched", 40, "decision.updated:d1");
+    await ob("keep-pending", "pending", 40, null);
+    await ob("keep-dead", "dead", 40, null);
+    await ob("keep-fresh", "dispatched", 29, null);
+    expect(OUTBOX_PURGEABLE_KEYS).toContain("release.observed:%");
+
+    const site = (
+      await sql<{ id: string }>`insert into site (key, name, adapter, base_url)
+        values ('s-retention', 'S', 'nexusphp', 'https://s.example') returning id`.execute(h.db)
+    ).rows[0]!.id;
+    const entity = (
+      await sql<{ id: string }>`insert into media_entity (kind, title) values ('movie', 'M') returning id`.execute(h.db)
+    ).rows[0]!.id;
+    const release = async (rid: string, bound: boolean) =>
+      (
+        await sql<{ id: string }>`insert into source_release (site_id, site_release_id, title, media_entity_id)
+          values (${site}, ${rid}, ${rid}, ${bound ? entity : null}) returning id`.execute(h.db)
+      ).rows[0]!.id;
+
+    // 识别尝试:5 次都过期 → 删最旧的 2 次,留最近 3 次;另一条资源只有 1 次过期的 → 保留
+    const r1 = await release("r1", true);
+    for (let i = 0; i < 5; i++) {
+      await sql`insert into resolution_attempt (source_release_id, outcome, evidence_version, reason, created_at)
+        values (${r1}, 'no_match', 'v1', ${`a${i}`}, ${ago(40 + i)})`.execute(h.db);
+    }
+    const r2 = await release("r2", false);
+    await sql`insert into resolution_attempt (source_release_id, outcome, evidence_version, created_at)
+      values (${r2}, 'no_match', 'v1', ${ago(90)})`.execute(h.db);
+
+    // 待识别条目:只删已识别 + 资源仍绑定作品 + 无补全记录引用 + 无未结事项的
+    const cand = async (rid: string, status: string, bound: boolean, days: number) => {
+      const sr = await release(rid, bound);
+      return (
+        await sql<{
+          id: string;
+        }>`insert into identity_candidate (source_release_id, status, reason, evidence_version, updated_at)
+          values (${sr}, ${status}, 'no_match', 'v1', ${ago(days)}) returning id`.execute(h.db)
+      ).rows[0]!.id;
+    };
+    await cand("c-del", "resolved", true, 61);
+    await cand("c-fresh", "resolved", true, 59);
+    await cand("c-unbound", "resolved", false, 61);
+    await cand("c-dismissed", "dismissed", true, 61);
+    await cand("c-archived", "archived", false, 61);
+    const enriched = await cand("c-enriched", "resolved", true, 61);
+    await sql`insert into metadata_enrichment (identity_candidate_id, normalized_title, query)
+      values (${enriched}, 'x', '{}'::jsonb)`.execute(h.db);
+
+    // 站点快照:过期的删,但每个站点最近一次成功的列表快照与运行中的保留
+    const snap = (kind: string, status: string, days: number) =>
+      sql`insert into site_snapshot (site_id, kind, status, started_at) values (${site}, ${kind}, ${status}, ${ago(days)})`.execute(
+        h.db,
+      );
+    await snap("periodic", "ok", 50); // 删
+    await snap("periodic", "ok", 40); // 最近一次成功的列表快照(站点刷新间隔按它判断):留
+    await snap("live_search", "ok", 35); // 删
+    await snap("periodic", "failed", 33); // 删
+    await snap("periodic", "running", 45); // 留
+    await snap("live_search", "ok", 10); // 未过期:留(实时补查不算列表快照)
+
+    const preview = await countRules(h.db, job("pipeline").rules!(ctx()));
+    expect(Object.fromEntries(preview.map((p) => [p.label, p.rows]))).toEqual({
+      已投递事件: 3,
+      识别尝试: 2,
+      已识别的待识别条目: 1,
+      站点快照: 3,
+    });
+    // 站点快照 3 行、batchSize 2:分两批
+    expect(await run("pipeline")).toBe("已投递事件 3、识别尝试 2、已识别的待识别条目 1、站点快照 3 行已删除");
+
+    const types = await sql<{ id: string }>`select type as id from outbox_event`.execute(h.db);
+    expect(types.rows.map((r) => r.id).sort()).toEqual(
+      ["keep-dead", "keep-decision-updated", "keep-fresh", "keep-lifecycle", "keep-pending"].sort(),
+    );
+    const attempts = await sql<{ reason: string | null }>`select reason from resolution_attempt
+      where source_release_id = ${r1} order by created_at desc`.execute(h.db);
+    expect(attempts.rows.map((r) => r.reason)).toEqual(["a0", "a1", "a2"]);
+    expect(
+      await ids(sql<{ id: string }>`select id from resolution_attempt where source_release_id = ${r2}`),
+    ).toHaveLength(1);
+    const cands = await sql<{ t: string }>`select r.title as t from identity_candidate c
+      join source_release r on r.id = c.source_release_id order by 1`.execute(h.db);
+    expect(cands.rows.map((r) => r.t)).toEqual(["c-archived", "c-dismissed", "c-enriched", "c-fresh", "c-unbound"]);
+    const snaps = await sql<{ k: string; s: string; d: number }>`select kind as k, status as s,
+        round(extract(epoch from ${now}::timestamptz - started_at) / 86400)::int as d
+      from site_snapshot order by started_at`.execute(h.db);
+    expect(snaps.rows).toEqual([
+      { k: "periodic", s: "running", d: 45 },
+      { k: "periodic", s: "ok", d: 40 },
+      { k: "live_search", s: "ok", d: 10 },
+    ]);
+    expect((await countRules(h.db, job("pipeline").rules!(ctx()))).every((p) => p.rows === 0)).toBe(true);
+  });
+
+  it("站点快照:有了更新的成功列表快照,旧的那次就可以删", async () => {
+    const site = (await sql<{ id: string }>`select id from site where key = 's-retention'`.execute(h.db)).rows[0]!.id;
+    await sql`insert into site_snapshot (site_id, kind, status, started_at) values (${site}, 'manual', 'partial', ${ago(2)})`.execute(
+      h.db,
+    );
+    const counts = await countRules(h.db, job("pipeline").rules!(ctx()));
+    expect(counts.find((p) => p.label === "站点快照")?.rows).toBe(1);
+  });
+
+  it("定时任务登记:流水数据清理每天低峰一次(cron)", () => {
+    expect(job("pipeline")).toMatchObject({
+      jobId: "storage.retention.pipeline",
+      schedule: { type: "cron", cron: "40 4 * * *" },
+    });
+  });
+
+  // ── 改写后的查询能走索引(小表上规划器会选全表扫描,关掉 seqscan 只为确认索引「可用」;
+  //    trigram GIN 只能位图扫描,别名用例同时关掉普通索引扫描,免得规划器整扫唯一索引)──
+
+  const plan = async (q: RawBuilder<unknown>, bitmapOnly = false) =>
+    await h.db.transaction().execute(async (trx) => {
+      await sql`set local enable_seqscan = off`.execute(trx);
+      if (bitmapOnly) await sql`set local enable_indexscan = off; set local enable_indexonlyscan = off`.execute(trx);
+      const r = await sql<{ "QUERY PLAN": string }>`explain ${q}`.execute(trx);
+      return r.rows.map((x) => x["QUERY PLAN"]).join("\n");
+    });
+
+  it("别名模糊匹配:`%` 写法走 title_alias_normalized_trgm_idx,只写 similarity() 用不上", async () => {
+    const e = (await sql<{ id: string }>`select id from media_entity limit 1`.execute(h.db)).rows[0]!.id;
+    await sql`insert into title_alias (media_entity_id, alias, normalized, source)
+      select ${e}, 'a' || i, 'alias title ' || i, 'tmdb' from generate_series(1, 50) i`.execute(h.db);
+    await sql`analyze title_alias`.execute(h.db);
+    const q = "alias title 7";
+    const after = await plan(
+      sql`select distinct media_entity_id from title_alias
+           where normalized = ${sql.lit(q)} or (normalized % ${sql.lit(q)} and similarity(normalized, ${sql.lit(q)}) >= 0.45)`,
+      true,
+    );
+    expect(after).toContain("title_alias_normalized_trgm_idx");
+    const before = await plan(
+      sql`select distinct media_entity_id from title_alias
+           where normalized = ${sql.lit(q)} or similarity(normalized, ${sql.lit(q)}) >= 0.45`,
+      true,
+    );
+    expect(before).not.toContain("title_alias_normalized_trgm_idx");
+    // 两种写法结果相同
+    const a = await sql<{ n: string }>`select normalized as n from title_alias
+      where normalized % ${q} and similarity(normalized, ${q}) >= 0.45 order by 1`.execute(h.db);
+    const b = await sql<{ n: string }>`select normalized as n from title_alias
+      where similarity(normalized, ${q}) >= 0.45 order by 1`.execute(h.db);
+    expect(a.rows).toEqual(b.rows);
+    expect(a.rows.length).toBeGreaterThan(0);
+  });
+
+  it("巡检回填:走部分索引 scheduled_job_run_harvest_idx", async () => {
+    const out = await plan(sql`select r.id from scheduled_job_run r join agent_run a on a.id = r.agent_run_id
+      where r.agent_status is null and r.agent_run_id is not null and a.status = any('{succeeded,failed}')`);
+    expect(out).toContain("scheduled_job_run_harvest_idx");
+  });
+});

apps/server/test/system/storage.test.ts

@@ -34,6 +34,7 @@ describe("预算档位", () => {
         "agent_event",
         "audit",
         "capability_stats",
+        "pipeline",
         "identity_pool",
         "analyze",
         "images",

apps/web/app/components/settings/system.tsx

@@ -80,6 +80,11 @@ const RETENTION_KEYS = [
   "audit",
   "capabilityStats",
   "identityPoolArchive",
+  "engineRun",
+  "outboxEvent",
+  "resolutionAttempt",
+  "identityResolved",
+  "siteSnapshot",
 ] as const;
 export const STORAGE_RETENTION_CARD: CardSpec = {
   anchor: "storage-retention",
@@ -99,6 +104,11 @@ export const STORAGE_RETENTION_CARD: CardSpec = {
       "未识别资源归档",
       "未命中的条目这么久没有新证据就归档，仍可手动指定",
     ),
+    dayField("retentionDays", "engineRun", "后台运行记录", "仍被任务记录引用的跟随任务保留"),
+    dayField("retentionDays", "outboxEvent", "已投递的内部事件"),
+    dayField("retentionDays", "resolutionAttempt", "识别尝试记录", "每个资源至少留最近 3 次"),
+    dayField("retentionDays", "identityResolved", "已识别的待识别条目", "忽略和归档的不删"),
+    dayField("retentionDays", "siteSnapshot", "站点快照", "每个站点最近一次成功的保留"),
   ],
   fromSettings: (s) =>
     Object.fromEntries(RETENTION_KEYS.map((k) => [`storage.retentionDays.${k}`, n(s.storage.retentionDays[k])])),

apps/web/app/components/status/storage-section.tsx

@@ -38,6 +38,7 @@ export const CLEANUP_LABEL: Record<StorageCleanupTarget, string> = {
   agent_event: "智能体事件",
   audit: "审计事件",
   capability_stats: "能力调用统计",
+  pipeline: "流水数据",
   identity_pool: "未识别资源归档",
   analyze: "数据库维护",
   images: "图片缓存容量",

packages/contracts/src/ops.ts

@@ -527,6 +527,7 @@ export const StorageCleanupTarget = z.enum([
   "agent_event",
   "audit",
   "capability_stats",
+  "pipeline",
   "identity_pool",
   "analyze",
   "images",

packages/contracts/src/settings.ts

@@ -559,6 +559,17 @@ export const StorageRetentionDays = z.object({
   capabilityStats: Days,
   /** 待识别池 unmatched 条目无新证据多少天后归档 */
   identityPoolArchive: Days,
+  // ── 2026-10-06 数据库清理(docs/runbook/storage.md「保留规则」);旧库缺这些字段时按缺省补 ──
+  /** 已结束的引擎 run(步骤级联)、运行元数据 / 事实 / 信号;仍被保留中的业务任务引用的跟随任务 */
+  engineRun: Days,
+  /** 已投递的 outbox 事件(只删无幂等键或一次性 / 带时间槽的幂等键) */
+  outboxEvent: Days,
+  /** 识别尝试记录;每条资源至少保留最近 3 次 */
+  resolutionAttempt: Days,
+  /** 已识别(资源已绑定作品)的待识别池条目;忽略、归档的不删 */
+  identityResolved: Days,
+  /** 站点列表快照;每个站点最近一次成功的保留 */
+  siteSnapshot: Days,
 });
 export type StorageRetentionDays = z.infer<typeof StorageRetentionDays>;
 
@@ -612,6 +623,11 @@ export const DEFAULT_STORAGE_SETTINGS: StorageSettings = {
     audit: 180,
     capabilityStats: 90,
     identityPoolArchive: 30,
+    engineRun: 14,
+    outboxEvent: 30,
+    resolutionAttempt: 30,
+    identityResolved: 60,
+    siteSnapshot: 30,
   },
   cacheTtl: {
     tmdbSearchDays: 7,

scripts/db-retention.ts

@@ -0,0 +1,59 @@
+// 数据保留清理(docs/runbook/storage.md「保留规则」):按 settings.storage.retentionDays 统计 / 执行各保留任务。
+// 规则与定时任务 storage.retention.* 是同一份代码(apps/server/src/modules/system/retention.ts)。
+// 默认只读(dry-run),输出每张表将删的行数;--apply 才删除(分批、每批单独事务,删得多的表随后 ANALYZE)。
+//
+// 用法(DATABASE_URL 必须显式给出,脚本不带默认库):
+//   DATABASE_URL=postgres://umt:***@<host>:5432/umt deno run -A scripts/db-retention.ts [--dry-run]   # 看每张表将删多少行
+//   DATABASE_URL=… deno run -A scripts/db-retention.ts --apply                                       # 执行
+//   DATABASE_URL=… deno run -A scripts/db-retention.ts --only task_run,pipeline [--apply]            # 只处理部分目标
+//   ENGINE_SCHEMA=openworkflow(缺省)
+//
+// 不包含:进程心跳(system.heartbeat_cleanup)、图片缓存与 kv_cache(cache 模块)、ANALYZE 任务(--apply 结束时对删过的表各自 ANALYZE)。
+// 对真实库先 dry-run,逐表看过再 --apply(docs/plan/lessons.md 第 22 条)。
+
+import { withStorageDefaults } from "@umt/contracts";
+import { createDb, sql } from "@umt/db";
+import { previewJob, RETENTION_JOBS } from "../apps/server/src/modules/system/retention.ts";
+
+const argv = [...Deno.args];
+const APPLY = argv.includes("--apply");
+if (APPLY && argv.includes("--dry-run")) {
+  console.error("--apply 与 --dry-run 不能同时给出");
+  Deno.exit(2);
+}
+const onlyArg = argv.indexOf("--only");
+const only = onlyArg >= 0 ? new Set((argv[onlyArg + 1] ?? "").split(",").filter(Boolean)) : null;
+const url = Deno.env.get("DATABASE_URL");
+if (!url) {
+  console.error("需要 DATABASE_URL(脚本不带默认库)");
+  Deno.exit(2);
+}
+const engineSchema = Deno.env.get("ENGINE_SCHEMA") ?? "openworkflow";
+
+const db = createDb({ connectionString: url, max: 2 });
+try {
+  const stored = await sql<{ value: unknown }>`select value from settings where key = 'storage'`.execute(db);
+  const settings = withStorageDefaults(stored.rows[0]?.value);
+  const now = new Date();
+  const jobs = RETENTION_JOBS.filter((j) => j.target !== "analyze" && (!only || only.has(j.target)));
+  console.log(`# db-retention (${APPLY ? "APPLY" : "DRY-RUN"}) ${now.toISOString()}`);
+  console.log(`# retentionDays ${JSON.stringify(settings.retentionDays)}\n`);
+  let total = 0;
+  for (const job of jobs) {
+    const counts = await previewJob(job, { db, engineSchema, settings, now });
+    console.log(`## ${job.target}(${job.jobId},保留 ${job.retention(settings) ?? "—"})`);
+    for (const c of counts) {
+      console.log(`  ${c.table.padEnd(34)} ${String(c.rows).padStart(8)}  ${c.label}`);
+      if (!c.cascade && job.rules) total += c.rows;
+    }
+    if (APPLY) {
+      const t0 = Date.now();
+      const summary = await job.run({ db, engineSchema, settings, now, batch: { budgetMs: 60 * 60_000 } });
+      console.log(`  → ${summary}(${Math.round((Date.now() - t0) / 100) / 10} 秒)`);
+    }
+    console.log("");
+  }
+  console.log(`# 合计将删 ${total} 行${APPLY ? "(已执行)" : "(dry-run,未改库;确认后加 --apply)"}`);
+} finally {
+  await db.destroy();
+}