merge: chore/drop-canonical-payload → test(删 patient_transactions.canonical_payload)

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

@@ -58,7 +58,7 @@ flowchart LR
 |---|---|---|
 | **① transforms** =「**形态改造**」*(manifest,per-host)* | 拆行 / 派生 / 关键词分流 / 多列推断 / 行过滤 | **6 个白名单算子**(`project` / `split_json_array` / `derive` / `route_by_pattern` / `pick_first_nonzero` / `filter`),无副作用纯函数,**不允许任意 JS / eval** |
 | **② assembler** =「**词汇翻译**」*(manifest,per-host)* | 字段名翻译(宿主 → canonical)+ 枚举码翻译(闭集字典 + `_default` 兜底)+ `emits` 推断 | manifest + yaml 驱动;**名字虽叫 assembler,本质是翻译器** |
-| **③ 事务合成** | 把 canonical 行写成 `patient_transactions`(append-only 账本) | `raw_payload` + `canonical_payload` 双留底;`source_event_id` 幂等键;`event_seq` 单调水位 |
+| **③ 事务合成** | 把 canonical 行写成 `patient_transactions`(append-only 账本) | `raw_payload` 留底(宿主原文,reparse 的输入);`source_event_id` 幂等键;`event_seq` 单调水位 |
 | **④ parser + Zod** =「**类型 + 语义 + 校验**」 | 从 transaction 把扁平行**构造成带类型的 `fact.content`**:单位 / 牙位 / 文本归一、code 纠偏(如 K00 误标→K08)、`FactContentSchema` 强校验 | **唯一会抛错拦脏数据的强校验闸**,字段漂移即拒 |
 | **⑤ fact 落地** | 写 `patient_facts`(版本流,supersede 旧版) | 唯一写入口 `FactWriter`,evidence 反指 transaction |
 

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

@@ -110,7 +110,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`(宿主原文)+ `canonical_payload`(assembler 翻译后中间态,审计/replay)+ `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`(宿主原文,唯一留底 + 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。
 

apps/pac-service/prisma/migrations/20260818210000_drop_transaction_canonical_payload/migration.sql

+-- 删除 patient_transactions.canonical_payload
+--
+-- 【为什么】它存 AssemblerEngine yaml 翻译后的中间态,设想用途是「审计回查」+「yaml 改后局部 replay」。
+--   2026-08-18 全仓库核查(含 pac-web / 原生 SQL / CLI / 测试 / test 分支):**1 处写、0 处读** ——
+--   两个设想用途都没有落地成代码,reparse 走的是 raw_payload 从头重跑。
+--   实测占用:测试服 patient_transactions 42 GB 中约 7.9 GB(按 subject_type 抽样外推)。
+--
+-- 【会不会丢证据】不会。canonical 是 raw_payload 经 field-mapping 推导出来的,
+--   需要时跑一次 reparse 即得 —— 那正是 reparse 做的事。真原始证据 raw_payload 原样保留。
+--
+-- 【执行成本】PostgreSQL 的 DROP COLUMN 只改系统目录、不重写表,42 GB 表上也是秒级、不锁长时间。
+--   ⚠️ 但空间不会立刻还给操作系统:已有行里那部分空间转为表内可复用空间,由后续写入吸收
+--   (效果 = 这张表在磁盘上停止增长一段时间)。要真正缩表需 VACUUM FULL / pg_repack,另行安排。
+ALTER TABLE "patient_transactions" DROP COLUMN "canonical_payload";

apps/pac-service/prisma/schema.prisma

@@ -548,11 +548,13 @@ model PatientTransaction {
   /// 宿主原生 100% 原文(adapter 收到啥存啥)。
   /// 唯一事件原始证据;canonical 视图由 parser 落到 patient_facts.content,
   /// before/after diff 由 fact 版本流(subject_id × version)推导
+  ///
+  /// 【为什么没有 canonical 留底】曾有 `canonical_payload` 列存 AssemblerEngine yaml 翻译后的中间态
+  ///   (设想用途:审计回查 + yaml 改后局部 replay)。2026-08-18 全仓库核查:**1 处写、0 处读** —— 两个
+  ///   设想用途都没有落地成代码,而 reparse 走的是本列(rawPayload)从头重跑。实测它占 patient_transactions
+  ///   约 7.9 GB(测试服),故删除。真需要 canonical 时从 rawPayload 跑一遍 field-mapping 即得 —— 那正是
+  ///   reparse 做的事,所以它是可推导的冗余,不是丢失的证据。
   rawPayload       Json   @map("raw_payload")
-  /// canonical row 留底(AssemblerEngine yaml 翻译后的中间产物,转 parser 用)。
-  /// 用途:① 审计 — fact 算错时回查中间态;② replay — yaml 改后局部重跑 parser 不必从 raw 重头算
-  /// nullable 兼容老数据(初期未实施 canonical 留底);新写入必须填
-  canonicalPayload Json?  @map("canonical_payload")
   /// payload 指纹(sha256 of rawRow);完整性 / 兜底校验用的独立字段。
   /// 注:source_event_id 由 TransactionSynthesizer 用 `来源:资源:source_unit:subjectId:updatedAt` 合成,
   ///     **不含本字段**(早期注释说"也作为 source_event_id 输入"已与代码不符,已更正)。

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

@@ -126,7 +126,7 @@ export class PipelineDispatcher {
         // ⭐ 提取真原文 link(W3 末加,合规留底):
         // - push: __source_row = event.payload(host event 原报文)
         // - mock pull: __source_row 未注入 → fallback 用 row 自身
-        // 提取后从 canonical 里 strip,避免 __source_row 流入 transaction.canonical_payload
+        // 提取后从 canonical 里 strip —— __source_row 是传输期的内部载荷,不该流进下游 parser
         const { __source_row: explicitSource, ...canonicalRow } = rawCanonicalRow as {
           __source_row?: Record<string, unknown>;
         } & Record<string, unknown>;

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)
- *   - 写 `canonicalPayload`(field-mapped 后的 row,审计 / replay 用)+ `rawPayload`(host 原文)
+ *   - 写 `rawPayload`(host 原文)—— 唯一留底,reparse 的输入
  *   - action / subjectType 仅做枚举合法性校验,值由 yaml 决定
  *
  * 三个上游入口共用:
@@ -28,7 +28,7 @@ export class TransactionSynthesizer {
    * 把 canonical row + emits 合成为 patient_transactions 写入对象。
    *
    * @param input.rawRow         host 原生行(rawPayload 用)
-   * @param input.canonicalRow   field-mapped 后的行(canonicalPayload + 派生 transaction 字段用)
+   * @param input.canonicalRow   field-mapped 后的行(派生 transaction 立柱字段用;**不再留底**,见 schema 注释)
    * @param input.emits          yaml 显式声明的 action/subjectType/occurredAtField(必须)
    * @param input.resource       canonical resource key(日志 + sourceEventId 用)
    * @param input.hostId         host UUID
@@ -134,7 +134,6 @@ export class TransactionSynthesizer {
       subjectType,
       subjectId,
       rawPayload: input.rawRow as Prisma.InputJsonValue,
-      canonicalPayload: input.canonicalRow as Prisma.InputJsonValue,
       payloadHash,
       occurredAt,
     };