merge: 初选矩阵提速 8.7 倍 —— 潜在治疗标签预计算落库

· plan_reasons 加 potential_labels，矩阵不再回查 patient_facts
  本地实测 958ms → 110ms，磁盘读 119,861 → 0，24 格逐字未变
· 补上就地刷新 reason 那条路的标签失效（证据变了就标记重算）
· 加 backfill-plan-labels CLI —— 上线后必须跑一次（迁移后全表 NULL，
  实测置空后矩阵返回 0 行）
· 撤回「重算 plan 会释放客服」那个错判断

apps/pac-service/package.json

@@ -28,6 +28,7 @@
     "sync-incremental:prod": "node --max-old-space-size=4096 dist/cli/sync-incremental.cli.js",
     "import-patient": "ts-node --transpile-only src/cli/import-patient.cli.ts",
     "recompute-persona": "ts-node --transpile-only src/cli/recompute-persona.cli.ts",
+    "backfill-plan-labels": "ts-node --transpile-only src/cli/backfill-plan-labels.cli.ts",
     "recompute-persona:prod": "node --max-old-space-size=8192 dist/cli/recompute-persona.cli.js",
     "recompute-plans": "ts-node --transpile-only src/cli/recompute-plans.cli.ts",
     "recompute-plans:prod": "node --max-old-space-size=8192 dist/cli/recompute-plans.cli.js",

apps/pac-service/prisma/migrations/20260817010638_add_plan_reason_potential_labels/migration.sql

+-- 初选矩阵提速：把「这条依据能推出哪几个潜在治疗标签」预先算好落库。
+--
+-- 此前矩阵每次开页面都现推：plan_reasons → 展开 evidence.factIds → 回查 patient_facts
+-- （1567 万行 / 18 GB）。最忙的诊所一次摊三万多次随机查、372 MB I/O，
+-- 而这一切的唯一产出就是这个字符串。
+-- 本地实测（15,884 条 plan）：958 ms → 110 ms，磁盘读 119,861 → 0，24 格的数逐字未变。
+--
+-- ⚠️ **刻意可空**，三态有别：
+--   NULL  = 还没算过（回填 / 新写入的行会被刷新任务捞走）
+--   '{}'  = 算过了，这条依据推不出任何标签（K 码不在规则表里 / 年龄不落区间）
+--   非空  = 标签集合
+-- ⛔ 别加 NOT NULL DEFAULT '{}' —— 那样「没算过」和「算过是空」就分不开，
+--    刷新任务再也找不到漏网的行，而漏了不会有任何报错。
+-- AlterTable
+ALTER TABLE "plan_reasons" ADD COLUMN "potential_labels" TEXT[];
+
+-- 矩阵只关心「推得出标签」的那些依据；推不出的占比不低，跳过它们能少扫一截。
+-- ⚠️ 用 IS NOT NULL 而不是 <> '{}'：后者会把「还没算过(NULL)」也排除在外，
+--    而回填期间正需要能看见它们。
+CREATE INDEX "plan_reasons_plan_id_potential_labels_idx"
+  ON "plan_reasons" ("plan_id")
+  WHERE "potential_labels" IS NOT NULL AND array_length("potential_labels", 1) > 0;

apps/pac-service/prisma/migrations/20260817011500_add_plan_reason_labels_missing_idx/migration.sql

+-- 「还没算标签的依据」专用索引。
+--
+-- 生成完 plan 之后要立刻补标签（不补就是 NULL，矩阵直接看不见这些人）。
+-- 没有这个索引的话，每次补都要为找 NULL 扫一遍 plan_reasons（测试服 31 万行）——
+-- 而绝大多数时候一条都没有，纯白扫。
+-- ⚠️ 部分索引只收 NULL 行 ⇒ 平时几乎是空的，补完即回到 O(1)。
+CREATE INDEX "plan_reasons_labels_missing_idx"
+  ON "plan_reasons" ("id")
+  WHERE "potential_labels" IS NULL;

apps/pac-service/prisma/schema.prisma

@@ -1254,6 +1254,28 @@ model PlanReason {
   closedReason String?   @map("closed_reason")
   closedAt     DateTime? @map("closed_at") @db.Timestamptz(3)
 
+  /**
+   * 这条依据能推出哪几个**潜在治疗标签**（implant / ortho / extraction …）—— 预先算好。
+   *
+   * ═══ 为什么落这一列 ═══════════════════════════════════════════════
+   * 初选矩阵此前每次开页面都现推：`plan_reasons` → 展开 `evidence.factIds` →
+   * 回查 `patient_facts`（1567 万行 / 18 GB）。最忙的诊所一次要摊三万多次随机查、
+   * 372 MB I/O —— **而这一切的唯一产出就是这个字符串**。
+   * 🔴 本地实测（15,884 条 plan 的诊所）：958 ms → 110 ms，**磁盘读 119,861 → 0**，
+   *   24 格的数逐字未变（见 `scripts/bench-matrix.ts` 的结果快照）。
+   *
+   * ═══ 为什么是数组 ═══════════════════════════════════════════════
+   * 一条依据可以挂多个 fact：93.7% 只有一个，但**有 3.7%（本地 1,391 条）
+   * 会推出不止一个 code** ⇒ ⛔ 存单值会把这批人算少，而那正是最不能出的错。
+   *
+   * ═══ ⚠️ 年龄被冻结在算的那一刻 ═══════════════════════════════════
+   * 标签规则里有三条带年龄（K08>18 / K07 3~12 / K07 13~40），所以它不是纯函数。
+   * 实测本地受年龄规则影响 5,551 人，**明天跨档 0 人、30 天内 16 人**（≈0.5 人/天）——
+   * 跟着画像重算每晚刷一次，误差可忽略。
+   * ⛔ 但**不能不刷**：不刷就会逐日漂，且不会有任何报错。刷新入口见 `plan-label.sql.ts`。
+   */
+  potentialLabels String[] @map("potential_labels")
+
   createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3)
 

