fix(sync): traceRawSourceTable 支持 union 回溯 — 修 refund_full 在 push/reparse 静默跳过

union-primary 资源(refund_full_rows ← union[status4, status3负额])此前回溯不到源表
(traceRawSourceTable 只认单 input/output),导致 push source=patient_settlement 时
refund_full assembler 不入选、reparse 也跳过 → **整单退费在这两条路径静默丢财务数据**。

修:union 各输入独立回溯,全部收敛到同一根表(两路都 → patient_settlement)才返回该根;
发散到多根则停在 union 输出(维持旧行为不误判)。export 供单测。

验证:108467 重推 patient_settlement → refund_full 落库(txn=1 新增,2 payment 幂等 dup);
单测 6 绿(union 收敛/发散/单链/route/lookup);全量 299 绿无回归。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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

@@ -2305,24 +2305,46 @@ interface TotalsBlock {
 
 /// 沿 transforms 的 output→input 链回溯,找到非任何 transform 产出的"原始源表"名。
 /// 例:diagnosis_rows ← _diagnosis_norm ← _diagnosis_raw ← fact_emr_treatment_out(终点=源表)。
-/// 用于 reparse:transaction.rawPayload 存的就是这张源表的行。
-function traceRawSourceTable(primaryTable: string, transforms: ReadonlyArray<unknown>): string {
-  const byOutput = new Map<string, string>(); // output 表名 → input 表名
+/// 用于 reparse(rawPayload 存的就是这张源表的行)+ push(source→匹配 assembler)。
+///
+/// union 多输入:各输入独立回溯,**全部收敛到同一根表**才返回该根(如 refund_full_rows
+///   ← union[_refund_status4, _refund_neg3] 两路都回到 patient_settlement → 返回它);
+///   若发散到多个根(语义歧义)→ 停在 union 输出本身(维持旧行为,不误判)。
+///   修复前 union 不被追 → refund_full 在 push/reparse 里回溯不到源表被静默跳过(丢财务数据)。
+export function traceRawSourceTable(primaryTable: string, transforms: ReadonlyArray<unknown>): string {
+  const byOutput = new Map<string, string>(); // output → input(单输入算子)
+  const unionInputs = new Map<string, string[]>(); // union output → inputs[]
   for (const tf of transforms) {
-    const t = tf as { input?: string; output?: string; outputs?: Array<{ output?: string }> };
+    const t = tf as {
+      kind?: string;
+      input?: string;
+      output?: string;
+      inputs?: string[];
+      outputs?: Array<{ output?: string }>;
+    };
+    if (t.kind === 'union' && t.output && Array.isArray(t.inputs)) {
+      unionInputs.set(t.output, t.inputs);
+      continue;
+    }
     if (t.output && t.input) byOutput.set(t.output, t.input);
     // route_by_pattern 多 output:每个 output 都回到同一 input
     if (Array.isArray(t.outputs) && t.input) {
       for (const o of t.outputs) if (o?.output) byOutput.set(o.output, t.input);
     }
   }
-  let cur = primaryTable;
-  const seen = new Set<string>();
-  while (byOutput.has(cur) && !seen.has(cur)) {
-    seen.add(cur);
-    cur = byOutput.get(cur)!;
-  }
-  return cur;
+  const resolve = (tbl: string, seen: Set<string>): string => {
+    if (seen.has(tbl)) return tbl; // 环保护
+    seen.add(tbl);
+    if (byOutput.has(tbl)) return resolve(byOutput.get(tbl)!, seen);
+    const ins = unionInputs.get(tbl);
+    if (ins && ins.length > 0) {
+      // 各输入独立回溯(各带 seen 副本,兄弟分支互不干扰);全一致才收敛,否则停在 union 输出
+      const roots = new Set(ins.map((i) => resolve(i, new Set(seen))));
+      return roots.size === 1 ? [...roots][0]! : tbl;
+    }
+    return tbl; // 已是根
+  };
+  return resolve(primaryTable, new Set<string>());
 }
 
 export interface PerResourceStats extends TotalsBlock {

apps/pac-service/tests/trace-raw-source.spec.ts

+/**
+ * traceRawSourceTable —— 回溯 assembler primary 表到「原始源表」,用于 push(source→匹配 assembler)
+ * 与 reparse(rawPayload 属哪张源表)。重点回归:**union 多输入必须能收敛到同一根**,
+ * 否则 union-primary 资源(refund_full)在 push/reparse 里被静默跳过(丢财务数据)。
+ *
+ * 跑:pnpm --filter @pac/service test -- trace-raw-source
+ */
+import { traceRawSourceTable } from '../src/modules/sync/cold-import/cold-import.service';
+
+// 结算链最小 transforms:消费(filter 链)+ 退费整单(filter×2 + union)+ 退费明细(filter+lookup)
+const transforms = [
+  { kind: 'filter', input: 'patient_settlement', output: '_pay_hdr', where: {} },
+  { kind: 'lookup', input: '_pay_hdr', output: 'payment_rows', from: 'settlement_modes', select: {} },
+  { kind: 'filter', input: 'patient_settlement', output: '_refund_status4', where: {} },
+  { kind: 'filter', input: 'patient_settlement', output: '_refund_neg3', where: {} },
+  { kind: 'union', inputs: ['_refund_status4', '_refund_neg3'], output: 'refund_full_rows' },
+  { kind: 'filter', input: 'patient_settlement_spec', output: '_refund_item_raw', where: {} },
+  { kind: 'lookup', input: '_refund_item_raw', output: 'refund_item_rows', from: 'patient_settlement', select: {} },
+  // route_by_pattern 多输出回同一 input
+  { kind: 'route_by_pattern', input: '_treat_raw', outputs: [{ output: '_actual_raw' }, { output: '_rec_raw' }] },
+];
+
+describe('traceRawSourceTable', () => {
+  it('单输入链:payment_rows → patient_settlement', () => {
+    expect(traceRawSourceTable('payment_rows', transforms)).toBe('patient_settlement');
+  });
+
+  it('union 收敛:refund_full_rows 两路都回到 patient_settlement(修复点)', () => {
+    expect(traceRawSourceTable('refund_full_rows', transforms)).toBe('patient_settlement');
+  });
+
+  it('lookup 链取 input(不取 from):refund_item_rows → patient_settlement_spec', () => {
+    expect(traceRawSourceTable('refund_item_rows', transforms)).toBe('patient_settlement_spec');
+  });
+
+  it('route 多输出回同一 input:_actual_raw → _treat_raw', () => {
+    expect(traceRawSourceTable('_actual_raw', transforms)).toBe('_treat_raw');
+  });
+
+  it('已是根表:原样返回', () => {
+    expect(traceRawSourceTable('patient_settlement', transforms)).toBe('patient_settlement');
+  });
+
+  it('union 输入发散到多根 → 停在 union 输出(不误判)', () => {
+    const diverge = [
+      { kind: 'filter', input: 'table_a', output: '_a', where: {} },
+      { kind: 'filter', input: 'table_b', output: '_b', where: {} },
+      { kind: 'union', inputs: ['_a', '_b'], output: 'merged' },
+    ];
+    expect(traceRawSourceTable('merged', diverge)).toBe('merged');
+  });
+});