Merge canonical source timestamp timezone fix into main

apps/pac-docs/content/docs/architecture/data-ingestion.mdx

@@ -54,6 +54,8 @@ flowchart TD
 
 时区、金额单位按宿主配置归一。原始影像地址按宿主数据保留；相对存储路径不等同可直接访问的网页链接。
 
+所有资源的 `createdAt`、`updatedAt` 与业务时间共用归一流程：无时区值按宿主时区解释，显式时区保持实际时刻。源修改时间用于新旧版本裁决；召回使用事实发生时间、计划时间等业务锚点，画像刷新使用 PAC 事实更新时间与交易水位。
+
 ## 摄入后续处理
 
 原始行 push 对受影响患者安排画像重算，画像处理后链式安排计划重算。接收计数、交易计数与事实计数不同：过滤、拆分、重复及无法识别的类型都会影响结果。

apps/pac-docs/content/docs/canonical-fields.mdx

@@ -24,9 +24,12 @@ icon: Table2
 | `patientExternalId` | ✅ | 所属患者 id（`patient` 主档自身可省）|
 | `clinicId` | ✅ | 诊所 id（`referral` / `patient` 主档可空）|
 | 〈发生时间〉 | ✅ | 每个主体一个主时间字段,名字见下表(如 `scheduledAt` / `startedAt` / `paidAt`)|
+| `createdAt` / `updatedAt` | — | 源记录创建 / 最后修改时刻；提供时用于事件版本识别与新旧版本判断 |
 
 > 另需 §1 的组织 id（`tenantId` / `sourceUnit`)。金额字段按接入配置的 `amountUnit` 原样发,PAC 归一;时间本地值即可。
 
+所有主体的发生、创建、修改时间统一归一：无时区值按宿主配置解释，带 `Z` 或时区偏移的值按其实际时刻处理。生日、回访日期等纯日期保持 `YYYY-MM-DD`。召回时间依据业务发生或计划时刻；源修改时间用于防止旧数据覆盖新版本。
+
 下面**只列各主体专有字段**(公共字段不再重复)。除身份/时间外几乎都可选 —— 映射不上的留空,PAC 自然 null。
 
 ---

apps/pac-docs/content/docs/ingestion.mdx

@@ -255,3 +255,16 @@ pnpm --filter @pac/service ingest-evidence -- --host=friday --sync-log=<uuid> --
 服务器构建产物用 `ingest-evidence:prod`。`--out` 导出完整正文，不覆盖已有文件，文件权限 0600；命令只读，不启动定时任务或自动重放。未指定 `--out` 时在终端输出正文。
 
 映射、时区、金额单位仍读取宿主当前配置，不保存配置快照。修改配置后的 reparse 按新口径计算；缺身份等源数据错误须修正后再推，原文留档不会自动补值或解除消费端重试阻塞。
+
+### 5.4 源修改时间校正
+
+旧源修改时间若被误当 UTC 保存，修复归一后的数据可能被判为旧版本。先核对并校正存量 `sourceUpdatedAt`，再进行必要的补推或重解析；不关闭乱序保护。
+
+```bash
+# 在 apps/pac-service 下执行；默认只读，必须指定宿主、租户和患者范围
+node dist/cli/backfill-source-updated-at.cli.js --host=<host> --tenant=<tenant> \
+  --dir=./data/<host> --patient=<PAC患者UUID,...>
+# 核对结果后，增加 --apply --audit-file=/private/new-audit.jsonl 才写入
+```
+
+工具按原始流水、原事件版本标记和现行转换/装配流程还原时间，只修正与旧误解析值精确吻合的水位。身份不符、证据不足或无法唯一还原的记录计入 `unverifiable`；并发变化时停止。原文、事件 ID、业务发生时间及事实内容保持原值，不产生新事实版本，不自动重算。写入前后时间保存到新建的 0600 审计文件。

apps/pac-service/src/cli/backfill-source-updated-at.cli.ts