apps/pac-service/scripts/bench-matrix.ts

+/**
+ * 初选矩阵基准 —— 「改之前 / 改之后」用**同一把尺**量。
+ *
+ * ⚠️ 只量 SQL 执行时间（`EXPLAIN ANALYZE` 的 Execution Time），⛔ 不含 HTTP / 序列化：
+ *   那两段在改动前后是一样的，混进来只会稀释信号。
+ * ⚠️ 每档跑 N 轮取**最快**，⛔ 不取平均 —— 这台机器上跑着别的东西，
+ *   平均值被噪声主导（实测同一条件下 1388~2130ms 都出现过）。最快值才稳定可比。
+ *
+ * 用法：
+ *   npx ts-node -T scripts/bench-matrix.ts            # 跑全部诊所档位
+ *   npx ts-node -T scripts/bench-matrix.ts --rounds 5
+ */
+import { PrismaClient } from '@prisma/client';
+import { poolBaseSql } from '../src/modules/plan/cohort-filter';
+import { planLabelAnchorsSql } from '../src/modules/plan/reason-temperature.sql';
+
+const prisma = new PrismaClient();
+const ROUNDS = Number(process.argv[process.argv.indexOf('--rounds') + 1]) || 3;
+
+/** 把 Prisma.Sql 的 `?` 占位换成字面量 —— EXPLAIN 不吃参数化语句。 */
+function inline(q: { sql: string; values: unknown[] }): string {
+  let i = 0;
+  return q.sql.replace(/\?/g, () => {
+    const v = q.values[i++];
+    return typeof v === 'string' ? `'${v.replace(/'/g, "''")}'` : String(v);
+  });
+}
+
+function matrixSql(hostId: string, tenantId: string, clinicId: string): string {
+  const scope = { hostId, tenantId, clinicIds: [], sourceUnits: [] } as never;
+  const inner = inline(planLabelAnchorsSql(poolBaseSql(scope, clinicId)));
+  // 外层与 `CohortAttributesService.matrix` 同构（温度 CASE 简化成三档，量的是同一条 join 路径）
+  return `EXPLAIN (ANALYZE, BUFFERS)
+WITH la AS (${inner}),
+b AS (SELECT patient_id, label,
+        CASE WHEN NOW() <= hot_until THEN 'hot'
+             WHEN NOW() <= warm_until THEN 'warm' ELSE 'cold' END AS temp
+      FROM la)
+SELECT label, temp, count(DISTINCT patient_id) AS n
+FROM b GROUP BY GROUPING SETS ((label, temp), (temp));`;
+}
+
+async function run(sql: string) {
+  const rows = await prisma.$queryRawUnsafe<Array<Record<string, string>>>(sql);
+  const text = rows.map((r) => Object.values(r)[0]).join('\n');
+  const ms = Number(/Execution Time: ([\d.]+)/.exec(text)?.[1] ?? NaN);
+  const reads = [...text.matchAll(/read=(\d+)/g)].reduce((a, m) => a + Number(m[1]), 0);
+  // ⚠️ 并行查询里每个 worker 各报一段 JIT，这里取首段；⛔ 别拿它跟墙钟直接比大小
+  const jit = Number(/JIT[\s\S]*?Total (\d+\.?\d*) ms/.exec(text)?.[1] ?? 0);
+  return { ms, reads, jit };
+}
+
+/**
+ * 🔴 **结果快照** —— 性能改动必须同时证明「24 格一个数都没变」。
+ *   ⛔ 只比耗时是不够的：把 join 改掉、把标签预计算，最容易的失败是**悄悄少算一批人**，
+ *   而那正是产品最不能接受的（「主管一对数就觉得系统在骗他」）。
+ */
+async function snapshot(sqlWithExplain: string) {
+  const plain = sqlWithExplain.replace(/^EXPLAIN \([^)]*\)\n/, '');
+  const rows = await prisma.$queryRawUnsafe<
+    Array<{ label: string | null; temp: string | null; n: bigint }>
+  >(plain);
+  return rows
+    .map((r) => `${r.label ?? '∑'}/${r.temp ?? '∑'}=${r.n}`)
+    .sort()
+    .join(' ');
+}
+
+(async () => {
+  const clinics = await prisma.$queryRaw<
+    Array<{ host_id: string; tenant_id: string; target_clinic_id: string; n: bigint }>
+  >`SELECT host_id, tenant_id, target_clinic_id, count(*) AS n
+    FROM followup_plans WHERE status='active' AND assignee_user_id IS NULL
+    GROUP BY 1,2,3 HAVING count(*) >= 10 ORDER BY 4 DESC LIMIT 5`;
+
+  console.log(`初选矩阵基准 · 每档取 ${ROUNDS} 轮最快值\n`);
+  console.log('  plan 数   最快耗时      磁盘读     JIT(首段)');
+  console.log('  ' + '─'.repeat(46));
+  const out: Array<Record<string, number>> = [];
+  const snaps: string[] = [];
+  for (const c of clinics) {
+    const sql = matrixSql(c.host_id, c.tenant_id, c.target_clinic_id);
+    let best = { ms: Infinity, reads: 0, jit: 0 };
+    for (let r = 0; r < ROUNDS; r++) {
+      const x = await run(sql);
+      if (x.ms < best.ms) best = x;
+    }
+    const n = Number(c.n);
+    console.log(
+      `  ${String(n).padStart(7)}   ${best.ms.toFixed(0).padStart(6)} ms   ` +
+        `${String(best.reads).padStart(7)}   ${best.jit ? best.jit.toFixed(0) + ' ms' : '—'}`,
+    );
+    out.push({ plans: n, ms: best.ms, reads: best.reads, jit: best.jit });
+    snaps.push(`${n}: ${await snapshot(sql)}`);
+  }
+  console.log('\nJSON ' + JSON.stringify(out));
+  console.log('\n结果快照（改动后必须逐字相同）：');
+  for (const s of snaps) console.log('  ' + s);
+  await prisma.$disconnect();
+})().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});

