feat(tenant): P0-2 摄入侧 source_unit 真填值 + manifest 切集团 slug

- manifest.schema: 加 identity_namespace_field
- tenant-resolver: 加 buildSourceUnitResolver(从 identity_namespace_field 解析)
- cold-import: source_unit 从 rawSource 解析,串入 patient upsert / stub /
  patientIndex(按 (source_unit,external_id) 复合键防撞号)/ patientIdResolver
- manifest.yaml: 删写死 tenant_map UUID,改 tenant_id: ruier-grp(集团 slug)+
  identity_namespace_field: brand
- 加数据迁移脚本(未执行)

⚠️ 前置审发现 patient_facts.subject_id(撞631)、patient_return_visits.external_id
(撞611)也是品牌级,合 tenant 会撞键 — 这俩也需 source_unit。迁移待补这两表后再跑。

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

apps/pac-service/data/jvs-dw/manifest.yaml

@@ -42,17 +42,12 @@ host_name:     jvs-dw
 display_name:  瑞尔集团 DW(单 host 多 brand)
 auto_sync:     true   # 纳入每日自动增量同步(scheduler 自动发现;取代 PAC_INCREMENTAL_HOSTS env)
 
-# ── tenant 路由(临时方案,W3 末)──
-# 理想态:DW 所有 ADS 表都带 tenant_id UUID 字段,PAC 直接 tenant_field=tenant_id pass-through。
-# 现状:DW 只有 fact_emr_treatment_out 一张表带 tenant_id,其他表只有 brand 中文字段。
-#   - 瑞尔 ↔ ba67e6cf30dc4f9c9c46adef188bbd04(DW 真实 host tenant UUID)
-#   - 瑞泰 ↔ 77057aed269f4a14957ae0ad0eff359a
-# 临时:yaml 维护 brand → DW UUID 映射,PAC 内部用 UUID 作 tenant_id(英文 / 跨表稳定)。
-# 后续:已反馈 DW 团队补全所有表的 tenant_id 字段;补完后改 tenant_field=tenant_id 删本映射。
-tenant_field:  brand
-tenant_map:
-  瑞尔: ba67e6cf30dc4f9c9c46adef188bbd04
-  瑞泰: 77057aed269f4a14957ae0ad0eff359a
+# ── 集团模型(tenant=集团 slug;品牌降为 source_unit 命名空间)──
+# tenant=集团:稳定 slug(各环境一致,不再写死 DW UUID);所有行归此集团。
+# 品牌(瑞尔/瑞泰)= patients.source_unit,经 identity_namespace_field 从 brand 列填,
+#   给 external_id 在集团内消歧(同号不同品牌=不同人,如 patient_id 261067 王辰/王文彦)。
+tenant_id:                ruier-grp   # 集团 = 租户(静态 slug)
+identity_namespace_field: brand       # source_unit ← 源 brand 字段(防撞号)
 
 amount_unit:   yuan
 timezone:      Asia/Shanghai

apps/pac-service/prisma/data-migrations/2026-06-27_tenant_to_group_jvs-dw.sql

