Merge push evidence and image path fixes from main into test

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

 ---
 title: 数据模型
-description: PAC 数据库 18 个 model 的层次、写入模式、隔离纪律与关键索引。
+description: PAC 数据库模型的层次、写入模式、隔离纪律与关键索引。
 icon: Database
 ---
 
@@ -8,7 +8,7 @@ icon: Database
 
 ---
 
-## 模型地图(18 model)
+## 模型地图
 
 隔离基线统一是 **`host_id + tenant_id`**(进每张业务表唯一键 + 每条查询)。
 
@@ -32,6 +32,7 @@ icon: Database
 | `plan_generation_logs` | 横切 L3 | append-only | 召回生成执行账本 |
 | `agent_invocations` | 横切 | append-only | AI 调用审计账本(跨所有层共享) |
 | `sync_logs` | 横切 L1 | append-only | 同步运行账本(cursor / reconcile / 计数) |
+| `ingest_evidence` | 摄入诊断 | 原文去重 + 有限保留 | `/push/rows` 已验签完整请求，包含成功行与入库前拒收行；不参与事实计算 |
 
 > PAC **不立** tenants / clinics / users 表 —— 这三个 id 都是宿主经换票 token 提供的 `TEXT`,PAC 不维护其主数据(见 [架构总览](/docs/architecture/overview) · 多租户与权限)。
 
@@ -59,6 +60,8 @@ icon: Database
 
 子表跟随父版本:`persona_features` 随 `personas`、`plan_scripts`/`plan_summaries` 随 `followup_plans`;父表 supersede 时子表自然进入历史。
 
+入站诊断表 `ingest_evidence` 按报文哈希合并接收次数，到期删除；它不是业务账本或事实版本流。
+
 **4 · 证据链与版本化。** 每个对外输出都能反查"为什么这么定":
 
 ```
@@ -110,7 +113,7 @@ patient_facts.transactionIds → patient_transactions.id → rawPayload
 
 **`patient_return_visits`** — 诊所回访任务记录(展示用)。唯一键 `(host_id, tenant_id, source_unit, external_id)`。**不是临床 fact、不进召回信号**;详情页"回访记录"按 `task_date` 倒序展示,避免客服重复外呼。
 
-**`patient_transactions`** — 操作账本(append-only)。`event_seq` BigInt 单调水位(供 persona / 游标消费);`source_event_id` 幂等键(Push 必带,Pull adapter 合成);`raw_payload`(宿主原文,唯一留底 + reparse 输入)+ `payload_hash`;`clinic_id` 立柱。`action` × `subject_type` 为 PAC 归一化封闭集 —— **23 个 action × 16 个 subject_type**(`@pac/types` `Action` / `SubjectType`)。
+**`patient_transactions`** — 操作账本(append-only)。`event_seq` BigInt 单调水位(供 persona / 游标消费);`source_event_id` 幂等键(Push 必带,Pull adapter 合成);`raw_payload`(单条宿主源行，push/pull 共用的 reparse 输入)+ `payload_hash`;`clinic_id` 立柱。`action` × `subject_type` 为 PAC 归一化封闭集 —— **23 个 action × 16 个 subject_type**(`@pac/types` `Action` / `SubjectType`)。
 
 **`patient_facts`** — 事实单元(版本流)。`subject_id` 业务身份跨版本稳定,`(…, subject_id, version)` 单调递增;`kind`(`actual`/`planned`,2)× `type`(**16 个 FactType**)× `status`(`active`/`superseded`/`cancelled`/`fulfilled`/`expired`/`invalidated`,6);`content` JSONB(per type zod 校验,规则引擎只读这里);`transaction_ids` 证据链。一个 transaction 可产 0/1/N 个 fact。
 
@@ -144,7 +147,9 @@ patient_facts.transactionIds → patient_transactions.id → rawPayload
 
 **`sync_logs`** — 同步运行账本。`direction`(pull/push)/ `resource` / `status`(running/success/partial/failed)/ `cursor_before`/`after`(pull 游标;push 留空)/ `triggered_by`(含 `reconcile:<date>` 对账)/ `fetched`、`transactions_written`、`duplicates`、`failed`、`facts_emitted` / `metadata`(cohort 进度 + 漂移审计 `suspectFields`)。
 
-Push 的 `fetched` 是收到的原生行数;主档 / 关系 / 回访直接 upsert 不计入流水数,原生行还可能被拆分或过滤,各计数不保证相加等于 `fetched`。有成功 upsert、新流水或幂等命中且部分失败时记 `partial`;入库前拒收行不另存原文或自动隔离。详见 [Push 回执](/docs/integration/channel-push#9-回执与重试)。
+Push 的 `fetched` 是收到的原生行数;主档 / 关系 / 回访直接 upsert 不计入流水数,原生行还可能被拆分或过滤,各计数不保证相加等于 `fetched`。有成功 upsert、新流水或幂等命中且部分失败时记 `partial`。`ingest_evidence_id` 指向请求原文；`metadata.rowFailures` 留行号、阶段、字段级原因，最多 500 项，截断时标记 `rowFailuresTruncated`。没有自动隔离或重试。详见 [Push 回执](/docs/integration/channel-push#9-回执与重试)。
+
+**`ingest_evidence`** — `/push/rows` 入站诊断原文。唯一键 `(host_id, source, payload_hash)`，哈希按完整原始正文计算；`raw_body` 为 TEXT，保留字段缺席 / null、空白、顺序和数字原文。重复报文共用一份正文，`received_count` 原子累加，每次运行仍生成独立同步记录。成功默认保留 7 天，有失败默认保留 30 天；清理时同步日志引用置空，定位信息保留。请求可跨品牌，仅按验签 `host_id` 归属，原文限服务器工具读取，不接入工作台查询。不保存映射、时区、金额单位配置快照。它补足入库前的原文缺口，不替代 `patient_transactions.raw_payload`；边界见 [入站证据](/docs/ingestion#53-入站证据与重解析边界)。
 
 ---
 

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

@@ -174,13 +174,13 @@ 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` | `fileUrl` | 宿主文件位置原值 → `file_url`；FRIDAY 从原表 `file_url` 取值，允许相对路径，不拼接、不验证可访问性 |
 | `image` | `fileName` / `fileType` | 文件名 / 文件类型 → `file_name` / `file_type`，有源值时保留 |
 | `diagnosis` | `sourceImageExternalId` | 精确指向同宿主、同品牌、同患者的影像 externalId → `source_image_external_id` |
 | `diagnosis` | `sourceAnalysisExternalId` | 产生此发现的稳定分析请求 ID → `source_analysis_external_id` |

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