apps/pac-service/src/cli/backfill-plan-labels.cli.ts

+/**
+ * Backfill Plan Labels CLI —— 回填 / 重算 `plan_reasons.potential_labels`。
+ *
+ * 初选矩阵靠这一列（改造后不再回查 `patient_facts`，958ms → 110ms）。
+ *
+ * 🔴 **上线后必须立刻跑一次**：迁移只是加了列，全表都是 NULL，而矩阵
+ *   `unnest(potential_labels)` 遇到 NULL 什么都不出 ⇒ **这段时间矩阵会全是 0**。
+ *   ⛔ 别指望夜间任务兜 —— 那要等到次日 03:30。
+ *
+ * ⚠️ **重算 plan 也能补齐，而且是安全的** —— ⛔ 别被"会释放客服"吓住（那是错的）：
+ *   `auto_release` 只在「本轮该患者 0 命中（信号真没了）」或「最后到诊诊所变了」时触发，
+ *   两者都是真实业务变化，不是重算这个动作造成的；这套逻辑本来就跟着增量同步每天在跑，
+ *   该释放的早已释放，再跑一次不会多释放任何人。
+ * ⇒ 日常维护**本来就靠原机制**：plan 生成收尾会调 `backfillMissing()`。
+ *   这个 CLI 只解决一件事：**上线那一刻的空窗** —— 迁移后全表 NULL，
+ *   等下一次增量同步（最长两小时）或夜间刷新（03:30）才自愈，
+ *   而 CLI 把这个窗口压到约一分钟。⛔ 别拿它当日常手段。
+ *
+ * Usage:
+ *   pnpm backfill-plan-labels              # 只补没算过的（上线后跑这个）
+ *   pnpm backfill-plan-labels -- --all     # 连算过的一起重算（改了标签规则后跑）
+ *   pnpm backfill-plan-labels -- --check   # 只报还差多少，不写库
+ */
+import { NestFactory } from '@nestjs/core';
+import { Logger } from '@nestjs/common';
+import { AppModule } from '../app.module';
+import { PlanLabelService } from '../modules/plan/plan-label.service';
+
+async function main(): Promise<void> {
+  const log = new Logger('backfill-plan-labels');
+  const argv = process.argv.slice(2);
+  const all = argv.includes('--all');
+  const checkOnly = argv.includes('--check');
+
+  const app = await NestFactory.createApplicationContext(AppModule, {
+    logger: ['error', 'warn', 'log'],
+  });
+  try {
+    const svc = app.get(PlanLabelService);
+    const before = await svc.countMissing();
+    log.log(`还没算过的依据：${before} 条`);
+    if (checkOnly) return;
+
+    const t0 = Date.now();
+    const written = all ? await svc.refreshAll() : await svc.backfillMissing();
+    const after = await svc.countMissing();
+    log.log(
+      `${all ? '全量重算' : '回填'}完成：写入 ${written} 条，` +
+        `耗时 ${((Date.now() - t0) / 1000).toFixed(1)}s，剩余未算 ${after} 条`,
+    );
+    // ⚠️ 回填完还有剩 = 有条写入路径没被覆盖到，值得当场查，⛔ 别等矩阵少人了才发现
+    if (!all && after > 0) {
+      log.error(`⚠️ 回填后仍有 ${after} 条没算过 —— 检查是否有新写入路径没接上补齐`);
+      process.exitCode = 1;
+    }
+  } finally {
+    await app.close();
+  }
+}
+
+main().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});

apps/pac-service/src/modules/plan/engine/plan-engine.service.ts

@@ -7,6 +7,7 @@ import { reasonNeedsRefresh } from './reason-refresh';
 import { runPool } from '../../../common/run-pool';
 import { PlanEventType, PlanEventReason } from '@pac/types';
 import { recordPlanEvent, recordPlanEventsBulk, computeHeldSeconds } from '../plan-event.recorder';