+/** 定向校正源更新时间水位。默认 dry-run；不重写原文/事件 ID/业务时间，不运行重算或 scheduler。 */
+import { appendFileSync, closeSync, openSync, readFileSync } from 'node:fs';
+import { resolve } from 'node:path';
+import { Logger } from '@nestjs/common';
+import { PrismaClient } from '@prisma/client';
+import { z } from 'zod';
+import { correctedSourceUpdatedAt, SourceUpdatedAtReplay } from './backfill-source-updated-at.lib';
+
+const option = (argv: string[], key: string) => argv.find(v => v.startsWith(`--${key}=`))?.slice(key.length + 3);
+
+async function main() {
+  const argv = process.argv.slice(2);
+  const hostName = option(argv, 'host'), tenantId = option(argv, 'tenant'), dir = option(argv, 'dir');
+  const file = option(argv, 'patients-file');
+  const patientIds = [...new Set([...(option(argv, 'patient') ?? '').split(','),
+    ...(file ? readFileSync(resolve(file), 'utf8').split(/\r?\n/).filter(v => !v.trim().startsWith('#')) : []),
+  ].map(v => v.trim()).filter(Boolean))];
+  if (!hostName || !tenantId || !dir || !patientIds.length || patientIds.some(id => !z.uuid().safeParse(id).success)) {
+    throw new Error('必须指定 --host、--tenant、--dir 和非空 PAC 患者 UUID 范围');
+  }
+  const apply = argv.includes('--apply'), auditFile = option(argv, 'audit-file');
+  if (apply && !auditFile) throw new Error('--apply 必须指定新的 --audit-file');
+  Logger.overrideLogger(['error']);
+  const db = new PrismaClient();
+  let fd: number | undefined;
+  const stats = { scanned: 0, unchanged: 0, changed: 0, written: 0, unverifiable: 0, conflicted: 0 };
+  const examples: Array<{ factId: string; before: Date | null; after: Date }> = [];
+  try {
+    const host = await db.host.findUniqueOrThrow({ where: { name: hostName }, select: { id: true, pullConfig: true } });
+    const config = (host.pullConfig ?? {}) as Record<string, unknown>;
+    const check = new SourceUpdatedAtReplay(resolve(dir));
+    if (check.manifest.host_name !== hostName) throw new Error('manifest 与宿主不符');
+    const replay = new SourceUpdatedAtReplay(resolve(dir), {
+      timezone: typeof config.timezone === 'string' ? config.timezone : check.manifest.timezone,
+      amountUnit: config.amountUnit === 'fen' || config.amountUnit === 'yuan' ? config.amountUnit : check.manifest.amount_unit,
+    });
+    const patients = await db.patient.findMany({ where: { hostId: host.id, tenantId, id: { in: patientIds } },
+      select: { id: true, externalId: true, sourceUnit: true } });
+    if (patients.length !== patientIds.length) throw new Error('患者不存在或跨宿主/租户');
+    if (apply) fd = openSync(resolve(auditFile!), 'wx', 0o600);
+    for (const patient of patients) {
+      let cursor: string | undefined;
+      for (;;) {
+        const facts = await db.patientFact.findMany({ where: { hostId: host.id, tenantId, patientId: patient.id,
+          sourceUpdatedAt: { not: null }, ...(cursor ? { id: { gt: cursor } } : {}) },
+          orderBy: { id: 'asc' }, take: 50, select: { id: true, sourceUnit: true, sourceUpdatedAt: true, updatedAt: true, transactionIds: true } });
+        if (!facts.length) break;
+        cursor = facts[facts.length - 1]!.id;
+        const txs = await db.patientTransaction.findMany({ where: { hostId: host.id, tenantId, patientId: patient.id,
+          id: { in: [...new Set(facts.flatMap(f => f.transactionIds))] } },
+          select: { id: true, hostId: true, tenantId: true, patientId: true, subjectId: true, subjectType: true,
+            clinicId: true, action: true, sourceEventId: true, rawPayload: true } });
+        const txMap = new Map(txs.map(t => [t.id, t]));
+        for (const fact of facts) {
+          stats.scanned++;
+          let next: Date | null;
+          try {
+            if (fact.sourceUnit !== patient.sourceUnit || !fact.transactionIds.length) throw new Error('身份/证据不足');
+            const pairs = fact.transactionIds.map(id => {
+              const tx = txMap.get(id);
+              if (!tx) throw new Error('证据缺失');
+              return replay.replay(tx, { sourceUnit: patient.sourceUnit, patientExternalId: patient.externalId });
+            });
+            next = correctedSourceUpdatedAt(fact.sourceUpdatedAt, pairs);
+          } catch { stats.unverifiable++; continue; }
+          if (!next) { stats.unchanged++; continue; }
+          stats.changed++;
+          if (examples.length < 20) examples.push({ factId: fact.id, before: fact.sourceUpdatedAt, after: next });
+          if (!apply) continue;
+          appendFileSync(fd!, JSON.stringify({ state: 'prepared', hostId: host.id, tenantId, patientId: patient.id,
+            factId: fact.id, before: fact.sourceUpdatedAt, after: next, updatedAt: fact.updatedAt }) + '\n');
+          // 证据或事实有并发变化时不覆盖；不触碰召回使用的业务时间和事实内容。
+          const updated = await db.patientFact.updateMany({ where: { id: fact.id, hostId: host.id, tenantId,
+            patientId: patient.id, sourceUpdatedAt: fact.sourceUpdatedAt, updatedAt: fact.updatedAt,
+            transactionIds: { equals: fact.transactionIds } }, data: { sourceUpdatedAt: next } });
+          if (updated.count !== 1) {
+            stats.conflicted++;
+            appendFileSync(fd!, JSON.stringify({ state: 'conflicted', factId: fact.id }) + '\n');
+            throw new Error('事实并发变化，停止后续写入');
+          }
+          stats.written++;
+          appendFileSync(fd!, JSON.stringify({ state: 'applied', factId: fact.id }) + '\n');
+        }
+      }
+    }
+    console.log(JSON.stringify({ mode: apply ? 'apply' : 'dry-run', host: hostName, ...stats, examples }));
+  } finally {
+    if (fd !== undefined) closeSync(fd);
+    await db.$disconnect();
+  }
+}
+
+if (require.main === module) main().catch(() => {
+  console.error('源更新时间修复未完成，请检查范围、连接或私有审计记录；已停止后续写入。');
+  process.exitCode = 1;
+});

