Merge DW EMR compatibility and lag alert fix into test

apps/pac-service/data/jvs-dw/assemblers/emr.yaml

@@ -48,3 +48,11 @@ join_arrays:
     foreign_key: emr_id
     same_source_row: true
     field_mapping: { toothPosition: tooth_position, message: treat_name }
+
+# 旧牙位编辑器会留下空元素/仅牙位元素，无正文不等于整份病历无效。
+# 非空元数据、未知字段和错误 message 类型仍拒绝；原始内容保留在 rawPayload。
+empty_narrative_entries:
+  examFindings:
+    empty_metadata_keys: &empty_editor_keys [englishValTopRight, englishVal, englishValBotRight, englishValBotLeft]
+  disposal:
+    empty_metadata_keys: *empty_editor_keys

apps/pac-service/src/modules/sync/assembler/assembler.schema.ts

@@ -111,6 +111,11 @@ export const AssemblerConfigSchema = z.object({
   /// 主表字段 → canonical 字段映射(canonical key ← host field)
   field_mapping: fieldMapping,
 
+  /// 显式兼容无正文的病历数组元素；仅白名单元数据为空时补 message:null。
+  empty_narrative_entries: z.record(z.string(), z.object({
+    empty_metadata_keys: z.array(z.string().min(1)),
+  })).optional(),
+
   /// 主表字段的枚举映射(key 是 canonical 字段名)
   enum_mapping: z.record(z.string(), enumMappingForField).optional(),
 

apps/pac-service/src/modules/sync/assembler/empty-narrative.ts

+/** 宿主在 assembler 显式声明才启用。保留牙位，绝不把未知字段/非空编辑器内容吞成空正文。 */
+export function normalizeEmptyNarrative(value: unknown, emptyMetadataKeys: string[]): unknown {
+  if (!Array.isArray(value)) return value;
+  const allowed = new Set(['toothPosition', ...emptyMetadataKeys]);
+  return value.map(entry => {
+    if (!entry || typeof entry !== 'object' || Array.isArray(entry) ||
+        Object.prototype.hasOwnProperty.call(entry, 'message')) return entry;
+    if (Object.keys(entry).some(key => !allowed.has(key))) return entry;
+    if (emptyMetadataKeys.some(key => entry[key] != null && entry[key] !== '' &&
+      !(Array.isArray(entry[key]) && entry[key].length === 0))) return entry;
+    return { ...entry, message: null };
+  });
+}

apps/pac-service/src/queues/dw-lag-monitor.service.ts

@@ -7,7 +7,7 @@ import { ColdImportService } from '../modules/sync/cold-import/cold-import.servi
 /**
  * DwLagMonitorService — 每小时检查 DW 增量数据滞后(W4 末)
  *
- * 输入:sync_logs 中最近一条成功的 incremental_bundle 的 cursor_after JSON
+ * 输入:sync_logs 中最近一条成功或部分成功的 incremental_bundle 的 cursor_after JSON
  *       例:{"fact_emr_treatment_out":"2026-05-26 15:30:00",...}
  *
  * 计算:max(cursor_after.*) vs now() → diff hours
@@ -62,7 +62,8 @@ export class DwLagMonitorService {
         where: {
           hostId: host.id,
           resource: 'incremental_bundle',
-          status: 'success',
+          // 与实际增量读取水位一致：partial 也已推进游标。
+          status: { in: ['success', 'partial'] },
           cursorAfter: { not: null },
         },
         orderBy: { startedAt: 'desc' },
@@ -71,6 +72,13 @@ export class DwLagMonitorService {
         this.logger.warn(`dw-lag: host=${host.name} 还没跑过增量(无 cursor)— 检查 sync-incremental cron`);
         continue;
       }
+      if (last.status === 'partial' && last.failed > 0) {
+        await this.alerter.send({
+          level: 'warning', title: 'DW 同步部分失败',
+          body: `最近一轮有 ${last.failed} 条处理失败；游标已推进不代表数据完整，请核查失败记录并定向补处理。`,
+          context: { host: host.name, syncLogId: last.id, failed: last.failed, startedAt: last.startedAt.toISOString() },
+        });
+      }
       let cursors: Record<string, string>;
       try {
         cursors = JSON.parse(last.cursorAfter) as Record<string, string>;
@@ -83,6 +91,10 @@ export class DwLagMonitorService {
           .map((v) => new Date(v).getTime())
           .filter((t) => !isNaN(t)),
       );
+      if (!Number.isFinite(maxCursorTs)) {
+        this.logger.warn(`dw-lag: host=${host.name} 无有效游标时间，无法判断滞后`);
+        continue;
+      }
       const diffHours = (now - maxCursorTs) / 3_600_000;
       const tag =
         diffHours > errorH ? '🔴' :

apps/pac-service/tests/dw-lag-partial.spec.ts

+import { DwLagMonitorService } from '../src/queues/dw-lag-monitor.service';
+import { Logger } from '@nestjs/common';
+beforeAll(() => Logger.overrideLogger(false));
+afterAll(() => Logger.overrideLogger(['log', 'warn', 'error']));
+async function check(age: number, status = 'partial', invalid = false) {
+  const latest = { id: 'run', status, failed: status === 'partial' ? 3 : 0, startedAt: new Date(),
+    cursorAfter: JSON.stringify({ emr: invalid ? 'invalid' : new Date(Date.now() - age * 3600000).toISOString() }) };
+  const old = { ...latest, status: 'success', failed: 0, cursorAfter: JSON.stringify({ emr: new Date(Date.now() - 25 * 3600000).toISOString() }) };
+  const prisma = { host: { findMany: jest.fn().mockResolvedValue([{ id: 'host', name: 'jvs-dw' }]) },
+    syncLog: { findFirst: jest.fn(async ({ where }) => where.status === 'success' ? old : latest) } };
+  const alert = { send: jest.fn().mockResolvedValue(undefined) };
+  const config = { getHostOpsConfig: () => ({ hasSqlSource: true, pushLagAlert: false }) };
+  await new DwLagMonitorService(prisma as any, alert as any, config as any).checkLag();
+  return { prisma, alert };
+}
+test('新 partial 替代旧 success，报告部分失败而非假滞后', async () => {
+  const { prisma, alert } = await check(3);
+  expect(prisma.syncLog.findFirst.mock.calls[0]![0].where.status).toEqual({ in: ['success', 'partial'] });
+  expect(alert.send.mock.calls.map(x => x[0].title)).toEqual(['DW 同步部分失败']);
+});
+test('partial 游标真的旧仍报滞后，失败不被掩盖', async () => {
+  const { alert } = await check(30);
+  expect(alert.send.mock.calls.map(x => x[0].title)).toEqual(['DW 同步部分失败', 'DW 数据滞后(注意)']);
+});
+test('success 正常不报警', async () => expect((await check(3, 'success')).alert.send).not.toHaveBeenCalled());
+test('无有效时间不抛 RangeError', async () => await expect(check(3, 'success', true)).resolves.toBeDefined());

apps/pac-service/tests/empty-narrative.spec.ts

+import { normalizeEmptyNarrative } from '../src/modules/sync/assembler/empty-narrative';
+import { EmrNarrativeEntrySchema } from '@pac/types';
+test.each([{ value: '不能丢的正文' }, { englishVal: ['nonempty'] }, { message: 7 }, { toothPosition: 7 }, null, 'text'])(
+  '非空或未知结构不能变为空正文 %p', entry => {
+    const [result] = normalizeEmptyNarrative([entry], ['englishVal']) as unknown[];
+    expect(EmrNarrativeEntrySchema.safeParse(result).success).toBe(false);
+  });
+test('字符串保持不变', () => expect(normalizeEmptyNarrative('正常文字', [])).toBe('正常文字'));

apps/pac-service/tests/emr-normalization.spec.ts

@@ -137,3 +137,18 @@ test('话术同时保留一般情况、辅助检查和未分类治疗原文，
   expect(text).toContain('治疗计划:复查后评估');
   expect(text).not.toContain('"message"');
 });
+
+test.each(['examine', 'dispose'])('5i5ya %s 兼容空旧编辑器、牙位和空对象，保留有效正文与原文', field => {
+  const entries = [{ englishVal: [], englishValTopRight: [], englishValBotRight: [], englishValBotLeft: [] },
+    { toothPosition: '16' }, {}, { toothPosition: '26', message: '原文' }];
+  const raw = { id: 'E-empty', patient_id: 'P1', organization_id: 'C1', brand: '瑞尔',
+    rq: '2026-10-17', created_date: '2026-10-17 12:00:00', updated_date: '2026-10-17 18:32:38',
+    [field]: JSON.stringify(entries) };
+  const saved = JSON.stringify(raw);
+  const { canonical, rawSource } = replay('jvs-dw', raw, 'fact_emr_treatment_out');
+  expect(canonical[field === 'examine' ? 'examFindings' : 'disposal']).toEqual([
+    { message: null }, { toothPosition: '16', message: null }, { message: null },
+    { toothPosition: '26', message: '原文' },
+  ]);
+  expect(JSON.stringify(rawSource)).toBe(saved);
+});

docs/qa/2026-10-17-dw-emr-alert-fix.md

+# 5i5ya 空病历元素兼容与游标告警修复
+
+## 问题与边界
+
+增量任务允许 partial 并继续使用其 cursorAfter，但滞后监控只读取 success，导致最近任务仅有少量坏记录时，误用更早成功任务的游标。监控现在与读取水位一致，并单独报告部分失败；游标推进不等于数据完整。召回计算耗时与并发策略不在本次修改范围。
+
+5i5ya 的 examine / dispose 存在缺少 message 的空元素：仅牙位，或旧编辑器的空数组。仅在 jvs-dw assembler 显式启用兼容，将这些元素的正文补为 null，保留牙位及完整原始记录。未知字段、非空编辑器内容、错误类型仍拒绝，不放宽通用病历协议。
+
+## 本地验证
+
+- 5 个相关测试套件，40 项通过；共享包构建、Prisma 生成、服务类型检查和构建通过。
+- 使用源端以下 3 条实际记录，在临时隔离数据库通过真实 assembler → transaction → parser → fact writer 落库，均成功；再次重放全部 unchanged，没有新增事实版本。临时数据库已删除。
+- 未启动画像或召回计算，未改动宿主游标。远端补处理前保留对应交易与事实快照。
+
+| 源病历 ID | 兼容字段 |
+| --- | --- |
+| a0edf43927ab4a508e141b29c1b00071 | examine |
+| 2210d9c4664147cbaaf75455718c77bd | dispose |
+| 7e4f2f9f2d3343e79017cef97f6f3e06 | dispose |
+
+发布顺序：独立 feature 合入 test，测试环境定向重放验证后，同一 feature 合入 main；不将 test 合回 main。历史 partial 日志保留原样，因此补处理成功后，部分失败告警仍可能指向旧任务，直到下一轮正常增量完成。