Merge PAC image analysis and push batch fixes into test

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

@@ -12,6 +12,8 @@ icon: Table2
 >
 > 真理源 = 各主体 parser 实际读取的字段(`apps/pac-service/.../pipeline/parsers/*.parser.ts`);摄入校验 schema(`canonical.ts`)只强校验身份三件套 + 少数关键字段,其余字段透传,以 parser 为准。
 
+> 影像分析与文件链接字段见下表；FRIDAY 原表推送格式见 [FRIDAY 推送数据契约](/docs/integration/friday-push-payload#10-影像分析)。
+
 ---
 
 ## 公共字段(所有主体)
@@ -172,6 +174,49 @@ icon: Table2
 | `doctorId` / `doctorName` | 医生 / 医生名 |
 | `encounterExternalId` | 关联接诊 |
 
+### 影像完整链接与分析来源
+
+影像：`image` → `image_record`。完整分析：`image_analysis` → `image_analysis_record` + `diagnosis_record`。分析问题复用现有诊断解析器，`codeSource=image_ai`；5i5ya 已有的逐条 `diagnosis` 输入保持兼容。
+
+| 所属 canonical | 字段 | 目标语义 / 事实 `content` 字段 |
+|---|---|---|
+| `image` | `fileUrl` | 完整可访问的文件链接 → `file_url`；FRIDAY 从 `fileUrlEcho` 取值，相对存储路径只留在 raw |
+| `image` | `fileName` / `fileType` | 文件名 / 文件类型 → `file_name` / `file_type`，有源值时保留 |
+| `diagnosis` | `sourceImageExternalId` | 精确指向同宿主、同品牌、同患者的影像 externalId → `source_image_external_id` |
+| `diagnosis` | `sourceAnalysisExternalId` | 产生此发现的稳定分析请求 ID → `source_analysis_external_id` |
+
+### `image_analysis` 影像分析 — 主时间 `analyzedAt`，允许未知
+
+| 字段 | 必填 | 说明 |
+|---|---|---|
+| `imageExternalId` | ✅ | 对应同作用域影像的 `externalId` |
+| `status` | ✅ | `pending` / `completed` / `failed` |
+| `analyzedAt` | | 实际分析时间；未知时留空，不产生有效诊断信号 |
+| `createdAt` / `updatedAt` | | 源记录创建 / 更新时间 |
+| `result` | | 分析报告原文，可为 JSON 字符串 |
+| `findings[]` | ✅ | 完整问题列表；无问题明确传 `[]` |
+| `findings[].externalId` | ✅ | 稳定的问题主键，同次分析不可重复 |
+| `findings[].patientExternalId` / `imageExternalId` | ✅ | 与分析记录的患者 / 影像一致 |
+| `findings[].name` | ✅ | 原中文问题名称 |
+| `findings[].code` | | 共享字典归一后的 K 码；未知留空 |
+| `findings[].toothPosition` / `excludedToothPositions` | | 标准牙位 / 排除牙位，分号分隔 |
+| `findings[].excluded` | | 是否排除整项，默认 `false` |
+
+问题中文名称归一后，复用共享 `diagnosis-kcode.yaml`：精确术语匹配 → 关键词兜底 → K 码。
+未命中时 `code=null`，保留原文。宿主自定义问题码仅留在 `raw_payload`，不参与 K 码归一。
+
+PAC 处理规则：
+
+- 一个分析 source：分析原行 + 完整 `problemList`；保留 `analysis_data` 原文。
+- 以分析 `check_id` 关联影像；校验宿主、品牌、患者及诊所归属。
+- 各宿主入口归一状态、字段和时区；后续统一处理 `image_analysis`。`completed` 且实际分析时间已知时生成有效信号；未知时间留底，不补成拍片/推送时间。
+- 问题名称走共享中文 → K 码字典，不增加宿主私有码对照；原名称保留。
+- 原牙位扣除 `exclude_teeth`，`ALL` 或扣空撤销对应信号。
+- 分析主键 + 问题主键确定稳定身份；事件去重覆盖业务内容，问题更新无需宿主生成版本号。
+- 已完成分析的完整问题列表更新时修订/撤销旧信号；未完成或失败的分析不清除旧有效结果。
+- URL 更新保持同一影像身份，不改变分析时间。
+- 完整 URL 保存到影像事实，通过稳定影像 ID 从分析结果反查；读权限沿用患者组织范围。
+
 ---
 
 ## 单据

apps/pac-docs/content/docs/event-enums.mdx

@@ -4,7 +4,7 @@ description: push 事件的 subjectType 与 action 全量封闭集速查。
 icon: ListTree
 ---
 
-> Push **形态 C** 事件的 `subjectType` + `action` 取值必须落在下表封闭集内(`subjectType` 16 个、`action` 23 个)。真理源:`packages/types`(`SubjectType` / `Action`)。
+> Push **形态 C** 事件的 `subjectType` + `action` 取值必须落在下表封闭集内。真理源:`packages/types`(`SubjectType` / `Action`)。
 >
 > 形态 A / 文件 / 直连由 PAC 内部归一,宿主无需关心本表。
 
@@ -35,6 +35,7 @@ icon: ListTree
 | `payment` 收款 | `payment_received` | 收款 |
 | `refund` 退费 | `refund_created` | 退费 |
 | `image` 影像资料 | `image_uploaded` | 影像上传 |
+| `image_analysis` 影像分析 | `image_analysis_recorded` | 分析状态 / 完整问题列表更新 |
 | `recharge` 充值 / 储值 | `recharge_received` | 充值 / 储值 / 办卡入账 |
 | `complaint` 投诉 | `complaint_filed` | 投诉提交 |
 | | `complaint_resolved` | 投诉结案 |

apps/pac-docs/content/docs/integration/friday-push-payload.mdx

@@ -419,3 +419,80 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
 | 私有字典不单独推 | `std_diag`(生产多品牌合计 1.5万~3万+行)/ `std_check_class`:纯翻译字典,**由宿主解析好 inline** 进 `diag[].std_code` / `med_check.class_name`。既省掉大表全量重推,也省掉 PAC 侧持久化字典存储——PAC 只留跨宿主的 `_shared` 临床字典 |
 | 支付通道 | `settlement_modes`(约 24% 结算单多通道拆付)暂不摄入:总值已在结算头 `net_receipts_this`,PAC 当前无支付方式分析需求。将来要"医保占比/分次支付"再摄入通道明细 |
 | 测试环境数据特征 | 部分租户为开发沙盒(诊所名如"XX专用诊所勿动"),字典类映射(治疗类别关键词等)待**生产数据**回流后再校准一轮 |
+
+
+---
+
+## 10. 影像分析
+
+新增 1 个 source：`med_check_analysis_req_record`。分析原表行附带完整 `problemList`；不单独推问题明细表。
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `id` | string | ✅ | 分析记录主键 |
+| `tenant_id` / `organization_id` / `patient_id` | string | ✅ | 原品牌 / 诊所 / 患者 ID |
+| `check_id` | string | ✅ | 对应 `med_check.id` |
+| `req_status` | string | ✅ | 原状态，全状态推送 |
+| `analysis_time` / `shooting_time` | string(datetime) | | 原分析 / 拍摄时间，允许空 |
+| `analysis_data` | string(JSON) | | 原分析 JSON 字符串，允许空 |
+| `recall_status` | string/number | | 原值 |
+| `created_gmt_at` / `updated_gmt_at` | string(datetime) | | 原创建 / 更新时间 |
+| `problemList` | array | ✅ | 对应分析的完整问题列表，无问题传 `[]` |
+
+`problemList` 元素按 `med_check_analysis_problem` 原行推送：
+
+| 字段 | 说明 |
+|---|---|
+| `id` | 原问题主键，字符串 |
+| `patient_id` / `check_id` | 原患者 / 影像 ID，与分析记录一致 |
+| `problem_code` | 原问题码，仅留底 |
+| `problem` | 原中文问题名称 |
+| `teeth` / `exclude_teeth` | 原牙位 / 排除项，含 `ALL`；原样保留 |
+| `recall_status` | 原值 |
+| `created_gmt_at` / `updated_gmt_at` | 原值，有则带上 |
+
+- 分析或问题明细变化：重推分析原行及完整 `problemList`。
+- 问题中文名称 → K 码使用 PAC 共享字典；牙位、状态、时间归一及影像关联由 PAC 处理。
+- `api_req` / `api_res` 调用日志不需推送。
+- 本新增 source 在 PAC 部署并联调后启用。
+
+### 影像文件链接补充
+
+`med_check` 原字段及 `file_url` 含义保持不变，新增以下字段：
+
+| 字段 | 类型 | 要求 |
+|---|---|---|
+| `fileUrlEcho` | string(url) | 完整可访问的 HTTP(S) 文件链接，含所需签名参数；沿用 SaaS 现有文件地址解析服务 |
+
+地址变化时按同一影像 `id` 重推；PAC 保留完整链接，并通过 `check_id` 关联分析结果。
+
+### 报文示例
+
+```json
+{
+  "source": "med_check_analysis_req_record",
+  "rows": [{
+    "id": "1001",
+    "tenant_id": "brand-demo",
+    "organization_id": "clinic-demo",
+    "patient_id": "4001",
+    "check_id": "3001",
+    "req_status": "NOTIFY_OK",
+    "analysis_time": "2026-10-12 10:05:00",
+    "shooting_time": "2026-10-12 10:00:00",
+    "analysis_data": null,
+    "created_gmt_at": "2026-10-12 10:01:00",
+    "updated_gmt_at": "2026-10-12 10:05:00",
+    "problemList": [{
+      "id": "2001",
+      "patient_id": "4001",
+      "check_id": "3001",
+      "problem_code": "P0010",
+      "problem": "龋齿",
+      "teeth": "16,26",
+      "exclude_teeth": "26",
+      "recall_status": 0
+    }]
+  }]
+}
+```

apps/pac-service/data/friday/assemblers/image.yaml

@@ -7,6 +7,7 @@ emits:
   action: image_uploaded
   subjectType: image
   occurredAtField: uploadedAt
+  contentVersion: true
 
 primary:
   table: med_check
@@ -22,8 +23,9 @@ field_mapping:
   uploadedAt:           shooting_time     # 拍摄时间
   encounterExternalId:  emr_id            # 病历号级挂载(= med_emr_info.emr_id,非 emr_sub_id)
   modality:             class_name        # 宿主 inline(class_code→std_check_class 已解析)→ 共享字典翻
-  fileUrl:              file_url          # OSS 相对路径(passthrough)
+  fileUrl:              fileUrlEcho       # 完整 HTTP(S) 地址；原存储路径仍在 raw
   fileName:             file_name
+  fileType:             file_type
 
 # ── 字典分层:真影像类型名共享(与 jvs-dw 实测同名);FRIDAY 特有名目留本地 ──
 dict_includes:

apps/pac-service/data/friday/assemblers/image_analysis.yaml

+# 原分析行 + 问题原行数组 → PAC 通用分析快照；私有 P 码只在 raw 中留底。
+canonical: image_analysis
+emits:
+  action: image_analysis_recorded
+  subjectType: image_analysis
+  occurredAtField: analyzedAt
+  contentVersion: true
+primary:
+  table: med_check_analysis_req_record
+  key: id
+field_mapping:
+  externalId: id
+  patientExternalId: patient_id
+  clinicId: organization_id
+  imageExternalId: check_id
+  analyzedAt: analysis_time
+  createdAt: created_gmt_at
+  updatedAt: updated_gmt_at
+  status: req_status
+  result: analysis_data
+  expectedFindingCount: _problem_count
+enum_mapping:
+  status:
+    UNDO: pending
+    OK: pending
+    FAIL: failed
+    NOTIFY_FAIL: failed
+    NOTIFY_OK: completed
+    _default: pending
+join_arrays:
+  findings:
+    table: image_analysis_problem_rows
+    foreign_key: analysis_id
+    same_source_row: true
+    field_mapping:
+      externalId: id
+      patientExternalId: patient_id
+      imageExternalId: check_id
+      code: _problem_normalized
+      name: problem
+      toothPosition: teeth
+      excludedToothPositions: _excluded_positions
+      excluded: exclude_teeth
+    dict_includes:
+      - ../../_shared/dict/diagnosis-kcode.yaml
+    enum_mapping:
+      code: { _default: '' }
+      excluded: { ALL: 'true', _default: 'false' }

apps/pac-service/data/friday/manifest.yaml

@@ -55,6 +55,7 @@ incremental:
     appointment_base:               { cursor_column: updated_gmt_at }
     med_emr_info:                   { cursor_column: updated_gmt_at }
     med_check:                      { cursor_column: updated_gmt_at }
+    med_check_analysis_req_record:   { cursor_column: updated_gmt_at }
     patient_settlement:             { cursor_column: updated_gmt_at }
     patient_settlement_spec:        { cursor_column: updated_gmt_at }
     customer_treat_plan_item:       { cursor_column: updated_gmt_at }
@@ -67,6 +68,7 @@ incremental:
 #   联系方式 1:N(customer_contacts→inline 默认号进主档)、计划头(customer_treat_plan→inline 进行)、
 #   支付通道(settlement_modes,金额已在结算头 net_receipts_this)。所有摄入路径(push/file/pull/reparse)同此形态。
 tables:
+  - { table: med_check_analysis_req_record, file: med_check_analysis_req_record.json } # 原分析行 + 完整 problemList
   - { table: customer_basic_info, file: customer_basic_info.csv }   # 患者主档(含 inline contacts_tel/relationship)
   - { table: appointment_base,    file: appointment_base.csv }      # 预约(原始,全状态)
   # ── EMR 病历链(W FRIDAY-EMR:Mongo med_emr_info status∈{3,4} 正式病历;export.sh 平铺导出)──
@@ -84,6 +86,31 @@ tables:
   - { table: customer_return_visit,   file: customer_return_visit.csv }      # 客户回访(含 inline task_date/task_status)
 
 transforms:
+  # 影像分析入口形态：只补字段名与原生状态；临床处理走共享 image_analysis 管线。
+  - kind: derive
+    input: med_check_analysis_req_record
+    output: med_check_analysis_req_record
+    fields:
+      _problem_count: { op: array_length, from: problemList }
+  - kind: split_json_array
+    input: med_check_analysis_req_record
+    output: image_analysis_problem_rows
+    array_field: problemList
+    parent_keys: { analysis_id: id }
+    element_fields:
+      id: id
+      patient_id: patient_id
+      check_id: check_id
+      problem: problem
+      teeth: teeth
+      exclude_teeth: exclude_teeth
+  - kind: derive
+    input: image_analysis_problem_rows
+    output: image_analysis_problem_rows
+    fields:
+      _problem_normalized: { op: normalize, from: problem }
+      _excluded_positions: { op: recode, from: exclude_teeth, map: { ALL: '' } }
+
   # ── 患者 phone:宿主已按「默认号→本人→最早」挑好、inline 进 customer_basic_info 的
   #    contacts_tel / relationship 两列(沿用宿主源列名;不再 lookup customer_contacts)。patient.yaml 直接读。──
 
@@ -437,6 +464,7 @@ transforms:
         from: follow_content
 
 assemblers:
+  - { file: assemblers/image_analysis.yaml }
   - { file: assemblers/patient.yaml }
   - { file: assemblers/appointment.yaml }
   - { file: assemblers/encounter.yaml }

apps/pac-service/src/modules/facts/image-links.ts

+import { FactStatus, FactType } from '@pac/types';
+import type { PrismaService } from '../../prisma/prisma.service';
+
+interface ImageLinkFact {
+  hostId: string;
+  tenantId: string;
+  sourceUnit: string;
+  patientId: string;
+  clinicId: string | null;
+  type: string;
+  content: unknown;
+}
+
+/** 读取时关联当前影像地址，地址刷新不改分析/诊断事实；全作用域匹配。 */
+export async function attachImageLinks<T extends ImageLinkFact>(prisma: PrismaService, facts: T[]): Promise<T[]> {
+  const reference = (fact: ImageLinkFact): string | null => {
+    const c = asObject(fact.content);
+    const id = c.source_image_external_id ??
+      (fact.type === FactType.IMAGE_ANALYSIS_RECORD ? c.image_external_id : null);
+    return typeof id === 'string' && id ? id : null;
+  };
+  const refs = facts.flatMap(f => {
+    const id = reference(f);
+    return id ? [{ fact: f, id }] : [];
+  });
+  if (!refs.length) return facts;
+  const key = (fact: ImageLinkFact, id: string) => JSON.stringify([
+    fact.hostId, fact.tenantId, fact.sourceUnit, fact.patientId, fact.clinicId, id,
+  ]);
+  const distinct = new Map(refs.map(ref => [key(ref.fact, ref.id), ref]));
+  const images = await prisma.patientFact.findMany({ where: {
+    type: FactType.IMAGE_RECORD, status: FactStatus.ACTIVE,
+    OR: [...distinct.values()].map(({ fact: f, id }) => ({
+      hostId: f.hostId, tenantId: f.tenantId, sourceUnit: f.sourceUnit,
+      patientId: f.patientId, clinicId: f.clinicId,
+      content: { path: ['image_external_id'], equals: id },
+    })),
+  } });
+  const urls = new Map<string, string>();
+  for (const image of images) {
+    const c = asObject(image.content);
+    if (typeof c.image_external_id === 'string' && typeof c.file_url === 'string' && /^https?:\/\//i.test(c.file_url)) {
+      urls.set(key(image, c.image_external_id), c.file_url);
+    }
+  }
+  return facts.map(f => {
+    const id = reference(f);
+    const url = id ? urls.get(key(f, id)) : null;
+    return url ? { ...f, content: { ...asObject(f.content), image_file_url: url } } : f;
+  });
+}
+
+function asObject(value: unknown): Record<string, unknown> {
+  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
+}

apps/pac-service/src/modules/patient/patient.service.ts

+import { attachImageLinks } from '../facts/image-links';
 import { Injectable, NotFoundException } from '@nestjs/common';
 import type { Prisma } from '@prisma/client';
 import type { PatientTimelineResponse } from '@pac/types';
@@ -338,7 +339,7 @@ export class PatientService {
         birthDate: patient.birthDate ? patient.birthDate.toISOString() : null,
         status: patient.active ? 'active' : 'archived',
       },
-      facts: facts.map((f) => ({
+      facts: (await attachImageLinks(this.prisma, facts)).map((f) => ({
         id: f.id,
         subjectId: f.subjectId,
         kind: f.kind as PatientTimelineResponse['facts'][number]['kind'],

apps/pac-service/src/modules/plan-aggregate/plan-aggregate.service.ts

+import { attachImageLinks } from '../facts/image-links';
 import { Injectable, NotFoundException } from '@nestjs/common';
 import type { Prisma } from '@prisma/client';
 import { calcAge, maskName, maskPhone } from '@pac/utils';
@@ -394,13 +395,14 @@ export class PlanAggregateService {
       orderBy: { occurredAt: 'desc' },
     });
     // 取头后的时刻可能已订正;按当前时刻重排,保持原 DESC 的 NULLS FIRST 契约。
-    return (await currentFactsByIdentity(this.prisma, history))
+    const facts = (await currentFactsByIdentity(this.prisma, history))
       .filter((fact) => ['active', 'fulfilled'].includes(fact.status))
       .sort((a, b) => {
         if (!a.occurredAt) return b.occurredAt ? -1 : 0;
         if (!b.occurredAt) return 1;
         return b.occurredAt.getTime() - a.occurredAt.getTime();
       });
+    return attachImageLinks(this.prisma, facts);
   }
 }
 

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

@@ -55,6 +55,8 @@ export const JoinArraySchema = z.object({
   field_mapping: fieldMapping,
   /// 数组元素的枚举映射(key 是数组元素的 canonical 字段名)
   enum_mapping: z.record(z.string(), enumMappingForField).optional(),
+  keyword_mapping: z.record(z.string(), keywordMappingForField).optional(),
+  keyword_strip_clauses: z.array(z.string().min(1)).optional(),
 });
 export type JoinArrayConfig = z.infer<typeof JoinArraySchema>;
 
@@ -78,6 +80,8 @@ export const EmitsSchema = z.object({
   occurredAtField: z.string().min(1),
   /// 可选;parser 可据此知道产 actual 还是 planned fact
   kind: FactKindSchema.optional(),
+  /// 原文内容变化独立留账；完整分析默认开启，已有主体沿用时间版本键。
+  contentVersion: z.boolean().optional(),
 });
 export type EmitsConfig = z.infer<typeof EmitsSchema>;
 

apps/pac-service/src/modules/sync/assembler/dict-include.ts

@@ -62,6 +62,7 @@ interface RawAssemblerYaml {
   enum_mapping?: Record<string, Record<string, string>>;
   keyword_mapping?: Record<string, unknown[]>;
   keyword_strip_clauses?: string[];
+  join_arrays?: Record<string, RawAssemblerYaml>;
   [k: string]: unknown;
 }
 
@@ -74,7 +75,11 @@ interface RawAssemblerYaml {
 export function resolveDictIncludes(raw: unknown, yamlPath: string): unknown {
   if (raw === null || typeof raw !== 'object') return raw;
   const cfg = raw as RawAssemblerYaml;
-  if (cfg.dict_includes === undefined) return raw;
+  const joins = cfg.join_arrays
+    ? { join_arrays: Object.fromEntries(Object.entries(cfg.join_arrays)
+        .map(([key, child]) => [key, resolveDictIncludes(child, yamlPath)])) }
+    : {};
+  if (cfg.dict_includes === undefined) return cfg.join_arrays ? { ...cfg, ...joins } : raw;
 
   const incPaths = z.array(z.string().min(1)).min(1).safeParse(cfg.dict_includes);
   if (!incPaths.success) {
@@ -131,6 +136,7 @@ export function resolveDictIncludes(raw: unknown, yamlPath: string): unknown {
   const { dict_includes: _drop, ...rest } = cfg;
   return {
     ...rest,
+    ...joins,
     ...(Object.keys(mergedEnum).length > 0 ? { enum_mapping: mergedEnum } : {}),
     ...(Object.keys(mergedKeyword).length > 0 ? { keyword_mapping: mergedKeyword } : {}),
     ...(mergedStrip.length > 0 ? { keyword_strip_clauses: mergedStrip } : {}),

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

@@ -502,12 +502,13 @@ export class ColdImportService {
 
     const agg = perResource.reduce(
       (a, s) => ({
+        upserted: a.upserted + s.patientsUpserted,
         txn: a.txn + s.transactionsWritten,
         dup: a.dup + s.duplicates,
         failed: a.failed + s.failed,
         facts: a.facts + s.factsCreated + s.factsSuperseded,
       }),
-      { txn: 0, dup: 0, failed: 0, facts: 0 },
+      { upserted: 0, txn: 0, dup: 0, failed: 0, facts: 0 },
     );
     // ── raw 层列漂移检测(form A 专属,最可靠):本次推送行的列集 vs 历史同源 rawPayload 列集 ──
     //   transform 会掩盖 assembler 层的删列(补 key),故在原生行层面直接对比。基线 = 已存 transaction,
@@ -555,9 +556,11 @@ export class ColdImportService {
     // 全批失败(非重复、无一落库)按 FAILED 算 —— 典型是宿主改了关键列(身份/时间)导致逐行合成失败,
     // 旧口径只看 firstError 会把这种批标成 SUCCESS,宿主毫无感知地持续丢数据。
     // 顺带收紧:部分行失败(failed>0)从 SUCCESS 改为 PARTIAL,让状态如实。
-    const allRowsFailed = agg.failed > 0 && agg.txn === 0 && agg.dup === 0;
+    // patient / relation / return_visit 直接 upsert，不生成 transaction；同样算有效进展。
+    const hasProgress = agg.upserted + agg.txn + agg.dup > 0;
+    const allRowsFailed = agg.failed > 0 && !hasProgress;
     const status =
-      (firstError && agg.txn === 0) || allRowsFailed
+      (firstError && !hasProgress) || allRowsFailed
         ? SyncStatus.FAILED
         : firstError || agg.failed > 0
           ? SyncStatus.PARTIAL
@@ -1724,8 +1727,6 @@ export class ColdImportService {
     stats.failed = assembled.stats.failed;
     if (assembled.rows.length > 0)
       stats.sampleCanonical.push(assembled.rows[0]!.canonical);
-    if (dryRun) return stats;
-
     // 本租户 externalId → PAC patient id 索引(本人 + 关系人解析共用)
     const indexByTenant = new Map<string, Map<string, string>>();
     const getIndex = async (tenantId: string) => {
@@ -1751,6 +1752,10 @@ export class ColdImportService {
         stats.failed++;
         continue;
       }
+      if (dryRun) {
+        stats.patientsUpserted++;
+        continue;
+      }
       const sourceUnit = sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
       // 本人不在索引 → 跟 processSubject 同处理:建空壳主档,边照常落。
@@ -1820,8 +1825,6 @@ export class ColdImportService {
     stats.failed = assembled.stats.failed;
     if (assembled.rows.length > 0)
       stats.sampleCanonical.push(assembled.rows[0]!.canonical);
-    if (dryRun) return stats;
-
     const indexByTenant = new Map<string, Map<string, string>>();
     const getIndex = async (tenantId: string) => {
       let idx = indexByTenant.get(tenantId);
@@ -1845,6 +1848,10 @@ export class ColdImportService {
         stats.failed++;
         continue;
       }
+      if (dryRun) {
+        stats.patientsUpserted++;
+        continue;
+      }
       const sourceUnit = sourceUnitResolver.resolve(rawSource);
       const idx = await getIndex(tenantId);
       // 同 processPatientRelations:本人不在索引就建空壳,不再静默丢(见那边注释)

apps/pac-service/src/modules/sync/pipeline/fact-content-schemas.ts

@@ -120,6 +120,8 @@ const DiagnosisRecordContent = z
     doctor_name: nullableString(),
     /// 反查源接诊;算法 JOIN 凭这字段
     source_encounter_external_id: z.string().min(1),
+    source_image_external_id: z.string().min(1).optional(),
+    source_analysis_external_id: z.string().min(1).optional(),
     /// 从 name_zh 抽的临床信号(严重度只在诊断原文里 —— 生产 severity 列 296 万条 100% null)
     clinical_signals: clinicalSignals(),
   })
@@ -232,6 +234,9 @@ const EmrRecordContent = z
 const ImageRecordContent = z
   .object({
     image_external_id: z.string().min(1),
+    file_url: z.string().url().regex(/^https?:\/\//i).optional(),
+    file_name: z.string().optional(),
+    file_type: z.string().optional(),
     encounter_external_id: nullableString(),
     modality: PACImageModalitySchema,
     tooth_positions: z.array(z.string()).nullable().optional().default(null),
@@ -243,6 +248,21 @@ const ImageRecordContent = z
   })
   .passthrough();
 
+const ImageAnalysisRecordContent = z.object({
+  analysis_external_id: z.string().min(1),
+  image_external_id: z.string().min(1),
+  code_source: z.literal('image_ai'),
+  status: z.enum(['pending', 'completed', 'failed']),
+  analyzed_at: nullableString(),
+  source_updated_at: nullableString(),
+  analysis_result: nullableString(),
+  findings: z.array(z.object({
+    externalId: z.string().min(1),
+    code: PACDiagnosisCodeSchema.nullable(),
+    name: z.string().min(1),
+  }).passthrough()),
+}).passthrough();
+
 /**
  * appointment_record — kind=planned/actual 区分
  *
@@ -467,6 +487,7 @@ export const FactContentSchemas = {
   [FactType.RECOMMENDATION_RECORD]: RecommendationRecordContent,
   [FactType.EMR_RECORD]: EmrRecordContent,
   [FactType.IMAGE_RECORD]: ImageRecordContent,
+  [FactType.IMAGE_ANALYSIS_RECORD]: ImageAnalysisRecordContent,
   [FactType.APPOINTMENT_RECORD]: AppointmentRecordContent,
   [FactType.CONSULTATION_RECORD]: ConsultationRecordContent,
   [FactType.VISIT_REGISTRATION_RECORD]: VisitRegistrationRecordContent,

apps/pac-service/src/modules/sync/pipeline/fact-writer.service.ts

 import { Injectable, Logger } from '@nestjs/common';
 import { createHash } from 'node:crypto';
-import { FactStatus } from '@pac/types';
+import { FactStatus, type FactType, type FactKind } from '@pac/types';
 import type { Prisma } from '@prisma/client';
 import { PrismaService } from '../../../prisma/prisma.service';
-import type { FactDraft } from './parsers/parser.interface';
+import type { FactDraft, FactSnapshotScope } from './parsers/parser.interface';
 import {
   FactContentSchemaError,
   validateFactContent,
@@ -43,6 +43,145 @@ export class FactWriter {
 
   constructor(private readonly prisma: PrismaService) {}
 
+  /** 完整源快照：加锁后比较锚点水位，补取消成员，复用同一事实版本写入器。 */
+  async writeSnapshots(snapshots: SnapshotWriteInput[]): Promise<BulkWriteOutcome> {
+    if (!snapshots.length) return { results: [], rejected: [] };
+    const selected = new Map<string, SnapshotWriteInput>();
+    for (const snapshot of snapshots) {
+      if (!snapshot.entries.some(e => e.draft.subjectId === snapshot.scope.anchorSubjectId)) {
+        throw new Error('snapshot anchor draft required');
+      }
+      for (const entry of snapshot.entries) {
+        if (entry.hostId !== snapshot.hostId || entry.tenantId !== snapshot.tenantId ||
+          entry.sourceUnit !== snapshot.sourceUnit || entry.patientId !== snapshot.patientId ||
+          entry.transactionId !== snapshot.transactionId) throw new Error('snapshot entry scope differs');
+      }
+      const key = JSON.stringify([snapshot.hostId, snapshot.tenantId, snapshot.sourceUnit, snapshot.scope.anchorSubjectId]);
+      selected.set(key, snapshot);
+    }
+    return this.prisma.$transaction(async (tx) => {
+      // ⛔ 不允许快照按成员分别降级写入，空快照和旧快照都必须经过同一锚点锁。
+      await tx.$executeRaw`
+        SELECT pg_advisory_xact_lock(hashtextextended(k, 0))
+        FROM (SELECT unnest(${[...selected.keys()].sort()}::text[]) AS k ORDER BY 1) AS locks
+      `;
+      const identity = (s: SnapshotWriteInput) => ({ hostId: s.hostId, tenantId: s.tenantId, sourceUnit: s.sourceUnit });
+      const anchors = await tx.patientFact.findMany({
+        where: { OR: snapshots.flatMap(s => [s.scope.anchorSubjectId,
+          ...(s.scope.references ?? []).map(ref => ref.subjectId)]
+          .map(subjectId => ({ ...identity(s), subjectId }))) },
+        orderBy: { version: 'desc' },
+      });
+      const keyOf = (hostId: string, tenantId: string, sourceUnit: string, subjectId: string) =>
+        JSON.stringify([hostId, tenantId, sourceUnit, subjectId]);
+      const latest = new Map<string, typeof anchors[number]>();
+      for (const anchor of anchors) {
+        const key = keyOf(anchor.hostId, anchor.tenantId, anchor.sourceUnit, anchor.subjectId);
+        if (!latest.has(key)) latest.set(key, anchor);
+      }
+      const transactionIds = [...new Set([
+        ...snapshots.map(s => s.transactionId),
+        ...[...latest.values()].flatMap(f => f.transactionIds),
+      ])];
+      const transactions = await tx.patientTransaction.findMany({
+        where: { id: { in: transactionIds } }, select: { id: true, eventSeq: true },
+      });
+      const sequences = new Map(transactions.map(t => [t.id, t.eventSeq]));
+      // 同批同锚点先按源更新时间裁决，再按已留账的事件序号。
+      // ⛔ 不允许旧报文重试仅因排在批尾而挤掉新快照（明细更新可不改父行时间）。
+      selected.clear();
+      for (const s of snapshots) {
+        const key = keyOf(s.hostId, s.tenantId, s.sourceUnit, s.scope.anchorSubjectId);
+        const previous = selected.get(key);
+        if (previous) {
+          if (previous.sourceUpdatedAt && s.sourceUpdatedAt &&
+            previous.sourceUpdatedAt.getTime() !== s.sourceUpdatedAt.getTime()) {
+            if (s.sourceUpdatedAt < previous.sourceUpdatedAt) continue;
+          } else {
+            const previousSeq = sequences.get(previous.transactionId);
+            const incomingSeq = sequences.get(s.transactionId);
+            if (previousSeq != null && incomingSeq != null && incomingSeq < previousSeq) continue;
+          }
+        }
+        selected.set(key, s);
+      }
+      const groups = [...selected.values()];
+      const results: FactWriteResult[] = [];
+      const rejected: FactReject[] = [];
+      const rejectSnapshot = (s: SnapshotWriteInput, message: string) => {
+        rejected.push({ type: s.entries[0]!.draft.type, subjectId: s.scope.anchorSubjectId,
+          transactionId: s.transactionId, issues: [message] });
+      };
+      const accepted = groups.filter(s => {
+        for (const reference of s.scope.references ?? []) {
+          const fact = latest.get(keyOf(s.hostId, s.tenantId, s.sourceUnit, reference.subjectId));
+          if (fact && (fact.patientId !== s.patientId || (fact.clinicId && fact.clinicId !== s.entries[0]?.draft.clinicId))) {
+            rejectSnapshot(s, 'snapshot reference ownership differs');
+            return false;
+          }
+        }
+        const anchor = latest.get(keyOf(s.hostId, s.tenantId, s.sourceUnit, s.scope.anchorSubjectId));
+        if (!anchor) return true;
+        if (anchor.patientId !== s.patientId || (anchor.clinicId && anchor.clinicId !== s.entries[0]?.draft.clinicId)) {
+          rejectSnapshot(s, 'snapshot identity changed');
+          return false;
+        }
+        const olderTime = !!(anchor.sourceUpdatedAt && s.sourceUpdatedAt && s.sourceUpdatedAt < anchor.sourceUpdatedAt);
+        const newerTime = !!(anchor.sourceUpdatedAt && s.sourceUpdatedAt && s.sourceUpdatedAt > anchor.sourceUpdatedAt);
+        const incomingSeq = sequences.get(s.transactionId);
+        const latestSeq = anchor.transactionIds.reduce<bigint>((max, id) => {
+          const seq = sequences.get(id);
+          return seq != null && seq > max ? seq : max;
+        }, 0n);
+        // 同一旧报文重试不能在时间戳不变时恢复已撤回的成员。
+        const olderEvidence = !newerTime && incomingSeq != null && incomingSeq < latestSeq;
+        if (!olderTime && !olderEvidence) return true;
+        results.push({ action: 'stale_skipped', factId: anchor.id, subjectId: anchor.subjectId, version: anchor.version });
+        return false;
+      });
+      const replacing = accepted.filter(s => s.scope.replace);
+      const members = replacing.length ? await tx.patientFact.findMany({
+        where: { status: FactStatus.ACTIVE, OR: replacing.map(s => ({
+          ...identity(s), type: s.scope.memberType,
+          content: { path: [s.scope.contentKey], equals: s.scope.externalId },
+        })) },
+      }) : [];
+      const entries: BulkEntry[] = [];
+      for (const s of accepted) {
+        const wanted = new Set(s.entries.map(e => e.draft.subjectId));
+        const groupEntries = [...s.entries];
+        let ownershipError = false;
+        for (const old of s.scope.replace ? members : []) {
+          const content = old.content as Record<string, unknown>;
+          if (old.hostId !== s.hostId || old.tenantId !== s.tenantId || old.sourceUnit !== s.sourceUnit ||
+            old.type !== s.scope.memberType || content[s.scope.contentKey] !== s.scope.externalId) continue;
+          if (old.patientId !== s.patientId || (old.clinicId && old.clinicId !== s.entries[0]?.draft.clinicId)) {
+            ownershipError = true;
+            break;
+          }
+          if (wanted.has(old.subjectId)) continue;
+          groupEntries.push({ ...s.entries[0]!, draft: {
+            subjectId: old.subjectId, type: old.type as FactType, kind: old.kind as FactKind,
+            status: FactStatus.CANCELLED, clinicId: old.clinicId, occurredAt: old.occurredAt,
+            plannedFor: old.plannedFor, validFrom: old.validFrom, validUntil: old.validUntil,
+            title: old.title, summary: old.summary, content: old.content as Prisma.InputJsonValue,
+          } });
+        }
+        if (ownershipError) { rejectSnapshot(s, 'snapshot member ownership changed'); continue; }
+        // 校验失败隔离整份分析，其余分析照常；不可只发布头或撤销部分成员。
+        try {
+          for (const entry of groupEntries) validateFactContent(entry.draft.type, entry.draft.subjectId, entry.draft.content);
+        } catch (err) {
+          rejectSnapshot(s, err instanceof Error ? err.message : String(err));
+          continue;
+        }
+        for (const entry of groupEntries) entries.push(entry);
+      }
+      const written = entries.length ? await this.bulkWrite(entries, tx) : { results: [], rejected: [] };
+      return { results: [...results, ...written.results], rejected: [...rejected, ...written.rejected] };
+    }, { maxWait: 30_000, timeout: 120_000 });
+  }
+
   async writeDraft(input: {
     draft: FactDraft;
     hostId: string;
@@ -233,7 +372,7 @@ export class FactWriter {
    *   对方收不到成功回执、无限重推同一批,实测卡了 13 小时一条没进。
    *   一行脏数据不该让同批另外 199 行陪葬 —— 隔离它、如实回报它,批照常走。
    */
-  async bulkWrite(entries: BulkEntry[]): Promise<BulkWriteOutcome> {
+  async bulkWrite(entries: BulkEntry[], client?: Prisma.TransactionClient): Promise<BulkWriteOutcome> {
     if (entries.length === 0) return { results: [], rejected: [] };
 
     // 假设全 batch 同 hostId+tenantId(processSubject 是按 host+tenant 跑的,符合)
@@ -274,7 +413,7 @@ export class FactWriter {
     }
     if (validatedEntries.length === 0) return { results: [], rejected };
 
-    return this.prisma.$transaction(async (tx) => {
+    const writeBatch = async (tx: Prisma.TransactionClient) => {
       await this.lockIdentities(tx, hostId, tenantId, validatedEntries.map((e) => ({
         sourceUnit: e.sourceUnit, subjectId: e.draft.subjectId,
       })));
@@ -465,7 +604,9 @@ export class FactWriter {
       }
       if (toCreate.length > 0) await tx.patientFact.createMany({ data: toCreate });
       return { results, rejected };
-    }, { maxWait: 30_000, timeout: 120_000 });
+    };
+    return client ? writeBatch(client)
+      : this.prisma.$transaction(writeBatch, { maxWait: 30_000, timeout: 120_000 });
   }
 
   /** 版本链无行时也能锁;一条 SQL 批量取锁,顺序固定避免交叉批死锁。 */
@@ -622,3 +763,14 @@ export interface BulkEntry {
   /// 源记录末次修改时间(canonicalRow.updatedAt);乱序防护裁决用,null = 源没给
   sourceUpdatedAt?: Date | null;
 }
+
+export interface SnapshotWriteInput {
+  hostId: string;
+  tenantId: string;
+  sourceUnit: string;
+  patientId: string;
+  transactionId: string;
+  sourceUpdatedAt?: Date | null;
+  scope: FactSnapshotScope;
+  entries: BulkEntry[];
+}

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

@@ -8,6 +8,8 @@ import {
   type FactReject,
   FactWriter,
   FactWriteResult,
+  type SnapshotWriteInput,
+  type BulkWriteOutcome,
 } from './fact-writer.service';
 import { FactContentSchemaError } from './fact-content-schemas';
 
@@ -20,7 +22,7 @@ import { FactContentSchemaError } from './fact-content-schemas';
  *   2. parser.parse(ctx) → FactDraft[]
  *   3. ClinicalSignalService.enrich —— 按 clinical-signals.yaml 配置的字段抽临床信号,
  *      命中则写进 draft.content.clinical_signals(**不新建衍生 fact**,见该 service 注释)
- *   4. 逐个调用 FactWriter.writeDraft
+ *   4. FactWriter 写版本流；完整源快照统一原子替换成员
  *   5. 汇总 metrics(created / superseded / unchanged / evidence_appended)
  *
  * 信号抽取放在这一层而不是各 parser 里:源字段跨 fact 类型且由 yaml 配置
@@ -28,8 +30,9 @@ import { FactContentSchemaError } from './fact-content-schemas';
  *
  * 设计:
  *   - 无 parser 的 action 静默跳过(返回 zero metrics)— transaction 已落账,稍后补 parser 重放即可
- *   - 单条 transaction 的多 fact 衍生**不开外层事务** — 每 draft 一次 $transaction(FactWriter 内部)
+ *   - 普通 transaction 的多 fact 衍生**不开外层事务** — 每 draft 一次 $transaction(FactWriter 内部)
  *     原因:外层事务跨多 subject 会放大锁范围;每 fact 独立写更适配未来 BullMQ jobName=subject_id 串行化
+ *   - 完整源快照按锚点加锁，头与成员同一事务；失败不可降级为逐成员写入
  *   - parser 抛错时单条 transaction 整体计 failed,不影响后续 transaction(由上层捕获)
  */
 @Injectable()
@@ -104,6 +107,20 @@ export class ParserPipeline {
         })
       )?.sourceUnit ?? '';
 
+    const scope = parser.snapshotScope?.({ transaction, canonicalRow });
+    if (scope) {
+      try {
+        const entries = drafts.map(draft => ({ draft, hostId: transaction.hostId,
+          tenantId: transaction.tenantId, sourceUnit: patientSourceUnit,
+          patientId: transaction.patientId!, transactionId: transaction.id, sourceUpdatedAt }));
+        applyOutcome(metrics, await this.writer.writeSnapshots([{ ...entries[0]!, scope, entries }]));
+      } catch (err) {
+        metrics.factsFailed++;
+        this.logger.error(`snapshot write failed: ${err instanceof Error ? err.message : String(err)}`);
+      }
+      return metrics;
+    }
+
     for (const draft of drafts) {
       try {
         const result = await this.writer.writeDraft({
@@ -180,6 +197,7 @@ export class ParserPipeline {
 
     // 1. 全部 tx 走 parser,收集 drafts(in-memory)
     const bulkEntries: BulkEntry[] = [];
+    const snapshots: SnapshotWriteInput[] = [];
     for (const item of items) {
       const parser = this.registry.get(item.transaction.action);
       if (!parser) {
@@ -196,8 +214,9 @@ export class ParserPipeline {
           item.transaction.occurredAt,
         );
         const sourceUpdatedAt = parseSourceUpdatedAt(item.canonicalRow);
+        const entries: BulkEntry[] = [];
         for (const draft of drafts) {
-          bulkEntries.push({
+          entries.push({
             draft,
             hostId: item.transaction.hostId,
             tenantId: item.transaction.tenantId,
@@ -207,6 +226,9 @@ export class ParserPipeline {
             sourceUpdatedAt,
           });
         }
+        const scope = parser.snapshotScope?.({ transaction: item.transaction, canonicalRow: item.canonicalRow });
+        if (scope && entries.length) snapshots.push({ ...entries[0]!, scope, entries });
+        else for (const entry of entries) bulkEntries.push(entry);
       } catch (err) {
         metrics.factsFailed++;
         this.logger.error(
@@ -216,6 +238,14 @@ export class ParserPipeline {
       }
     }
 
+    if (snapshots.length) {
+      try {
+        applyOutcome(metrics, await this.writer.writeSnapshots(snapshots));
+      } catch (err) {
+        metrics.factsFailed += snapshots.length;
+        this.logger.error(`snapshot batch rolled back: ${err instanceof Error ? err.message : String(err)}`);
+      }
+    }
     if (bulkEntries.length === 0) return metrics;
 
     // 2. 一次 bulk write,失败降级 per-entry(写一份保证收尾)
@@ -307,6 +337,21 @@ export class ParserPipeline {
   }
 }
 
+function applyOutcome(metrics: PipelineRunMetrics, outcome: BulkWriteOutcome): void {
+  metrics.factsFailed += outcome.rejected.length;
+  for (const reject of outcome.rejected) metrics.factRejects.push(reject);
+  for (const result of outcome.results) {
+    metrics.writes.push(result);
+    switch (result.action) {
+      case 'created': metrics.factsCreated++; break;
+      case 'superseded': metrics.factsSuperseded++; break;
+      case 'unchanged': metrics.factsUnchanged++; break;
+      case 'evidence_appended': metrics.factsEvidenceAppended++; break;
+      case 'stale_skipped': metrics.factsStaleSkipped++; break;
+    }
+  }
+}
+
 export interface BatchItem {
   transaction: {
     id: string;

apps/pac-service/src/modules/sync/pipeline/parsers/diagnosis.parser.ts

@@ -87,6 +87,8 @@ export class DiagnosisParser implements Parser {
           doctor_id: c.doctorId ? String(c.doctorId) : null, // host user_id Int64 → string
           doctor_name: (c.doctorName as string | undefined) ?? null,
           source_encounter_external_id: sourceEncounter ?? externalId,
+          ...(c.sourceImageExternalId ? { source_image_external_id: String(c.sourceImageExternalId) } : {}),
+          ...(c.sourceAnalysisExternalId ? { source_analysis_external_id: String(c.sourceAnalysisExternalId) } : {}),
         },
       },
     ];

apps/pac-service/src/modules/sync/pipeline/parsers/image-analysis.parser.ts

+import { Injectable } from '@nestjs/common';
+import { Action, FactKind, FactStatus, FactType, ImageAnalysisCanonicalSchema } from '@pac/types';
+import { DiagnosisParser } from './diagnosis.parser';
+import type { FactDraft, FactSnapshotScope, Parser, ParserContext } from './parser.interface';
+import { toothSet } from './tooth-position.util';
+
+/** 通用宿主影像分析：标准输入 → 源快照 + 现有诊断事实。 */
+@Injectable()
+export class ImageAnalysisParser implements Parser {
+  readonly name = Action.IMAGE_ANALYSIS_RECORDED;
+
+  constructor(private readonly diagnosis: DiagnosisParser) {}
+
+  snapshotScope(ctx: ParserContext): FactSnapshotScope {
+    const c = ImageAnalysisCanonicalSchema.parse(ctx.canonicalRow);
+    return {
+      anchorSubjectId: `${FactType.IMAGE_ANALYSIS_RECORD}:${c.externalId}`,
+      memberType: FactType.DIAGNOSIS_RECORD,
+      contentKey: 'source_analysis_external_id',
+      externalId: c.externalId,
+      replace: c.status === 'completed' && !!validDate(c.analyzedAt),
+      references: [{ subjectId: `${FactType.IMAGE_RECORD}:${c.imageExternalId}` }],
+    };
+  }
+
+  parse(ctx: ParserContext): FactDraft[] {
+    const c = ImageAnalysisCanonicalSchema.parse(ctx.canonicalRow);
+    const analyzedAt = validDate(c.analyzedAt);
+    const drafts: FactDraft[] = [{
+      subjectId: `${FactType.IMAGE_ANALYSIS_RECORD}:${c.externalId}`,
+      kind: FactKind.ACTUAL,
+      type: FactType.IMAGE_ANALYSIS_RECORD,
+      status: FactStatus.ACTIVE,
+      clinicId: c.clinicId,
+      occurredAt: analyzedAt,
+      title: '影像分析',
+      summary: c.findings.map(f => f.name).join('、') || null,
+      content: {
+        analysis_external_id: c.externalId,
+        image_external_id: c.imageExternalId,
+        code_source: 'image_ai',
+        status: c.status,
+        analyzed_at: analyzedAt?.toISOString() ?? null,
+        source_updated_at: validDate(c.updatedAt)?.toISOString() ?? null,
+        analysis_result: c.result ?? null,
+        findings: c.findings.map(f => ({ ...f, code: f.code || null })),
+      },
+    }];
+
+    // 未完成或时间未知只留源快照，不制造新的临床信号，也不清掉旧有效结果。
+    if (c.status !== 'completed' || !analyzedAt) return drafts;
+    for (const finding of c.findings) {
+      const teeth = toothSet(finding.toothPosition);
+      for (const excluded of toothSet(finding.excludedToothPositions)) teeth.delete(excluded);
+      const externalId = `image_ai:${encodeURIComponent(c.externalId)}:${encodeURIComponent(finding.externalId)}`;
+      const diagnosis = this.diagnosis.parse({
+        ...ctx,
+        transaction: { ...ctx.transaction, occurredAt: analyzedAt },
+        canonicalRow: {
+          externalId,
+          code: finding.code,
+          name: finding.name,
+          codeSource: 'image_ai',
+          toothPosition: [...teeth].sort().join(';') || null,
+          sourceImageExternalId: c.imageExternalId,
+          sourceAnalysisExternalId: c.externalId,
+          sourceEncounterExternalId: c.encounterExternalId ?? null,
+        },
+      });
+      if (finding.excluded || (!!finding.toothPosition?.trim() && teeth.size === 0)) {
+        for (const draft of diagnosis) draft.status = FactStatus.CANCELLED;
+      }
+      drafts.push(...diagnosis);
+    }
+    return drafts;
+  }
+}
+
+function validDate(value: unknown): Date | null {
+  if (typeof value !== 'string' || !value.trim()) return null;
+  const date = new Date(value);
+  return Number.isFinite(date.getTime()) ? date : null;
+}

apps/pac-service/src/modules/sync/pipeline/parsers/image.parser.ts

@@ -7,7 +7,7 @@ import { normalizeToothPosition } from './tooth-position.util';
  * ImageParser — `image_uploaded` 解析器
  *
  * 产 1 个 `image_record`(actual),只装 metadata + 医生 finding,**不持文件本体**。
- * 影像 AI 识别结果归 agent_invocations.output,不进 fact 层。
+ * 宿主分析另走通用 image_analysis；PAC 自身模型产物归 agent_invocations.output。
  */
 @Injectable()
 export class ImageParser implements Parser {
@@ -48,6 +48,9 @@ export class ImageParser implements Parser {
           encounter_external_id: encounterExternalId,
           doctor_id: c.doctorId ? String(c.doctorId) : null,
           doctor_name: (c.doctorName as string | undefined) ?? null,
+          ...(c.fileUrl ? { file_url: String(c.fileUrl) } : {}),
+          ...(c.fileName ? { file_name: String(c.fileName) } : {}),
+          ...(c.fileType ? { file_type: String(c.fileType) } : {}),
         },
       },
     ];

apps/pac-service/src/modules/sync/pipeline/parsers/parser.interface.ts

@@ -74,4 +74,15 @@ export interface Parser {
   /// 标识用(日志 / 调试),通常等于 Action 字符串
   readonly name: string;
   parse(ctx: ParserContext): FactDraft[];
+  /// 完整源快照的成员替换；统一由 FactWriter 原子更新，不在宿主入口删除事实。
+  snapshotScope?(ctx: ParserContext): FactSnapshotScope;
+}
+
+export interface FactSnapshotScope {
+  anchorSubjectId: string;
+  memberType: FactType;
+  contentKey: string;
+  externalId: string;
+  replace: boolean;
+  references?: Array<{ subjectId: string }>;
 }

apps/pac-service/src/modules/sync/pipeline/parsers/parser.registry.ts

@@ -10,6 +10,7 @@ import { OrderParser } from './order.parser';
 import { PaymentParser } from './payment.parser';
 import { EmrParser } from './emr.parser';
 import { ImageParser } from './image.parser';
+import { ImageAnalysisParser } from './image-analysis.parser';
 import { RechargeParser } from './recharge.parser';
 import { ComplaintParser } from './complaint.parser';
 import { ReferralParser } from './referral.parser';
@@ -50,6 +51,7 @@ export class ParserRegistry {
     private readonly consultationParser: ConsultationParser,
     private readonly visitRegistrationParser: VisitRegistrationParser,
     private readonly visitReceptionParser: VisitReceptionParser,
+    private readonly imageAnalysisParser: ImageAnalysisParser,
   ) {
     // 接诊(只元数据)
     this.register(Action.DOCTOR_ENCOUNTER_STARTED, encounterParser);
@@ -78,6 +80,7 @@ export class ParserRegistry {
 
     // 影像
     this.register(Action.IMAGE_UPLOADED, imageParser);
+    this.register(Action.IMAGE_ANALYSIS_RECORDED, imageAnalysisParser);
 
     // 接待流程
     this.register(Action.VISIT_REGISTERED, visitRegistrationParser);

apps/pac-service/src/modules/sync/pipeline/parsers/tooth-position.util.ts

@@ -74,6 +74,6 @@ export function toothSet(input: string | null | undefined): Set<string> {
   const norm = normalizeToothPosition(input);
   if (!norm) return new Set();
   return new Set(
-    norm.split(';').map((t) => t.trim()).filter(Boolean).filter(isValidToothToken),
+    norm.split(/[;,；，]/).map((t) => t.trim()).filter(Boolean).filter(isValidToothToken),
   );
 }

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

@@ -12,6 +12,7 @@ import { OrderParser } from './parsers/order.parser';
 import { PaymentParser } from './parsers/payment.parser';
 import { EmrParser } from './parsers/emr.parser';
 import { ImageParser } from './parsers/image.parser';
+import { ImageAnalysisParser } from './parsers/image-analysis.parser';
 import { RechargeParser } from './parsers/recharge.parser';
 import { ComplaintParser } from './parsers/complaint.parser';
 import { ReferralParser } from './parsers/referral.parser';
@@ -35,6 +36,7 @@ import { VisitReceptionParser } from './parsers/visit-reception.parser';
     PaymentParser,
     EmrParser,
     ImageParser,
+    ImageAnalysisParser,
     RechargeParser,
     ComplaintParser,
     ReferralParser,

apps/pac-service/src/modules/sync/pipeline/transaction-synthesizer.ts

@@ -120,7 +120,11 @@ export class TransactionSynthesizer {
     const updatedAtMark =
       (c.updatedAt as string | undefined) ?? (c.createdAt as string | undefined) ?? '';
     // source_unit 并进幂等键:集团内同 subjectId 跨品牌(瑞尔/瑞泰 encounter 1011032)是两条不同事件
-    const sourceEventId = `${input.sourceContext}:${input.resource}:${input.sourceUnit}:${subjectId}:${updatedAtMark}`;
+    // 完整分析快照默认按内容留账；原表与 canonical push 入口语义一致。
+    const versionByContent = emits.contentVersion ?? (subjectType === SubjectType.IMAGE_ANALYSIS);
+    const contentVersion = versionByContent
+      ? `:${this.sha256(JSON.stringify(sortKeys(input.rawRow)))}` : '';
+    const sourceEventId = `${input.sourceContext}:${input.resource}:${input.sourceUnit}:${subjectId}:${updatedAtMark}${contentVersion}`;
 
     const payloadHash = this.sha256(JSON.stringify(input.rawRow));
 
@@ -157,3 +161,14 @@ export class TransactionSynthesizer {
     return createHash('sha256').update(s).digest('hex');
   }
 }
+
+/** 内容身份不受对象键顺序影响；数组保留源顺序。 */
+function sortKeys(value: unknown): unknown {
+  if (value instanceof Date) return value.toISOString();
+  if (Array.isArray(value)) return value.map(sortKeys);
+  if (value && typeof value === 'object') {
+    return Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b))
+      .map(([key, item]) => [key, sortKeys(item)]));
+  }
+  return value;
+}

apps/pac-service/src/modules/sync/push/push-receiver.service.ts

@@ -65,13 +65,14 @@ export class PushReceiverService {
 
     const agg = r.perResource.reduce(
       (a, s) => ({
+        upserted: a.upserted + s.patientsUpserted,
         txn: a.txn + s.transactionsWritten,
         dup: a.dup + s.duplicates,
         failed: a.failed + s.failed,
         factsFailed: a.factsFailed + s.factsFailed,
         stubs: a.stubs + s.patientStubsCreated,
       }),
-      { txn: 0, dup: 0, failed: 0, factsFailed: 0, stubs: 0 },
+      { upserted: 0, txn: 0, dup: 0, failed: 0, factsFailed: 0, stubs: 0 },
     );
 
     // fact.content 被 schema 拒收的明细 —— 必须透出去,否则宿主只能看到"成功但没进数据"。
@@ -85,10 +86,10 @@ export class PushReceiverService {
       );
     }
 
-    // ── 关键字段守卫:整批全失败(非重复、无一落库)→ 拒收报错,别让宿主静默丢数据 ──
+    // ── 关键字段守卫:整批全失败(无 upsert / transaction / 重复命中)→ 拒收报错 ──
     // 典型成因:宿主改了身份/时间列名 → 逐行合成失败。失败行不落任何库(raw 只活在 txn),
     // 拒绝是无损的;宿主修好后整批重推即可(幂等)。SyncLog 已标 FAILED + 告警(ingestRawTables 内)。
-    if (rows.length > 0 && agg.txn === 0 && agg.dup === 0 && agg.failed > 0) {
+    if (rows.length > 0 && agg.upserted === 0 && agg.txn === 0 && agg.dup === 0 && agg.failed > 0) {
       throw new BizError(
         ApiCode.SYNC_CONTRACT_DRIFT,
         `整批 ${agg.failed}/${rows.length} 行全部失败 —— 疑似关键字段缺失/改名,请核对数据字典` +

apps/pac-service/src/modules/sync/transforms/operators/derive.op.ts

 import type { DeriveExpr, DeriveOp } from '../transforms.schema';
-import { isEmptyText, type Row } from '../row';
+import { isEmptyText, tryParseJsonArray, type Row } from '../row';
 
 /**
  * derive — 给行加新字段(in-place,output 可以同 input)。
@@ -64,6 +64,12 @@ export function runDerive(op: DeriveOp, rows: Row[]): Row[] {
 
 function evalExpr(expr: DeriveExpr, row: Row): unknown {
   switch (expr.op) {
+    case 'array_length': {
+      const value = row[expr.from];
+      // 空串/缺失不能冒充明确的空快照。
+      if (typeof value === 'string' && !value.trim()) return null;
+      return tryParseJsonArray(value)?.length ?? null;
+    }
     case 'substring': {
       const v = row[expr.from];
       if (typeof v !== 'string') return v ?? null;

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

@@ -205,6 +205,7 @@ export const DeriveExprSchema = z.discriminatedUnion('op', [
   DeriveNormalizeSchema,
   DeriveCoalesceSchema,
   DeriveRecodeSchema,
+  z.object({ op: z.literal('array_length'), from: z.string().min(1) }),
 ]);
 export type DeriveExpr = z.infer<typeof DeriveExprSchema>;
 

apps/pac-service/tests/canonical-fact-layer.spec.ts

@@ -22,6 +22,7 @@ import {
   PACTreatmentStatuses,
   PACImageModalities,
   FDI_TOOTH_POSITIONS,
+  ImageAnalysisCanonicalSchema,
 } from '@pac/types';
 import { FactContentSchemas } from '../src/modules/sync/pipeline/fact-content-schemas';
 import {
@@ -193,7 +194,9 @@ describe('canonical-fact-layer | 闸 4:yaml enum_mapping 目标 ∈ canonical-co
         });
         continue;
       }
-      const allowed = closedSets[canonicalField];
+      const allowed = canonicalField === 'status' && config.canonical === 'image_analysis'
+        ? new Set<string>(ImageAnalysisCanonicalSchema.shape.status.options)
+        : closedSets[canonicalField];
       if (!allowed) continue; // 不在我们校验范围内的字段(host-specific transform 等)
       test(`${host}/${file}:enum_mapping.${canonicalField} 目标 ∈ canonical-codes`, () => {
         const bad: string[] = [];

apps/pac-service/tests/image-analysis-fixtures.ts

+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import * as yaml from 'js-yaml';
+import { Action, type ImageAnalysisCanonical } from '@pac/types';
+import { AssemblerEngine } from '../src/modules/sync/assembler/assembler-engine';
+import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.schema';
+import { resolveDictIncludes } from '../src/modules/sync/assembler/dict-include';
+import { TransformEngine } from '../src/modules/sync/transforms/transform-engine';
+import { TransformsSchema } from '../src/modules/sync/transforms/transforms.schema';
+import type { ParserContext } from '../src/modules/sync/pipeline/parsers/parser.interface';
+
+export function analysisRow(overrides: Record<string, unknown> = {}) {
+  return {
+    id: '1001', tenant_id: 'brand-a', organization_id: 'clinic-a', patient_id: '4001', check_id: '3001',
+    req_status: 'NOTIFY_OK', analysis_time: '2026-10-12 10:05:00',
+    created_gmt_at: '2026-10-12 10:01:00', updated_gmt_at: '2026-10-12 10:06:00',
+    analysis_data: '{"sourceResult":true}',
+    problemList: [{ id: '2001', patient_id: '4001', check_id: '3001',
+      problem_code: 'P0002', problem: '龋齿', teeth: '16,26', exclude_teeth: '26' }],
+    ...overrides,
+  };
+}
+
+export function assembleFriday(rows: Record<string, unknown>[], source = 'med_check_analysis_req_record') {
+  const dir = join(__dirname, '../data/friday');
+  const manifest = yaml.load(readFileSync(join(dir, 'manifest.yaml'), 'utf8')) as { transforms: unknown };
+  const path = join(dir, 'assemblers', source === 'med_check' ? 'image.yaml' : 'image_analysis.yaml');
+  const config = AssemblerConfigSchema.parse(resolveDictIncludes(yaml.load(readFileSync(path, 'utf8')), path));
+  const tables = new TransformEngine().run({ tables: { [source]: rows }, transforms: TransformsSchema.parse(manifest.transforms) });
+  return new AssemblerEngine().assemble({ tables, config,
+    normalize: { amountUnit: 'yuan', timezone: 'Asia/Shanghai' }, hostId: 'host-demo' });
+}
+
+export function canonicalAnalysis(overrides: Record<string, unknown> = {}): ImageAnalysisCanonical {
+  const result = assembleFriday([analysisRow(overrides)]);
+  if (result.errors.length) throw new Error(JSON.stringify(result.errors.map(e => e.reason)));
+  return result.rows[0]!.canonical as ImageAnalysisCanonical;
+}
+
+export function parserContext(canonicalRow: Record<string, unknown>, overrides: Partial<ParserContext['transaction']> = {}): ParserContext {
+  return { canonicalRow, transaction: {
+    id: 'tx-demo', hostId: 'host-demo', tenantId: 'tenant-demo', patientId: 'patient-demo',
+    action: Action.IMAGE_ANALYSIS_RECORDED, subjectType: 'image_analysis', subjectId: String(canonicalRow.externalId),
+    occurredAt: new Date('2026-10-12T02:05:00Z'), clinicId: 'clinic-a', ...overrides,
+  } };
+}

apps/pac-service/tests/image-analysis-ingestion.spec.ts

+import { Logger } from '@nestjs/common';
+import { Action, FactStatus, FactType, ImageAnalysisCanonicalSchema } from '@pac/types';
+import { AssemblerEngine } from '../src/modules/sync/assembler/assembler-engine';
+import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.schema';
+import { DiagnosisParser } from '../src/modules/sync/pipeline/parsers/diagnosis.parser';
+import { ImageAnalysisParser } from '../src/modules/sync/pipeline/parsers/image-analysis.parser';
+import { ImageParser } from '../src/modules/sync/pipeline/parsers/image.parser';
+import { TransactionSynthesizer } from '../src/modules/sync/pipeline/transaction-synthesizer';
+import { validateFactContent } from '../src/modules/sync/pipeline/fact-content-schemas';
+import { analysisRow, assembleFriday, canonicalAnalysis, parserContext } from './image-analysis-fixtures';
+
+Logger.overrideLogger(false);
+const parser = new ImageAnalysisParser(new DiagnosisParser());
+
+describe('image analysis shared ingestion', () => {
+  test('中文决定 K 码，矛盾的私有问题码只在原文留底；保留完整分析原文', () => {
+    const raw = analysisRow();
+    const assembled = assembleFriday([raw]);
+    expect(assembled.rows[0]!.rawSource).toEqual(raw);
+    const canonical = assembled.rows[0]!.canonical as Record<string, unknown>;
+    const facts = parser.parse(parserContext(canonical));
+    expect(facts[1]!.content).toMatchObject({ code: 'K02', code_source: 'image_ai', name_zh: '龋齿',
+      tooth_position: '16', source_image_external_id: '3001', source_analysis_external_id: '1001' });
+    expect(facts[0]!.content).toMatchObject({ analysis_result: raw.analysis_data });
+    expect(facts[1]!.occurredAt?.toISOString()).toBe('2026-10-12T02:05:00.000Z');
+    for (const fact of facts) expect(() => validateFactContent(fact.type, fact.subjectId, fact.content)).not.toThrow();
+  });
+
+  test('关键词兜底与医生诊断使用同一中文字典，不新增宿主特判', () => {
+    const raw = analysisRow();
+    raw.problemList = [{ ...raw.problemList[0]!, problem: '后牙区深龋', problem_code: 'anything' }];
+    const facts = parser.parse(parserContext(canonicalAnalysis(raw)));
+    expect(facts[1]!.content).toMatchObject({ code: 'K02', name_zh: '后牙区深龋' });
+  });
+
+  test.each(['恒牙萌出空间不足', '囊肿', '牙槽骨吸收', '可疑龋'])('未命中的 %s 保留原文，不用 P 码补出诊断', name => {
+    const raw = analysisRow();
+    raw.problemList = [{ ...raw.problemList[0]!, problem: name, problem_code: 'P0010' }];
+    const facts = parser.parse(parserContext(canonicalAnalysis(raw)));
+    expect(facts[1]!.content).toMatchObject({ code: null, name_zh: name, code_source: 'image_ai' });
+  });
+
+  test('乳牙牙位、全部排除、扣空均保留稳定问题身份', () => {
+    const raw = analysisRow();
+    const before = parser.parse(parserContext(canonicalAnalysis(raw)))[1]!;
+    for (const exclude of ['ALL', '16,26']) {
+      raw.problemList = [{ ...raw.problemList[0]!, exclude_teeth: exclude }];
+      const cancelled = parser.parse(parserContext(canonicalAnalysis(raw)))[1]!;
+      expect(cancelled.subjectId).toBe(before.subjectId);
+      expect(cancelled.status).toBe(FactStatus.CANCELLED);
+    }
+    raw.problemList = [{ ...raw.problemList[0]!, teeth: '51,61', exclude_teeth: '61' }];
+    expect(parser.parse(parserContext(canonicalAnalysis(raw)))[1]!.content).toMatchObject({ tooth_position: '51' });
+  });
+
+  test('改中文名称和牙位只升同一问题版本，不变更业务身份', () => {
+    const raw = analysisRow();
+    const first = parser.parse(parserContext(canonicalAnalysis(raw)))[1]!;
+    raw.problemList = [{ ...raw.problemList[0]!, problem: '缺失牙', teeth: '36', exclude_teeth: '' }];
+    const next = parser.parse(parserContext(canonicalAnalysis(raw)))[1]!;
+    expect(next.subjectId).toBe(first.subjectId);
+    expect(next.content).toMatchObject({ code: 'K08', tooth_position: '36' });
+  });
+
+  test.each(['UNDO', 'OK', 'FAIL', 'NOTIFY_FAIL'])('%s 只留分析快照，不发布或替换临床信号', req_status => {
+    const ctx = parserContext(canonicalAnalysis({ req_status }));
+    expect(parser.parse(ctx)).toHaveLength(1);
+    expect(parser.snapshotScope(ctx).replace).toBe(false);
+  });
+
+  test('实际分析时间未知只留底，不用事务时间冒充分析时间', () => {
+    const ctx = parserContext(canonicalAnalysis({ analysis_time: null }));
+    expect(parser.parse(ctx)).toHaveLength(1);
+    expect(parser.parse(ctx)[0]!.occurredAt).toBeNull();
+    expect(parser.snapshotScope(ctx).replace).toBe(false);
+  });
+
+  test('明确空列表保留分析主体并声明完整替换', () => {
+    const ctx = parserContext(canonicalAnalysis({ problemList: [] }));
+    expect(parser.parse(ctx)[0]!.type).toBe(FactType.IMAGE_ANALYSIS_RECORD);
+    expect(parser.snapshotScope(ctx).replace).toBe(true);
+  });
+
+  test.each([undefined, null, '', 'invalid', [null], [{ id: 'x' }]])('缺失/破损明细 %p 不能冒充空列表', problemList => {
+    expect(assembleFriday([analysisRow({ problemList })]).stats.failed).toBe(1);
+  });
+
+  test.each(['patient_id', 'check_id'])('明细 %s 不一致拒绝整份分析', field => {
+    const raw = analysisRow();
+    raw.problemList = [{ ...raw.problemList[0]!, [field]: 'different' }];
+    expect(assembleFriday([raw]).stats.failed).toBe(1);
+  });
+
+  test('品牌间同分析 ID 的明细仍只关联同一原文', () => {
+    const a = analysisRow();
+    const b = analysisRow({ tenant_id: 'brand-b', problemList: [{ ...a.problemList[0]!, problem: '残根' }] });
+    const rows = assembleFriday([a, b]).rows;
+    expect(rows).toHaveLength(2);
+    expect((rows[0]!.canonical as ReturnType<typeof canonicalAnalysis>).findings[0]!.name).toBe('龋齿');
+    expect((rows[1]!.canonical as ReturnType<typeof canonicalAnalysis>).findings[0]!.name).toBe('残根');
+  });
+
+  test('第二宿主换字段名和时区后复用同一标准解析器', () => {
+    const config = AssemblerConfigSchema.parse({ canonical: 'image_analysis', primary: { table: 'other', key: 'caseId' },
+      emits: { action: Action.IMAGE_ANALYSIS_RECORDED, subjectType: 'image_analysis', occurredAtField: 'analyzedAt' },
+      field_mapping: { externalId: 'caseId', patientExternalId: 'person', clinicId: 'center', imageExternalId: 'picture',
+        analyzedAt: 'finishedAt', status: 'state', expectedFindingCount: 'count', findings: 'findings' },
+      enum_mapping: { status: { done: 'completed' } } });
+    const input = { caseId: '1001', person: '4001', center: 'clinic-a', picture: '3001',
+      finishedAt: '2026-10-12 02:05:00', state: 'done', count: 1,
+      findings: [{ externalId: '2001', patientExternalId: '4001', imageExternalId: '3001',
+        code: 'K02', name: '龋齿', toothPosition: '16;26', excludedToothPositions: '26', excluded: false }] };
+    const result = new AssemblerEngine().assemble({ tables: { other: [input] }, config,
+      normalize: { amountUnit: 'fen', timezone: 'UTC' }, hostId: 'other-host' });
+    const canonical = ImageAnalysisCanonicalSchema.parse(result.rows[0]!.canonical);
+    const diagnosis = parser.parse(parserContext(canonical))[1]!;
+    const friday = parser.parse(parserContext(canonicalAnalysis()))[1]!;
+    expect(diagnosis.content).toEqual(friday.content);
+    expect(diagnosis.occurredAt).toEqual(friday.occurredAt);
+  });
+
+  test('地址保存完整签名；旧相对路径报文继续接受；伪完整地址拒绝', () => {
+    const raw = { id: '3001', tenant_id: 'brand-a', organization_id: 'clinic-a', patient_id: '4001',
+      shooting_time: '2026-10-12 10:00:00', class_name: '全景片', file_url: 'old/path.jpg',
+      fileUrlEcho: 'https://files.example.com/path.jpg?signature=token&expires=123', file_name: 'path.jpg' };
+    const result = assembleFriday([raw], 'med_check');
+    const ctx = parserContext(result.rows[0]!.canonical as Record<string, unknown>);
+    expect(new ImageParser().parse(ctx)[0]!.content).toMatchObject({ file_url: raw.fileUrlEcho });
+    expect(assembleFriday([{ ...raw, fileUrlEcho: undefined }], 'med_check').stats.failed).toBe(0);
+    for (const fileUrlEcho of ['relative/path', 'javascript:alert(1)', '//files.example.com/x']) {
+      expect(assembleFriday([{ ...raw, fileUrlEcho }], 'med_check').stats.failed).toBe(1);
+    }
+  });
+
+  test('内容版本区分仅明细/链接变更，稳定对象键顺序；原时间键模式不变', () => {
+    const row = canonicalAnalysis();
+    const raw = analysisRow();
+    const synthesizer = new TransactionSynthesizer();
+    const emit = { action: Action.IMAGE_ANALYSIS_RECORDED, subjectType: 'image_analysis' as const,
+      occurredAtField: 'analyzedAt', contentVersion: true };
+    const synth = (rawRow: Record<string, unknown>, contentVersion = true) => synthesizer.synthesize({
+      rawRow, canonicalRow: row, emits: { ...emit, contentVersion }, resource: 'image_analysis',
+      hostId: 'host-demo', tenantId: 'tenant-demo', sourceUnit: 'brand-a', sourceContext: 'friday',
+      patientIdResolver: () => 'patient-demo',
+    })!;
+    expect(synth(raw).sourceEventId).toBe(synth(Object.fromEntries(Object.entries(raw).reverse())).sourceEventId);
+    expect(synth(raw).sourceEventId).not.toBe(synth({ ...raw, problemList: [] }).sourceEventId);
+    expect(synth(raw, false).sourceEventId).toBe(synth({ ...raw, problemList: [] }, false).sourceEventId);
+    const standardPush = (rawRow: Record<string, unknown>) => synthesizer.synthesize({
+      rawRow, canonicalRow: row, emits: { action: Action.IMAGE_ANALYSIS_RECORDED,
+        subjectType: 'image_analysis', occurredAtField: 'analyzedAt' }, resource: 'image_analysis',
+      hostId: 'host-demo', tenantId: 'tenant-demo', sourceUnit: 'brand-a', sourceContext: 'friday',
+      patientIdResolver: () => 'patient-demo',
+    })!;
+    expect(standardPush(raw).sourceEventId).toBe(synth(raw).sourceEventId);
+    expect(standardPush(raw).sourceEventId).not.toBe(standardPush({ ...raw, problemList: [] }).sourceEventId);
+  });
+});

apps/pac-service/tests/image-analysis-postgres.spec.ts

+import { Logger } from '@nestjs/common';
+import { PrismaClient } from '@prisma/client';
+import { Action, FactStatus, FactType } from '@pac/types';
+import { FactWriter, type SnapshotWriteInput } from '../src/modules/sync/pipeline/fact-writer.service';
+import { ImageAnalysisParser } from '../src/modules/sync/pipeline/parsers/image-analysis.parser';
+import { DiagnosisParser } from '../src/modules/sync/pipeline/parsers/diagnosis.parser';
+import { ImageParser } from '../src/modules/sync/pipeline/parsers/image.parser';
+import { TransactionSynthesizer } from '../src/modules/sync/pipeline/transaction-synthesizer';
+import { ParserPipeline } from '../src/modules/sync/pipeline/parser-pipeline.service';
+import type { ParserRegistry } from '../src/modules/sync/pipeline/parsers/parser.registry';
+import { ClinicalSignalService } from '../src/modules/clinical-signals/clinical-signal.service';
+import { attachImageLinks } from '../src/modules/facts/image-links';
+import type { PrismaService } from '../src/prisma/prisma.service';
+import { analysisRow, canonicalAnalysis, parserContext } from './image-analysis-fixtures';
+
+const url = process.env.PAC_IMAGE_TEST_DATABASE_URL;
+const describePg = url ? describe : describe.skip;
+
+describePg('image analysis | isolated PostgreSQL', () => {
+  let prisma: PrismaClient;
+  let writer: FactWriter;
+  let hostId: string;
+  let patientId: string;
+  let patientB: string;
+  const parser = new ImageAnalysisParser(new DiagnosisParser());
+  const synthesizer = new TransactionSynthesizer();
+  const tenantId = 'test-tenant';
+
+  beforeAll(async () => {
+    const parsed = new URL(url!);
+    if (!['localhost', '127.0.0.1'].includes(parsed.hostname) || !parsed.pathname.startsWith('/pac_image_ci_')) {
+      throw new Error('requires an isolated localhost pac_image_ci_ database');
+    }
+    Logger.overrideLogger(false);
+    prisma = new PrismaClient({ datasourceUrl: url });
+    writer = new FactWriter(prisma as unknown as PrismaService);
+    const host = await prisma.host.create({ data: { name: 'image-ci', appId: 'image-ci', appSecretHash: 'test-only' } });
+    hostId = host.id;
+    const a = await prisma.patient.create({ data: { hostId, tenantId, sourceUnit: 'brand-a', externalId: '4001' } });
+    const b = await prisma.patient.create({ data: { hostId, tenantId, sourceUnit: 'brand-b', externalId: '4001' } });
+    patientId = a.id;
+    patientB = b.id;
+    // Prisma schema 不表达的 partial unique，按正式事实身份键补齐。
+    await prisma.$executeRawUnsafe("CREATE UNIQUE INDEX image_ci_active_fact ON patient_facts(host_id,tenant_id,source_unit,subject_id) WHERE status='active'");
+    await prisma.$executeRawUnsafe("CREATE UNIQUE INDEX image_ci_event ON patient_transactions(host_id,tenant_id,source_event_id) WHERE source_event_id IS NOT NULL");
+  });
+  beforeEach(async () => {
+    await prisma.patientFact.deleteMany({ where: { hostId } });
+    await prisma.patientTransaction.deleteMany({ where: { hostId } });
+  });
+  afterAll(async () => { await prisma?.$disconnect(); });
+
+  async function snapshot(overrides: Record<string, unknown> = {}, sourceUnit = 'brand-a'): Promise<SnapshotWriteInput> {
+    const rawRow = analysisRow({ ...overrides, tenant_id: sourceUnit });
+    const canonicalRow = canonicalAnalysis(rawRow);
+    const pid = sourceUnit === 'brand-a' ? patientId : patientB;
+    const data = synthesizer.synthesize({ rawRow, canonicalRow,
+      resource: 'image_analysis', hostId, tenantId, sourceUnit, sourceContext: 'test',
+      patientIdResolver: () => pid,
+      emits: { action: Action.IMAGE_ANALYSIS_RECORDED, subjectType: 'image_analysis',
+        occurredAtField: 'analyzedAt', contentVersion: true } })!;
+    await prisma.patientTransaction.createMany({ data: [data], skipDuplicates: true });
+    const transaction = await prisma.patientTransaction.findFirstOrThrow({ where: {
+      hostId, tenantId, sourceEventId: data.sourceEventId,
+    } });
+    const ctx = parserContext(canonicalRow, { id: transaction.id, hostId, tenantId, patientId: pid });
+    const sourceUpdatedAt = canonicalRow.updatedAt ? new Date(canonicalRow.updatedAt) : null;
+    const entries = parser.parse(ctx).map(draft => ({ draft, hostId, tenantId, sourceUnit, patientId: pid,
+      transactionId: transaction.id, sourceUpdatedAt }));
+    return { ...entries[0]!, entries, scope: parser.snapshotScope(ctx) };
+  }
+
+  async function activeDiagnoses(sourceUnit = 'brand-a') {
+    return prisma.patientFact.findMany({ where: { hostId, tenantId, sourceUnit,
+      type: FactType.DIAGNOSIS_RECORD, status: FactStatus.ACTIVE } });
+  }
+
+  async function image(fileUrl: string, sourceUnit = 'brand-a', owner = patientId) {
+    const tx = await prisma.patientTransaction.create({ data: {
+      hostId, tenantId, patientId: owner, clinicId: 'clinic-a', action: Action.IMAGE_UPLOADED,
+      subjectType: 'image', subjectId: '3001', rawPayload: {}, occurredAt: new Date(),
+      payloadHash: 'test-only',
+    } });
+    const draft = new ImageParser().parse(parserContext({ externalId: '3001', modality: 'pano', fileUrl },
+      { hostId, tenantId, patientId: owner, id: tx.id }))[0]!;
+    await writer.writeDraft({ draft, hostId, tenantId, sourceUnit, patientId: owner, transactionId: tx.id });
+  }
+
+  function pipeline() {
+    const registry = { get: (action: string) => action === Action.IMAGE_ANALYSIS_RECORDED
+      ? parser : action === Action.IMAGE_UPLOADED ? new ImageParser() : null };
+    const signals = new ClinicalSignalService();
+    signals.onModuleInit();
+    return new ParserPipeline(registry as unknown as ParserRegistry, writer,
+      prisma as unknown as PrismaService, signals);
+  }
+
+  async function pipelineItem(input: SnapshotWriteInput) {
+    const row = await prisma.patientTransaction.findUniqueOrThrow({ where: { id: input.transactionId } });
+    return { transaction: { ...row, action: row.action as Action },
+      canonicalRow: canonicalAnalysis(row.rawPayload as Record<string, unknown>) };
+  }
+
+  test('统一编排器单条 / 批量入口同样修订、撤回并拦住旧报文', async () => {
+    const p = pipeline();
+    const first = await pipelineItem(await snapshot());
+    const created = await p.runForTransaction(first);
+    expect(created.factsCreated).toBe(2);
+    expect(created.factsFailed).toBe(0);
+    const empty = await pipelineItem(await snapshot({ problemList: [] }));
+    const cleared = await p.runForBatch([empty]);
+    expect(cleared.factsSuperseded).toBe(2);
+    expect(cleared.factsFailed).toBe(0);
+    expect(await activeDiagnoses()).toHaveLength(0);
+    expect((await p.runForBatch([first])).factsStaleSkipped).toBe(1);
+    expect(await activeDiagnoses()).toHaveLength(0);
+  });
+
+  test('同批坏分析隔离，健康分析与普通影像事实继续走统一编排', async () => {
+    await image('https://files.example.com/other.jpg', 'brand-a', patientB);
+    const bad = await pipelineItem(await snapshot());
+    const healthy = await pipelineItem(await snapshot({ id: '1002', check_id: '3002',
+      problemList: [{ ...analysisRow().problemList[0]!, check_id: '3002' }] }));
+    const imageTx = await prisma.patientTransaction.create({ data: {
+      hostId, tenantId, patientId, clinicId: 'clinic-a', subjectType: 'image', subjectId: '3002',
+      action: Action.IMAGE_UPLOADED, occurredAt: new Date('2026-10-12T02:00:00Z'),
+      rawPayload: {}, payloadHash: 'test-only',
+    } });
+    const outcome = await pipeline().runForBatch([bad, healthy, {
+      transaction: { ...imageTx, action: imageTx.action as Action },
+      canonicalRow: { externalId: '3002', modality: 'pano', uploadedAt: '2026-10-12T02:00:00Z',
+        fileUrl: 'https://files.example.com/healthy.jpg' },
+    }]);
+    expect(outcome.factsFailed).toBe(1);
+    expect(outcome.factRejects).toHaveLength(1);
+    expect(outcome.factsCreated).toBe(3);
+    const facts = await activeDiagnoses();
+    expect(facts).toHaveLength(1);
+    expect((await attachImageLinks(prisma as unknown as PrismaService, facts))[0]!.content)
+      .toMatchObject({ source_analysis_external_id: '1002', image_file_url: 'https://files.example.com/healthy.jpg' });
+  });
+
+  test('完整摄入、更新、撤回、旧报文重试和相同报文幂等', async () => {
+    const first = await snapshot();
+    await writer.writeSnapshots([first]);
+    expect(await activeDiagnoses()).toHaveLength(1);
+    await writer.writeSnapshots([await snapshot({ problemList: [] })]);
+    expect(await activeDiagnoses()).toHaveLength(0);
+    const retry = await writer.writeSnapshots([await snapshot()]);
+    expect(retry.results[0]!.action).toBe('stale_skipped');
+    expect(await activeDiagnoses()).toHaveLength(0);
+    const count = await prisma.patientFact.count();
+    await writer.writeSnapshots([await snapshot({ problemList: [] })]);
+    expect(await prisma.patientFact.count()).toBe(count);
+    const audit = await prisma.patientTransaction.findUniqueOrThrow({ where: { id: first.transactionId } });
+    expect(audit.rawPayload).toMatchObject({ analysis_data: '{"sourceResult":true}', problemList: expect.any(Array) });
+  });
+
+  test('同问题改码和牙位只更新同一身份；ALL 取消信号', async () => {
+    await writer.writeSnapshots([await snapshot()]);
+    const old = (await activeDiagnoses())[0]!;
+    const raw = analysisRow();
+    const problem = { ...raw.problemList[0]!, problem: '缺失牙', teeth: '36', exclude_teeth: '' };
+    await writer.writeSnapshots([await snapshot({ problemList: [problem], updated_gmt_at: '2026-10-12 10:07:00' })]);
+    const next = (await activeDiagnoses())[0]!;
+    expect(next.subjectId).toBe(old.subjectId);
+    expect(next.version).toBe(old.version + 1);
+    expect(next.content).toMatchObject({ code: 'K08', tooth_position: '36' });
+    await writer.writeSnapshots([await snapshot({ problemList: [{ ...problem, exclude_teeth: 'ALL' }], updated_gmt_at: '2026-10-12 10:08:00' })]);
+    expect(await activeDiagnoses()).toHaveLength(0);
+  });
+
+  test('失败/缺分析时间不会清空已有有效结果', async () => {
+    await writer.writeSnapshots([await snapshot()]);
+    for (const change of [{ req_status: 'FAIL' }, { analysis_time: null }]) {
+      await writer.writeSnapshots([await snapshot({ ...change, problemList: [], updated_gmt_at: '2026-10-12 10:07:00' })]);
+      expect(await activeDiagnoses()).toHaveLength(1);
+    }
+  });
+
+  test('品牌同号独立；清除 A 不影响 B', async () => {
+    await writer.writeSnapshots([await snapshot(), await snapshot({}, 'brand-b')]);
+    await writer.writeSnapshots([await snapshot({ problemList: [] })]);
+    expect(await activeDiagnoses('brand-a')).toHaveLength(0);
+    expect(await activeDiagnoses('brand-b')).toHaveLength(1);
+  });
+
+  test.each([false, true])('同时间戳批内旧报文在末尾，仍按留账顺序选新快照（已有锚点=%s）', async existing => {
+    const old = await snapshot();
+    if (existing) await writer.writeSnapshots([old]);
+    const next = await snapshot({ problemList: [] });
+    await writer.writeSnapshots([next, old]);
+    expect(await activeDiagnoses()).toHaveLength(0);
+    const head = await prisma.patientFact.findFirstOrThrow({ where: {
+      hostId, type: FactType.IMAGE_ANALYSIS_RECORD, status: FactStatus.ACTIVE,
+    } });
+    expect(head.transactionIds).toContain(next.transactionId);
+  });
+
+  test('并发旧/新快照只留下新版本，锁与取消在同一事务', async () => {
+    const old = await snapshot();
+    const next = await snapshot({ problemList: [], updated_gmt_at: '2026-10-12 10:08:00' });
+    await Promise.all([writer.writeSnapshots([next]), writer.writeSnapshots([old])]);
+    expect(await activeDiagnoses()).toHaveLength(0);
+    const head = await prisma.patientFact.findFirstOrThrow({ where: { hostId, sourceUnit: 'brand-a',
+      type: FactType.IMAGE_ANALYSIS_RECORD, status: FactStatus.ACTIVE } });
+    expect(head.sourceUpdatedAt?.toISOString()).toBe('2026-10-12T02:08:00.000Z');
+  });
+
+  test('快照中任一成员不合法时整体回滚，不能先发布头或先撤旧结果', async () => {
+    await writer.writeSnapshots([await snapshot()]);
+    const invalid = await snapshot({ problemList: [], updated_gmt_at: '2026-10-12 10:08:00' });
+    invalid.entries.push({ ...invalid.entries[0]!, draft: { ...invalid.entries[0]!.draft,
+      subjectId: 'diagnosis_record:bad', type: FactType.DIAGNOSIS_RECORD,
+      content: { code: 'P0010', source_encounter_external_id: 'x' } } });
+    const count = await prisma.patientFact.count();
+    expect((await writer.writeSnapshots([invalid])).rejected).toHaveLength(1);
+    expect(await prisma.patientFact.count()).toBe(count);
+    expect(await activeDiagnoses()).toHaveLength(1);
+  });
+
+  test('读接口获取最新文件地址；刷新地址不增加分析版本', async () => {
+    await image('https://files.example.com/old.jpg');
+    await writer.writeSnapshots([await snapshot()]);
+    const rows = await activeDiagnoses();
+    expect((await attachImageLinks(prisma as unknown as PrismaService, rows))[0]!.content).toMatchObject({ image_file_url: 'https://files.example.com/old.jpg' });
+    await image('https://files.example.com/new.jpg?signature=value');
+    expect((await attachImageLinks(prisma as unknown as PrismaService, rows))[0]!.content).toMatchObject({ image_file_url: 'https://files.example.com/new.jpg?signature=value' });
+    expect((await activeDiagnoses())[0]!.version).toBe(rows[0]!.version);
+  });
+
+  test('已知影像归属冲突时拒绝临床写入，读关联也不跨患者取地址', async () => {
+    await image('https://files.example.com/other-patient.jpg', 'brand-a', patientB);
+    const input = await snapshot();
+    expect((await writer.writeSnapshots([input])).rejected[0]!.issues).toContain('snapshot reference ownership differs');
+    expect(await activeDiagnoses()).toHaveLength(0);
+    const fakeFact = { ...input.entries[1]!.draft, hostId, tenantId, sourceUnit: 'brand-a',
+      patientId, clinicId: 'clinic-a' };
+    expect((await attachImageLinks(prisma as unknown as PrismaService, [fakeFact]))[0]!.content).not.toHaveProperty('image_file_url');
+  });
+});

apps/pac-service/tests/push-upsert-batch-status.spec.ts

+import { Logger } from '@nestjs/common';
+import { ApiCode, SyncStatus } from '@pac/types';
+import { ColdImportService } from '../src/modules/sync/cold-import/cold-import.service';
+import { PushReceiverService } from '../src/modules/sync/push/push-receiver.service';
+import { AssemblerEngine } from '../src/modules/sync/assembler/assembler-engine';
+import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.schema';
+import type { AssemblerConfig } from '../src/modules/sync/assembler/assembler.schema';
+import type { ColdImportManifest } from '../src/modules/sync/cold-import/manifest.schema';
+import { TransformEngine } from '../src/modules/sync/transforms/transform-engine';
+
+Logger.overrideLogger(false);
+
+type UpsertResource = 'patient' | 'patient_relation' | 'patient_return_visit';
+
+function setup(resource: UpsertResource) {
+  const prisma = {
+    host: { findFirst: jest.fn().mockResolvedValue({ id: 'host-demo' }) },
+    syncLog: {
+      create: jest.fn().mockResolvedValue({ id: 'sync-demo' }),
+      update: jest.fn().mockResolvedValue({}),
+    },
+    patientTransaction: { findMany: jest.fn().mockResolvedValue([]) },
+    patient: {
+      findMany: jest.fn().mockResolvedValue([
+        { id: 'patient-1', externalId: 'p1', sourceUnit: 'brand-demo' },
+        { id: 'patient-2', externalId: 'p2', sourceUnit: 'brand-demo' },
+      ]),
+      upsert: jest.fn().mockResolvedValue({ id: 'patient-1' }),
+    },
+    patientProfile: { upsert: jest.fn().mockResolvedValue({}) },
+    patientRelation: { upsert: jest.fn().mockResolvedValue({}) },
+    patientReturnVisit: { upsert: jest.fn().mockResolvedValue({}) },
+  };
+  const alerter = { send: jest.fn().mockResolvedValue(undefined) };
+  const queue = { enqueuePersonaRecompute: jest.fn().mockResolvedValue(undefined) };
+  const cold = new ColdImportService(prisma as never, {} as never, {} as never,
+    new AssemblerEngine(), {} as never, new TransformEngine(), alerter as never);
+  const config = AssemblerConfigSchema.parse({ canonical: resource,
+    primary: { table: 'native', key: 'id' },
+    field_mapping: resource === 'patient' ? { externalId: 'id' } : resource === 'patient_relation'
+      ? { patientExternalId: 'customer_id', relatedExternalId: 'referee_patient_id', relationship: 'relationship' }
+      : { externalId: 'id', patientExternalId: 'customer_id' },
+  });
+  // 文件 / 外部依赖替身；装配、直接 upsert、批次状态及接收回执均运行正式实现。
+  const dependencies = cold as unknown as {
+    resolveHostDir(): string;
+    readManifest(): ColdImportManifest;
+    loadAllAssemblers(): AssemblerConfig[];
+    hydratePushLookupTables(): Promise<void>;
+    detectRawColumnDrift(): Promise<[]>;
+  };
+  jest.spyOn(dependencies, 'resolveHostDir').mockReturnValue('/virtual');
+  jest.spyOn(dependencies, 'readManifest').mockReturnValue({
+    tenant_id: 'friday-market', identity_namespace_field: 'tenant_id',
+    amount_unit: 'yuan', timezone: 'Asia/Shanghai', transforms: [],
+  } as never);
+  jest.spyOn(dependencies, 'loadAllAssemblers').mockReturnValue([config]);
+  jest.spyOn(dependencies, 'hydratePushLookupTables').mockResolvedValue(undefined);
+  jest.spyOn(dependencies, 'detectRawColumnDrift').mockResolvedValue([]);
+  const receiver = new PushReceiverService(prisma as never, {} as never, cold, queue as never);
+  const good = { id: 'p1', tenant_id: 'brand-demo', customer_id: 'p1',
+    referee_patient_id: 'p2', relationship: 'friend' };
+  const bad = resource === 'patient' ? { ...good, id: null }
+    : { ...good, customer_id: null };
+  const send = (rows: Record<string, unknown>[], dryRun = false) => receiver.receiveRows({
+    hostId: 'host-demo', hostName: 'friday', source: 'native', rows, dryRun,
+  });
+  return { prisma, cold, queue, alerter, good, bad, send, dependencies };
+}
+
+describe.each<UpsertResource>(['patient', 'patient_relation', 'patient_return_visit'])(
+  'push batch | direct upsert %s', resource => {
+    test('有效行已写入，坏行保留 failed 计数；SyncLog 为 partial，接收器不抛整批错误', async () => {
+      const s = setup(resource);
+      const result = await s.send([s.good, s.bad]);
+      expect(result).toMatchObject({ accepted: 2, failed: 1, transactionsWritten: 0, duplicates: 0 });
+      expect(s.prisma.syncLog.update.mock.calls[0]![0].data).toMatchObject({
+        status: SyncStatus.PARTIAL, failed: 1, errorMessage: null,
+      });
+      expect(s.queue.enqueuePersonaRecompute).not.toHaveBeenCalled();
+    });
+
+    test('全是坏行仍明确拒收；不能把真正全失败改成成功', async () => {
+      const s = setup(resource);
+      await expect(s.send([s.bad])).rejects.toMatchObject({ code: ApiCode.SYNC_CONTRACT_DRIFT });
+      expect(s.prisma.syncLog.update.mock.calls[0]![0].data.status).toBe(SyncStatus.FAILED);
+    });
+
+    test('全有效直接 upsert 即使没有 transaction，仍为 success', async () => {
+      const s = setup(resource);
+      expect(await s.send([s.good])).toMatchObject({ failed: 0, transactionsWritten: 0 });
+      expect(s.prisma.syncLog.update.mock.calls[0]![0].data.status).toBe(SyncStatus.SUCCESS);
+    });
+
+    test('预检与正式批次分类一致，不执行业务写入', async () => {
+      const s = setup(resource);
+      expect(await s.send([s.good, s.bad], true)).toMatchObject({ dryRun: true, failed: 1 });
+      expect(s.prisma.syncLog.update.mock.calls[0]![0].data.status).toBe(SyncStatus.PARTIAL);
+      for (const model of [s.prisma.patient, s.prisma.patientProfile,
+        s.prisma.patientRelation, s.prisma.patientReturnVisit]) {
+        expect(model.upsert).not.toHaveBeenCalled();
+      }
+      expect(s.queue.enqueuePersonaRecompute).not.toHaveBeenCalled();
+    });
+  },
+);
+
+test.each<UpsertResource>(['patient_relation', 'patient_return_visit'])(
+  '%s 预检仍校验租户归属，不能把身份不完整的行算成成功', async resource => {
+    const s = setup(resource);
+    jest.spyOn(s.dependencies, 'readManifest').mockReturnValue({
+      tenant_field: 'tenant_id', identity_namespace_field: 'tenant_id',
+      amount_unit: 'yuan', timezone: 'Asia/Shanghai', transforms: [],
+    } as never);
+    await expect(s.send([{ ...s.good, tenant_id: null }], true))
+      .rejects.toMatchObject({ code: ApiCode.SYNC_CONTRACT_DRIFT });
+    expect(s.prisma.syncLog.update.mock.calls[0]![0].data.status).toBe(SyncStatus.FAILED);
+    expect(s.prisma.patient.findMany).not.toHaveBeenCalled();
+  },
+);
+
+test('复现生产：200 条关系中 197 条有效、3 条身份缺失，不再报整批失败', async () => {
+  const s = setup('patient_relation');
+  const rows = [...Array.from({ length: 197 }, () => ({ ...s.good })),
+    { ...s.good, referee_patient_id: null }, { ...s.good, referee_patient_id: null }, s.bad];
+  const result = await s.send(rows);
+  expect(result).toMatchObject({ accepted: 200, failed: 3, transactionsWritten: 0 });
+  expect(s.prisma.patientRelation.upsert).toHaveBeenCalledTimes(197);
+  expect(s.prisma.syncLog.update.mock.calls[0]![0].data).toMatchObject({
+    status: SyncStatus.PARTIAL, fetched: 200, failed: 3, errorMessage: null,
+  });
+});
+
+test.each([
+  { transactionsWritten: 1, duplicates: 0 },
+  { transactionsWritten: 0, duplicates: 1 },
+])('普通事件新增 / 重复命中仍算进展：%j', async progress => {
+  const s = setup('patient_relation');
+  const stats = { resource: 'diagnosis', patientsUpserted: 0, ...progress,
+    failed: 1, factsFailed: 0, factRejects: [], patientStubsCreated: 0, sampleCanonical: [] };
+  jest.spyOn(s.cold, 'ingestRawTables').mockResolvedValue({
+    syncLogId: 'sync-demo', perResource: [stats], touched: [], mappingMisses: [], suspectFields: [],
+  } as never);
+  expect(await s.send([s.good, s.bad])).toMatchObject({ ...progress, failed: 1 });
+});

apps/pac-web/src/components/plan-detail/facts-timeline.tsx

@@ -314,6 +314,8 @@ function TimelineRow({ fact }: { fact: AdaptedFact }) {
 
   const { title, note } = factSummary(fact);
   const rightLines = rightColumn(fact);
+  const imageUrl = fact.content?.image_file_url ??
+    (fact.type === 'image_record' ? fact.content?.file_url : null);
 
   return (
     <div className="relative pb-3">
@@ -338,6 +340,10 @@ function TimelineRow({ fact }: { fact: AdaptedFact }) {
           {note && (
             <div className="text-[11px] text-slate-500 leading-snug mt-0.5 line-clamp-2">{note}</div>
           )}
+          {typeof imageUrl === 'string' && /^https?:\/\//i.test(imageUrl) && (
+            <a href={imageUrl} target="_blank" rel="noopener noreferrer"
+              className="text-[11px] text-brand-600 hover:underline">查看影像</a>
+          )}
         </div>
         {rightLines.length > 0 && (
           <div className="flex flex-col items-end gap-0.5 text-[11.5px] text-slate-600 tabular-nums whitespace-nowrap">
@@ -517,6 +523,7 @@ const FACT_META: Record<string, FactMeta> = {
   encounter_record: { label: '接诊', tone: 'slate', Icon: UserRound },
   emr_record: { label: '病历', tone: 'slate', Icon: FileText },
   image_record: { label: '影像', tone: 'slate', Icon: ImageIcon },
+  image_analysis_record: { label: '影像分析', tone: 'slate', Icon: ImageIcon },
   appointment_record: { label: '预约', tone: 'slate', Icon: Calendar },
   consultation_record: { label: '咨询', tone: 'slate', Icon: MessageCircle },
   visit_registration_record: { label: '挂号', tone: 'slate', Icon: ClipboardList },

docs/qa/2026-10-13-friday-push-batch-retry-audit.md

+# Friday SaaS → PAC 批次失败与消费阻塞核查
+
+## 核查范围
+
+- PAC 修复分支：`codex/friday-image-push-contract`，尚未提交、部署。
+- 消费端：`https://code.jwsmed.com/fiday-saas/open-connector`，读取远端 `master@1d04897c8107adc6db07998ffdd843e297efe3c2`。未修改该仓库。
+- PAC 生产数据库、日志和运行代码已只读核对；Friday SaaS 生产数据库不可访问，未查询实际消费位点、积压数量或部署版本。
+
+## PAC 生产证据
+
+2026-10-12 23:46（北京时间）读取 PAC 生产库：
+
+| source | 结果 |
+|---|---|
+| `customer_basic_info` | 18:42–18:45，共 50 批、10,000 行，均成功 |
+| `customer_referee_circle` | 1 批成功、15 批失败；最新失败 22:56，每批 200 行、3 条失败 |
+| 预约、病历、影像等其它 source | 当时未发现生产 push 记录 |
+
+22:56 对应生产应用日志有 3 条身份校验错误：2 条 `relatedExternalId=null`，1 条 `patientExternalId=null`。关系表已有 341 条记录落库，不能将该批描述为全部 200 条都未写入。
+
+生产运行代码的整批拒收条件为 `txn=0 && duplicates=0 && failed>0`。关系、主档、回访直接 upsert，不生成 transaction；该判断遗漏了成功 upsert 的进展。
+
+## PAC 修复
+
+- 批次状态和接收器拒收判断同时计入成功 upsert；有进展且部分行失败时为 `partial`，正常返回 `failed` 计数。
+- 真正没有任何有效进展、存在失败行时仍拒收，保持全失败守卫。
+- 关系与回访预检按有效身份、租户统计预期 upsert，业务数据库不写入。
+- 回归覆盖生产的“200 行、197 条成功、3 条身份缺失”，以及主档、关系、回访的全成功 / 部分失败 / 全失败 / 预检路径。
+
+## 消费端代码结论
+
+依据 `PushJob.java` 和 `EventLogStore.java`：
+
+1. `consumer_offset` 按 `consumer_id` 保存一个 `last_event_id`，未按 source 拆分。
+2. 从事件表按 `id > last_event_id`、`ORDER BY id` 取一批，再按 source 分组发送。
+3. 任一组失败立即退出，保留原位点；全部组成功才推进到本批最大事件 ID。
+4. 非成功业务码，或成功业务码下 `data.failed>0`，都判失败。退避指数增长，最高 1 小时；没有自动跳过坏事件的分支。
+5. `EventScanJob` 的 `sync_cursor` 按 source 独立维护；扫描写事件和投递消费是两条独立流程。消费阻塞不等于源扫描停止，其它消费者的位点也独立。
+
+因此，按当前 master，一批无效关系事件可以挡住 PAC 消费者后面的所有订阅事件。该机制与生产观察到的关系表失败及一小时重试一致；生产具体位点、后面有哪些事件、积压多少，以及部署代码是否与 master 一致，仍未直接核实。
+
+**只部署 PAC 的批次误判修复不能保证解除阻塞**：回执仍如实保留 `failed=3`，该版本消费者继续判失败。需修正或隔离已经入队的无效关系事件，或者让消费端按行隔离失败并保留审计。只改未来扫描的过滤条件不会改写已有事件 payload。

packages/types/src/enums/index.ts

@@ -205,6 +205,7 @@ export const Action = {
   EMR_SUBMITTED: 'emr_submitted', // 病历提交
   EMR_REJECTED: 'emr_rejected', // 病历打回
   IMAGE_UPLOADED: 'image_uploaded', // 影像上传
+  IMAGE_ANALYSIS_RECORDED: 'image_analysis_recorded', // 宿主影像分析记录
 
   // ── 产品收集 ──
   RECHARGE_RECEIVED: 'recharge_received', // 充值入账
@@ -235,6 +236,7 @@ export const ActionSchema = z.enum([
   'emr_submitted',
   'emr_rejected',
   'image_uploaded',
+  'image_analysis_recorded',
   // 产品收集
   'recharge_received',
   'complaint_filed',
@@ -261,6 +263,7 @@ export const SubjectType = {
   PAYMENT: 'payment', // 收款
   REFUND: 'refund', // 退费
   IMAGE_STUDY: 'image', // 影像资料 — W3 末统一命名,对齐 FactType.IMAGE_RECORD
+  IMAGE_ANALYSIS: 'image_analysis', // 宿主提供的影像分析
   // ── 产品收集 ──
   RECHARGE: 'recharge', // 充值/储值/福利卡办理
   COMPLAINT: 'complaint', // 投诉
@@ -282,6 +285,7 @@ export const SubjectTypeSchema = z.enum([
   'payment',
   'refund',
   'image',
+  'image_analysis',
   // 产品收集
   'recharge',
   'complaint',
@@ -347,6 +351,7 @@ export const FactType = {
   REFUND_RECORD: 'refund_record', // 退费记录
   EMR_RECORD: 'emr_record', // 病历记录(自由文本)
   IMAGE_RECORD: 'image_record', // 影像记录
+  IMAGE_ANALYSIS_RECORD: 'image_analysis_record', // 分析原文、完成状态及问题快照
   APPOINTMENT_RECORD: 'appointment_record', // 预约(kind=planned/actual 区分)
 
   // ── 产品收集 ──
@@ -370,6 +375,7 @@ export const FactTypeSchema = z.enum([
   'refund_record',
   'emr_record',
   'image_record',
+  'image_analysis_record',
   'appointment_record',
   // 产品收集
   'recharge_record',

packages/types/src/schemas/canonical.ts

@@ -352,10 +352,58 @@ export const ImageStudyCanonicalSchema = z
     clinicId: z.string().min(1),
     uploadedAt: isoDateTime,
     modality: z.string().optional().nullable(),
+    fileUrl: z.string().url().regex(/^https?:\/\//i).optional().nullable(),
+    fileName: z.string().optional().nullable(),
+    fileType: z.string().optional().nullable(),
   })
   .passthrough();
 export type ImageStudyCanonical = z.infer<typeof ImageStudyCanonicalSchema>;
 
+/** 宿主分析的通用投影；私有问题码不进入临床归一。 */
+export const ImageAnalysisCanonicalSchema = z.object({
+  externalId: z.string().min(1),
+  patientExternalId: z.string().min(1),
+  clinicId: z.string().min(1),
+  imageExternalId: z.string().min(1),
+  analyzedAt: optionalIsoDateTime,
+  createdAt: optionalIsoDateTime,
+  updatedAt: optionalIsoDateTime,
+  status: z.enum(['pending', 'completed', 'failed']),
+  result: z.string().optional().nullable(),
+  expectedFindingCount: z.number().int().nonnegative().optional(),
+  findings: z.array(z.object({
+    externalId: z.string().min(1),
+    patientExternalId: z.string().min(1),
+    imageExternalId: z.string().min(1),
+    code: z.string().optional().nullable(),
+    name: z.string().min(1),
+    toothPosition: z.string().optional().nullable(),
+    excludedToothPositions: z.string().optional().nullable(),
+    excluded: z.preprocess((v) => v === 'true' ? true : v === 'false' ? false : v,
+      z.boolean()).default(false),
+  }).passthrough()),
+}).passthrough().superRefine((row, ctx) => {
+  for (const field of ['analyzedAt', 'createdAt', 'updatedAt'] as const) {
+    if (row[field] && !Number.isFinite(new Date(row[field]).getTime())) {
+      ctx.addIssue({ code: 'custom', path: [field], message: 'invalid datetime' });
+    }
+  }
+  if (row.expectedFindingCount !== undefined && row.findings.length !== row.expectedFindingCount) {
+    ctx.addIssue({ code: 'custom', path: ['findings'], message: 'incomplete finding snapshot' });
+  }
+  const ids = new Set<string>();
+  for (const [i, finding] of row.findings.entries()) {
+    if (ids.has(finding.externalId)) {
+      ctx.addIssue({ code: 'custom', path: ['findings', i, 'externalId'], message: 'duplicate finding id' });
+    }
+    ids.add(finding.externalId);
+    if (finding.patientExternalId !== row.patientExternalId || finding.imageExternalId !== row.imageExternalId) {
+      ctx.addIssue({ code: 'custom', path: ['findings', i], message: 'finding ownership differs from analysis' });
+    }
+  }
+});
+export type ImageAnalysisCanonical = z.infer<typeof ImageAnalysisCanonicalSchema>;
+
 // ─────────────────────────────────────────────────────────
 // v2.1 拆出:diagnosis / treatment / recommendation 独立 canonical 资源
 //   transaction 层只校验"身份 + 时间 + 诊所"三件套;
@@ -522,6 +570,7 @@ export const CanonicalResourceSchemas = {
   payment: PaymentCanonicalSchema,
   refund: RefundCanonicalSchema,
   image: ImageStudyCanonicalSchema,
+  image_analysis: ImageAnalysisCanonicalSchema,
   // 产品收集
   recharge: RechargeCanonicalSchema,
   complaint: ComplaintCanonicalSchema,
@@ -649,6 +698,12 @@ export const CanonicalResourceMeta: Record<
     datetimeFields: ['uploadedAt'],
     booleanFields: [],
   },
+  image_analysis: {
+    moneyFields: [],
+    moneyArrayFields: [],
+    datetimeFields: ['analyzedAt', 'createdAt', 'updatedAt'],
+    booleanFields: [],
+  },
   // ── 产品收集 ──
   recharge: {
     moneyFields: ['amount'],