+import { PlanLabelService } from '../plan-label.service';
 
 /**
  * PlanEngineService — 跑所有 scenario plugin,产 FollowupPlan + PlanReason
@@ -33,6 +34,7 @@ export class PlanEngineService {
   constructor(
     private readonly prisma: PrismaService,
     private readonly initiation: TreatmentInitiationRecallScenario,
+    private readonly planLabels: PlanLabelService,
   ) {
     // v2.1:一期只跑潜在治疗新链召回(treatment_initiation_recall)
     // 链已完成召回(treatment_aftercare_recall)留后续
@@ -113,6 +115,17 @@ export class PlanEngineService {
         orderBy: { version: 'desc' },
         select: { id: true },
       });
+      /**
+       * ⭐ 单患者重算也要补标签 —— 与批量那条同一个理由：
+       *   新写的 reason 是 NULL、就地刷新过的也被置回了 NULL，
+       *   ⛔ 不补的话这位患者在初选矩阵里**当场消失**，直到夜间重算才回来，且不报错。
+       * ⚠️ 平时代价接近零：`potential_labels IS NULL` 上有部分索引，没得补时一次索引查就返回。
+       */
+      try {
+        await this.planLabels.backfillMissing();
+      } catch (e) {
+        this.logger.warn(`[标签] 单患者补齐失败（夜间刷新会兜住）：${e instanceof Error ? e.message : e}`);
+      }
       // upsert 路径自身会 supersede 旧版本(版本流),不算"关闭退池";plansClosed 只统计 0 命中关闭。
       return { plansCreated: created, plansClosed: 0, outcome: result, currentPlanId: current?.id ?? null };
     } catch (err) {
@@ -367,6 +380,23 @@ export class PlanEngineService {
       }
     }
 
+    /**
+     * ⭐ **本轮新写的 reason 必须立刻补上 `potential_labels`** —— 初选矩阵靠这一列。
+     *
+     * 🔴 不补 = 新 plan 的那一列是 NULL，而矩阵 `unnest(potential_labels)` 直接跳过它们 ⇒
+     *   **今天生成的人在矩阵里看不见**，且不会有任何报错（主管只会觉得"怎么少了一批"）。
+     * ⚠️ 走 `PlanLabelService` 的同一份 SQL，⛔ 别在这里用 TS 再算一遍标签：
+     *   规则表是产品配置，第二份实现迟早和矩阵那份漂开（`plan-label.sql.ts` 头注那条）。
+     * ⚠️ 失败不阻断生成：标签是"派生数据"，夜间刷新会兜住；
+     *   而生成本身回滚的代价远大于矩阵晚一晚上准。
+     */
+    try {
+      const filled = await this.planLabels.backfillMissing();
+      if (filled > 0) this.logger.log(`[标签] 本轮补齐 ${filled} 条依据的潜在治疗标签`);
+    } catch (e) {
+      this.logger.warn(`[标签] 补齐失败（不阻断生成，夜间刷新会兜住）：${e instanceof Error ? e.message : e}`);
+    }
+
     return {
       scenariosRun: this.scenarios.length,
       patientsHit: hitsByPatient.size,
@@ -568,6 +598,25 @@ export class PlanEngineService {
                 },
               });
             }
