fix(tenant): dispatcher source_unit 串通 + reconcile 守卫升级(零碎待办)

- pipeline-dispatcher:加 identityNamespaceField,patient upsert/index/patientIdResolver
  全按 (source_unit,external_id) 复合键(pull/push/reconcile 多命名空间 host 也正确;
  单命名空间默认 '' 不变)
- reconcile 守卫:除 distinct tenant 外也查 distinct source_unit —— 合集团后 jvs-dw
  只剩 1 tenant 但 2 source_unit,旧守卫拦不住会撞号;现多命名空间一律转 cold-import

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/pac-service/src/modules/sync/pipeline/pipeline-dispatcher.service.ts

@@ -8,6 +8,17 @@ import { ParserPipeline } from './parser-pipeline.service';
 import type { EmitsConfig } from '../assembler/assembler.schema';
 import { mergePatientPreferences } from './patient-preferences.util';
 
+/** 患者索引 key:集团内 external_id 跨 source_unit 撞号,按 (source_unit, external_id) 建键。 */
+function patientIndexKey(sourceUnit: string, externalId: string): string {
+  return `${sourceUnit}#${externalId}`;
+}
+/** 从一行(canonical 或 raw)按命名空间字段取 source_unit;字段未声明 → ''(单命名空间)。 */
+function resolveSourceUnit(row: Record<string, unknown>, field?: string): string {
+  if (!field) return '';
+  const v = row[field];
+  return v === undefined || v === null ? '' : String(v).trim();
+}
+
 /**
  * PipelineDispatcher — "canonical tables → transaction + fact" 的共享内核。
  *
@@ -47,6 +58,9 @@ export class PipelineDispatcher {
     tables: Record<string, unknown[]>;
     hostId: string;
     tenantId: string;
+    /** 患者命名空间字段(集团模型,= manifest.identity_namespace_field;如 'brand')。
+     *  不传 = 单命名空间 host,source_unit=''。多品牌 host(jvs-dw reconcile)必传,否则撞号。 */
+    identityNamespaceField?: string;
     /** 用于合成 source_event_id;**跨 run 稳定**否则永远不重复 */
     sourceContext: string;
     syncLogId?: string;
@@ -67,7 +81,8 @@ export class PipelineDispatcher {
     const patientRows = (input.tables.patients ?? []) as Record<string, unknown>[];
     for (const row of patientRows) {
       try {
-        const patient = await this.upsertPatient(row, input.hostId, input.tenantId);
+        const sourceUnit = resolveSourceUnit(row, input.identityNamespaceField);
+        const patient = await this.upsertPatient(row, input.hostId, input.tenantId, sourceUnit);
         m.patientsUpserted++;
         // 患者主档变更也算"事件",触发 persona 重算(do_not_contact_status 等画像依赖它)
         if (patient?.id) touchedPatients.add(patient.id);
@@ -104,6 +119,7 @@ export class PipelineDispatcher {
           __source_row?: Record<string, unknown>;
         } & Record<string, unknown>;
         const rawRow = explicitSource ?? canonicalRow;
+        const sourceUnit = resolveSourceUnit(rawRow, input.identityNamespaceField);
         const txn = this.synthesizer.synthesize({
           rawRow,
           canonicalRow,
@@ -111,7 +127,7 @@ export class PipelineDispatcher {
           resource,
           hostId: input.hostId,
           tenantId: input.tenantId,
-          patientIdResolver: (extId) => patientIndex.get(extId),
+          patientIdResolver: (extId) => patientIndex.get(patientIndexKey(sourceUnit, extId)),
           sourceContext: input.sourceContext,
         });
         if (!txn) {
@@ -214,6 +230,7 @@ export class PipelineDispatcher {
     row: Record<string, unknown>,
     hostId: string,
     tenantId: string,
+    sourceUnit: string,
   ): Promise<{ id: string }> {
     const externalId = String(row.externalId ?? '').trim();
     if (!externalId) throw new Error('patient row 缺 externalId');
@@ -241,10 +258,9 @@ export class PipelineDispatcher {
         row.referralAmount != null ? Math.round(Number(row.referralAmount) * 100) : null,
     };
 
-    // sourceUnit:''(过渡)— 现 tenant 仍按命名空间隔离;真填值随 tenant→集团 remap 一起做
     const patient = await this.prisma.patient.upsert({
-      where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit: '', externalId } },
-      create: { hostId, tenantId, externalId, ...patientData },
+      where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId } },
+      create: { hostId, tenantId, sourceUnit, externalId, ...patientData },
       update: patientData,
     });
     await this.prisma.patientProfile.upsert({
@@ -261,9 +277,10 @@ export class PipelineDispatcher {
   ): Promise<Map<string, string>> {
     const rows = await this.prisma.patient.findMany({
       where: { hostId, tenantId },
-      select: { id: true, externalId: true },
+      select: { id: true, externalId: true, sourceUnit: true },
     });
-    return new Map(rows.map((r) => [r.externalId, r.id]));
+    // 集团内 external_id 跨 source_unit 撞号 → 索引按 (source_unit, external_id) 建键
+    return new Map(rows.map((r) => [patientIndexKey(r.sourceUnit, r.externalId), r.id]));
   }
 }
 

apps/pac-service/src/modules/sync/reconcile/reconcile.orchestrator.ts

@@ -47,16 +47,24 @@ export class ReconcileOrchestrator {
     //     · 多 tenant(如 jvs-dw 瑞尔+瑞泰)→ 单 tenant dispatcher 无法按行派 → 拒绝,指向 `pnpm sync`
     //       (ColdImportService 走 buildTenantResolver 按 brand 派,是多 tenant 的正确补数路径)
     //     · 空 host(尚无患者)→ 退化到 demo/mock 占位(首次导入边界,reconcile 极少在空 host 上跑)
+    // 集团模型:多命名空间维度从 tenant 移到了 source_unit。守卫要同时查 tenant 和 source_unit ——
+    //   否则 jvs-dw(1 tenant=ruier-grp,2 source_unit=瑞尔/瑞泰)会绕过守卫、走单命名空间 dispatcher 撞号。
     const hostTenants = await this.prisma.patient.findMany({
       where: { hostId: host.id },
       distinct: ['tenantId'],
       select: { tenantId: true },
     });
-    if (hostTenants.length > 1) {
+    const hostSourceUnits = await this.prisma.patient.findMany({
+      where: { hostId: host.id },
+      distinct: ['sourceUnit'],
+      select: { sourceUnit: true },
+    });
+    if (hostTenants.length > 1 || hostSourceUnits.length > 1) {
       throw new BadRequestException(
-        `host=${host.name} 存在多个 tenant(${hostTenants.map((t) => t.tenantId).join(', ')}),` +
-          `ReconcileOrchestrator 走单 tenant dispatcher,无法按行派 tenant → 会写错 tenant、造重复患者。` +
-          `请改用 \`pnpm sync -- --dir=<manifest_dir>\`(ColdImportService 按 brand 派 tenant)做时间窗补数。`,
+        `host=${host.name} 存在多个 tenant/命名空间(tenant=${hostTenants.map((t) => t.tenantId).join(',')};` +
+          `source_unit=${hostSourceUnits.map((s) => s.sourceUnit).join(',')}),` +
+          `ReconcileOrchestrator 走单命名空间 dispatcher,无法按行派 → 会撞号、造重复患者。` +
+          `请改用 \`pnpm sync -- --dir=<manifest_dir>\`(ColdImportService 按行派 tenant + source_unit)做时间窗补数。`,
       );
     }
     const tenantId =