@@ -140,8 +140,8 @@ pnpm sync -- --dir=./data/<host> --full          # 忽略游标强制全量(灾
 **入站失败与原文留存**:
 
 - 校验失败的记录拒收,有效记录继续写入;Push 的批次分类与重试见 [Push 回执](/docs/integration/channel-push#9-回执与重试)。
-- 主档、关系、回访直接 upsert,不生成原始流水;入库前拒收行仅有失败计数和错误日志,不保存完整原文、不进入自动隔离 / 重试队列。
-- 已写入的 `patient_transactions` 保留 `rawPayload`;事实校验拒收用 `factsFailed` / `factRejects` 报告。宿主保留源事件及 payload,负责修正重推或持久化隔离。
+- 主档、关系、回访直接 upsert，不生成事实流水；`/push/rows` 经入口校验后由 `ingest_evidence` 留完整请求原文，含入库前拒收行。没有自动隔离 / 重试队列。
+- 已写入的 `patient_transactions` 继续保留单条源行 `rawPayload`，供 push、pull 的现有 reparse 共用；事实校验拒收用 `factsFailed` / `factRejects` 报告。两种原文的边界见 [§5.3](#53-入站证据与重解析边界)。宿主仍保留源事件及 payload，负责修正重推或持久化隔离。
 
 **幂等(数据库双 UNIQUE)** —— 任何并发 / 重跑结果一致:
 
@@ -231,3 +231,27 @@ docker compose -f docker-compose.prod.yml exec -T pac-service tail -f /tmp/repar
 | env `PAC_REPARSE_BATCH` | 全量时每批患者数(默认 3000,控内存) |
 
 > **典型场景**:源数据中文诊断名变体繁多、多数无标准 ICD 码,全靠 enum / keyword 白名单翻译;漏映射会导致 `code=null`、进不了对应病种召回。修法:在 `diagnosis.yaml` 补 keyword → `pnpm reparse:prod -- --subject-type=diagnosis`,即可把该病种此前漏召的存量患者补回召回池。十万级患者全量 reparse 内存恒定在 1G 以内,单患者 / 小批为秒级。
+
+### 5.3 入站证据与重解析边界
+
+| 原文 | 用途 | 后续处理 |
+|---|---|---|
+| `patient_transactions.rawPayload` | 已入库事件的单条宿主源行；push、pull 保持同一语义 | 现有 `reparse` 读取，复用 transforms、assembler、`processSubject` 和事实版本流 |
+| `ingest_evidence.rawBody` | `/push/rows` 的完整请求，补足主档 / 关系 / 回访及入库前拒收行的原文缺口 | 本次提供留存、失败定位及只读导出；未新增重放执行器 |
+
+原文留存不改变流水的根数据形状：`rawPayload` 仍是原始源行，不改成请求包或证据引用。未来从证据重放时，只恢复 `{source, rows}` 并调用现有 `ingestRawTables` 入口；临床事件仍写原始流水，再走同一事实管道，直接 upsert 资源沿用原路径。不另建 push 的事实解析或患者写入链路。
+
+**范围与保留**：验签、请求结构校验、并发闸通过后，先保存原文再执行业务摄入。成功、失败、预检都保存；留档失败时不执行业务摄入或确认成功。`/push/events`、未验签、格式非法、超限请求不纳入本次留存。相同宿主、source、完整正文共用一份证据，接收次数及最近接收时间累加。
+
+默认成功保留 7 天、有失败保留 30 天，按最近接收延长；失败标记不因后续重推成功而缩短保留期。环境变量 `PAC_INGEST_EVIDENCE_SUCCESS_DAYS` / `PAC_INGEST_EVIDENCE_FAILURE_DAYS` 可配置 1–3650 天，失败保留期不短于成功。每天上海时间 04:30 分批清理，单次最多 10,000 份；`PAC_SCHEDULER_DISABLED=1` 时不自动清理。删除正文后，`SyncLog.ingestEvidenceId` 置空，日志及失败定位仍保留。
+
+**服务器侧查询**：推送记录含 `ingestEvidenceId`，原文不经宿主 HTTP API 返回。按验签宿主查询同步日志或证据 id；有同步日志时同时返回 `metadata.rowFailures`：0 基行号、资源、阶段、字段级原因，最多 500 项，截断标记 `rowFailuresTruncated`。无法定位单行的批次异常仍保留完整正文及批次错误。
+
+```bash
+pnpm --filter @pac/service ingest-evidence -- --host=friday --sync-log=<uuid> --out=/path/request.json
+# 尚未生成同步日志的早期异常，可改用 --evidence=<uuid>
+```
+
+服务器构建产物用 `ingest-evidence:prod`。`--out` 导出完整正文，不覆盖已有文件，文件权限 0600；命令只读，不启动定时任务或自动重放。未指定 `--out` 时在终端输出正文。
+
+映射、时区、金额单位仍读取宿主当前配置，不保存配置快照。修改配置后的 reparse 按新口径计算；缺身份等源数据错误须修正后再推，原文留档不会自动补值或解除消费端重试阻塞。

apps/pac-docs/content/docs/integration/channel-push.mdx

@@ -250,7 +250,7 @@ const sig  = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
 | `failed` | 装配 / 写入失败计数,按派生资源累计 | 修正后重推,或由宿主持久化隔离后处理;不能直接忽略 |
 | `factsFailed` | 流水已保存,但部分派生事实被校验拒收的数量 | 看 `factRejects` 定位与修正 |
 | `mappingMisses` | 存在未映射的原值 | 暂停推送,发样本给 PAC |
-| `suspectFields` | 与历史相比列发生增减 | 确认表结构是否变更,同步 PAC |
+| `suspectFields` | 必填映射列整批缺席，或与历史原文相比列发生增减 | 确认表结构是否变更,同步 PAC；可选字段无历史基线且未提供时不计缺列 |
 
 一条原生行可能拆成多条流水或被业务规则过滤,上述计数不保证相加等于 `accepted`。
 `status` 在推送记录中查询,不是 `/push/rows` 回执中的字段。
@@ -297,8 +297,9 @@ const sig  = require('crypto').createHmac('sha256', SECRET).update(`${ts}.${body
 
 **失败留档与重试责任**:
 
-- PAC 当前记录失败计数和错误日志,不保存入库前被拒行的完整原文,不提供这些行的逐条主键清单,也没有自动隔离 / 重试队列。
-- 主档、关系、回访仅保存当前态;这些直接 upsert 资源不生成原始流水。只有已写入的 `patient_transactions` 保留 `rawPayload`。
+- `/push/rows` 经验签、请求结构校验及并发闸后，先留完整请求原文，再处理数据；成功、部分失败、整批失败及预检均留档。未通过这些入口检查的请求不留正文。
+- 推送记录返回 `ingestEvidenceId`；逐行装配、身份、租户或写入失败的定位保存在 PAC 同步日志。原文仅供 PAC 服务器侧按宿主查询，不经宿主 API 返回。保留期与查询方式见 [入站证据](/docs/ingestion#53-入站证据与重解析边界)。
+- 主档、关系、回访仍直接 upsert；临床事件仍写入 `patient_transactions.rawPayload`，现有 reparse 继续读取该字段。留档不改变宿主 payload、幂等键或回执口径，也不增加自动隔离 / 重试队列。
 - 宿主保留 outbox / 事件日志及原始 payload。数据错误修正后重推;按业务策略隔离失败事件时,先持久化并留审计,再决定是否推进位点。
 - 同一批重推具有幂等性,但不修正原数据只会重复同一错误。统一位点的消费端若遇到 `failed>0` 就整批重试,后续事件会被该批阻挡。
 

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

@@ -191,7 +191,7 @@ icon: FileJson
 | `emr_id` | string | | → 病历号(`med_emr_info.emr_id`) |
 | `class_name` | string | | **影像类型名(宿主 inline)**——原 `class_code` 经自家 `std_check_class` 解析成类型名(口内像照片/小牙片/全景片/CT影像截图/口腔扫描…);PAC 再映射成 modality |
 | `file_name` / `file_type` | string | | 文件名 / 类型 |
-| `file_url` | string | | OSS 相对路径 |
+| `file_url` | string | | 宿主原文件位置，允许 OSS 相对路径；原样推送，PAC 保留原值 |
 | `files_size` | number | | 字节数 |
 | `shooting_time` | string(datetime) | ✅ | 拍摄时间 |
 | `created_gmt_at` / `updated_gmt_at` | string(datetime) | ✅ | |
@@ -323,7 +323,7 @@ icon: FileJson
 | `return_visit_method` | string/number | | 回访方式——`1`打电话 `2`发微信 `3`发短信 |
 | `created_gmt_at` / `updated_gmt_at` | string(datetime) | ✅ | 创建 / 末次修改时间 |
 
-**可一并 inline 的 `customer_task` 列**(PAC 当前不映射、不保存这些额外原文;
+**可一并 inline 的 `customer_task` 列**(PAC 当前不映射到回访业务字段；请求留档仍保留原文，
 需要使用时另行补映射并重推):`suggested_return_person`(建议回访人)、`actual_return_person` /
 `actual_return_time`(实际回访人/时间)、`customer_status`(客户状态)、`is_first`(初诊/复诊)。
 
@@ -378,8 +378,8 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
 ```
 
 两者返回**同一份数据**:按时间倒序最近 N 批(默认 50、上限 200),每批含
-`source / status / dryRun / fetched / transactionsWritten / duplicates / failed / errorMessage`。
-可事后查询批次状态与失败计数;`errorMessage` 是批次级错误,逐行校验原因看 PAC 日志。当前不返回失败行主键或原文。预检批次由 `dryRun:true` 标识,不与正式推送混淆。
+`source / status / dryRun / fetched / transactionsWritten / duplicates / failed / errorMessage / ingestEvidenceId`。
+可事后查询批次状态与失败计数；`errorMessage` 是批次级错误，`ingestEvidenceId` 供 PAC 侧定位完整请求原文，到期清理后为空。原文及逐行失败定位仅供 PAC 服务器侧排查，宿主 API 不返回正文或失败行主键。预检批次由 `dryRun:true` 标识。
 
 ### 8.3 其他可用面
 
@@ -456,15 +456,11 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
 - `api_req` / `api_res` 调用日志不需推送。
 - 本新增 source 在 PAC 部署并联调后启用。
 
-### 影像文件链接补充
+### 影像文件位置
 
-`med_check` 原字段及 `file_url` 含义保持不变，新增以下字段：
+`med_check.file_url` 按宿主原值推送，允许相对路径；无需补充完整链接。`fileUrlEcho` 作废，PAC 不映射此字段。
 
-| 字段 | 类型 | 要求 |
-|---|---|---|
-| `fileUrlEcho` | string(url) | 完整可访问的 HTTP(S) 文件链接，含所需签名参数；沿用 SaaS 现有文件地址解析服务 |
-
-地址变化时按同一影像 `id` 重推；PAC 保留完整链接，并通过 `check_id` 关联分析结果。
+文件位置变化时按同一影像 `id` 重推；PAC 保留原值，通过 `check_id` 关联分析结果。展示时的地址解析另行处理。
 
 ### 报文示例
 
@@ -503,6 +499,6 @@ GET /pac/v1/admin/host/self/push-logs?limit=50
 
 - `code=0` 表示请求处理完成;同时检查 `failed` / `factsFailed`。部分成功记为 `partial`,有效行正常写入,失败行拒收。
 - 主档、关系、回访直接 upsert;成功也可能 `transactionsWritten=0, duplicates=0`。`accepted` 仅为收到的原生行数。
-- 入库前拒收的行不保存完整原文、不进入 PAC 自动隔离 / 重试队列;宿主保留事件 payload,修正后重推或持久化隔离处理。
+- `/push/rows` 经入口校验后完整留档，包含入库前被拒的原生行；宿主继续按本契约推送，无需调整格式。留档不自动修正或重试坏数据，宿主仍保留事件 payload，修正后重推或持久化隔离处理。
 - `failed>0` 时不能仅凭 `code=0` 推进位点;不处理坏事件而整批重试会阻挡后续推送。
 - 统一口径与示例见 [Push 通道 · 回执与重试](/docs/integration/channel-push#9-回执与重试)。

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

@@ -23,7 +23,7 @@ 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:              fileUrlEcho       # 完整 HTTP(S) 地址；原存储路径仍在 raw
+  fileUrl:              file_url         # 宿主原值；可为相对存储路径，不拼接或校验可访问性
   fileName:             file_name
   fileType:             file_type
 

apps/pac-service/package.json

@@ -46,7 +46,9 @@
     "stale-scan:prod": "node dist/cli/stale-scan.cli.js",
     "openapi:dump": "ts-node --transpile-only src/cli/dump-openapi.cli.ts",
     "refresh-clinic-names": "ts-node --transpile-only src/cli/refresh-clinic-names.cli.ts",
-    "golden": "ts-node --transpile-only tests/golden/run.ts"
+    "golden": "ts-node --transpile-only tests/golden/run.ts",
+    "ingest-evidence": "ts-node --transpile-only src/cli/ingest-evidence.cli.ts",
+    "ingest-evidence:prod": "node dist/cli/ingest-evidence.cli.js"
   },
   "prisma": {
     "seed": "ts-node --transpile-only prisma/seed.ts"

apps/pac-service/prisma/migrations/20261013090000_ingest_evidence/migration.sql

+CREATE TABLE "ingest_evidence" (
+  "id" UUID NOT NULL,
+  "host_id" UUID NOT NULL,
+  "source" TEXT NOT NULL,
+  "payload_hash" VARCHAR(64) NOT NULL,
+  "raw_body" TEXT NOT NULL,
+  "row_count" INTEGER NOT NULL,
+  "received_count" INTEGER NOT NULL DEFAULT 1,
+  "first_received_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "last_received_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "has_failures" BOOLEAN NOT NULL DEFAULT false,
+  "expires_at" TIMESTAMPTZ(3) NOT NULL,
+  CONSTRAINT "ingest_evidence_pkey" PRIMARY KEY ("id"),
+  CONSTRAINT "ingest_evidence_host_id_fkey" FOREIGN KEY ("host_id") REFERENCES "hosts"("id") ON DELETE CASCADE ON UPDATE CASCADE,
+  CONSTRAINT "ingest_evidence_counts_check" CHECK ("row_count" >= 0 AND "received_count" > 0)
+);
+CREATE UNIQUE INDEX "ingest_evidence_host_id_source_payload_hash_key" ON "ingest_evidence"("host_id", "source", "payload_hash");
+CREATE INDEX "ingest_evidence_expires_at_idx" ON "ingest_evidence"("expires_at");
+ALTER TABLE "sync_logs" ADD COLUMN "ingest_evidence_id" UUID;
+CREATE INDEX "sync_logs_ingest_evidence_id_idx" ON "sync_logs"("ingest_evidence_id");
+ALTER TABLE "sync_logs" ADD CONSTRAINT "sync_logs_ingest_evidence_id_fkey"
+  FOREIGN KEY ("ingest_evidence_id") REFERENCES "ingest_evidence"("id") ON DELETE SET NULL ON UPDATE CASCADE;

apps/pac-service/prisma/schema.prisma

@@ -153,6 +153,7 @@ model Host {
   transactions         PatientTransaction[]
   facts                PatientFact[]
   syncLogs             SyncLog[]
+  ingestEvidence       IngestEvidence[]
   personas             Persona[]
   personaRecomputeLogs PersonaRecomputeLog[]
   followupPlans        FollowupPlan[]
@@ -1929,15 +1930,35 @@ model AgentInvocation {
   @@map("agent_invocations")
 }
 
+/// Push 原生行入站证据；完整原报文独立于事实流水，不保存宿主配置快照。
+/// 一批可以跨品牌，所有归属按验签 host 隔离；仅服务器侧排查工具读取原文。
+/// 不参与事实解析；已入库事件仍以 PatientTransaction.rawPayload 为 reparse 输入。
+model IngestEvidence {
+  id              String   @id @default(uuid()) @db.Uuid
+  hostId          String   @map("host_id") @db.Uuid
+  source          String
+  payloadHash     String   @map("payload_hash") @db.VarChar(64)
+  rawBody         String   @map("raw_body") @db.Text
+  rowCount        Int      @map("row_count")
+  receivedCount   Int      @default(1) @map("received_count")
+  firstReceivedAt DateTime @default(now()) @map("first_received_at") @db.Timestamptz(3)
+  lastReceivedAt  DateTime @default(now()) @map("last_received_at") @db.Timestamptz(3)
+  hasFailures     Boolean  @default(false) @map("has_failures")
+  expiresAt       DateTime @map("expires_at") @db.Timestamptz(3)
+  host            Host     @relation(fields: [hostId], references: [id], onDelete: Cascade)
+  syncLogs        SyncLog[]
+
+  @@unique([hostId, source, payloadHash])
+  @@index([expiresAt])
+  @@map("ingest_evidence")
+}
+
 /// 事实接入 orchestration 日志。同步引擎每次 pull / 接收 push 都落一条记录,
 /// 用于追踪进度、定位失败、计量数据量。跟 PersonaRecomputeLog(画像重算)/ PlanGenerationLog(召回生成)
 /// 平行,不混用 — 各 orchestration 域一张账本,字段贴近自身语义。
 ///
-/// 故意只到 platform + tenant — 一次批量 pull 把整个 tenant 所有 clinic 的数据一并拉来,
 /// run 级别没有 clinic 概念。落到 patient_transactions 的 clinic_id 是 per-record 业务字段。
-///
-/// 单患者 pull 例外:patient_id 立柱(可空),NULL = 批量 run,非 NULL = 单患者即时 pull。
-/// 单患者 pull 路径**绝不写 cursor_after**(避免污染批量水位),靠 DB 唯一约束兜底重复。
+/// 单患者 pull 的 patient_id 立柱(可空)，该路径绝不写 cursor_after(避免污染批量水位)。
 model SyncLog {
   id        String  @id @default(uuid()) @db.Uuid
   /// 接入宿主(FK → platforms.id)
@@ -1946,6 +1967,10 @@ model SyncLog {
   /// 单患者 pull 时填写,批量 run 留 null。非 NULL 时 cursor_before/cursor_after 也应为 null。
   patientId String? @map("patient_id") @db.Uuid
 
+  /// /push/rows 的已验签请求原文。到期清理后引用置空，失败定位仍保留。
+  ingestEvidenceId String? @map("ingest_evidence_id") @db.Uuid
+  ingestEvidence   IngestEvidence? @relation(fields: [ingestEvidenceId], references: [id], onDelete: SetNull)
+
   /// 同步方向:pull(PAC 拉)/ push(宿主推)。mock 数据走 pull 路径(adapter 内部造数据)
   direction String
   /// 同步资源(对齐 patient_transactions 的 subject_type 维度:
@@ -1980,9 +2005,9 @@ model SyncLog {
   /// 幂等去重命中数(source_event_id 已存在,跳过未入库)。
   /// 主档/关系/回访直接 upsert、原生行拆分/过滤时,计数不保证相加等于 fetched。
   duplicates          Int     @default(0)
-  /// 装配/写入失败计数;拒收行不另存原文,当前没有 failed_transactions 隔离表。
+  /// 装配/写入失败计数;原文见 ingestEvidence，当前没有自动隔离/重试队列。
   failed              Int     @default(0)
-  /// 批次级错误主因;逐行校验原因进应用日志,不提供失败行原文或主键清单。
+  /// 批次级错误主因;逐行校验定位见 metadata.rowFailures，原文不经宿主 API 返回。
   errorMessage        String? @map("error_message") @db.Text
 
   startedAt DateTime  @default(now()) @map("started_at") @db.Timestamptz(3)
@@ -1991,6 +2016,8 @@ model SyncLog {
   /// cohort batch / assembler 进度 + 资源画像(JSONB,无强 schema)
   /// 形态见 packages/types schema(可选,目前用 unknown);典型字段:
   ///   { cohortTotal, cohortDone, lastAssembler, rssPeakMb, batchTimings: [...] }
+  /// /push/rows: rowFailures = [{rowIndex(0 基,无法定位为 null),resource,stage,reason,fields?}]，最多 500。
+  /// rowFailuresTruncated 表示定位样本被截断；原始正文仍完整保留于 ingestEvidence。
   metadata Json?
 
   host         Host                 @relation(fields: [hostId], references: [id])
@@ -2001,6 +2028,7 @@ model SyncLog {
   @@index([status])
   /// 单患者 pull 反查:"该患者最近被刷新过几次"
   @@index([patientId, startedAt])
+  @@index([ingestEvidenceId])
   /// MIGRATION 20260528000000 + 20260703021623:partial UNIQUE (host_id)
   ///   WHERE status='running' AND direction='pull' — 同 host 同时只能有 1 个 running 的
   ///   **pull** 行(存量/增量 sync 单飞锁);push 批次不受此锁(并发由 PushController 信号量管)。

apps/pac-service/src/cli/ingest-evidence.cli.ts

+/** 服务器侧证据导出。只读，不启动 Nest，不注册定时任务，不自动回放。 */
+import { PrismaClient } from '@prisma/client';
+import { existsSync, writeFileSync } from 'node:fs';
+import { loadEnvFile } from 'node:process';
+
+export async function readEvidence(prisma: Pick<PrismaClient, 'host' | 'syncLog'>,
+  hostName: string, syncLogId: string) {
+  const host = await prisma.host.findUnique({ where: { name: hostName }, select: { id: true } });
+  if (!host) throw new Error('宿主不存在');
+  const log = await prisma.syncLog.findFirst({ where: { id: syncLogId, hostId: host.id },
+    select: { id: true, resource: true, status: true, metadata: true, ingestEvidence: true } });
+  if (!log) throw new Error('该宿主没有此同步记录');
+  if (!log.ingestEvidence) throw new Error('该批次未留原文，或原文已到期清理');
+  if (log.ingestEvidence.hostId !== host.id) throw new Error('证据宿主归属不匹配');
+  // 不解析正文，保留字段缺席、空值、数字精度、空白及原始字段顺序。
+  return { syncLogId: log.id, source: log.resource, status: log.status,
+    metadata: log.metadata, evidence: log.ingestEvidence };
+}
+
+/** source 配置错误或早期异常可能尚未生成 SyncLog；仍可按证据 id 回看原文。 */
+export async function readEvidenceById(prisma: Pick<PrismaClient, 'host' | 'ingestEvidence'>,
+  hostName: string, id: string) {
+  const host = await prisma.host.findUnique({ where: { name: hostName }, select: { id: true } });
+  if (!host) throw new Error('宿主不存在');
+  const evidence = await prisma.ingestEvidence.findFirst({ where: { id, hostId: host.id } });
+  if (!evidence) throw new Error('该宿主没有此证据，或原文已到期清理');
+  return { evidence };
+}
+
+async function main(): Promise<void> {
+  const args = new Map(process.argv.slice(2).filter(a => a.startsWith('--')).map(a => {
+    const pos = a.indexOf('=');
+    return pos < 0 ? [a.slice(2), ''] : [a.slice(2, pos), a.slice(pos + 1)];
+  }));
+  if (args.has('help')) {
+    console.log('pnpm ingest-evidence -- --host=friday (--sync-log=<UUID> | --evidence=<UUID>) [--out=/path/request.json]');
+    return;
+  }
+  const host = args.get('host');
+  const logId = args.get('sync-log');
+  const evidenceId = args.get('evidence');
+  const id = logId || evidenceId;
+  if (!host || !id || !!logId === !!evidenceId || !/^[a-f\d]{8}(?:-[a-f\d]{4}){3}-[a-f\d]{12}$/i.test(id)) {
+    throw new Error('需要 --host，并指定合法 UUID --sync-log 或 --evidence（二选一）');
+  }
+  if (existsSync('.env')) loadEnvFile('.env');
+  const prisma = new PrismaClient();
+  try {
+    const result = logId ? await readEvidence(prisma, host, logId) : await readEvidenceById(prisma, host, id);
+    const out = args.get('out');
+    if (out) {
+      // 不覆盖已有文件；导出的患者原文仅文件所有者可读。
+      writeFileSync(out, result.evidence.rawBody, { encoding: 'utf8', flag: 'wx', mode: 0o600 });
+      const { rawBody: _rawBody, ...evidence } = result.evidence;
+      console.log(JSON.stringify({ ...result, evidence, exportedTo: out }, null, 2));
+    } else {
+      console.log(JSON.stringify(result, null, 2));
+    }
+  } finally {
+    await prisma.$disconnect();
+  }
+}
+
+if (require.main === module) {
+  void main().catch(error => { console.error((error as Error).message); process.exitCode = 1; });
+}

apps/pac-service/src/modules/admin/hosts.service.ts

@@ -439,6 +439,7 @@ export class HostsService {
         duplicates: r.duplicates ?? 0,
         failed: r.failed ?? 0,
         errorMessage: r.errorMessage,
+        ingestEvidenceId: r.ingestEvidenceId ?? null,
         startedAt: r.startedAt.toISOString(),
         endedAt: r.endedAt?.toISOString() ?? null,
       })),

apps/pac-service/src/modules/sync/assembler/row-rejections.ts

+import type { AssemblerResult } from './assembler-engine';
+
+export interface RowRejection {
+  rawSource: unknown;
+  stage: 'assemble' | 'identity' | 'tenant' | 'write' | 'pipeline';
+  reason: string;
+  fields?: string[];
+}
+
+// 原文引用仅在本次调用中存在，不进入 PerResourceStats 的 JSON 响应或应用日志。
+const rejected = new WeakMap<object, RowRejection[]>();
+const assemblerSources = new WeakMap<object, unknown>();
+const truncated = new WeakSet<object>();
+
+export function withRejectedSource<T extends object>(error: T, source: unknown): T {
+  assemblerSources.set(error, source);
+  return error;
+}
+export const ROW_REJECTION_LIMIT = 500;
+
+export function recordRowRejection(stats: object, entry: RowRejection): void {
+  const entries = rejected.get(stats) ?? [];
+  if (entries.length < ROW_REJECTION_LIMIT) entries.push(entry);
+  else truncated.add(stats);
+  rejected.set(stats, entries);
+}
+
+export function recordAssemblerRejections(stats: object, assembled: AssemblerResult): void {
+  for (const error of assembled.errors ?? []) {
+    const issues = Array.isArray(error.issues) ? error.issues : [];
+    const fields = issues.flatMap((issue: unknown) => {
+      if (typeof issue !== 'object' || issue === null || !('path' in issue) || !Array.isArray(issue.path)) return [];
+      return [issue.path.join('.')];
+    });
+    recordRowRejection(stats, { rawSource: assemblerSources.get(error), stage: 'assemble',
+      reason: error.reason, ...(fields.length ? { fields } : {}) });
+  }
+}
+
+export function rowRejections(stats: object): readonly RowRejection[] {
+  return rejected.get(stats) ?? [];
+}
+
+export function rowRejectionsTruncated(stats: object): boolean {
+  return truncated.has(stats);
+}
+
+export function writeFailureReason(error: unknown): string {
+  const code = typeof error === 'object' && error !== null && 'code' in error
+    ? String(error.code) : '';
+  return /^P\d{4}$/.test(code) ? `写入失败(${code})` : '写入或解析失败';
+}

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

@@ -234,7 +234,8 @@ const EmrRecordContent = z
 const ImageRecordContent = z
   .object({
     image_external_id: z.string().min(1),
-    file_url: z.string().url().regex(/^https?:\/\//i).optional(),
+    /// 原样留存宿主文件位置，不在摄入层要求完整链接。
+    file_url: z.string().optional(),
     file_name: z.string().optional(),
     file_type: z.string().optional(),
     encounter_external_id: nullableString(),

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

@@ -48,7 +48,7 @@ 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) } : {}),
+          ...(typeof c.fileUrl === 'string' ? { file_url: c.fileUrl } : {}),
           ...(c.fileName ? { file_name: String(c.fileName) } : {}),
           ...(c.fileType ? { file_type: String(c.fileType) } : {}),
         },

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

@@ -12,7 +12,7 @@ import type { EmitsConfig } from '../assembler/assembler.schema';
  * **v2.1 关键变更**(canonical-fact-layer.md §五):
  *   - 删除 resource → action / subjectType 硬映射(原 actionForResource / subjectTypeForResource)
  *   - 改读 yaml 显式声明的 `emits` 三件套(action / subjectType / occurredAtField)
- *   - 写 `rawPayload`(host 原文)—— 唯一留底,reparse 的输入
+ *   - 写 `rawPayload`(单条 host 原始源行)—— push/pull 共用的 reparse 输入
  *   - action / subjectType 仅做枚举合法性校验,值由 yaml 决定
  *
  * 三个上游入口共用:

apps/pac-service/src/modules/sync/push/ingest-evidence.service.ts

+import { Injectable, Logger } from '@nestjs/common';
+import { Cron } from '@nestjs/schedule';
+import { createHash, randomUUID } from 'node:crypto';
+import { Prisma } from '@prisma/client';
+import { PrismaService } from '../../../prisma/prisma.service';
+import { schedulerDisabled } from '../../../queues/scheduler-switch';
+
+export function evidenceRetentionDays(name: string, fallback: number): number {
+  const value = process.env[name];
+  if (value === undefined || value === '') return fallback;
+  const days = Number(value);
+  if (!Number.isSafeInteger(days) || days < 1 || days > 3650) {
+    throw new Error(`${name} 必须是 1–3650 的整数`);
+  }
+  return days;
+}
+
+@Injectable()
+export class IngestEvidenceService {
+  private readonly logger = new Logger(IngestEvidenceService.name);
+  constructor(private readonly prisma: PrismaService) {}
+
+  /** 已验签的原生行请求，处理前必须可靠保存；正文不含鉴权 Headers。 */
+  async capture(input: { hostId: string; source: string; rawBody: string; rowCount: number }): Promise<string> {
+    const now = new Date();
+    const successDays = evidenceRetentionDays('PAC_INGEST_EVIDENCE_SUCCESS_DAYS', 7);
+    const failureDays = Math.max(successDays,
+      evidenceRetentionDays('PAC_INGEST_EVIDENCE_FAILURE_DAYS', 30));
+    const expiresAt = new Date(now.getTime() + successDays * 86400000);
+    const failureExpiresAt = new Date(now.getTime() + failureDays * 86400000);
+    const hash = createHash('sha256').update(input.rawBody, 'utf8').digest('hex');
+    // 原子合并重推计数；不用“先查后写”，并发重推也只保留一份原文。
+    const rows = await this.prisma.$queryRaw<Array<{ id: string }>>(Prisma.sql`
+      INSERT INTO ingest_evidence
+        (id,host_id,source,payload_hash,raw_body,row_count,received_count,
+         first_received_at,last_received_at,has_failures,expires_at)
+      VALUES (${randomUUID()}::uuid,${input.hostId}::uuid,${input.source},${hash},
+        ${input.rawBody},${input.rowCount},1,${now},${now},false,${expiresAt})
+      ON CONFLICT (host_id,source,payload_hash) DO UPDATE SET
+        received_count=ingest_evidence.received_count+1,
+        last_received_at=GREATEST(ingest_evidence.last_received_at,EXCLUDED.last_received_at),
+        expires_at=GREATEST(ingest_evidence.expires_at,EXCLUDED.expires_at,
+          CASE WHEN ingest_evidence.has_failures THEN ${failureExpiresAt} ELSE EXCLUDED.expires_at END)
+      RETURNING id`);
+    return rows[0]!.id;
+  }
+
+  async markFailed(hostId: string, id: string): Promise<void> {
+    const days = Math.max(evidenceRetentionDays('PAC_INGEST_EVIDENCE_SUCCESS_DAYS', 7),
+      evidenceRetentionDays('PAC_INGEST_EVIDENCE_FAILURE_DAYS', 30));
+    const expiresAt = new Date(Date.now() + days * 86400000);
+    await this.prisma.$executeRaw(Prisma.sql`
+      UPDATE ingest_evidence SET has_failures=true,expires_at=GREATEST(expires_at,${expiresAt})
+      WHERE id=${id}::uuid AND host_id=${hostId}::uuid`);
+  }
+
+  /** 分批清理；删除正文后同步记录、失败定位仍保留，引用置空。 */
+  async pruneExpired(limit = 1000): Promise<number> {
+    if (!Number.isSafeInteger(limit) || limit < 1 || limit > 1000) throw new Error('清理批量需为 1–1000');
+    return this.prisma.$executeRaw(Prisma.sql`
+      DELETE FROM ingest_evidence WHERE id IN (
+        SELECT id FROM ingest_evidence WHERE expires_at < now()
+        ORDER BY expires_at LIMIT ${limit} FOR UPDATE SKIP LOCKED)`);
+  }
+
+  @Cron('30 4 * * *', { name: 'ingest-evidence-retention', timeZone: 'Asia/Shanghai' })
+  async prune(): Promise<void> {
+    if (schedulerDisabled()) return;
+    let count = 0;
+    for (let i = 0; i < 10; i++) {
+      const deleted = await this.pruneExpired();
+      count += deleted;
+      if (deleted < 1000) break;
+    }
+    if (count) this.logger.log(`已清理 ${count} 份到期入站证据`);
+  }
+}

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

@@ -13,6 +13,7 @@ import { ColdImportService } from '../cold-import/cold-import.service';
 import { QueueProducer } from '../../../queues/queue-producer.service';
 import type { EmitsConfig } from '../assembler/assembler.schema';
 import type { PushEvent, PushEventsResponse, PushRowsResponse } from './push.schema';
+import { IngestEvidenceService } from './ingest-evidence.service';
 
 /**
  * PushReceiver — 处理已验签的 push 事件批。
@@ -36,6 +37,7 @@ export class PushReceiverService {
     private readonly dispatcher: PipelineDispatcher,
     private readonly coldImport: ColdImportService,
     private readonly queue: QueueProducer,
+    private readonly evidence: IngestEvidenceService,
   ) {}
 
   /**
@@ -49,17 +51,23 @@ export class PushReceiverService {
     hostName: string;
     source: string;
     rows: Record<string, unknown>[];
+    rawBody?: string; // 控制器传入验签原文，不重新序列化
     /// 预检(?dryRun=1):不写业务数据、不派生事实、不触发重算;仍保留预检推送记录。
     dryRun?: boolean;
   }): Promise<PushRowsResponse> {
     const { hostId, hostName, source, rows } = input;
     const dryRun = !!input.dryRun;
 
+    const evidenceId = await this.evidence.capture({ hostId, source, rowCount: rows.length,
+      rawBody: input.rawBody ?? JSON.stringify({ source, rows }) });
+    let hasFailures = false;
+    try {
       const r = await this.coldImport.ingestRawTables({
         hostName,
         source,
         rows,
         dryRun,
+        ingestEvidenceId: evidenceId,
         triggeredBy: `${dryRun ? 'push-dryrun' : 'push'}:${hostName}:${source}`,
       });
 
@@ -75,6 +83,8 @@ export class PushReceiverService {
         { upserted: 0, txn: 0, dup: 0, failed: 0, factsFailed: 0, stubs: 0 },
       );
 
+      hasFailures = agg.failed > 0 || agg.factsFailed > 0 || r.hasFailures;
+
       // fact.content 被 schema 拒收的明细 —— 必须透出去,否则宿主只能看到"成功但没进数据"。
       // 2026-07-29 事故:4 行负金额结算被拒,回执却是 success/failed=0,对方无从判断,
       // 只看到超时 → 重推同一批 13 小时。定位信息(哪条 subject、哪个字段)不含 content 原文。
@@ -155,6 +165,12 @@ export class PushReceiverService {
         dryRun,
         samples,
       };
+    } catch (error) {
+      hasFailures = true;
+      throw error;
+    } finally {
+      if (hasFailures) await this.evidence.markFailed(hostId, evidenceId);
+    }
   }
 
   async receive(input: {

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

@@ -141,10 +141,10 @@ export class PushController {
   @Public()
   @HttpCode(200)
   @ApiOperation({
-    summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签;?dryRun=1 预检不落库)',
+    summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签;?dryRun=1 预检不写业务数据)',
     description:
       'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。Body: { source, rows[] }。' +
-      ' 加 ?dryRun=1 走**预检**:跑完整管线(transforms+装配+归一)但不落库、不触发重算,' +
+      ' 加 ?dryRun=1 走**预检**:跑 transforms+装配+归一，不写业务数据、不派生事实、不触发重算；保留预检记录及请求原文,' +
       '响应带各资源样本 canonical —— 对接自测先预检到 failed=0 且 mappingMisses/suspectFields 为 0,再正式推。',
   })
   @ApiQuery({
@@ -180,6 +180,7 @@ export class PushController {
         hostName,
         source: parsed.source,
         rows: parsed.rows,
+        rawBody,
         dryRun,
       }),
     );

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

@@ -118,7 +118,7 @@ export const PushRowsResponseSchema = z.object({
   patientStubsCreated: z.number().int().describe('新建空壳患者主档数(持续 >0 = 患者号对不上)'),
   /// 宿主自检信号:样本批推完看这两个数,>0 先停下联系 PAC。
   mappingMisses: z.number().int().describe('映射覆盖缺口种类数(有原值落 _default)'),
-  suspectFields: z.number().int().describe('疑似宿主改表项数(列缺席/新增,vs 历史基线)'),
+  suspectFields: z.number().int().describe('疑似宿主改表项数(必填列整批缺席，或历史原文列增减)'),
   /// 预检模式(?dryRun=1):跑 transforms+装配+归一+tenant 解析,不写业务数据、不派生事实,
   /// 仍保留预检推送记录。返回预计数,正式推送仍须检查失败计数。
   dryRun: z.boolean().describe('true = 预检,未写业务数据,仍保留预检推送记录'),

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

@@ -17,6 +17,7 @@ import { ReconcileOrchestrator } from './reconcile/reconcile.orchestrator';
 import { PushController } from './push/push.controller';
 import { PushReceiverService } from './push/push-receiver.service';
 import { HmacVerifier } from './push/hmac-verifier.service';
+import { IngestEvidenceService } from './push/ingest-evidence.service';
 
 /**
  * SyncModule — 数据接入主模块。
@@ -55,6 +56,7 @@ import { AdminModule } from '../admin/admin.module';
     PullOrchestrator,
     ReconcileOrchestrator,
     PushReceiverService,
+    IngestEvidenceService,
     HmacVerifier,
   ],
   exports: [

apps/pac-service/tests/assembler-column-drift.spec.ts

+import { Logger } from '@nestjs/common';
+import { AssemblerEngine } from '../src/modules/sync/assembler/assembler-engine';
+import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.schema';
+import { ColdImportService } from '../src/modules/sync/cold-import/cold-import.service';
+
+Logger.overrideLogger(false);
+
+const config = AssemblerConfigSchema.parse({ canonical: 'image', primary: { table: 'native', key: 'id' },
+  field_mapping: { externalId: 'id', patientExternalId: 'person', clinicId: 'center', uploadedAt: 'time',
+    fileUrl: 'new_optional_path', modality: 'type', extraHostField: 'unmapped_optional' } });
+const good = { id: 'i1', person: 'p1', center: 'c1', time: '2026-10-13T01:00:00Z' };
+const assemble = (row: Record<string, unknown>) => new AssemblerEngine().assemble({ config,
+  tables: { native: [row] }, normalize: { timezone: 'UTC', amountUnit: 'fen' }, hostId: 'other-host' });
+
+test('另一宿主的新增可选列 / passthrough 缺席不误判改表', () => {
+  const result = assemble(good);
+  expect(result.stats.failed).toBe(0);
+  expect(result.stats.suspectFields).toEqual([]);
+});
+
+test('必填原列整批缺席仍告警，且不接收不完整身份', () => {
+  const row: Record<string, unknown> = { ...good };
+  delete row.person;
+  const result = assemble(row);
+  expect(result.stats.failed).toBe(1);
+  expect(result.stats.suspectFields).toEqual([
+    { resource: 'image', field: 'patientExternalId', hostField: 'person', total: 1, reason: 'column_absent' },
+  ]);
+});
+
+test('已提供的可选字段类型错误仍拒收，不能把可选误当不校验', () => {
+  expect(assemble({ ...good, new_optional_path: 123 }).stats.failed).toBe(1);
+});
+
+test('历史原文已存在的列消失仍能检测，不关闭历史漂移告警', async () => {
+  const baseline = { ...good, new_optional_path: 'old/path.jpg' };
+  const db = { patientTransaction: { findMany: jest.fn().mockResolvedValue(
+    Array.from({ length: 5 }, () => ({ rawPayload: baseline }))) } };
+  const cold = new ColdImportService(db as never, {} as never, {} as never, {} as never,
+    {} as never, {} as never, {} as never);
+  const detector = cold as unknown as { detectRawColumnDrift(args: {
+    hostId: string; source: string; subjectTypes: string[]; pushedCols: Set<string>;
+    rowCount: number; excludeSyncLogId: string;
+  }): Promise<unknown[]> };
+  expect(await detector.detectRawColumnDrift({ hostId: 'other-host', source: 'native',
+    subjectTypes: ['image'], pushedCols: new Set(Object.keys(good)), rowCount: 200,
+    excludeSyncLogId: 'this-batch' })).toEqual([{ resource: 'native', field: 'new_optional_path',
+      hostField: 'new_optional_path', total: 200, reason: 'column_removed' }]);
+});

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

@@ -119,17 +119,45 @@ describe('image analysis shared ingestion', () => {
     expect(diagnosis.occurredAt).toEqual(friday.occurredAt);
   });
 
-  test('地址保存完整签名；旧相对路径报文继续接受；伪完整地址拒绝', () => {
+  test.each(['old/path.jpg', 'https://files.example.com/path.jpg?signature=token&expires=123',
+    'oss://bucket/影像 图.jpg', ''])('影像文件位置 %p 从宿主 file_url 原样保留', file_url => {
     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' };
+      shooting_time: '2026-10-12 10:00:00', class_name: '全景片', file_url,
+      fileUrlEcho: 'https://ignored.example.com/echo.jpg', file_name: 'path.jpg' };
     const result = assembleFriday([raw], 'med_check');
+    expect(result.stats.failed).toBe(0);
+    expect(result.stats.suspectFields).toEqual([]);
+    expect(result.rows[0]!.rawSource).toEqual(raw);
     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);
-    }
+    const fact = new ImageParser().parse(ctx)[0]!;
+    expect(fact.content).toMatchObject({ file_url });
+    expect(() => validateFactContent(fact.type, fact.subjectId, fact.content)).not.toThrow();
+  });
+
+  test('复现告警批次：200 条相对路径影像无 fileUrlEcho，全接收且不报缺列', () => {
+    const rows = Array.from({ length: 200 }, (_, i) => ({ id: String(i), tenant_id: 'brand-a',
+      organization_id: 'clinic-a', patient_id: '4001', shooting_time: '2026-10-12 10:00:00',
+      class_name: '全景片', file_url: `old/${i}.jpg` }));
+    const result = assembleFriday(rows, 'med_check');
+    expect(result.rows).toHaveLength(200);
+    expect(result.stats.failed).toBe(0);
+    expect(result.stats.suspectFields).toEqual([]);
+    result.rows.forEach((row, i) => {
+      const fact = new ImageParser().parse(parserContext(row.canonical as Record<string, unknown>))[0]!;
+      expect(fact.content).toMatchObject({ file_url: rows[i]!.file_url });
+      expect(() => validateFactContent(fact.type, fact.subjectId, fact.content)).not.toThrow();
+    });
+  });
+
+  test.each([undefined, null])('影像可选文件位置 %p 不拒收也不报缺列', file_url => {
+    const row: Record<string, unknown> = { id: '3001', tenant_id: 'brand-a', organization_id: 'clinic-a', patient_id: '4001',
+      shooting_time: '2026-10-12 10:00:00', file_url };
+    if (file_url === undefined) delete row.file_url;
+    const result = assembleFriday([row], 'med_check');
+    expect(result.stats.failed).toBe(0);
+    expect(result.stats.suspectFields).toEqual([]);
+    expect(new ImageParser().parse(parserContext(result.rows[0]!.canonical as Record<string, unknown>))[0]!.content)
+      .not.toHaveProperty('file_url');
   });
 
   test('内容版本区分仅明细/链接变更，稳定对象键顺序；原时间键模式不变', () => {

apps/pac-service/tests/ingest-evidence-postgres.spec.ts

+import { PrismaClient } from '@prisma/client';
+import { IngestEvidenceService } from '../src/modules/sync/push/ingest-evidence.service';
+import type { PrismaService } from '../src/prisma/prisma.service';
+import { readEvidence, readEvidenceById } from '../src/cli/ingest-evidence.cli';
+
+const url = process.env.PAC_EVIDENCE_TEST_DATABASE_URL;
+const describePg = url ? describe : describe.skip;
+
+describePg('ingest evidence | isolated PostgreSQL', () => {
+  let db: PrismaClient;
+  let service: IngestEvidenceService;
+  let hostId: string;
+  let otherHostId: string;
+  const source = 'customer_referee_circle';
+  const body = '{ "source":"customer_referee_circle", "rows":[{"id":9007199254740993123,"customer_id":null},{"id":"b"}] }';
+
+  beforeAll(async () => {
+    const parsed = new URL(url!);
+    if (!['localhost', '127.0.0.1'].includes(parsed.hostname) || !parsed.pathname.startsWith('/pac_evidence_ci_')) {
+      throw new Error('requires isolated localhost pac_evidence_ci_ database');
+    }
+    db = new PrismaClient({ datasourceUrl: url });
+    service = new IngestEvidenceService(db as unknown as PrismaService);
+    hostId = (await db.host.create({ data: { name: 'evidence-a', appId: 'evidence-a', appSecretHash: 'test-only' } })).id;
+    otherHostId = (await db.host.create({ data: { name: 'evidence-b', appId: 'evidence-b', appSecretHash: 'test-only' } })).id;
+  });
+  beforeEach(async () => { await db.syncLog.deleteMany(); await db.ingestEvidence.deleteMany(); });
+  afterAll(async () => { await db?.$disconnect(); });
+
+  const capture = (rawBody = body, host = hostId, src = source) =>
+    service.capture({ hostId: host, source: src, rawBody, rowCount: 2 });
+
+  test('保存原始正文，保留大整数、字段缺席、null 和空白；不存配置快照', async () => {
+    const id = await capture();
+    const row = await db.ingestEvidence.findUniqueOrThrow({ where: { id } });
+    expect(row.rawBody).toBe(body);
+    expect(row.rawBody).toContain('9007199254740993123');
+    expect(row.rowCount).toBe(2);
+    expect(row.hasFailures).toBe(false);
+    expect(row.expiresAt.getTime() - row.lastReceivedAt.getTime()).toBe(7 * 86400000);
+    expect(Object.keys(row)).not.toContain('timezone');
+  });
+
+  test('并发同一报文只保留一份，接收次数原子累加', async () => {
+    const ids = await Promise.all(Array.from({ length: 12 }, () => capture()));
+    expect(new Set(ids).size).toBe(1);
+    expect(await db.ingestEvidence.count()).toBe(1);
+    expect((await db.ingestEvidence.findUniqueOrThrow({ where: { id: ids[0] } })).receivedCount).toBe(12);
+  });
+
+  test('内容变化、宿主变化、source 变化分别留证据', async () => {
+    await capture(); await capture(body+' '); await capture(body, otherHostId); await capture(body, hostId, 'other');
+    expect(await db.ingestEvidence.count()).toBe(4);
+  });
+
+  test('失败延长保留期，跨宿主不能标记；重推不降回成功保留期', async () => {
+    const id = await capture();
+    await service.markFailed(otherHostId, id);
+    expect((await db.ingestEvidence.findUniqueOrThrow({ where: { id } })).hasFailures).toBe(false);
+    await service.markFailed(hostId, id);
+    const failed = await db.ingestEvidence.findUniqueOrThrow({ where: { id } });
+    expect(failed.hasFailures).toBe(true);
+    expect(failed.expiresAt.getTime() - Date.now()).toBeGreaterThan(29 * 86400000);
+    await capture();
+    expect((await db.ingestEvidence.findUniqueOrThrow({ where: { id } })).expiresAt.getTime())
+      .toBeGreaterThanOrEqual(failed.expiresAt.getTime());
+  });
+
+  test('重复批次关联多次同步；按宿主导出原文，不串其它宿主', async () => {
+    const id = await capture();
+    const logs = await Promise.all([0, 1].map(() => db.syncLog.create({ data: {
+      hostId, tenantId: 't', resource: source, direction: 'push', status: 'partial',
+      ingestEvidenceId: id, metadata: { rowFailures: [{ rowIndex: 0, stage: 'assemble' }] },
+    } })));
+    expect((await readEvidence(db, 'evidence-a', logs[0]!.id)).evidence.rawBody).toBe(body);
+    await expect(readEvidence(db, 'evidence-b', logs[0]!.id)).rejects.toThrow('该宿主没有此同步记录');
+    expect(await db.syncLog.count({ where: { ingestEvidenceId: id } })).toBe(2);
+  });
+
+  test('到期清理正文、保留同步日志与定位；未到期证据保留', async () => {
+    const oldId = await capture();
+    const newId = await capture(body+' ');
+    const log = await db.syncLog.create({ data: { hostId, tenantId: 't', resource: source,
+      direction: 'push', status: 'partial', ingestEvidenceId: oldId,
+      metadata: { rowFailures: [{ rowIndex: 0 }] } } });
+    await db.ingestEvidence.update({ where: { id: oldId }, data: { expiresAt: new Date(0) } });
+    expect(await service.pruneExpired()).toBe(1);
+    expect(await db.ingestEvidence.findUnique({ where: { id: newId } })).not.toBeNull();
+    const kept = await db.syncLog.findUniqueOrThrow({ where: { id: log.id } });
+    expect(kept.ingestEvidenceId).toBeNull();
+    expect(kept.metadata).toEqual({ rowFailures: [{ rowIndex: 0 }] });
+    await expect(readEvidence(db, 'evidence-a', log.id)).rejects.toThrow('原文已到期清理');
+  });
+
+  test('未生成同步日志的早期异常也可读取，仍限制宿主归属', async () => {
+    const id = await capture();
+    await service.markFailed(hostId, id);
+    expect((await readEvidenceById(db, 'evidence-a', id)).evidence.rawBody).toBe(body);
+    await expect(readEvidenceById(db, 'evidence-b', id)).rejects.toThrow('该宿主没有此证据');
+  });
+});

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

@@ -7,6 +7,7 @@ import { AssemblerConfigSchema } from '../src/modules/sync/assembler/assembler.s
 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';
+import { TransactionSynthesizer } from '../src/modules/sync/pipeline/transaction-synthesizer';
 
 Logger.overrideLogger(false);
 
@@ -33,7 +34,7 @@ function setup(resource: UpsertResource) {
   };
   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,
+  const cold = new ColdImportService(prisma as never, new TransactionSynthesizer(), {} as never,
     new AssemblerEngine(), {} as never, new TransformEngine(), alerter as never);
   const config = AssemblerConfigSchema.parse({ canonical: resource,
     primary: { table: 'native', key: 'id' },
@@ -57,7 +58,8 @@ function setup(resource: UpsertResource) {
   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 evidence = { capture: jest.fn().mockResolvedValue('evidence-demo'), markFailed: jest.fn().mockResolvedValue(undefined) };
+  const receiver = new PushReceiverService(prisma as never, {} as never, cold, queue as never, evidence 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 }
@@ -65,7 +67,7 @@ function setup(resource: UpsertResource) {
   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 };
+  return { prisma, cold, queue, alerter, good, bad, send, dependencies, evidence, receiver };
 }
 
 describe.each<UpsertResource>(['patient', 'patient_relation', 'patient_return_visit'])(
@@ -90,6 +92,8 @@ describe.each<UpsertResource>(['patient', 'patient_relation', 'patient_return_vi
       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);
+      expect(s.evidence.capture).toHaveBeenCalledTimes(1);
+      expect(s.evidence.markFailed).not.toHaveBeenCalled();
     });
 
     test('预检与正式批次分类一致，不执行业务写入', async () => {
@@ -129,6 +133,87 @@ test('复现生产：200 条关系中 197 条有效、3 条身份缺失，不再
   expect(s.prisma.syncLog.update.mock.calls[0]![0].data).toMatchObject({
     status: SyncStatus.PARTIAL, fetched: 200, failed: 3, errorMessage: null,
   });
+  expect(s.prisma.syncLog.update.mock.calls[0]![0].data.metadata.rowFailures).toMatchObject([
+    { rowIndex: 197, fields: ['relatedExternalId'] },
+    { rowIndex: 198, fields: ['relatedExternalId'] },
+    { rowIndex: 199, fields: ['patientExternalId'] },
+  ]);
+  expect(s.evidence.markFailed).toHaveBeenCalledWith('host-demo', 'evidence-demo');
+});
+
+test('定位样本截断仍明确标记，完整原生请求不截断', async () => {
+  const s = setup('patient_relation');
+  // 内部摄入调用验证保护上限；HTTP 层仍按既有契约限制单批行数。
+  const rows = Array.from({ length: 501 }, () => ({ ...s.bad }));
+  await expect(s.send(rows)).rejects.toMatchObject({ code: ApiCode.SYNC_CONTRACT_DRIFT });
+  const metadata = s.prisma.syncLog.update.mock.calls[0]![0].data.metadata;
+  expect(metadata.rowFailures).toHaveLength(500);
+  expect(metadata.rowFailuresTruncated).toBe(true);
+  expect(JSON.parse(s.evidence.capture.mock.calls[0]![0].rawBody).rows).toHaveLength(501);
+});
+
+test('push 留请求证据后仍走通用流水合成，rawPayload 保留单条宿主原始行', async () => {
+  const s = setup('patient');
+  const raw = { id: 'a1', customer_id: 'p1', clinic_id: 'c1', tenant_id: 'brand-demo',
+    scheduled_at: '2026-10-13 09:00:00', status: 'scheduled', private_note: 'source-only', nested: { field: null } };
+  const config = AssemblerConfigSchema.parse({ canonical: 'appointment',
+    primary: { table: 'appointment_rows', key: 'id' },
+    field_mapping: { externalId: 'id', patientExternalId: 'customer_id', clinicId: 'clinic_id',
+      scheduledAt: 'scheduled_at', status: 'status' },
+    emits: { action: 'appointment_created', subjectType: 'appointment', occurredAtField: 'scheduledAt' },
+  });
+  jest.spyOn(s.dependencies, 'loadAllAssemblers').mockReturnValue([config]);
+  jest.spyOn(s.dependencies, 'readManifest').mockReturnValue({
+    tenant_id: 'friday-market', identity_namespace_field: 'tenant_id',
+    amount_unit: 'yuan', timezone: 'Asia/Shanghai', transforms: [{ kind: 'project',
+      input: 'native', output: 'appointment_rows', keep: ['id', 'customer_id', 'clinic_id', 'scheduled_at', 'status'] }],
+  } as never);
+  const written: Array<{ rawPayload: unknown }> = [];
+  const writes = s.cold as unknown as { flushBatchedWrite(input: {
+    buffer: Array<{ tx: { rawPayload: unknown } }>; stats: { transactionsWritten: number }
+  }): Promise<void> };
+  jest.spyOn(writes, 'flushBatchedWrite').mockImplementation(async input => {
+    written.push(...input.buffer.map(row => row.tx));
+    input.stats.transactionsWritten += input.buffer.length;
+  });
+  expect(await s.send([raw])).toMatchObject({ transactionsWritten: 1, failed: 0 });
+  expect(written).toHaveLength(1);
+  expect(written[0]!.rawPayload).toEqual(raw);
+  expect(written[0]!.rawPayload).not.toHaveProperty('__source_row');
+  expect(written[0]!.rawPayload).not.toHaveProperty('rows');
+  expect(raw.scheduled_at).toBe('2026-10-13 09:00:00');
+  expect(s.evidence.capture).toHaveBeenCalledTimes(1);
+});
+
+test('证据保存失败时不执行业务摄入，也不确认接收成功', async () => {
+  const s = setup('patient_relation');
+  s.evidence.capture.mockRejectedValue(new Error('evidence store unavailable'));
+  await expect(s.send([s.good])).rejects.toThrow('evidence store unavailable');
+  expect(s.prisma.syncLog.create).not.toHaveBeenCalled();
+  expect(s.prisma.patientRelation.upsert).not.toHaveBeenCalled();
+});
+
+test('不改签名原报文，缺席与 null 可区分；原文不进入错误 metadata 或正常回执', async () => {
+  const s = setup('patient_relation');
+  const missing = { ...s.good, private_note: 'private-source-evidence' };
+  delete (missing as Partial<typeof missing>).referee_patient_id;
+  const rows = [missing, { ...s.good, referee_patient_id: null }];
+  const body = '{ "source": "native", "rows": '+JSON.stringify(rows)+' }';
+  await expect(s.receiver.receiveRows({ hostId: 'host-demo', hostName: 'friday',
+    source: 'native', rows, rawBody: body })).rejects.toMatchObject({ code: ApiCode.SYNC_CONTRACT_DRIFT });
+  expect(s.evidence.capture.mock.calls[0]![0].rawBody).toBe(body);
+  const metadata = s.prisma.syncLog.update.mock.calls[0]![0].data.metadata;
+  expect(metadata.rowFailures.map((r: { rowIndex: number }) => r.rowIndex)).toEqual([0, 1]);
+  expect(JSON.stringify(metadata)).not.toContain('private-source-evidence');
+  expect(s.prisma.syncLog.create.mock.calls[0]![0].data.ingestEvidenceId).toBe('evidence-demo');
+});
+
+test('重复对象与宿主同名内部字段也不伪造源行位置', async () => {
+  const s = setup('patient_relation');
+  const row = { ...s.bad, __source_row: { customer_id: 'forged', referee_patient_id: 'forged' } };
+  await expect(s.send([row, row])).rejects.toMatchObject({ code: ApiCode.SYNC_CONTRACT_DRIFT });
+  expect(s.prisma.syncLog.update.mock.calls[0]![0].data.metadata.rowFailures
+    .map((r: { rowIndex: number }) => r.rowIndex)).toEqual([0, 1]);
 });
 
 test.each([

packages/types/src/schemas/canonical.ts

@@ -352,7 +352,8 @@ 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(),
+    /// 宿主文件位置原值，可为相对路径；展示时再决定如何解析。
+    fileUrl: z.string().optional().nullable(),
     fileName: z.string().optional().nullable(),
     fileType: z.string().optional().nullable(),
   })