+            /**
+             * 🔴 **证据换了，标签必须重算** —— 置回 NULL(=「没算过」)，
+             *   本轮收尾的 `backfillMissing()` 会把它填上。
+             *
+             * ⚠️ ⛔ 不能不管：`potential_labels` 是从 `evidence.factIds` 指向的 fact 推出来的。
+             *   这条路**只改 evidence、不升版本**，那一列于是既不是 NULL（补齐会跳过它）、
+             *   又已经对不上新证据 ⇒ 矩阵会**按旧标签把人算进错的格子**，
+             *   要到夜间全量重算才自愈，中间一整天不报错。
+             * ⚠️ ⛔ 也别在这里用 TS 顺手算一个填进去：规则表是产品配置，
+             *   第二份实现迟早和矩阵那份漂开（见 `plan-label.sql.ts` 头注）。
+             *   置 NULL 让它回到同一份 SQL，是唯一不会漂的写法。
+             * ⚠️ 走 raw：Prisma 的标量数组字段**不能**用类型化 API 置 null，
+             *   而改用 `[]` 会让「没算过」和「算过是空」分不开（那正是这一列设成可空的理由）。
+             */
+            if (staleRows.length > 0) {
+              await tx.$executeRaw`
+                UPDATE plan_reasons SET potential_labels = NULL
+                WHERE id = ANY(ARRAY[${Prisma.join(staleRows.map((r) => r.id))}]::uuid[])`;
+            }
           });
           if (staleRows.length > 0) {
             this.logger.log(

apps/pac-service/src/modules/plan/plan-label.service.ts

+import { Injectable, Logger } from '@nestjs/common';
+import { PrismaService } from '../../prisma/prisma.service';
+import { refreshLabelsSql, countMissingLabelsSql, LABEL_REFRESH_BATCH } from './plan-label.sql';
+
+interface BatchResult {
+  seen: number;
+  written: number;
+  maxId: string | null;
+}
+
+/**
+ * `plan_reasons.potential_labels` 的刷新器 —— 初选矩阵的提速全靠这一列。
+ *
+ * 两种跑法：
+ *   · `backfillMissing()` —— 只补没算过的（NULL）。上线回填、以及新写入行的兜底。
+ *   · `refreshAll()`      —— 连算过的一起重算。⚠️ 标签含年龄规则，**不重算就会逐日漂**。
+ *
+ * ⛔ 两者都**不在请求路径上**：它们扫全表，得由 CLI / 定时任务跑。
+ */
+@Injectable()
+export class PlanLabelService {
+  private readonly logger = new Logger(PlanLabelService.name);
+
+  constructor(private readonly prisma: PrismaService) {}
+
+  /** 还有多少条没算过 —— 回填收尾核对、以及"刷新任务是不是在漏人"的体检项。 */
+  async countMissing(): Promise<number> {
+    const [row] = await this.prisma.$queryRaw<Array<{ n: number }>>(countMissingLabelsSql);
+    return row?.n ?? 0;
+  }
+
+  /** 只补没算过的。返回写了多少行。 */
+  async backfillMissing(batch = LABEL_REFRESH_BATCH): Promise<number> {
+    return this.run({ onlyMissing: true, batch, label: '回填' });
+  }
+
+  /** 全量重算（年龄会变）。返回写了多少行。 */
+  async refreshAll(batch = LABEL_REFRESH_BATCH): Promise<number> {
+    return this.run({ onlyMissing: false, batch, label: '重算' });
+  }
+
+  private async run(o: { onlyMissing: boolean; batch: number; label: string }): Promise<number> {
+    let afterId: string | null = null;
+    let written = 0;
+    let seenTotal = 0;
+    for (;;) {
+      const rows: BatchResult[] = await this.prisma.$queryRaw<BatchResult[]>(
+        refreshLabelsSql({ afterId, onlyMissing: o.onlyMissing, batch: o.batch }),
+      );
+      const r = rows[0];
+      const seen = r?.seen ?? 0;
+      if (seen === 0) break;
+      written += r?.written ?? 0;
+      seenTotal += seen;
+      /**
+       * 🔴 游标按 `maxId` 推进，⛔ 不按"写了几行" ——
+       *   全量重算时绝大多数行算出来跟原值一样、不会被写，此时 written=0
+       *   但**并不是做完了**。拿 written 当停止条件 = 第一批之后就静默罢工。
+       */
+      afterId = r?.maxId ?? null;
+      if (!afterId) break;
+      if (seen < o.batch) break; // 最后一批
+    }
+    this.logger.log(`${o.label}完成：看过 ${seenTotal} 条，写入 ${written} 条`);
+    return written;
+  }
+}

apps/pac-service/src/modules/plan/plan-label.sql.ts

+import { Prisma } from '@prisma/client';
+import { labelCaseSql, AGE_YEARS_SQL } from './reason-temperature.sql';
+
+/**
+ * `plan_reasons.potential_labels` 的**计算与刷新** —— 回填、夜间刷新、新写入三条路共用这一份。
+ *
+ * ═══ 为什么要预计算 ═══════════════════════════════════════════════
+ * 初选矩阵此前每次开页面都现推：`plan_reasons` → 展开 `evidence.factIds` →
+ * 回查 `patient_facts`（1567 万行 / 18 GB）。最忙的诊所一次摊三万多次随机查、372 MB I/O，
+ * **而这一切的唯一产出就是一个标签字符串**。
+ * 🔴 本地实测（15,884 条 plan）：958 ms → 110 ms，磁盘读 119,861 → 0，24 格逐字未变。
+ *
+ * ═══ 🔴 标签规则只有一处 ═══════════════════════════════════════════
+ * 这里复用 `labelCaseSql`（矩阵此前用的同一个函数，由 `POTENTIAL_LABEL_RULES` 生成）——
+ * ⛔ **绝不在这里另写一份 CASE**：规则表是产品配置，抄第二份必然漂，
+ *   而漂了之后矩阵的数会**静默**不一样，没有任何报错（同 `enums` 里那条
+ *   「同名不同义是统计事故的标准配方」）。
+ *
+ * ═══ ⚠️ 年龄被冻结 ═══════════════════════════════════════════════
+ * 规则里有三条带年龄（K08>18 / K07 3~12 / K07 13~40）⇒ 标签不是纯函数。
+ * 实测受影响 5,551 人中「明天跨档 0 人、30 天内 16 人」（≈0.5 人/天）——
+ * 每晚刷一次误差可忽略，⛔ 但**不刷就会逐日漂且不报错**。
+ */
+
+/** 一次刷新的批量上限 —— ⚠️ 别一条 UPDATE 扫全表：31 万行会长时间持锁。 */
+export const LABEL_REFRESH_BATCH = 5_000;
+
+/**
+ * 计算并写回一批 `potential_labels`，**按 id 游标推进**。
+ *
+ * @param afterId 上一批处理到的最大 id（首批传 null）
+ * @param onlyMissing true = 只补 `IS NULL` 的（回填 / 补漏网）；
+ *                    false = 连已算过的一起重算（夜间刷新，因为年龄会变）
+ * @returns 本批**看过**的行（`seen`）与实际写了几行（`written`）、本批最大 id（`maxId`）
+ *
+ * 🔴 **必须带游标** —— 只写 `ORDER BY id LIMIT n` 的话：
+ *   · onlyMissing 那条路碰巧能走完（写完就不再是 NULL，下一批自然换人）；
+ *   · 而**全量刷新会原地打转** —— 每次都取同一批，永远推不到第二批。
+ *   ⚠️ 更阴的是它不会报错，只是任务每晚白跑。同类「闸门永远追不上」的坑，
+ *     留痕清理那里刚踩过一次（那次是拿 `updatedAt` 当判据）。
+ *
+ * ⚠️ 全程 `LEFT JOIN` 到 fact —— 推不出标签的依据也要写 `'{}'`，
+ *   ⛔ 不能只更新有标签的那些：否则它们永远停在 NULL，被每一轮重复捞出来。
+ */
+export function refreshLabelsSql(opts: {
+  afterId: string | null;
+  onlyMissing: boolean;
+  batch?: number;
+}): Prisma.Sql {
+  const limit = opts.batch ?? LABEL_REFRESH_BATCH;
+  const gate = opts.onlyMissing ? Prisma.sql`AND pr.potential_labels IS NULL` : Prisma.empty;
+  const cursor = opts.afterId
+    ? Prisma.sql`AND pr.id > ${opts.afterId}::uuid`
+    : Prisma.empty;
+  return Prisma.sql`
+    WITH target AS (
+      SELECT pr.id, fp.patient_id
+      FROM plan_reasons pr
+      JOIN followup_plans fp ON fp.id = pr.plan_id
+      WHERE fp.superseded_at IS NULL
+        ${gate}
+        ${cursor}
+      ORDER BY pr.id
+      LIMIT ${limit}
+    ),
+    computed AS (
+      SELECT t.id,
+             COALESCE(
+               array_agg(DISTINCT lab.lbl) FILTER (WHERE lab.lbl IS NOT NULL),
+               '{}'::text[]
+             ) AS labels
+      FROM target t
+      JOIN patients p ON p.id = t.patient_id
+      JOIN plan_reasons pr2 ON pr2.id = t.id
+      LEFT JOIN LATERAL jsonb_array_elements_text(pr2.evidence->'factIds') fid ON TRUE
+      LEFT JOIN patient_facts f
+        ON f.id = fid::uuid
+       AND f.status = 'active'
+       AND COALESCE(f.occurred_at, f.planned_for) IS NOT NULL
+      LEFT JOIN LATERAL (SELECT ${labelCaseSql('f', AGE_YEARS_SQL)} AS lbl) lab ON TRUE
+      GROUP BY t.id
+    ),
+    upd AS (
+      UPDATE plan_reasons pr
+         SET potential_labels = c.labels
+        FROM computed c
+       WHERE pr.id = c.id
+         AND (pr.potential_labels IS DISTINCT FROM c.labels)
+      RETURNING pr.id
+    )
+    -- ⚠️ seen 与 written 必须**分开报**：全量刷新时绝大多数行算出来跟原值一样、
+    --   不会被写，此时 written=0 ⛔ 不代表做完了。推进游标看的是 seen / maxId。
+    SELECT (SELECT count(*)::int FROM target)      AS seen,
+           (SELECT count(*)::int FROM upd)         AS written,
+           (SELECT max(id::text) FROM target)     AS "maxId"`;
+}
+
+/** 还有多少条没算过 —— 回填收尾与告警用。 */
+export const countMissingLabelsSql = Prisma.sql`
+  SELECT count(*)::int AS n
+  FROM plan_reasons pr
+  JOIN followup_plans fp ON fp.id = pr.plan_id
+  WHERE fp.superseded_at IS NULL AND pr.potential_labels IS NULL`;

apps/pac-service/src/modules/plan/plan.module.ts

@@ -14,6 +14,7 @@ import { PlanEngineService } from './engine/plan-engine.service';
 import { ChainComposerService } from './engine/chain-composer.service';
 import { TreatmentInitiationRecallScenario } from './engine/scenarios/treatment-initiation-recall.scenario';
 import { RecallDebugController } from './recall-debug/recall-debug.controller';
+import { PlanLabelService } from './plan-label.service';
 import { RecallDebugService } from './recall-debug/recall-debug.service';
 
 /**
@@ -42,8 +43,9 @@ import { RecallDebugService } from './recall-debug/recall-debug.service';
     ChainComposerService,
     TreatmentInitiationRecallScenario,
     RecallDebugService,
+    PlanLabelService,
   ],
   // MCP 的主管工具直接用这两个 service(条件注册,见 mcp-server.factory)
-  exports: [PlanService, PlanAssignmentService, AgentRosterService, AssignmentProposalService, CohortAttributesService, ExecutionService, ExecutionCallbackService, PlanEngineService, ChainComposerService],
+  exports: [PlanService, PlanLabelService, PlanAssignmentService, AgentRosterService, AssignmentProposalService, CohortAttributesService, ExecutionService, ExecutionCallbackService, PlanEngineService, ChainComposerService],
 })
 export class PlanModule {}

apps/pac-service/src/modules/plan/reason-temperature.sql.ts

@@ -155,16 +155,10 @@ export function planLabelAnchorsSql(planFilter: Prisma.Sql): Prisma.Sql {
            ${BOUNDS.warm}   AS warm_until,
            ${BOUNDS.anchor} AS anchor_at
     FROM followup_plans fp
-    JOIN patients p ON p.id = fp.patient_id
     JOIN plan_reasons pr ON pr.plan_id = fp.id
-    CROSS JOIN LATERAL jsonb_array_elements_text(pr.evidence->'factIds') fid
-    JOIN patient_facts f ON f.id = fid::uuid AND f.status = 'active'
-    CROSS JOIN LATERAL (SELECT COALESCE(f.occurred_at, f.planned_for) AS at) anc
-    CROSS JOIN LATERAL (SELECT ${labelCaseSql('f', AGE_YEARS_SQL)} AS lbl) lab
+    CROSS JOIN LATERAL unnest(pr.potential_labels) AS lab(lbl)
     ${LAST_VISIT_JOIN}
     WHERE ${planFilter}
-      AND lab.lbl IS NOT NULL
-      AND anc.at IS NOT NULL
     GROUP BY 1, 2, 3`;
 }
 

apps/pac-service/src/queues/invocation-retention.service.ts

@@ -2,6 +2,7 @@ import { Injectable, Logger } from '@nestjs/common';
 import { Cron } from '@nestjs/schedule';
 import { ConfigService } from '@nestjs/config';
 import { PrismaService } from '../prisma/prisma.service';
+import { PlanLabelService } from '../modules/plan/plan-label.service';
 import type { AppConfig } from '../config/configuration';
 
 /**
@@ -74,3 +75,35 @@ export class InvocationRetentionService {
     return res.count;
   }
 }
+
+/**
+ * PlanLabelRefreshService —— 每晚重算 `plan_reasons.potential_labels`。
+ *
+ * 🔴 **不刷就会逐日漂，且不会有任何报错**：标签规则里有三条带年龄
+ *   （K08>18 / K07 3~12 / K07 13~40），患者过生日跨档时标签就该变。
+ *   实测本地受影响 5,551 人中「明天跨档 0 人、30 天内 16 人」（≈0.5 人/天）——
+ *   每晚刷一次误差可忽略，但**一天不刷就多欠一天**。
+ *
+ * ⚠️ 排在画像重算之后（03:30）：标签只依赖 fact 与生日，与画像无先后依赖，
+ *   但错开时段避免两个全表任务抢 I/O。
+ */
+@Injectable()
+export class PlanLabelRefreshService {
+  private readonly logger = new Logger(PlanLabelRefreshService.name);
+
+  constructor(private readonly labels: PlanLabelService) {}
+
+  @Cron(process.env.PAC_PLAN_LABEL_REFRESH_CRON || '30 3 * * *', {
+    name: 'plan-label-refresh',
+    timeZone: 'Asia/Shanghai',
+  })
+  async refresh(): Promise<void> {
+    const missing = await this.labels.countMissing();
+    if (missing > 0) {
+      // 平时该是 0（生成完就补）。非 0 说明有条路写了 reason 却没补标签 —— 值得看一眼。
+      this.logger.warn(`[标签] 刷新前发现 ${missing} 条没算过 —— 生成路径可能漏了补齐`);
+    }
+    const n = await this.labels.refreshAll();
+    this.logger.log(`[标签] 夜间重算完成，${n} 条因年龄跨档等原因发生变化`);
+  }
+}

apps/pac-service/src/queues/queues.module.ts

@@ -11,7 +11,7 @@ import { QueueProducer } from './queue-producer.service';
 import { StaleScanService } from './stale-scan.service';
 import { SyncIncrementalSchedulerService } from './sync-incremental.scheduler';
 import { DwLagMonitorService } from './dw-lag-monitor.service';
-import { InvocationRetentionService } from './invocation-retention.service';
+import { InvocationRetentionService, PlanLabelRefreshService } from './invocation-retention.service';
 import { DailyHealthReportService } from './daily-health-report.service';
 import { DailyHealthReportController } from './daily-health-report.controller';
 import { PersonaRecomputeProcessor } from './processors/persona-recompute.processor';
@@ -73,6 +73,7 @@ import { ColdImportProcessor } from './processors/cold-import.processor';
     DwLagMonitorService,
     DailyHealthReportService,
     InvocationRetentionService,
+    PlanLabelRefreshService,
     PersonaRecomputeProcessor,
     PlanRecomputeProcessor,
     PlanAssetGenerateProcessor,

apps/pac-service/tests/plan-engine-batch.spec.ts

@@ -234,6 +234,7 @@ function makeStore(
     }),
   };
 