+-- 一次性数据迁移(jvs-dw 专属):品牌级 tenant → 集团级 tenant。
+--
+-- 背景:租户模型改造 P0。把两个品牌 tenant(瑞尔 ba67e6cf… / 瑞泰 77057aed…)
+--       合并为一个集团 tenant slug 'ruier-grp',品牌降为 patients.source_unit。
+--
+-- 安全性(见 docs/tenant-group-org-model-refactor.md §2/§9):
+--   "换槽位不加信息" — 把 brand 从 tenant_id 槽搬到 source_unit 槽。
+--   472 撞号(同 external_id 不同品牌 = 不同人)由 source_unit 分开,不发生任何合并。
+--   顺序:先回填 source_unit(读旧 tenant_id),再 remap tenant_id。
+--
+-- 可逆:source_unit ↔ 旧 tenant 映射保留;回退即按 source_unit 反推旧 UUID。
+-- 幂等:WHERE 限定旧 UUID,重复执行不会二次改(旧 UUID 已不存在)。
+
+BEGIN;
+
+-- ① 回填 patients.source_unit(从旧品牌 tenant_id)
+UPDATE patients SET source_unit = '瑞尔' WHERE tenant_id = 'ba67e6cf30dc4f9c9c46adef188bbd04';
+UPDATE patients SET source_unit = '瑞泰' WHERE tenant_id = '77057aed269f4a14957ae0ad0eff359a';
+
+-- ② 14 张带 tenant_id 的表:品牌 UUID → 集团 slug 'ruier-grp'
+DO $$
+DECLARE t text;
+BEGIN
+  FOREACH t IN ARRAY ARRAY[
+    'agent_invocations','followup_plans','patient_facts','patient_relations',
+    'patient_return_visits','patient_transactions','patients','persona_recompute_logs',
+    'personas','plan_executions','plan_generation_logs','plan_scripts',
+    'plan_summaries','sync_logs'
+  ] LOOP
+    EXECUTE format(
+      'UPDATE %I SET tenant_id = ''ruier-grp'' WHERE tenant_id IN (%L, %L)',
+      t, 'ba67e6cf30dc4f9c9c46adef188bbd04', '77057aed269f4a14957ae0ad0eff359a'
+    );
+  END LOOP;
+END $$;
+
+COMMIT;

apps/pac-service/src/modules/sync/cold-import/cold-import.service.ts

@@ -35,9 +35,19 @@ import {
 } from './clickhouse-source.service';
 import { AlertService } from '../../../common/alerting/alert.service';
 import { TransformEngine } from '../transforms/transform-engine';
-import { buildTenantResolver, type TenantResolver } from './tenant-resolver';
+import {
+  buildTenantResolver,
+  buildSourceUnitResolver,
+  type TenantResolver,
+  type SourceUnitResolver,
+} from './tenant-resolver';
 import { mergePatientPreferences } from '../pipeline/patient-preferences.util';
 
