merge: gap 集合式形态 + 对拍工具(默认 legacy) → test

apps/pac-service/package.json

@@ -30,6 +30,8 @@
     "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",
+    "verify-gap-equivalence": "ts-node --transpile-only src/cli/verify-gap-equivalence.cli.ts",
+    "verify-gap-equivalence:prod": "node --max-old-space-size=8192 dist/cli/verify-gap-equivalence.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",
     "timeline": "ts-node --transpile-only src/cli/timeline.cli.ts",

apps/pac-service/src/cli/verify-gap-equivalence.cli.ts

+/**
+ * verify-gap-equivalence — gap 计算形态对拍(legacy ↔ setbased)
+ *
+ * 为什么必须有这个工具:
+ *   `buildGapCore` 是【召回】与【画像】共用的单一真理源,改错的后果是**静默少召** ——
+ *   不报错、不炸测试,几个月后才由一线反馈冒出来。而现有 spec(arch-denture /
+ *   polish-implies / restoration-in-place / treated-evidence / review-implies)
+ *   全是纯 JS 常量与正则断言,**不碰 SQL 行为**,重写后照样全绿 → 对这次改动的保护 ≈ 0。
+ *   所以正确性只能靠"两版跑同一份数据、逐 (患者×信号×牙位) 差分"来证。
+ *
+ * 怎么保证可信:
+ *   ① 两版在**同一个 REPEATABLE READ 事务**里跑 —— 否则并发增量摄入会造出假差异。
+ *   ② SQL 来自 scenario 自己的 `buildScenarioSql()`,不是这里另抄一份 ——
+ *      另抄就变成"验证我抄得对不对",而不是验证线上行为。
+ *   ③ `--self` 模式让 legacy 跟自己对拍,先证明工具本身可信(必须零差异)。
+ *
+ * Usage:
+ *   pnpm verify-gap-equivalence -- --host=jvs-dw                 # 全部 11 个子场景
+ *   pnpm verify-gap-equivalence -- --host=jvs-dw --sub=impacted_tooth
+ *   pnpm verify-gap-equivalence -- --host=jvs-dw --self          # 自对拍(工具自检)
+ *   pnpm verify-gap-equivalence -- --host=jvs-dw --bench         # 只测耗时,不差分
+ *
+ * 退出码:0 = 零差异;1 = 有差异或出错(可直接进 CI / 部署脚本)。
+ */
+import { NestFactory } from '@nestjs/core';
+import { Prisma } from '@prisma/client';
+import { lookupDxTreatment } from '@pac/types';
+import { AppModule } from '../app.module';
+import { PrismaService } from '../prisma/prisma.service';
+import { TreatmentInitiationRecallScenario } from '../modules/plan/engine/scenarios/treatment-initiation-recall.scenario';
+import type { ScenarioScope } from '../modules/plan/engine/scenario.interface';
+import type { GapVariant } from '../modules/clinical-gap/potential-treatment-gap.sql';
+
+interface Args {
+  host: string;
+  sub?: string;
+  self: boolean;
+  bench: boolean;
+  samples: number;
+}
+
+function parseArgs(argv: string[]): Args {
+  const a: Args = { host: 'demo', self: false, bench: false, samples: 20 };
+  for (const s of argv) {
+    if (s.startsWith('--host=')) a.host = s.slice('--host='.length);
+    else if (s.startsWith('--sub=')) a.sub = s.slice('--sub='.length);
+    else if (s.startsWith('--samples=')) a.samples = Number(s.slice('--samples='.length)) || 20;
+    else if (s === '--self') a.self = true;
+    else if (s === '--bench') a.bench = true;
+  }
+  return a;
+}
+
+interface DiffRow {
+  side: string;
+  patient_id: string;
+  signal_fact_id: string;
+  tooth: string | null;
+}
+
+async function bootstrap(): Promise<number> {
+  const args = parseArgs(process.argv.slice(2));
+  // 报告一律走 console:Nest 的 logger 级别会被 createApplicationContext 全局压掉,
+  // 而这个 CLI 的输出就是它的全部产物 —— 不能被日志级别吃掉。
+  const out = (m: string): void => console.log(m);
+  const bad_ = (m: string): void => console.error(m);
+  const app = await NestFactory.createApplicationContext(AppModule, {
+    logger: ['warn', 'error'],
+  });
+
+  let bad = 0;
+  try {
+    const prisma = app.get(PrismaService);
+    const scenario = app.get(TreatmentInitiationRecallScenario);
+
+    const host = await prisma.host.findUnique({ where: { name: args.host } });
+    if (!host) throw new Error(`Host '${args.host}' not found`);
+    const tenants = await prisma.patient.findMany({
+      where: { hostId: host.id },
+      select: { tenantId: true },
+      distinct: ['tenantId'],
+    });
+    if (!tenants.length) throw new Error('No tenants found for host');
+
+    // 🔴 now 固定一次:两版必须拿同一个时间锚,否则 cooldown 边界上的信号会来回抖。
+    const now = new Date();
+
+    const entries = Object.entries(TreatmentInitiationRecallScenario.SUB_SCENARIOS).filter(
+      ([k]) => !args.sub || k === args.sub,
+    );
+    if (!entries.length) throw new Error(`--sub=${args.sub} 不是已知子场景`);
+
+    for (const t of tenants) {
+      const scope: ScenarioScope = { hostId: host.id, tenantId: t.tenantId, now };
+      out(`▶ host=${args.host} tenant=${t.tenantId} 子场景=${entries.length} 个`);
+
+      for (const [subKey, cfg] of entries) {
+        const rule = lookupDxTreatment(cfg.primaryCode);
+        if (!rule) throw new Error(`${subKey}.primaryCode=${cfg.primaryCode} 不在 DiagnosisTreatmentMap`);
+        const leftVariant: GapVariant = 'legacy';
+        const rightVariant: GapVariant = args.self ? 'legacy' : 'setbased';
+        const left = scenario.buildScenarioSql(scope, cfg.primaryCode, rule, leftVariant);
+        const right = scenario.buildScenarioSql(scope, cfg.primaryCode, rule, rightVariant);
+
+        // ── 计时:两版各单独跑一次(不在同一事务里,避免第二次白蹭第一次的缓存判断失真;
+        //    对拍另开一个事务)。work_mem 与线上一致。
+        const timeOne = async (sql: Prisma.Sql): Promise<{ ms: number; rows: number }> => {
+          const t0 = Date.now();
+          const [, rows] = await prisma.$transaction([
+            prisma.$executeRaw`SET LOCAL work_mem = '256MB'`,
+            prisma.$queryRaw<{ n: bigint }[]>(
+              Prisma.sql`SELECT count(*)::bigint AS n FROM (${sql}) q`,
+            ),
+          ]);
+          return { ms: Date.now() - t0, rows: Number(rows[0]?.n ?? 0) };
+        };
+        const l = await timeOne(left);
+        const r = await timeOne(right);
+        const speed = r.ms > 0 ? (l.ms / r.ms).toFixed(2) : 'n/a';
+        out(
+          `  ${subKey.padEnd(24)} ${leftVariant}=${String(l.ms).padStart(7)}ms/${l.rows}行  ` +
+            `${rightVariant}=${String(r.ms).padStart(7)}ms/${r.rows}行  ×${speed}`,
+        );
+        if (args.bench) continue;
+
+        // ── 差分:两版同一快照,双向 EXCEPT ALL(ALL 保留重复,行数不一致也会暴露)──
+        const key = Prisma.raw('patient_id, signal_fact_id, tooth');
+        const diffSql = Prisma.sql`
+          WITH lg AS (${left}), sb AS (${right}),
+          d1 AS (SELECT ${key} FROM lg EXCEPT ALL SELECT ${key} FROM sb),
+          d2 AS (SELECT ${key} FROM sb EXCEPT ALL SELECT ${key} FROM lg)
+          SELECT 'only_legacy'::text AS side, ${key} FROM d1
+          UNION ALL
+          SELECT 'only_setbased'::text AS side, ${key} FROM d2`;
+        const diffs = await prisma.$transaction(
+          async (tx) => {
+            await tx.$executeRaw`SET TRANSACTION ISOLATION LEVEL REPEATABLE READ`;
+            await tx.$executeRaw`SET LOCAL work_mem = '256MB'`;
+            return tx.$queryRaw<DiffRow[]>(diffSql);
+          },
+          { timeout: 60 * 60 * 1000, maxWait: 60_000 },
+        );
+
+        if (diffs.length === 0) {
+          out(`  ${subKey.padEnd(24)} ✅ 零差异`);
+        } else {
+          bad++;
+          const onlyL = diffs.filter((d) => d.side === 'only_legacy').length;
+          const onlyR = diffs.filter((d) => d.side === 'only_setbased').length;
+          bad_(
+            `  ${subKey.padEnd(24)} ❌ 差异 ${diffs.length} 条(只在 legacy=${onlyL} / 只在 setbased=${onlyR})`,
+          );
+          for (const d of diffs.slice(0, args.samples)) {
+            bad_(`      ${d.side} patient=${d.patient_id} sig=${d.signal_fact_id} tooth=${d.tooth ?? 'NULL'}`);
+          }
+        }
+      }
+    }
+  } catch (e) {
+    bad_(e instanceof Error ? (e.stack ?? e.message) : String(e));
+    bad++;
+  } finally {
+    await app.close();
+  }
+
+  if (bad === 0) {
+    console.log('══ 全部零差异 ══');
+    return 0;
+  }
+  console.error(`══ ${bad} 个子场景不一致 / 出错 ══`);
+  return 1;
+}
+
+bootstrap().then((code) => process.exit(code));