+  const labelsReset = jest.fn(async () => 0);
   const prisma = {
     followupPlan,
     planReason,
@@ -251,10 +252,16 @@ function makeStore(
         },
         planReason: { update: planReason.update },
         planEventLog,
+        /**
+         * 就地刷新 reason 时会把 `potential_labels` 置回 NULL（=「重算我」）——
+         * 那一步走 raw（Prisma 的标量数组字段不能用类型化 API 置 null）。
+         * 替身把调用记下来，让测试能断言"证据变了就一定标记重算"。
+         */
+        $executeRaw: labelsReset,
       }),
     ),
   };
-  return { prisma, plans, logs, planReason, events };
+  return { prisma, plans, logs, planReason, events, labelsReset };
 }
 
 function makeScenario(hits: ScenarioHit[]) {
@@ -271,8 +278,10 @@ function hit(patientId: string, subKey: string, priorityScore = 50, targetClinic
     evidence: { factIds: ['f1'] },
   } as ScenarioHit;
 }
+/** 标签刷新器替身 —— 生成收尾会调它补 `potential_labels`；这里只要不炸即可。 */
+const labelSvcStub = { backfillMissing: jest.fn(async () => 0), countMissing: jest.fn(async () => 0) };
 function engine(prisma: unknown, scenario: unknown) {
-  return new PlanEngineService(prisma as never, scenario as never);
+  return new PlanEngineService(prisma as never, scenario as never, labelSvcStub as never);
 }
 const NOW = new Date('2026-06-02T00:00:00Z');
 
