merge: perf/db

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

apps/server/src/modules/file-manager/identify.ts

@@ -240,7 +240,8 @@ export async function identifyCore(
       const rows = await sql<{ media_entity_id: string }>`
         select distinct a.media_entity_id from title_alias a join media_entity e on e.id = a.media_entity_id
          where e.status not in ('merged', 'archived')
-           and (a.normalized = ${q} or (length(${q}) >= 3 and similarity(a.normalized, ${q}) >= ${TRGM}))
+           and (a.normalized = ${q}
+                ${[...q].length >= 3 ? sql`or (a.normalized % ${q} and similarity(a.normalized, ${q}) >= ${TRGM})` : sql``})
          limit 20`.execute(db);
       for (const r of rows.rows) ids.add(r.media_entity_id);
     }

apps/server/src/modules/schedules/scheduler.ts

@@ -289,7 +289,10 @@ export class Scheduler {
     if (manual || status !== job.last_status || status === "failed") this.publish(job.id, status);
   }
 
-  /** 巡检入队的 agent_run 结束后:回填终态与摘要;配置了 reportAsDecision 的写一条 info 待处理 */
+  /**
+   * 巡检入队的 agent_run 结束后:回填终态与摘要;配置了 reportAsDecision 的写一条 info 待处理。
+   * 条件与部分索引 scheduled_job_run_harvest_idx(迁移 0089)一致,每 15 秒一次不再全表扫运行历史。
+   */
   async harvest(): Promise<number> {
     const rows = await sql<{
       id: string;
@@ -304,7 +307,7 @@ export class Scheduler {
         from scheduled_job_run r
         join scheduled_job j on j.id = r.job_id
         join agent_run a on a.id = r.agent_run_id
-       where r.agent_status is null and a.status = any(${TERMINAL_AGENT})
+       where r.agent_status is null and r.agent_run_id is not null and a.status = any(${TERMINAL_AGENT})
        limit 50`.execute(this.db);
     for (const r of rows.rows) {
       const summary = (r.summary ?? "").trim() || `智能体运行结束：${r.status}`;

apps/server/src/modules/search/media.ts

@@ -47,7 +47,8 @@ export async function searchLocalMedia(
              a.alias as matched
         from title_alias a join media_entity e on e.id = a.media_entity_id
        where e.status <> 'merged'
-         and (a.normalized = ${n} or a.normalized like ${`${n}%`} or similarity(a.normalized, ${n}) >= 0.3)
+         and (a.normalized = ${n} or a.normalized like ${`${n}%`}
+              or (a.normalized % ${n} and similarity(a.normalized, ${n}) >= 0.3))
        order by e.id, score desc
        limit 200`.execute(ctx.db);
     rows.push(...r.rows);

apps/server/src/modules/sources-identity/capabilities.ts

@@ -39,7 +39,8 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: ()
         select distinct on (e.id) e.*, greatest(similarity(a.normalized, ${n}), case when a.normalized = ${n} then 1 else 0 end) as score,
                a.alias as matched
           from title_alias a join media_entity e on e.id = a.media_entity_id
-         where e.status <> 'merged' and (a.normalized = ${n} or similarity(a.normalized, ${n}) >= 0.3)
+         where e.status <> 'merged'
+           and (a.normalized = ${n} or (a.normalized % ${n} and similarity(a.normalized, ${n}) >= 0.3))
            ${input.kind ? sql`and e.kind = ${input.kind}` : sql``}
            ${input.year ? sql`and (e.year is null or abs(e.year - ${input.year}) <= 1)` : sql``}
          order by e.id, score desc`.execute(db);

apps/server/src/modules/sources-identity/routes/identity.ts

@@ -389,7 +389,7 @@ export function identityRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
     if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId);
     if (query.q) {
       const n = normalizeTitle(query.q);
-      q = q.where(sql<boolean>`(normalized = ${n} or similarity(normalized, ${n}) >= 0.3)`);
+      q = q.where(sql<boolean>`(normalized = ${n} or (normalized % ${n} and similarity(normalized, ${n}) >= 0.3))`);
     }
     const rows = await q
       .orderBy("created_at")

apps/server/src/modules/sources-identity/routes/sources.ts

@@ -852,7 +852,11 @@ export function sourcesRoutes(app: ModuleApp, rt: () => SourcesRuntime) {
     if (query.resolutionStatus) q = q.where("resolution_status", "=", query.resolutionStatus);
     if (query.q)
       q = q.where((eb) =>
-        eb.or([eb("title", "ilike", `%${query.q}%`), sql<boolean>`similarity(title, ${query.q}) >= 0.3`]),
+        eb.or([
+          eb("title", "ilike", `%${query.q}%`),
+          // `%` 让 source_release_title_trgm_idx 可用(两个分支都走索引,BitmapOr)
+          sql<boolean>`(title % ${query.q} and similarity(title, ${query.q}) >= 0.3)`,
+        ]),
       );
     const rows = await q
       .orderBy("first_seen_at", "desc")

apps/server/src/modules/sources-identity/service/resolve.ts

@@ -164,9 +164,11 @@ async function localCandidates(
     for (const r of row) if (r.media_entity_id) ids.add(r.media_entity_id);
   }
   for (const q of queryTitles(parsed)) {
+    // `%` 让 title_alias_normalized_trgm_idx 可用(阈值 0.3 ≤ TRGM_THRESHOLD,结果与只写 similarity 相同)
     const rows = await sql<{ media_entity_id: string }>`
       select distinct media_entity_id from title_alias
-       where normalized = ${q} or (length(${q}) >= 3 and similarity(normalized, ${q}) >= ${TRGM_THRESHOLD})
+       where normalized = ${q}
+          ${[...q].length >= 3 ? sql`or (normalized % ${q} and similarity(normalized, ${q}) >= ${TRGM_THRESHOLD})` : sql``}
        limit 20`.execute(exec);
     for (const r of rows.rows) ids.add(r.media_entity_id);
   }

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 deleteInBatches(c.db, t("observation"), sql`created_at < ${cutoff}`, c.batch);
-      return summaryOf("观察记录", r);
+      const [r] = await applyRules(c, this.rules!(c));
+      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 deleteInBatches(c.db, t("sse_event"), sql`at < ${cutoff}`, c.batch);
-      return summaryOf("SSE 事件", r);
+      const [r] = await applyRules(c, this.rules!(c));
+      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 e = (n: string) => sql.raw(`"${c.engineSchema}"."${n}"`);
-      const task = await deleteInBatches(
-        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 ? "，未删完，下次继续" : ""}`;
+      const [task, runs, meta, facts, signals, ledger] = await applyRules(c, taskRunRules(c));
+      const all = [task!, runs!, meta!, facts!, signals!, ledger!];
+      return `业务任务 ${task!.deleted}、引擎 run ${runs!.deleted}、运行元数据 ${meta!.deleted + facts!.deleted + signals!.deleted + ledger!.deleted} 行已删除${truncNote(all)}`;
     },
   },
   {
@@ -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);
-      const msg = await deleteInBatches(
-        c.db,
-        t("agent_run_message"),
-        sql`created_at < ${cutoff} and status = 'consumed' and ${ended}`,
-        c.batch,
-      );
-      return `智能体事件 ${ev.deleted}、消息 ${msg.deleted} 行已删除${ev.truncated || msg.truncated ? "，未删完，下次继续" : ""}`;
+      return [
+        { label: "智能体事件", table: q("agent_run_event"), where: sql`x.at < ${cutoff} and ${ended}` },
+        {
+          label: "智能体消息",
+          table: q("agent_run_message"),
+          where: sql`x.created_at < ${cutoff} and x.status = 'consumed' and ${ended}`,
+        },
+      ];
+    },
+    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 deleteInBatches(
-        c.db,
-        t("audit_event"),
-        sql`at < ${cutoff} and target_type is distinct from 'capability'`,
-        c.batch,
-      );
-      return summaryOf("审计事件", r);
+      const [r] = await applyRules(c, this.rules!(c));
+      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(
-        c.db,
-        t("audit_event"),
-        sql`at < ${cutoff} and target_type = 'capability'`,
-        c.batch,
-      );
-      const inv = await deleteInBatches(
-        c.db,
-        t("capability_invocation"),
-        sql`created_at < ${cutoff} and status <> 'running'`,
-        c.batch,
-      );
-      return `能力审计 ${a.deleted}、调用记录 ${inv.deleted} 行已删除${a.truncated || inv.truncated ? "，未删完，下次继续" : ""}`;
+      return [
+        { label: "能力审计", table: q("audit_event"), where: sql`x.at < ${cutoff} and x.target_type = 'capability'` },
+        {
+          label: "能力调用记录",
+          table: q("capability_invocation"),
+          where: sql`x.created_at < ${cutoff} and x.status <> 'running'`,
+        },
+      ];
+    },
+    async run(c) {
+      const [a, inv] = await applyRules(c, this.rules!(c));
+      return `能力审计 ${a!.deleted}、调用记录 ${inv!.deleted} 行已删除${truncNote([a!, inv!])}`;
+    },
+  },
+  {
+    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/src/modules/system/storage.ts

@@ -248,10 +248,59 @@ export async function storageHealth(d: StorageDeps): Promise<{ level: HealthLeve
   return { level: l.level, alerts: l.alerts };
 }
 
+type Statements = SystemStorage["db"]["statements"];
+const STATEMENT_TOP = 5;
+/** 平均耗时排行只看调用过这么多次的语句(迁移、一次性维护语句不进榜) */
+const SLOW_MIN_CALLS = 10;
+
+/** pg_stat_statements 前几名;没装扩展或 postgres 没预加载时 enabled=false(不报错) */
+export async function statementStats(db: Kysely<DB>): Promise<Statements> {
+  const off = (note: string): Statements => ({ enabled: false, note, since: null, slowest: [], frequent: [] });
+  const has = await sql<{ r: string | null }>`select to_regclass('public.pg_stat_statements')::text as r`
+    .execute(db)
+    .catch(() => null);
+  if (!has?.rows[0]?.r) return off("没有安装 pg_stat_statements 扩展");
+  // 装了扩展但 postgres 没预加载时,查视图会报错
+  try {
+    await sql`select 1 from public.pg_stat_statements limit 1`.execute(db);
+  } catch {
+    return off("postgres 没有预加载 pg_stat_statements（shared_preload_libraries），重启数据库后生效");
+  }
+  type Row = { query: string; calls: string; total: number; mean: number; rows: string };
+  const top = async (order: "mean" | "calls") =>
+    (
+      await sql<Row>`
+        select left(regexp_replace(query, '[[:space:]]+', ' ', 'g'), 500) as query, calls::text,
+               total_exec_time as total, mean_exec_time as mean, rows::text
+          from public.pg_stat_statements
+         where dbid = (select oid from pg_database where datname = current_database())
+           and query not ilike '%pg_stat_statements%'
+           ${order === "mean" ? sql`and calls >= ${SLOW_MIN_CALLS} order by mean_exec_time desc` : sql`order by calls desc`}
+         limit ${STATEMENT_TOP}`.execute(db)
+    ).rows.map((r) => ({
+      query: r.query,
+      calls: Number(r.calls),
+      totalMs: Math.round(r.total * 100) / 100,
+      meanMs: Math.round(r.mean * 1000) / 1000,
+      rows: Number(r.rows),
+    }));
+  const info = await sql<{ since: Date | null }>`select stats_reset as since from public.pg_stat_statements_info`
+    .execute(db)
+    .catch(() => null);
+  const since = info?.rows[0]?.since;
+  return {
+    enabled: true,
+    note: null,
+    since: since ? new Date(since).toISOString() : null,
+    slowest: await top("mean"),
+    frequent: await top("calls"),
+  };
+}
+
 export async function buildStorage(
   d: StorageDeps & { engineSchema: string; refresh?: boolean },
 ): Promise<SystemStorage> {
-  const [bytes, scan, tables, kv, jobs] = await Promise.all([
+  const [bytes, scan, tables, kv, jobs, statements] = await Promise.all([
     dbSize(d.db).catch(() => null),
     dirScan(d.paths, { wait: true, force: d.refresh ?? false }).catch(() => null),
     sql<{
@@ -287,6 +336,7 @@ export async function buildStorage(
       .where("id", "in", Object.values(CLEANUP_JOB_IDS))
       .execute()
       .catch(() => []),
+    statementStats(d.db),
   ]);
   const lastSummaries = new Map<string, string | null>();
   if (jobs.length) {
@@ -324,6 +374,7 @@ export async function buildStorage(
         lastAutovacuumAt: iso(t.vac),
         lastAnalyzeAt: iso(t.ana),
       })),
+      statements,
     },
     kvCache: kv.rows.map((r) => ({ region: r.region, entries: r.entries, expired: r.expired, bytes: Number(r.bytes) })),
     images: { ...l.images, files: scan ? scan.images.files : null },