apps/pac-service/src/modules/clinical-gap/potential-treatment.selector.ts

 import { Injectable } from '@nestjs/common';
+import { Prisma } from '@prisma/client';
 import { lookupDxTreatment, resolverCategoriesFor } from '@pac/types';
 import { PrismaService } from '../../prisma/prisma.service';
 import {
   buildGapCore,
   GAP_FLAGS_BY_PRIMARY,
   GAP_PRIMARY_GROUPS,
+  gapVariant,
 } from './potential-treatment-gap.sql';
 
 /**
@@ -43,21 +45,23 @@ export class PotentialTreatmentSelector {
       if (!rule) continue;
       const resolverCats = resolverCategoriesFor(primaryCode) as readonly string[];
       const cfgFlags = GAP_FLAGS_BY_PRIMARY[primaryCode] ?? {};
-      const gap = buildGapCore({ rule, cfgFlags, allCodes, resolverCats });
+      const gap = buildGapCore({ rule, cfgFlags, allCodes, resolverCats, variant: gapVariant() });
 
-      const rows = await this.prisma.$queryRaw<RawGapRow[]>`
-        SELECT
+      // 投影列(两形态共用;tooth 单列,取法不同)
+      const projection = Prisma.sql`
           sig.id                       AS fact_id,
           sig.content->>'code'         AS code,
           sig.content->>'name_zh'      AS name_zh,
           sig.type                     AS signal_type,
-          ${gap.toothOutput}           AS tooth,
           sig.content->>'confidence'   AS confidence,
           EXTRACT(DAY FROM ${now}::timestamptz - COALESCE(sig.occurred_at, sig.planned_for))::int AS days_since,
-          COALESCE(sig.occurred_at, sig.planned_for) AS anchor_at
+          COALESCE(sig.occurred_at, sig.planned_for) AS anchor_at`;
+      // ⚠️ 画像是**逐患者**调用(全量 54.7 万次),这里的 scope 恒为 1 个患者 ——
+      //   gap_scope 只有一行,各分支走 (patient_id, type, status) 索引,形态不会退化成全表扫。
+      const queryBody = (joinAddon: Prisma.Sql, gapAddon: Prisma.Sql): Prisma.Sql => Prisma.sql`
         FROM patients p
         JOIN patient_facts sig ON sig.patient_id = p.id
-        ${gap.lateralJoin}
+        ${joinAddon}
         WHERE p.host_id = ${hostId}::uuid
           AND p.tenant_id = ${tenantId}
           AND p.id = ${patientId}::uuid
@@ -68,8 +72,26 @@ export class PotentialTreatmentSelector {
           AND COALESCE(sig.occurred_at, sig.planned_for) IS NOT NULL
           ${gap.restorationIneligibleFrag}
           ${gap.congenitalFrag}
-          ${gap.gapWhere}
-      `;
+          ${gapAddon}`;
+
+      const sb = gap.setBased;
+      const sql = sb
+        ? Prisma.sql`
+        WITH gap_cand AS MATERIALIZED (
+          SELECT ${projection}${sb.candExtraCols}
+          ${queryBody(Prisma.empty, sb.candWhere)}
+        )${sb.postCtes}
+        SELECT c.fact_id, c.code, c.name_zh, c.signal_type, c.confidence, c.days_since, c.anchor_at,
+               ${sb.toothOutput} AS tooth
+        FROM gap_cand c
+        ${sb.remJoin}
+        WHERE TRUE ${sb.outerWhere}`
+        : Prisma.sql`
+        SELECT ${projection},
+               ${gap.toothOutput}     AS tooth
+        ${queryBody(gap.lateralJoin, gap.gapWhere)}`;
+
+      const rows = await this.prisma.$queryRaw<RawGapRow[]>(sql);
       for (const r of rows) {
         out.push({
           primaryCode,

apps/pac-service/tests/gap-setbased-parity.spec.ts

+/**
+ * gap 集合式形态的**结构对拍**(纯 SQL 文本层,不连库)
+ *
+ * 定位:数据层的等价性由 `pnpm verify-gap-equivalence`(逐 患者×信号×牙位 差分)证明,
+ *   本 spec 只守一件单元测试能守住的事 —— **两种形态的分支集合不许走散**。
+ *   典型事故:后来人给 legacy 加了第 14 条 resolved 分支,忘了同步 setbased →
+ *   线上悄悄少销一类证据 → 静默多召 / 少召。那种漏法 tsc 和现有 spec 全都发现不了,
+ *   但分支计数会当场炸。
+ *
+ * ⚠️ 本 spec 断言的是"两边都改了",不是"改对了"。改完仍必须跑 verify-gap-equivalence。
+ */
+import { lookupDxTreatment, resolverCategoriesFor } from '@pac/types';
+import {
+  buildGapCore,
+  GAP_FLAGS_BY_PRIMARY,
+  GAP_PRIMARY_GROUPS,
+} from '../src/modules/clinical-gap/potential-treatment-gap.sql';
+
+const PRIMARY_CODES = Object.keys(GAP_PRIMARY_GROUPS);
+const WHOLE_MOUTH = ['K05', 'K07'];
+
+function core(primaryCode: string, variant: 'legacy' | 'setbased') {
+  const rule = lookupDxTreatment(primaryCode);
+  if (!rule) throw new Error(`no rule for ${primaryCode}`);
+  const grp = GAP_PRIMARY_GROUPS[primaryCode];
+  return buildGapCore({
+    rule,
+    cfgFlags: GAP_FLAGS_BY_PRIMARY[primaryCode] ?? {},
+    allCodes: [...grp.dxCodes, ...grp.recCodes],
+    resolverCats: resolverCategoriesFor(primaryCode) as readonly string[],
+    variant,
+  });
+}
+
+const count = (hay: string, needle: RegExp): number => (hay.match(needle) ?? []).length;
+
+describe('gap 集合式 ↔ 逐行形态:结构对拍', () => {
+  it('variant 默认 legacy;只有显式 setbased 才产出集合式拼装件', () => {
+    const g = core('K08', 'legacy');
+    expect(g.setBased).toBeUndefined();
+    expect(core('K08', 'setbased').setBased).toBeDefined();
+  });
+
+  describe.each(PRIMARY_CODES)('%s', (code) => {
+    const isWhole = WHOLE_MOUTH.includes(code);
+
+    it('全口码不进集合式(原样走 legacy),牙位码必须有 resolved 预聚合链', () => {
+      const g = core(code, 'setbased');
+      if (isWhole) {
+        // 全口场景 legacy 的 lateral 本来就会被 PG 的 useless-left-join removal 摘掉 →
+        // 集合式零收益;实测硬套进来还慢 2~3 倍(见 buildGapSetBased 里的早退注释)。
+        expect(g.setBased).toBeUndefined();
+      } else {
+        const sb = g.setBased!;
+        expect(sb.postCtes.sql).toContain('gap_scope');
+        expect(sb.postCtes.sql).toContain('gap_resolved');
+        expect(sb.postCtes.sql).toContain('gap_rem');
+        expect(sb.remJoin.sql).toContain('LEFT JOIN gap_rem');
+      }
+    });
+
+    if (!WHOLE_MOUTH.includes(code)) {
+      it('两种形态的 resolved 分支条数必须一致(加分支只改一边 = 静默错召)', () => {
+        const legacy = core(code, 'legacy').lateralJoin.sql;
+        const setbased = core(code, 'setbased').setBased!.postCtes.sql;
+        // legacy 分支用裸 UNION 分隔;setbased 用 UNION ALL(先聚合后去重,不需要 UNION 的排序去重)
+        const legacyBranches = count(legacy, /\bUNION\b(?!\s+ALL)/g) + 1;
+        const setBranches = count(setbased, /\bUNION ALL\b/g) + 1;
+        expect(setBranches).toBe(legacyBranches);
+      });
+
+      it('每个分支都挂了患者收窄(漏一个就全表扫 34GB patient_facts)', () => {
+        const sql = core(code, 'setbased').setBased!.postCtes.sql;
+        const branches = sql
+          .slice(sql.indexOf('FROM ('), sql.indexOf(') sb('))
+          .split(/\bUNION ALL\b/);
+        expect(branches.length).toBeGreaterThan(1);
+        for (const b of branches) {
+          expect(b).toContain('IN (SELECT patient_id FROM gap_scope)');
+        }
+      });
+
+      it('gate 列永不为 NULL —— 时间门分支带 IS NOT NULL,无门分支写死 infinity', () => {
+        const sql = core(code, 'setbased').setBased!.postCtes.sql;
+        const branches = sql
+          .slice(sql.indexOf('FROM ('), sql.indexOf(') sb('))
+          .split(/\bUNION ALL\b/);
+        for (const b of branches) {
+          const ungated = b.includes(`'infinity'::timestamptz AS gate`);
+          const gated = /IS NOT NULL/.test(b);
+          // 二者必居其一:否则全 NULL 组会被 max() 聚成 NULL、当成"无门恒过"→ 误销 → 静默少召
+          expect(ungated || gated).toBe(true);
+        }
+      });
+
+      it('严格 > 只出现在「更晚结构诊断」一条分支上', () => {
+        const sql = core(code, 'setbased').setBased!.postCtes.sql;
+        expect(count(sql, /TRUE AS strict/g)).toBe(1);
+        expect(sql).toContain('CASE WHEN r.strict THEN r.gate > c.gap_anchor ELSE r.gate >= c.gap_anchor END');
+      });
+
+      it('牙位顺序与重复原样保留(tooth 串会落进 plan_reasons 给客服看)', () => {
+        const sql = core(code, 'setbased').setBased!.postCtes.sql;
+        expect(sql).toContain('WITH ORDINALITY');
+        expect(sql).toContain('array_agg(u.x ORDER BY u.ord)');
+      });
+
+      it('gap_rem 无行要补空数组,不能留 NULL', () => {
+        const sb = core(code, 'setbased').setBased!;
+        expect(sb.toothOutput.sql).toContain("COALESCE(gap_rem.remaining_teeth, ARRAY[]::text[])");
+        expect(sb.outerWhere.sql).toContain("COALESCE(gap_rem.remaining_teeth, ARRAY[]::text[])");
+      });
+    }
+  });
+
+  it('病历号等值相关(整颌活动义齿)只在 K08 出现,且走 enc 列而非时间门', () => {
+    const k08 = core('K08', 'setbased').setBased!.postCtes.sql;
+    expect(k08).toContain("adx.content->>'emr_external_id' AS enc");
+    expect(k08).toContain("r.enc IS NULL OR r.enc = c.gap_sig_enc");
+    const k02 = core('K02', 'setbased').setBased!.postCtes.sql;
+    expect(k02).not.toContain("AS enc,\n             FALSE AS strict");
+    expect(count(k02, /adx\./g)).toBe(0);
+  });
+
+  it('「建议优先于诊断」分支只对 diagnosis_record 信号生效(ndx 标志)', () => {
+    const sql = core('K08', 'setbased').setBased!.postCtes.sql;
+    expect(count(sql, /TRUE AS ndx/g)).toBe(1);
+    expect(sql).toContain("NOT r.ndx OR c.gap_sig_type = 'diagnosis_record'");
+  });
+
+  it('牙位级规则若被加上 excludeIfEverTreated,集合式必须直接炸而不是静默算错', () => {
+    const rule = { ...lookupDxTreatment('K08')!, excludeIfEverTreated: true };
+    expect(() =>
+      buildGapCore({
+        rule,
+        cfgFlags: GAP_FLAGS_BY_PRIMARY.K08,
+        allCodes: ['K08'],
+        resolverCats: resolverCategoriesFor('K08') as readonly string[],
+        variant: 'setbased',
+      }),
+    ).toThrow(/excludeIfEverTreated/);
+  });
+});