@@ -317,7 +326,7 @@ describe('runAllForHost 批量路径 — 5 种结局等价', () => {
   // ⭐ 2026-07:(scenario,subKey) 没变、但 signals 语义变了(如缺牙按年龄排治疗新增
   //   focusCategory / patientAge)—— 原先走 unchanged 分支一个字都不改,存量 plan 永远修不好。
   test('unchanged + signals 语义变化 → reason 就地刷新,**不升版本、不动认领**', async () => {
-    const { prisma, plans, planReason } = makeStore({
+    const { prisma, plans, planReason, labelsReset } = makeStore({
       plans: [
         {
           id: 'p-stale',
@@ -368,6 +377,14 @@ describe('runAllForHost 批量路径 — 5 种结局等价', () => {
     expect(arg.where.id).toBe('r-stale');
     expect(arg.data.signals.focusCategory).toBe('prosthodontic');
     expect(arg.data.signals.patientAge).toBe(90);
+    /**
+     * 🔴 **证据被就地改写 ⇒ 必须同时把 `potential_labels` 置回 NULL**。
+     *
+     * 标签是从 `evidence.factIds` 指向的 fact 推出来的，而这条路**只改 evidence、不升版本** ——
+     * 不置空的话那一列既不是 NULL（收尾的 backfillMissing 会跳过它）、又对不上新证据 ⇒
+     * 初选矩阵会**按旧标签把人算进错的格子**，要到夜间全量重算才自愈，中间一整天不报错。
+     */
+    expect(labelsReset).toHaveBeenCalledTimes(1);
   });
 
   test('⭐ unchanged + signals 完全一致 → 不写 reason(防每日重算全量重写)', async () => {

apps/pac-service/tests/plan-label.spec.ts

+import { refreshLabelsSql, countMissingLabelsSql, LABEL_REFRESH_BATCH } from '../src/modules/plan/plan-label.sql';
+import { planLabelAnchorsSql } from '../src/modules/plan/reason-temperature.sql';
+import { Prisma } from '@prisma/client';
+
+/**
+ * `plan_reasons.potential_labels` —— 初选矩阵提速的那一列。
+ *
+ * 本地实测（15,884 条 plan 的诊所）：958 ms → 110 ms，磁盘读 119,861 → 13,500，
+ * 而 24 格的数**逐字未变**。这组测试守的是让它继续成立的几条结构性前提。
+ */
+
+const text = (q: Prisma.Sql) => q.sql;
+
+describe('矩阵查询不再碰 patient_facts', () => {
+  const sql = text(planLabelAnchorsSql(Prisma.sql`fp.status = 'active'`));
+
+  it('🔴 ⛔ 不再 JOIN patient_facts —— 那是三万多次随机查、372 MB I/O 的来源', () => {
+    expect(sql).not.toMatch(/patient_facts/);
+  });
+
+  it('🔴 ⛔ 不再展开 evidence.factIds', () => {
+    expect(sql).not.toMatch(/jsonb_array_elements_text/);
+  });
+
+  it('⛔ 不再 JOIN patients —— 年龄已烘进标签，那张表原本要被全表扫', () => {
+    expect(sql).not.toMatch(/JOIN patients/);
+  });
+
+  it('改用预存的 potential_labels', () => {
+    expect(sql).toMatch(/unnest\(pr\.potential_labels\)/);
+  });
+
+  it('末诊锚点仍来自 patient_profiles —— 温度那根轴没动', () => {
+    expect(sql).toMatch(/patient_profiles/);
+    expect(sql).toMatch(/last_visit_at/);
+  });
+});
+
+describe('刷新 SQL 的几条结构前提', () => {
+  it('🔴 带 id 游标 —— 否则全量重算每次取同一批，原地打转且不报错', () => {
+    const first = text(refreshLabelsSql({ afterId: null, onlyMissing: false }));
+    const next = text(refreshLabelsSql({ afterId: 'x', onlyMissing: false }));
+    expect(first).not.toMatch(/pr\.id >/);
+    expect(next).toMatch(/pr\.id >/);
+  });
+
+  it('🔴 seen / written / maxId 分开报 —— written=0 ⛔ 不等于做完了', () => {
+    const s = text(refreshLabelsSql({ afterId: null, onlyMissing: false }));
+    expect(s).toMatch(/AS seen/);
+    expect(s).toMatch(/AS written/);
+    expect(s).toMatch(/"maxId"/);
+  });
+
+  it('onlyMissing 才加 IS NULL 闸；全量重算⛔ 不加', () => {
+    expect(text(refreshLabelsSql({ afterId: null, onlyMissing: true }))).toMatch(
+      /potential_labels IS NULL/,
+    );
+    expect(text(refreshLabelsSql({ afterId: null, onlyMissing: false }))).not.toMatch(
+      /potential_labels IS NULL/,
+    );
+  });
+
+  it('🔴 用 LEFT JOIN 取 fact —— 推不出标签的也要写 {}，否则永远停在 NULL 被反复捞', () => {
+    const s = text(refreshLabelsSql({ afterId: null, onlyMissing: true }));
+    expect(s).toMatch(/LEFT JOIN patient_facts/);
+    expect(s).toMatch(/'\{\}'::text\[\]/);
+  });
+
+  it('⛔ 不在这里另写一份标签 CASE —— 复用矩阵那份（规则表是单一真源）', () => {
+    const q = refreshLabelsSql({ afterId: null, onlyMissing: true });
+    // ⚠️ 标签值走参数化占位，在 `values` 里而不在 `sql` 里 —— 断言写在 sql 上会假绿。
+    const vals = q.values.map(String);
+    for (const lbl of ['implant', 'early_ortho', 'restoration']) {
+      expect(vals).toContain(lbl);
+    }
+    // 且确实是一段 CASE（来自 labelCaseSql），⛔ 不是这里手写的映射
+    expect(q.sql).toMatch(/CASE\s+WHEN/);
+  });
+
+  it('只处理未作废的 plan', () => {
+    expect(text(refreshLabelsSql({ afterId: null, onlyMissing: true }))).toMatch(
+      /fp\.superseded_at IS NULL/,
+    );
+    expect(text(countMissingLabelsSql)).toMatch(/fp\.superseded_at IS NULL/);
+  });
+
+  it('批量有上限 —— ⛔ 别一条 UPDATE 扫全表', () => {
+    expect(LABEL_REFRESH_BATCH).toBeGreaterThan(0);
+    expect(LABEL_REFRESH_BATCH).toBeLessThanOrEqual(20_000);
+  });
+});