apps/pac-service/src/cli/backfill-source-updated-at.lib.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import * as yaml from 'js-yaml';
+import { AssemblerEngine } from '../modules/sync/assembler/assembler-engine';
+import { AssemblerConfigSchema } from '../modules/sync/assembler/assembler.schema';
+import { resolveDictIncludes } from '../modules/sync/assembler/dict-include';
+import { normalizeDatetime, type NormalizeContext } from '../modules/sync/assembler/field-mapper';
+import { ColdImportManifestSchema } from '../modules/sync/cold-import/manifest.schema';
+import { traceRawSourceTable } from '../modules/sync/cold-import/cold-import.service';
+import { buildSourceUnitResolver, buildTenantResolver } from '../modules/sync/cold-import/tenant-resolver';
+import { TransformEngine } from '../modules/sync/transforms/transform-engine';
+
+export interface SourceTimeEvidence {
+  hostId: string;
+  tenantId: string;
+  patientId: string | null;
+  subjectId: string;
+  subjectType: string;
+  clinicId: string;
+  action: string;
+  sourceEventId: string | null;
+  rawPayload: unknown;
+}
+export interface SourceTimePair { legacy: Date; normalized: Date }
+
+/** 仅重建原表证据的更新时间；与摄入共用 manifest/transform/assembler，不运行 parser 或写库。 */
+export class SourceUpdatedAtReplay {
+  readonly manifest;
+  private readonly configs;
+
+  constructor(dir: string, private readonly normalize?: NormalizeContext) {
+    this.manifest = ColdImportManifestSchema.parse(yaml.load(readFileSync(join(dir, 'manifest.yaml'), 'utf8')));
+    this.configs = this.manifest.assemblers.map(ref => {
+      const file = join(dir, ref.file);
+      return AssemblerConfigSchema.parse(resolveDictIncludes(yaml.load(readFileSync(file, 'utf8')), file));
+    }).filter(config => config.emits);
+  }
+
+  replay(evidence: SourceTimeEvidence, expected: { sourceUnit: string; patientExternalId: string }): SourceTimePair {
+    if (!evidence.rawPayload || typeof evidence.rawPayload !== 'object' || Array.isArray(evidence.rawPayload)) {
+      throw new Error('缺少原表证据');
+    }
+    const row = evidence.rawPayload as Record<string, unknown>;
+    if (buildTenantResolver(this.manifest).resolve(row) !== evidence.tenantId ||
+      buildSourceUnitResolver(this.manifest).resolve(row) !== expected.sourceUnit) throw new Error('证据租户/源组织不符');
+    const matches: SourceTimePair[] = [];
+    const context = this.normalize ?? { amountUnit: this.manifest.amount_unit, timezone: this.manifest.timezone };
+    for (const config of this.configs.filter(c => c.emits!.action === evidence.action && c.emits!.subjectType === evidence.subjectType)) {
+      const sourceField = config.field_mapping.updatedAt;
+      const original = sourceField ? row[sourceField] : undefined;
+      if (typeof original !== 'string' || !original.trim()) continue;
+      const transforms = this.manifest.transforms ?? [];
+      const tables: Record<string, Record<string, unknown>[]> = { [traceRawSourceTable(config.primary.table, transforms)]: [row] };
+      for (const op of transforms) if (op.kind !== 'union' && !tables[op.input]) tables[op.input] = [];
+      const result = new AssemblerEngine().assemble({
+        tables: new TransformEngine().run({ tables, transforms }), config, normalize: context, hostId: evidence.hostId,
+      });
+      if (result.errors.length) continue;
+      for (const item of result.rows) {
+        const c = item.canonical as Record<string, unknown>;
+        if (String(c.externalId) !== evidence.subjectId || String(c.patientExternalId) !== expected.patientExternalId ||
+          c.clinicId !== evidence.clinicId) continue;
+        const normalized = new Date(c.updatedAt as string);
+        const directlyNormalized = new Date(normalizeDatetime(original, context.timezone));
+        // 不猜测 transform 改写过的更新时间，也不依赖修复工具进程自己的时区。
+        if (!Number.isFinite(normalized.getTime()) || normalized.getTime() !== directlyNormalized.getTime()) continue;
+        // 原事件 ID 保留了当次装配实际采用的版本标记，可区分“旧裸时间”与“已归一时间”。
+        // 只接受能与原文/本次归一值精确对应的标记；不按资源名或统一减八小时猜测。
+        const mark = evidence.sourceEventId?.replace(/:[a-f0-9]{64}$/, '');
+        const legacy = mark?.endsWith(`:${original}`) ? new Date(normalizeDatetime(original, 'UTC'))
+          : mark?.endsWith(`:${String(c.updatedAt)}`) ? normalized : null;
+        if (legacy && Number.isFinite(legacy.getTime())) matches.push({ legacy, normalized });
+      }
+    }
+    if (matches.length !== 1) throw new Error('证据无法唯一还原更新时间');
+    return matches[0]!;
+  }
+}
+
+/** 只有全部证据均支持旧 UTC 误解，且当前水位精确吻合旧值时，才允许校正。 */
+export function correctedSourceUpdatedAt(current: Date | null, evidence: SourceTimePair[]): Date | null {
+  if (!current || !evidence.length) return null;
+  const legacy = evidence.reduce((max, e) => Math.max(max, e.legacy.getTime()), -Infinity);
+  const normalized = evidence.reduce((max, e) => Math.max(max, e.normalized.getTime()), -Infinity);
+  if (!Number.isFinite(legacy) || !Number.isFinite(normalized) || current.getTime() !== legacy || legacy === normalized) return null;
+  return new Date(normalized);
+}