+/** 患者索引 key:集团内 external_id 可跨 source_unit 撞号,索引按 (source_unit, external_id) 建键。 */
+function patientIndexKey(sourceUnit: string, externalId: string): string {
+  return `${sourceUnit}#${externalId}`;
+}
+
 /**
  * ColdImportService(v2 — AssemblerEngine 驱动)
  *
@@ -56,6 +66,9 @@ import { mergePatientPreferences } from '../pipeline/patient-preferences.util';
 @Injectable()
 export class ColdImportService {
   private readonly logger = new Logger(ColdImportService.name);
+  /// 当次 import 的 source_unit 解析器(从 manifest.identity_namespace_field 构建)。
+  /// cold-import 按 host 串行,实例字段安全;未声明命名空间字段时恒返回 ''。
+  private sourceUnitResolver: SourceUnitResolver = { resolve: () => '' };
 
   constructor(
     private readonly prisma: PrismaService,
@@ -100,6 +113,7 @@ export class ColdImportService {
     const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
     if (!host) throw new Error(`reparse: host '${opts.hostName}' not found`);
     const tenantResolver = buildTenantResolver(manifest);
+    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
     const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
     const subjectCfgs = this.loadAllAssemblers(absDir, manifest).filter((c) => !!c.emits);
     const targetCfgs = opts.subjectTypes?.length
@@ -278,6 +292,7 @@ export class ColdImportService {
     const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
     if (!host) throw new Error(`ingestRawTables: host '${opts.hostName}' not found`);
     const tenantResolver = buildTenantResolver(manifest);
+    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
     const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
     const transforms = manifest.transforms ?? [];
 
@@ -514,6 +529,7 @@ export class ColdImportService {
     }
     // v2.3:per-row tenant resolver(支持单 host 多 tenant)
     const tenantResolver = buildTenantResolver(manifest);
+    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
     const knownTenants = tenantResolver.knownTenants();
     const seenTenants = new Set<string>(); // 跑过程动态收集(pass-through 模式不预知)
     this.logger.log(
@@ -938,6 +954,7 @@ export class ColdImportService {
     }
 
     const tenantResolver = buildTenantResolver(manifest);
+    this.sourceUnitResolver = buildSourceUnitResolver(manifest);
     const seenTenants = new Set<string>();
 
     // 3. 单患者 SQL — 改写所有 queries 加 WHERE patient_id+brand
@@ -1260,6 +1277,8 @@ export class ColdImportService {
       const externalId = String(c.externalId).trim();
       // v2.3:per-row tenant 解析(从 host 真原文 rawSource 读 brand 字段)
       const tenantId = tenantResolver.resolve(rawSource);
+      // 集团模型:source_unit(发证命名空间)从同一 rawSource 解析,给 external_id 消歧
+      const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
       if (!tenantId) {
         stats.failed++;
         this.logger.warn(
@@ -1314,9 +1333,9 @@ export class ColdImportService {
             this.prisma.patient.upsert({
               where: {
                 // sourceUnit:''(过渡)— 真填值随 tenant→集团 remap 一起做
-                hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit: '', externalId },
+                hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId },
               },
-              create: { hostId, tenantId, externalId, ...patientData },
+              create: { hostId, tenantId, sourceUnit, externalId, ...patientData },
               update: patientData,
             }),
           'patient upsert',
@@ -1390,10 +1409,11 @@ export class ColdImportService {
         stats.failed++;
         continue;
       }
+      const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
-      const patientId = idx.get(patientExternalId);
+      const patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
       if (!patientId) continue; // 本人不在 PAC(非 active client)→ 无处挂靠,跳过
-      const relatedPatientId = idx.get(relatedExternalId) ?? null;
+      const relatedPatientId = idx.get(patientIndexKey(sourceUnit, relatedExternalId)) ?? null;
 
       try {
         await this.withDbRetry(
@@ -1472,8 +1492,9 @@ export class ColdImportService {
         stats.failed++;
         continue;
       }
+      const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
-      const patientId = idx.get(patientExternalId);
+      const patientId = idx.get(patientIndexKey(sourceUnit, patientExternalId));
       if (!patientId) continue; // 本人不在 PAC → skip
 
       const data = {
@@ -1588,11 +1609,12 @@ export class ColdImportService {
           patientIndex = await this.buildPatientIndex(hostId, tenantId);
           patientIndexByTenant.set(tenantId, patientIndex);
         }
-        // stub auto-create
+        const sourceUnit = this.sourceUnitResolver.resolve(rawSource);
+        // stub auto-create(索引/键带 source_unit,集团内 external_id 撞号消歧)
         const patientExternalId = canonicalRow.patientExternalId as string | undefined;
-        if (patientExternalId && !patientIndex.has(patientExternalId)) {
-          const stubId = await this.ensurePatientStub(hostId, tenantId, patientExternalId);
-          patientIndex.set(patientExternalId, stubId);
+        if (patientExternalId && !patientIndex.has(patientIndexKey(sourceUnit, patientExternalId))) {
+          const stubId = await this.ensurePatientStub(hostId, tenantId, sourceUnit, patientExternalId);
+          patientIndex.set(patientIndexKey(sourceUnit, patientExternalId), stubId);
         }
         const txn = this.synthesizer.synthesize({
           rawRow: rawSource,
@@ -1601,7 +1623,7 @@ export class ColdImportService {
           resource: config.canonical,
           hostId,
           tenantId,
-          patientIdResolver: (extId) => patientIndex!.get(extId),
+          patientIdResolver: (extId) => patientIndex!.get(patientIndexKey(sourceUnit, extId)),
           sourceContext,
         });
         if (!txn) {
@@ -1876,11 +1898,12 @@ export class ColdImportService {
       () =>
         this.prisma.patient.findMany({
           where: { hostId, tenantId },
-          select: { id: true, externalId: true },
+          select: { id: true, externalId: true, sourceUnit: true },
         }),
       'buildPatientIndex',
     );
-    return new Map(rows.map((p) => [p.externalId, p.id]));
+    // 集团内 external_id 可跨 source_unit 撞号 → 索引按 (source_unit, external_id) 建键
+    return new Map(rows.map((p) => [patientIndexKey(p.sourceUnit, p.externalId), p.id]));
   }
 
   /// W4 末:本批 fact 引用的 patient 主档没拉到 → 即时建空 stub
@@ -1890,14 +1913,15 @@ export class ColdImportService {
   private async ensurePatientStub(
     hostId: string,
     tenantId: string,
+    sourceUnit: string,
     externalId: string,
   ): Promise<string> {
     const patient = await this.withDbRetry(
       () =>
         this.prisma.patient.upsert({
-          // sourceUnit:''(过渡)— stub 与真主档同用 '',不产生双行;真填值随 remap 一起处理
-          where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit: '', externalId } },
-          create: { hostId, tenantId, externalId, active: true },
+          // source_unit 与真主档同源(从同一 rawSource.brand 解析)→ stub 与真主档命中同一行,不双行
+          where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId } },
+          create: { hostId, tenantId, sourceUnit, externalId, active: true },
           update: {}, // 有则 noop(姓名等真实主档 upsert 走 processPatients 路径)
         }),
       'ensurePatientStub patient',

apps/pac-service/src/modules/sync/cold-import/manifest.schema.ts

@@ -132,6 +132,12 @@ export const ColdImportManifestSchema = z
     tenant_map: z.record(z.string(), z.string().min(1)).optional(),
     brand_to_tenant: z.record(z.string(), z.string()).optional(),
 
+    /// **患者命名空间字段**(集团模型):tenant=集团后,external_id 在集团内可撞号
+    /// (不同源组织发的同号 = 不同人)。声明用哪个**源行字段**填 patients.source_unit 消歧。
+    /// 通用 — 各 host "在哪层发患者号"不同:jvs-dw='brand'(瑞尔/瑞泰);别的 host 可能=区域/诊所/不设。
+    /// 不设时 source_unit='',适合单命名空间 host。
+    identity_namespace_field: z.string().min(1).optional(),
+
     /// 金额单位(yuan = 元小数,fen = 分整数)
     amount_unit: z.enum(['fen', 'yuan']),
     /// 时区(用于解析无 tz 的 datetime 字符串)

apps/pac-service/src/modules/sync/cold-import/tenant-resolver.ts

@@ -110,3 +110,28 @@ export function buildTenantResolver(manifest: ColdImportManifest): TenantResolve
     'buildTenantResolver: manifest 必须含 tenant_id / tenant_field / brand_to_tenant 之一',
   );
 }
+
+/**
+ * SourceUnitResolver — 从源行解析 patients.source_unit(发证命名空间)。
+ *
+ * 与 tenant 解耦:tenant=集团(静态),source_unit=源行的命名空间字段值(如 brand)。
+ * manifest 未声明 identity_namespace_field → 恒返回 ''(单命名空间 host,external_id 集团内本就唯一)。
+ * 返回 '' 而非 null:source_unit 是 NOT NULL 列,'' = 无命名空间。
+ */
+export interface SourceUnitResolver {
+  resolve(row: Record<string, unknown>): string;
+}
+
+export function buildSourceUnitResolver(manifest: ColdImportManifest): SourceUnitResolver {
+  const field = manifest.identity_namespace_field;
+  if (!field) {
+    return { resolve: () => '' };
+  }
+  return {
+    resolve(row: Record<string, unknown>): string {
+      const v = row[field];
+      if (v === undefined || v === null) return '';
+      return String(v).trim();
+    },
+  };
+}