apps/pac-service/src/modules/sync/assembler/field-mapper.ts

@@ -179,7 +179,10 @@ export function normalizeCanonical(
   }
 
   // ─ timezone:无 tz datetime → 补 offset ─
-  for (const k of meta.datetimeFields) {
+  // createdAt/updatedAt 是所有资源共用的事件版本/回退时间，不依赖各资源是否在
+  // schema 显式列出它们。否则 passthrough 的裸时间会被 synthesizer/parser 按进程时区解释。
+  // birthDate/taskDate 等纯日期仍只按各自契约处理，不按字段后缀猜测时间类型。
+  for (const k of new Set(['createdAt', 'updatedAt', ...meta.datetimeFields])) {
     if (k in out && typeof out[k] === 'string') {
       out[k] = normalizeDatetime(out[k] as string, ctx.timezone);
     }
@@ -282,7 +285,7 @@ function coerceArray(v: unknown): unknown[] | null {
 export function normalizeDatetime(s: string, timezone: string): string {
   const trimmed = s.trim();
   // 已有 tz
-  if (/[Zz]$/.test(trimmed) || /[+\-]\d{2}:?\d{2}$/.test(trimmed)) {
+  if (/[Zz]$/.test(trimmed) || /[+-]\d{2}:?\d{2}$/.test(trimmed)) {
     return trimmed.replace(' ', 'T');
   }
   // 仅日期 → 当地零点(见上方注释:直通会被 JS 当成 UTC 零点)

apps/pac-service/tests/backfill-source-updated-at.spec.ts

+import { join } from 'node:path';
+import { Logger } from '@nestjs/common';
+import { correctedSourceUpdatedAt, SourceUpdatedAtReplay, type SourceTimeEvidence } from '../src/cli/backfill-source-updated-at.lib';
+
+const wall = '2026-10-16 10:47:34';
+const old = new Date('2026-10-16T10:47:34Z');
+const correct = new Date('2026-10-16T02:47:34Z');
+const expected = { sourceUnit: 'brand-1', patientExternalId: '205835' };
+const make = (timestamp = wall): SourceTimeEvidence => ({
+  hostId: 'host-1', tenantId: 'friday-market', patientId: 'patient-uuid', clinicId: 'clinic-1',
+  action: 'doctor_encounter_started', subjectType: 'encounter', subjectId: 'visit-1',
+  sourceEventId: `push:friday:encounter:brand-1:visit-1:${timestamp}`,
+  rawPayload: { id: 'visit-1', tenant_id: 'brand-1', patient_appointment_id: '205835',
+    organization_id: 'clinic-1', appointment_status: '4', in_time: timestamp,
+    created_gmt_at: timestamp, updated_gmt_at: timestamp },
+});
+let replay: SourceUpdatedAtReplay;
+beforeAll(() => {
+  jest.spyOn(Logger.prototype, 'log').mockImplementation(() => {});
+  jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => {});
+  replay = new SourceUpdatedAtReplay(join(__dirname, '../data/friday'));
+});
+afterAll(() => jest.restoreAllMocks());
+
+test('真实 manifest/transform/assembler 从原文还原两个时间，精确校正旧水位', () => {
+  const pair = replay.replay(make(), expected);
+  expect(pair).toEqual({ legacy: old, normalized: correct });
+  expect(correctedSourceUpdatedAt(old, [pair])).toEqual(correct);
+  expect(correctedSourceUpdatedAt(correct, [pair])).toBeNull();
+});
+
+test.each(['2026-10-16T02:47:34Z', '2026-10-16T10:47:34+08:00'])('显式时区 %s 保持原值', timestamp => {
+  const pair = replay.replay(make(timestamp), expected);
+  expect(pair).toEqual({ legacy: correct, normalized: correct });
+  expect(correctedSourceUpdatedAt(correct, [pair])).toBeNull();
+});
+
+test('已归一事件标记不是旧 UTC 误解，不能再偏移一次', () => {
+  const tx = { ...make(), sourceEventId: 'push:friday:encounter:brand-1:visit-1:2026-10-16T10:47:34+08:00' };
+  expect(replay.replay(tx, expected)).toEqual({ legacy: correct, normalized: correct });
+});
+
+test('宿主配置为东京时区时使用九小时偏差，不写死八小时', () => {
+  const tokyo = new SourceUpdatedAtReplay(join(__dirname, '../data/friday'), { timezone: 'Asia/Tokyo', amountUnit: 'fen' });
+  expect(tokyo.replay(make(), expected)).toEqual({ legacy: old, normalized: new Date('2026-10-16T01:47:34Z') });
+});
+
+test('不符合原文身份或缺少版本标记时拒绝猜测', () => {
+  expect(() => replay.replay(make(), { ...expected, sourceUnit: 'other-brand' })).toThrow();
+  expect(() => replay.replay(make(), { ...expected, patientExternalId: 'other-patient' })).toThrow();
+  expect(() => replay.replay({ ...make(), clinicId: 'other-clinic' }, expected)).toThrow();
+  expect(() => replay.replay({ ...make(), tenantId: 'other-tenant' }, expected)).toThrow();
+  expect(() => replay.replay({ ...make(), sourceEventId: null }, expected)).toThrow();
+  expect(() => replay.replay({ ...make(), sourceEventId: 'unknown-format' }, expected)).toThrow();
+});
+
+test('多证据按正确时间取最大值；未知现值、无证据、非法日期均不修', () => {
+  const pairs = [{ legacy: old, normalized: correct },
+    { legacy: new Date('2026-10-16T11:00:00Z'), normalized: new Date('2026-10-16T03:00:00Z') }];
+  expect(correctedSourceUpdatedAt(pairs[1]!.legacy, pairs)).toEqual(pairs[1]!.normalized);
+  expect(correctedSourceUpdatedAt(new Date('2026-10-17T11:00:00Z'), pairs)).toBeNull();
+  expect(correctedSourceUpdatedAt(null, pairs)).toBeNull();
+  expect(correctedSourceUpdatedAt(old, [])).toBeNull();
+  expect(correctedSourceUpdatedAt(old, [{ legacy: new Date('invalid'), normalized: correct }])).toBeNull();
+});

apps/pac-service/tests/canonical-source-timezone.spec.ts

+import { readFileSync, readdirSync } from 'node:fs';
+import { join } from 'node:path';
+import * as yaml from 'js-yaml';
+import { CanonicalResourceMeta, type CanonicalResourceKey } from '@pac/types';
+import { AssemblerEngine } from '../src/modules/sync/assembler/assembler-engine';
+import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.schema';
+import { applyMapping, normalizeCanonical } from '../src/modules/sync/assembler/field-mapper';
+import { resolveDictIncludes } from '../src/modules/sync/assembler/dict-include';
+import { TransactionSynthesizer } from '../src/modules/sync/pipeline/transaction-synthesizer';
+import { ParserPipeline } from '../src/modules/sync/pipeline/parser-pipeline.service';
+import { EncounterParser } from '../src/modules/sync/pipeline/parsers/encounter.parser';
+
+const context = { timezone: 'Asia/Shanghai', amountUnit: 'fen' as const };
+const resources = Object.keys(CanonicalResourceMeta) as CanonicalResourceKey[];
+const wall = '2026-10-16 10:47:34';
+const utc = '2026-10-16T02:47:34.000Z';
+
+// 按所有资源断言真实时刻，避免仅补 encounter 元信息后其他 passthrough 资源继续漏转。
+describe.each(resources)('%s 通用源时间', (resource) => {
+  test('创建/修改时间均按宿主时区解释，原文保持不变', () => {
+    const raw = Object.freeze({ createdAt: wall, updatedAt: wall });
+    const normalized = normalizeCanonical(raw, resource, context);
+    expect(new Date(normalized.createdAt as string).toISOString()).toBe(utc);
+    expect(new Date(normalized.updatedAt as string).toISOString()).toBe(utc);
+    expect(raw).toEqual({ createdAt: wall, updatedAt: wall });
+    expect(normalizeCanonical(normalized, resource, context)).toEqual(normalized);
+  });
+
+  test('显式时区不重复偏移，宿主时区没有写死为东八区', () => {
+    for (const value of ['2026-10-16T02:47:34Z', '2026-10-16T11:47:34+09:00']) {
+      const out = normalizeCanonical({ createdAt: value, updatedAt: value }, resource, context);
+      expect(new Date(out.createdAt as string).toISOString()).toBe(utc);
+      expect(new Date(out.updatedAt as string).toISOString()).toBe(utc);
+    }
+    const out = normalizeCanonical({ updatedAt: wall }, resource, { ...context, timezone: 'Asia/Tokyo' });
+    expect(new Date(out.updatedAt as string).toISOString()).toBe('2026-10-16T01:47:34.000Z');
+  });
+});
+
+test('纯日期、空时间、Date 对象和缺省字段不变', () => {
+  const instant = new Date(utc);
+  const out = normalizeCanonical({ birthDate: '2026-10-16', taskDate: '2026-10-16',
+    createdAt: instant, updatedAt: null }, 'patient_return_visit', context);
+  expect(out).toEqual({ birthDate: '2026-10-16', taskDate: '2026-10-16', createdAt: instant, updatedAt: null });
+  expect(normalizeCanonical({ updatedAt: '' }, 'emr', context)).toEqual({ updatedAt: '' });
+  expect(normalizeCanonical({}, 'emr', context)).toEqual({});
+});
+
+const configs = ['friday', 'jvs-dw'].flatMap(host => {
+  const dir = join(__dirname, '../data', host, 'assemblers');
+  return readdirSync(dir).filter(f => f.endsWith('.yaml')).map(file => ({ host, file,
+    config: AssemblerConfigSchema.parse(resolveDictIncludes(yaml.load(readFileSync(join(dir, file), 'utf8')), join(dir, file))),
+  }));
+});
+
+test.each(configs)('$host/$file 源创建/修改时间统一归一', ({ config }) => {
+  const raw: Record<string, unknown> = {};
+  for (const field of ['createdAt', 'updatedAt', 'sourceCreatedAt', 'sourceUpdatedAt']) {
+    const source = config.field_mapping[field];
+    if (source) raw[source] = wall;
+  }
+  const out = normalizeCanonical(applyMapping(raw, config.field_mapping), config.canonical as CanonicalResourceKey, context);
+  for (const field of ['createdAt', 'updatedAt', 'sourceCreatedAt', 'sourceUpdatedAt']) {
+    if (config.field_mapping[field]) expect(new Date(out[field] as string).toISOString()).toBe(utc);
+  }
+});
+
+function encounter() {
+  const config = configs.find(c => c.host === 'friday' && c.config.canonical === 'encounter')!.config;
+  const raw = { id: 'visit-1', patient_appointment_id: 'patient-1', organization_id: 'clinic-1',
+    in_time: wall, created_gmt_at: wall, updated_gmt_at: wall };
+  const result = new AssemblerEngine().assemble({ tables: { [config.primary.table]: [raw] }, config,
+    normalize: context, hostId: 'host-1' });
+  expect(result.errors).toEqual([]);
+  const canonicalRow = result.rows[0]!.canonical as Record<string, unknown>;
+  const synthesize = (row: Record<string, unknown>) => new TransactionSynthesizer().synthesize({ rawRow: raw,
+    canonicalRow: row, emits: config.emits, resource: 'encounter', hostId: 'host-1', tenantId: 'tenant-1',
+    sourceUnit: 'brand-1', sourceContext: 'push:host', patientIdResolver: () => 'patient-uuid' })!;
+  return { raw, canonicalRow, synthesize };
+}
+
+test('真实 assembler → synthesizer：业务时间与版本时间同归一，重复摄入键稳定', () => {
+  const { raw, canonicalRow, synthesize } = encounter();
+  const a = synthesize(canonicalRow), b = synthesize(canonicalRow);
+  expect(a.occurredAt).toEqual(new Date(utc));
+  expect(a.sourceEventId).toBe(b.sourceEventId);
+  expect(a.rawPayload).toEqual(raw);
+  // 业务时刻缺失时，修改/创建时间回退也不能按服务器 UTC 误解。
+  expect(synthesize({ ...canonicalRow, startedAt: null }).occurredAt).toEqual(new Date(utc));
+  expect(synthesize({ ...canonicalRow, startedAt: null, updatedAt: null }).occurredAt).toEqual(new Date(utc));
+});
+
+test.each(['single', 'batch'] as const)('归一后进入 %s parser，乱序裁决拿到正确 UTC 更新时间', async mode => {
+  const { canonicalRow } = encounter();
+  const result = { action: 'created', factId: 'fact-1', subjectId: 'encounter_record:visit-1', version: 1 };
+  const writer = { writeDraft: jest.fn().mockResolvedValue(result),
+    bulkWrite: jest.fn().mockResolvedValue({ results: [result], rejected: [] }) };
+  const prisma = { patient: { findUnique: async () => ({ sourceUnit: 'brand-1' }),
+    findMany: async () => [{ id: 'patient-uuid', sourceUnit: 'brand-1' }] } };
+  const pipeline = new ParserPipeline({ get: () => new EncounterParser() } as never, writer as never,
+    prisma as never, { enrichAll: (drafts: unknown) => drafts } as never);
+  const item = { canonicalRow, transaction: { id: 'tx-1', hostId: 'host-1', tenantId: 'tenant-1',
+    patientId: 'patient-uuid', action: 'doctor_encounter_started' as const, subjectType: 'encounter',
+    subjectId: 'visit-1', clinicId: 'clinic-1', occurredAt: new Date(utc) } };
+  const metrics = mode === 'single' ? await pipeline.runForTransaction(item) : await pipeline.runForBatch([item]);
+  const entry = mode === 'single' ? writer.writeDraft.mock.calls[0]![0] : writer.bulkWrite.mock.calls[0]![0][0];
+  expect(entry.sourceUpdatedAt).toEqual(new Date(utc));
+  expect(entry.draft.occurredAt).toEqual(new Date(utc));
+  expect(metrics.factsFailed).toBe(0);
+  expect(metrics.factsCreated).toBe(1);
+});

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

@@ -83,7 +83,7 @@ test.each(['native', 'json', 'legacy-text'])(
     expect(readEmrNarrative(content.treatment_notes)).toEqual([{ toothPosition: null, message: '采集资料' }]);
     expect(content.treatment_plan).toBeNull(); // treat 原文不能变成未来 planned 事实
     expect((tx.occurredAt as Date).toISOString()).toBe('2023-04-08T18:46:00.000Z');
-    expect(canonical.updatedAt).toBe(fridayRaw.updated_gmt_at); // 保持幂等标记形态
+    expect(canonical.updatedAt).toBe('2023-04-09T03:10:00+08:00'); // 源修改时间也按宿主时区归一
     expect(JSON.stringify(rawSource)).toBe(saved);
     expect(JSON.stringify(raw)).toBe(saved);
     expect(content.exam_findings).not.toContain('"value"');
@@ -101,7 +101,7 @@ test('5i5ya：检查影像原文不变，本次治疗/未来计划分开，病
     plan: JSON.stringify([{ toothPosition: '46', treatName: '复查后评估' }]),
   }, 'fact_emr_treatment_out');
   expect((tx.occurredAt as Date).toISOString()).toBe('2025-03-03T02:34:29.000Z');
-  expect(canonical.updatedAt).toBe('2025-03-03 11:00:00');
+  expect(canonical.updatedAt).toBe('2025-03-03T11:00:00+08:00');
   expect(JSON.parse(content.exam_findings as string)).toEqual(exam);
   expect(readEmrNarrative(content.disposal)).toEqual([{ toothPosition: null, message: '本次处置原文' }]);
   expect(emrNarrativeText(content.treatment_notes)).toBe('46 采集资料');

packages/types/src/schemas/canonical.ts

@@ -585,7 +585,8 @@ export type CanonicalResourceKey = keyof typeof CanonicalResourceSchemas;
  * FieldMapper.normalize() 用这份元信息按需做类型归一化。
  *
  * money 字段 = 该字段值在 amountUnit=yuan 时需要 ×100 转分。
- * datetime 字段 = 无 tz 时按 timezone 配置补 offset。
+ * datetime 字段 = 资源专属时间，无 tz 时按 timezone 配置补 offset。
+ * 通用 createdAt/updatedAt 由 FieldMapper 对所有资源统一归一，不依赖这里逐项声明。
  * boolean 字段 = host 用 0/1/"Y"/"N"/"true"/"false" 都接收,统一转 bool。
  */
 export const CanonicalResourceMeta: Record<
@@ -668,7 +669,7 @@ export const CanonicalResourceMeta: Record<
   emr: {
     moneyFields: [],
     moneyArrayFields: [],
-    // createdAt 是 DW 的 occurredAtField；updatedAt 是幂等标记，保持既有形态。
+    // createdAt 是 DW 的 occurredAtField；通用创建/修改时间也统一归一后供幂等与乱序裁决使用。
     datetimeFields: ['submittedAt', 'createdAt'],
     booleanFields: [],
   },