Day 23 完成了生命徵象硬門控（Vital-Sign Hard Gate）：作者合成警示一旦觸發，就把較不急迫候選直接排除。六筆合成案例中，這個做法修正了三筆原本過輕的 Top-1，卻也在一筆邊界案例中永久刪掉合成參考候選。

今天換一個方向。安全聯集門控（Safety Union Gate）不把「安全」定義成更積極地刪除，而是先把一般檢索、主訴、生命徵象與固定高風險保留候選合併，再依來源加權排序。這個方法至少保證：原本的基礎候選仍在集合內，後續排序器有機會重新考慮它。

不過，保留候選不等於 Top-1 一定正確，更不等於臨床安全。本篇八筆案例、主訴路由、生命徵象門檻、候選分數、加權值與合成參考級數都是作者為工程測試建立的合成資料。它們不是韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS）的正式規則，也不是真實病患、護理師或專家標籤。

下圖用概念示意呈現四條候選來源進入同一個聯集；重複候選合併，但 Base 候選不會因聯集而消失。圖中沒有正式醫療規則或病患資訊。

上圖右側仍由臨床人員判讀，因為本系列建立的是研究與教學用決策支援元件，不是自動診斷或取代正式檢傷流程的系統。

本篇目標

完成本篇後，你會得到五個可驗證產物：

1. 一份鎖定四種門控方法共同條件的公平比較契約。
2. 一個實作 U = B ∪ C ∪ V ∪ H、去重與 Base 保留不變量的 Safety Union Gate。
3. 六筆完整沿用 Day 23 輸入、分數與合成參考級數的回歸案例。
4. 兩筆刻意暴露固定保留集合與錯誤主訴路由代價的新案例。
5. 一份包含候選集合、來源加權、Top-1 方向與候選保留狀態的公開追蹤紀錄（Trace）。

本篇不執行大型語言模型（Large Language Model, LLM）生成，也不把八筆結果報成準確率。今天只回答一個範圍更小的問題：固定相同候選分數與 Base Top-2 後，四種門控動作如何改變候選集合與合成 Top-1？

本篇會用到的名詞

候選不是最後判定。它只是仍能進入排序與後續推理的知識單元。Top-1 則是某一條方法排序後的第一名；本篇把它和作者指定的合成參考級數比較，不能直接解讀成病患檢傷結果。

為什麼生命徵象不是唯一入口

KTAS 官方介紹把 KTAS 描述為症狀導向的檢傷工具：先依第一印象判斷，再從病患主訴出發，套用共同主要考量與主訴特定次要考量。這支持「主訴路徑不能被生命徵象路徑取代」，但官方簡介不足以重建完整數值規則與主訴目錄。KTAS 官方介紹與五級分類標準

Moon 等人的公開研究也把生命徵象列為主要考量之一，並同時列出疼痛分數、出血相關狀況與受傷機轉等其他主要考量；血糖與脫水程度則是次要考量例子。這支持「生命徵象只是整體演算法的一部分」，不能支持本篇作者合成路由就是正式 KTAS 實作。Moon 等人：Triage accuracy and causes of mistriage using the Korean Triage and Acuity Scale

因此，Day 24 的設計不是宣稱已經重建 KTAS，而是測試一項一般性的知識工程問題：當不同來源都可能帶來重要候選時，能否先保留證據，再把來源可信度交給排序階段處理？

先分清楚五種資料角色

本篇刻意不把所有東西都叫做「資料」：

六筆 Day 23 案例會完整沿用原來的 inputs、candidate_scores、synthetic_reference_level 與情境文字。Day 24 只替它們加入作者合成主訴路由，再新增兩筆代價案例。這樣可以確認結果差異來自新的候選集合與排序規則，不是偷偷改寫舊案例。

四個集合各自負責什麼

B：共同的 Base Top-2

先把五筆公開級數定義依作者合成分數排序，取前兩名形成 B。令五級候選全集為 D = {1, 2, 3, 4, 5}，共同分數為 s(d)，則：

B = Top-2(D, s)

四種方法都從同一個 B 與同一組 s(d) 開始。這是公平比較的基礎。Day 23 的原始程式會先保有五級候選再執行 Hard Gate；Day 24 為了比較「只刪除、只加權與聯集新增」三種動作，明確把共同起點鎖成 Base Top-2。六筆回歸案例的 Top-1 仍和 Day 23 公開結果一致。

C：作者合成的主訴候選

C 由合成主訴路由產生。例如「高風險主訴、生命徵象沒有警示」案例把 C 設為 {2, 3}，並指定第二級為主訴優先候選。這是在測試主訴路徑能否補入或提高候選，不表示 {2, 3} 是正式 KTAS 對應。

主訴缺失或無法路由時，C 是空集合。程式不會猜一個主訴，也不會把缺失改寫成低風險。

V：合成生命徵象警示候選

V 沿用 Day 23 的六條作者合成生命徵象門檻。若門檻產生第二級上限，V 是 {1, 2}；若產生第一級上限，V 是 {1}。沒有警示，或只有缺失且沒有其他已觀察警示時，V 是空集合。

Day 23 的收縮壓（Systolic Blood Pressure, SBP）、心率（Heart Rate, HR）、呼吸速率（Respiratory Rate, RR）與周邊血氧飽和度（Peripheral Oxygen Saturation, SpO2）門檻只是反事實工程規則。本篇不新增正式數值，也不把缺失值補成正常。

H：固定高風險保留集合

H 固定為 {1, 2}。它像一道候選保險：即使 B、C、V 都沒有帶入第一或第二級，聯集仍讓兩級候選進入排序。

這項設計很積極，所以必須把代價一起測試。H 不是正式 KTAS 規則，也沒有經過臨床校準；固定加權可能讓原本符合合成參考的第四級被第二級超越。若文章只展示 H 修正檢傷不足，卻不建立這種反例，就無法評估工程取捨。

Safety Union 的集合不變量

四個集合去重後形成 U：

U = B ∪ C ∪ V ∪ H

符號逐項代表：

• B 是共同 Base Top-2。
• C 是作者合成主訴候選。
• V 是作者合成生命徵象警示候選。
• H 是固定高風險保留集合。
• ∪ 是聯集；相同級數只保留一次。
• U 是進入 Safety Union 排序的最終候選集合。

因為 U 明確包含 B，所以一定滿足：

B ⊆ U

這就是本篇唯一稱為「安全不變量」的性質：Safety Union 的集合運算不會刪除任何 Base 候選。它沒有保證參考候選一定在 B，也沒有保證 U 的 Top-1 一定正確。

以低血氧合成案例為例，B = {4, 2}、C = {2, 4}、V = {1, 2}、H = {1, 2}。去重後：

U = {1, 2, 4}

四個來源一共提供八次級數出現，但 U 只有三筆不同候選。B 原有的第四級沒有因生命徵象警示而消失。

下圖以同一筆合成案例示範 B、C、V、H 去重後形成 U。請注意 Base 的第四級仍在聯集中，接下來只是重新排序。

上圖左側四條線可以同時指向同一級。程式以 level 作為去重鍵，不會建立四份第二級文件；來源則保存在加權條件中，讓排序器知道這筆候選得到哪些訊號。

去重後如何做來源感知排序

Safety Union 沿用每級作者合成分數 s(d)，再加入三種固定權重：

s'(d) = s(d)
        + 0.25 × I(d = complaint_priority)
        + 0.15 × I(d ∈ V)
        + 0.10 × I(d ∈ H)

各符號的意思如下：

• d 是某一個候選級數。
• s(d) 是該級原本的共同合成分數。
• s'(d) 是來源加權後的排序分數。
• complaint_priority 是 C 中唯一的主訴優先級；C 中其他級數不拿 0.25。
• I(條件) 是指示函數：條件成立時為 1，否則為 0。
• d ∈ V 表示候選同時得到生命徵象警示支持。
• d ∈ H 表示候選屬於固定高風險保留集合。

低血氧合成案例的第二級原分數是 0.91，同時是主訴優先、V 與 H 成員，因此：

s'(2) = 0.91 + 0.25 × 1 + 0.15 × 1 + 0.10 × 1
      = 1.41

第四級原分數雖然是 0.96，但沒有任何來源加權，所以 s'(4) = 0.96。排序後第二級成為 Top-1，第四級仍留在 U 中。

這些加權值是刻意鎖定的測試常數，不是機率、臨床風險或學習後參數。分數超過 1 也不代表超過 100% 信心；它只是加法排序分數。

四種方法如何公平比較

四條路徑使用相同的五級合成分數、相同 Base Top-2 與相同合成參考級數，只改變門控行為：

若分數同分，四條路徑都以數字較小、也就是較急迫的級數優先。案例原始分數本身不允許同分，這項規則只處理加權後可能出現的平手。

下圖固定同一個 Base Top-2 與同一組分數，只改變候選動作。請觀察 Hard Gate 的候選數變少、Soft Gate 的候選數不變，以及 Safety Union 的候選數增加。

上圖的 Hard Gate、Soft Gate 與 Safety Union 在這筆案例都把第二級排到 Top-1，但理由不同：Hard Gate 刪掉第四級，Soft Gate 保留第四級但提高第二級，Safety Union 則同時保留第四級、加入第一級，並替第二級累加三個來源訊號。

缺失值如何傳遞

Day 24 不因加入 C 與 H 就改寫 Day 23 的生命徵象缺失政策：

1. 某個生命徵象缺失時，依賴該欄位的規則不執行。
2. 其他已觀察欄位若觸發警示，V 仍依最急迫的已觸發上限建立。
3. 沒有已觀察警示但存在缺失時，生命徵象狀態保留為 indeterminate_missing_values，V 是空集合。
4. C 只依作者合成主訴路由建立，不用生命徵象缺失猜測主訴。
5. 主訴也缺失或無法路由時，C 是空集合，不把缺失當成低風險。

這表示各證據路徑可以各自承認「不知道」。H 仍會固定加入 {1, 2}，但它只保護候選可見性；若加權太強，仍可能造成合成檢傷過度。

八筆案例如何覆蓋成功與代價

前六筆沿用 Day 23，另外加入 C；後兩筆專門測試 Safety Union 的代價：

這八筆不是隨機抽樣，也沒有代表任何疾病盛行率。把「修正五筆不足」和「產生兩筆過度」放在同一組鎖定案例中，是為了讓程式變更時能重跑相同邊界，而不是估計哪一種門控在臨床上比較好。

先在自己的專案資料夾建立本篇完整檔案

接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾，依下列順序建立檔案；每個程式碼區塊都是該檔案的完整內容，不含省略號。

本篇沿用 Day 13 的 Poetry 與重現性工具、Day 17 的五筆公開級數定義，以及 Day 23 已完整建立的六筆合成案例、生命徵象門檻與公開結果。Day 24 保留六筆案例的輸入、共同分數與 synthetic_reference_level，再加入作者合成主訴路由及兩筆代價案例；不讀取病患列、護理師標籤或專家標籤。以下是 Day 24 新增或修改後的完整設定、合成案例覆寫、RAG 套件入口、Safety Union 核心、執行入口與測試；公開摘要、完整追蹤紀錄與執行清單（Run Manifest）都由執行入口自動產生，不需要手動建立。

先從專案根目錄建立需要的資料夾：

mkdir -p configs/rag tests/fixtures src/triage_rag/rag scripts tests results/public results/runs/day-24

如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案，把對應區塊完整貼入後儲存。

檔案 1：建立 configs/rag/day-24-safety-union-gate.json

鎖定 B、C、V、H、Safety Union 不變量、四路徑公平比較、三種固定來源加權、上游雜湊、八筆預期與非臨床限制。

請在文字編輯器建立 configs/rag/day-24-safety-union-gate.json，貼入以下完整內容並儲存：

{
  "schema_version": 1,
  "experiment_id": "day-24-synthetic-safety-union-gate",
  "scope": "author_written_synthetic_safety_union_microbenchmark_not_patient_triage_or_clinical_evaluation",
  "sources": {
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa",
    "day23_config_path": "configs/rag/day-23-vital-hard-gate.json",
    "day23_config_sha256": "72c2b964e8e1080c27f44002577334e7e6e1789dbfecd2fe5405ac8c250365d9",
    "day23_cases_path": "tests/fixtures/day-23-synthetic-gate-cases.json",
    "day23_cases_sha256": "f40c9e3dd9a1679c58e6664e94b90bc215adc882537d39d3a086746a3e77179e",
    "day23_public_result_path": "results/public/day-23-vital-hard-gate.json",
    "day23_public_result_sha256": "e2533938753e362fedffd6964b1209eed9aa6334ccb945c296541a603a85aa93",
    "day24_cases_path": "tests/fixtures/day-24-synthetic-union-cases.json",
    "day24_cases_sha256": "94a997fff4e70105249af6f1808eb748c835bcee96feea334c90a1dc894722fe"
  },
  "candidate_contract": {
    "valid_levels": [1, 2, 3, 4, 5],
    "direction": "smaller_number_is_more_urgent",
    "candidate_source": "the_five_public_level_definition_chunks_only",
    "base_top_k": 2,
    "base_definition": "top_2_levels_by_shared_author_written_synthetic_score",
    "complaint_set_definition": "author_written_synthetic_route_levels_only",
    "vital_set_definition": "levels_1_through_synthetic_vital_ceiling_when_alarm_else_empty",
    "reserve_levels": [1, 2],
    "union_expression": "U = B union C union V union H",
    "deduplication_key": "level",
    "safety_invariant": "B_is_a_subset_of_U_and_union_never_deletes_a_base_candidate"
  },
  "comparison_contract": {
    "methods": ["none", "hard", "soft", "safety_union"],
    "shared_score_policy": "all_methods_start_from_the_same_author_written_synthetic_scores_and_same_base_top_2",
    "none_policy": "rank_base_candidates_by_raw_score",
    "hard_policy": "remove_base_candidates_with_level_greater_than_synthetic_vital_ceiling",
    "soft_policy": "keep_base_candidates_and_add_vital_bonus_without_adding_or_deleting_candidates",
    "safety_union_policy": "rank_the_deduplicated_union_with_additive_source_bonuses",
    "tie_break_policy": "smaller_level_number_first",
    "bonuses": {
      "complaint_priority": 0.25,
      "vital_membership": 0.15,
      "reserve_membership": 0.1
    }
  },
  "evaluation": {
    "expected_case_count": 8,
    "expected_reused_day23_case_count": 6,
    "expected_new_cost_case_count": 2,
    "top_k_trace": 3,
    "expected_direction_counts": {
      "none": {"exact": 3, "undertriage": 5, "overtriage": 0},
      "hard": {"exact": 5, "undertriage": 2, "overtriage": 1},
      "soft": {"exact": 5, "undertriage": 2, "overtriage": 1},
      "safety_union": {"exact": 6, "undertriage": 0, "overtriage": 2}
    },
    "expected_reference_removed_counts": {"hard": 1, "soft": 0, "safety_union": 0},
    "expected_base_preserved_by_union_count": 8,
    "direction_definition": {
      "exact": "candidate_top1_equals_synthetic_reference_level",
      "undertriage": "candidate_top1_is_numerically_greater_than_synthetic_reference_level",
      "overtriage": "candidate_top1_is_numerically_smaller_than_synthetic_reference_level"
    },
    "claim_boundary": "counts_describe_only_eight_author_written_contract_cases_and_are_not_accuracy_or_safety_estimates"
  },
  "outputs": {
    "public_summary_path": "results/public/day-24-safety-union-gate.json",
    "run_output_root": "results/runs/day-24",
    "result_filename": "safety-union-gate-results.json"
  },
  "limitations": [
    "八筆案例、主訴路由、候選分數與 synthetic_reference_level 全由作者為工程測試撰寫，不是真實病患、護理師標籤或專家標籤。",
    "生命徵象門檻沿用 Day 23 的反事實工程規則，不是 KTAS 官方規則，也不能用於臨床判斷。",
    "主訴候選 C 與固定高風險保留集合 H 是教學用集合，不代表完整 KTAS 主訴目錄、正式修飾因子或套用順序。",
    "加權值是用來暴露候選保留、排序改變與檢傷過度代價的固定測試常數，沒有經過臨床校準。",
    "安全聯集只保證不刪除 Base 候選，不保證 Top-1 正確，也不等於臨床安全。",
    "本篇只報告八筆鎖定案例的描述性計數，不宣稱分類準確率、召回率、檢傷不足率、檢傷過度率或臨床效能。",
    "系統只能作為研究與教學用決策支援元件，不能取代護理師、醫師或正式檢傷流程。"
  ]
}

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 2：建立 tests/fixtures/day-24-synthetic-union-cases.json

為六筆 Day 23 案例加入作者合成主訴路由，另保存兩筆檢傷過度代價案例與四方法逐筆鎖定預期。

請在文字編輯器建立 tests/fixtures/day-24-synthetic-union-cases.json，貼入以下完整內容並儲存：

{
  "schema_version": 1,
  "scope": "author_written_synthetic_safety_union_stress_cases_not_patient_records_or_clinical_labels",
  "score_semantics": "每級分數是為了固定四條路徑共同排序訊號的合成檢索分數；不是模型信心、機率或臨床風險。",
  "reference_semantics": "synthetic_reference_level 是作者為工程測試指定的預期級數；不是護理師標籤、專家標籤或 KTAS 判定。",
  "reused_day23_cases": [
    {
      "case_id": "alarm-low-saturation-rescue",
      "complaint_route": {
        "status": "routed",
        "candidate_levels": [2, 4],
        "priority_level": 2
      },
      "expected": {
        "base_levels": [4, 2],
        "vital_levels": [1, 2],
        "hard_levels": [2],
        "union_levels": [1, 2, 4],
        "top1_by_method": {"none": 4, "hard": 2, "soft": 2, "safety_union": 2},
        "direction_by_method": {"none": "undertriage", "hard": "exact", "soft": "exact", "safety_union": "exact"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "alarm-unresponsive-rescue",
      "complaint_route": {
        "status": "routed",
        "candidate_levels": [1, 3],
        "priority_level": 1
      },
      "expected": {
        "base_levels": [3, 1],
        "vital_levels": [1],
        "hard_levels": [1],
        "union_levels": [1, 2, 3],
        "top1_by_method": {"none": 3, "hard": 1, "soft": 1, "safety_union": 1},
        "direction_by_method": {"none": "undertriage", "hard": "exact", "soft": "exact", "safety_union": "exact"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "multiple-alarms-most-severe-wins",
      "complaint_route": {
        "status": "routed",
        "candidate_levels": [1, 2],
        "priority_level": 1
      },
      "expected": {
        "base_levels": [2, 1],
        "vital_levels": [1],
        "hard_levels": [1],
        "union_levels": [1, 2],
        "top1_by_method": {"none": 2, "hard": 1, "soft": 1, "safety_union": 1},
        "direction_by_method": {"none": "undertriage", "hard": "exact", "soft": "exact", "safety_union": "exact"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "no-alarm-high-risk-complaint",
      "complaint_route": {
        "status": "routed",
        "candidate_levels": [2, 3],
        "priority_level": 2
      },
      "expected": {
        "base_levels": [3, 2],
        "vital_levels": [],
        "hard_levels": [2, 3],
        "union_levels": [1, 2, 3],
        "top1_by_method": {"none": 3, "hard": 3, "soft": 3, "safety_union": 2},
        "direction_by_method": {"none": "undertriage", "hard": "undertriage", "soft": "undertriage", "safety_union": "exact"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "missing-saturation-is-indeterminate",
      "complaint_route": {
        "status": "routed",
        "candidate_levels": [2, 4],
        "priority_level": 2
      },
      "expected": {
        "base_levels": [4, 2],
        "vital_levels": [],
        "hard_levels": [2, 4],
        "union_levels": [1, 2, 4],
        "top1_by_method": {"none": 4, "hard": 4, "soft": 4, "safety_union": 2},
        "direction_by_method": {"none": "undertriage", "hard": "undertriage", "soft": "undertriage", "safety_union": "exact"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "rescued_undertriage"
      }
    },
    {
      "case_id": "boundary-alarm-removes-reference",
      "complaint_route": {
        "status": "routed",
        "candidate_levels": [3, 2],
        "priority_level": 3
      },
      "expected": {
        "base_levels": [3, 2],
        "vital_levels": [1, 2],
        "hard_levels": [2],
        "union_levels": [1, 2, 3],
        "top1_by_method": {"none": 3, "hard": 2, "soft": 2, "safety_union": 3},
        "direction_by_method": {"none": "exact", "hard": "overtriage", "soft": "overtriage", "safety_union": "exact"},
        "hard_reference_retained": false,
        "safety_reference_retained": true,
        "safety_outcome": "restored_hard_removed_reference"
      }
    }
  ],
  "new_cases": [
    {
      "case_id": "missing-complaint-reserve-overtriage",
      "scenario": "主訴路由缺失且生命徵象未警示；固定高風險保留集合改變 Top-1，用來暴露檢傷過度代價。",
      "inputs": {"Mental": 1, "SBP": 120, "HR": 80, "RR": 18, "Saturation": 98},
      "candidate_scores": {"1": 0.55, "2": 0.9, "3": 0.72, "4": 0.98, "5": 0.8},
      "synthetic_reference_level": 4,
      "complaint_route": {
        "status": "unroutable_missing_complaint",
        "candidate_levels": [],
        "priority_level": null
      },
      "expected": {
        "base_levels": [4, 2],
        "vital_levels": [],
        "hard_levels": [2, 4],
        "union_levels": [1, 2, 4],
        "top1_by_method": {"none": 4, "hard": 4, "soft": 4, "safety_union": 2},
        "direction_by_method": {"none": "exact", "hard": "exact", "soft": "exact", "safety_union": "overtriage"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "reserve_cost_overtriage"
      }
    },
    {
      "case_id": "complaint-router-false-positive",
      "scenario": "合成主訴路由誤把第二級設為優先，安全聯集保留參考候選但加權後產生檢傷過度。",
      "inputs": {"Mental": 1, "SBP": 124, "HR": 84, "RR": 16, "Saturation": 99},
      "candidate_scores": {"1": 0.6, "2": 0.9, "3": 0.74, "4": 0.96, "5": 0.5},
      "synthetic_reference_level": 4,
      "complaint_route": {
        "status": "routed_synthetic_false_positive",
        "candidate_levels": [2, 4],
        "priority_level": 2
      },
      "expected": {
        "base_levels": [4, 2],
        "vital_levels": [],
        "hard_levels": [2, 4],
        "union_levels": [1, 2, 4],
        "top1_by_method": {"none": 4, "hard": 4, "soft": 4, "safety_union": 2},
        "direction_by_method": {"none": "exact", "hard": "exact", "soft": "exact", "safety_union": "overtriage"},
        "hard_reference_retained": true,
        "safety_reference_retained": true,
        "safety_outcome": "complaint_route_cost_overtriage"
      }
    }
  ]
}

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 3：建立 src/triage_rag/rag/__init__.py

更新 RAG 子套件入口，保留 Day 21 至 23 公開函式並加入 Day 24 合併、契約、聯集、四路徑 trace 與聚合函式。

請在文字編輯器建立 src/triage_rag/rag/__init__.py，貼入以下完整內容並儲存：

"""Retrieval-augmented generation pipelines."""

from triage_rag.rag.flat import (
    FlatRagContractError,
    build_context,
    build_user_message,
    evaluate_answer,
    validate_flat_contract,
    validate_public_cases,
)
from triage_rag.rag.hierarchical import (
    HierarchicalRagContractError,
    build_hierarchical_context,
    build_hierarchical_user_message,
    expand_selected_children,
    expected_parent_ids,
    group_context_for_generation,
    rank_expanded_children,
    rank_parents,
    validate_hierarchical_contract,
)
from triage_rag.rag.vital_gate import (
    VitalHardGateContractError,
    direction_against_synthetic_reference,
    evaluate_vital_gate,
    rank_level_candidates,
    run_vital_hard_gate_case,
    select_level_candidates,
    summarize_vital_hard_gate_results,
    validate_vital_hard_gate_contract,
)
from triage_rag.rag.safety_union import (
    SafetyUnionContractError,
    build_safety_union_sets,
    merge_safety_union_cases,
    run_safety_union_case,
    summarize_safety_union_results,
    validate_safety_union_contract,
)

__all__ = [
    "FlatRagContractError",
    "build_context",
    "build_user_message",
    "evaluate_answer",
    "validate_flat_contract",
    "validate_public_cases",
    "HierarchicalRagContractError",
    "build_hierarchical_context",
    "build_hierarchical_user_message",
    "expand_selected_children",
    "expected_parent_ids",
    "group_context_for_generation",
    "rank_expanded_children",
    "rank_parents",
    "validate_hierarchical_contract",
    "VitalHardGateContractError",
    "direction_against_synthetic_reference",
    "evaluate_vital_gate",
    "rank_level_candidates",
    "run_vital_hard_gate_case",
    "select_level_candidates",
    "summarize_vital_hard_gate_results",
    "validate_vital_hard_gate_contract",
    "SafetyUnionContractError",
    "build_safety_union_sets",
    "merge_safety_union_cases",
    "run_safety_union_case",
    "summarize_safety_union_results",
    "validate_safety_union_contract",
]

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 4：建立 src/triage_rag/rag/safety_union.py

驗證合成範圍與公平比較，重用六筆 Day 23 案例，建立 B、C、V、H 與 U，實作無門控、Hard、Soft、Safety Union 排序及描述性聚合。

請在文字編輯器建立 src/triage_rag/rag/safety_union.py，貼入以下完整內容並儲存：

"""Deterministic Safety Union Gate for the synthetic Day 24 microbenchmark."""

from __future__ import annotations

import copy
from collections import Counter
from typing import Any, Dict, List, Mapping, Sequence

from triage_rag.rag.vital_gate import (
    VALID_LEVELS,
    direction_against_synthetic_reference,
    evaluate_vital_gate,
)


JsonObject = Dict[str, Any]
EXPECTED_SCOPE = (
    "author_written_synthetic_safety_union_microbenchmark_"
    "not_patient_triage_or_clinical_evaluation"
)
EXPECTED_FIXTURE_SCOPE = (
    "author_written_synthetic_safety_union_stress_cases_"
    "not_patient_records_or_clinical_labels"
)
FORBIDDEN_CASE_FIELDS = {
    "KTAS_RN",
    "KTAS_expert",
    "error_group",
    "Length of stay_min",
    "Disposition",
    "Diagnosis in ED",
    "result",
}


class SafetyUnionContractError(ValueError):
    """Raised when the Day 24 safety or comparison contract drifts."""


def _is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _assert_no_forbidden_fields(value: Any, *, location: str = "case") -> None:
    if isinstance(value, Mapping):
        overlap = sorted(FORBIDDEN_CASE_FIELDS.intersection(value))
        if overlap:
            raise SafetyUnionContractError(f"{location} 含禁止欄位：{overlap}")
        for key, nested in value.items():
            _assert_no_forbidden_fields(nested, location=f"{location}.{key}")
    elif isinstance(value, list):
        for index, nested in enumerate(value):
            _assert_no_forbidden_fields(nested, location=f"{location}[{index}]")


def merge_safety_union_cases(
    day23_fixture: Mapping[str, Any],
    day24_fixture: Mapping[str, Any],
) -> List[JsonObject]:
    """Attach Day 24 complaint routes to six Day 23 cases, then add two costs."""

    day23_cases = day23_fixture.get("cases")
    reused = day24_fixture.get("reused_day23_cases")
    new_cases = day24_fixture.get("new_cases")
    if not isinstance(day23_cases, list) or not isinstance(reused, list):
        raise SafetyUnionContractError("Day 23 cases 與 reused_day23_cases 必須是陣列")
    if not isinstance(new_cases, list):
        raise SafetyUnionContractError("new_cases 必須是陣列")
    day23_by_id = {str(case.get("case_id")): case for case in day23_cases}
    if len(day23_by_id) != len(day23_cases):
        raise SafetyUnionContractError("Day 23 case_id 不得重複")

    merged: List[JsonObject] = []
    reused_ids = []
    for overlay in reused:
        case_id = str(overlay.get("case_id"))
        reused_ids.append(case_id)
        if case_id not in day23_by_id:
            raise SafetyUnionContractError(f"找不到要沿用的 Day 23 案例：{case_id}")
        case = copy.deepcopy(day23_by_id[case_id])
        case.pop("expected", None)
        case["origin"] = "reused_day23_case"
        case["complaint_route"] = copy.deepcopy(overlay.get("complaint_route"))
        case["expected"] = copy.deepcopy(overlay.get("expected"))
        merged.append(case)

    expected_order = [str(case.get("case_id")) for case in day23_cases]
    if reused_ids != expected_order:
        raise SafetyUnionContractError(
            "Day 24 必須依原順序完整沿用六筆 Day 23 案例"
        )
    for source in new_cases:
        case = copy.deepcopy(source)
        case["origin"] = "new_day24_cost_case"
        merged.append(case)
    if len({str(case.get("case_id")) for case in merged}) != len(merged):
        raise SafetyUnionContractError("合併後的 case_id 不得重複")
    return merged


def validate_safety_union_contract(
    config: Mapping[str, Any],
    day23_config: Mapping[str, Any],
    day23_fixture: Mapping[str, Any],
    day23_public_result: Mapping[str, Any],
    day24_fixture: Mapping[str, Any],
    candidates: Sequence[Mapping[str, Any]],
) -> JsonObject:
    """Validate synthetic scope, fair comparison, routes, and Day 23 reuse."""

    if config.get("scope") != EXPECTED_SCOPE:
        raise SafetyUnionContractError("Day 24 scope 不允許改成病患或臨床評估")
    if day24_fixture.get("scope") != EXPECTED_FIXTURE_SCOPE:
        raise SafetyUnionContractError("Day 24 fixture 必須明示為作者合成案例")

    candidate_contract = config.get("candidate_contract")
    comparison = config.get("comparison_contract")
    if not isinstance(candidate_contract, Mapping) or not isinstance(comparison, Mapping):
        raise SafetyUnionContractError("缺少候選或公平比較契約")
    required_candidate_values = {
        "valid_levels": list(VALID_LEVELS),
        "direction": "smaller_number_is_more_urgent",
        "candidate_source": "the_five_public_level_definition_chunks_only",
        "base_top_k": 2,
        "base_definition": "top_2_levels_by_shared_author_written_synthetic_score",
        "complaint_set_definition": "author_written_synthetic_route_levels_only",
        "vital_set_definition": "levels_1_through_synthetic_vital_ceiling_when_alarm_else_empty",
        "reserve_levels": [1, 2],
        "union_expression": "U = B union C union V union H",
        "deduplication_key": "level",
        "safety_invariant": "B_is_a_subset_of_U_and_union_never_deletes_a_base_candidate",
    }
    drift = {
        key: {"actual": candidate_contract.get(key), "expected": expected}
        for key, expected in required_candidate_values.items()
        if candidate_contract.get(key) != expected
    }
    if drift:
        raise SafetyUnionContractError(f"候選聯集契約漂移：{drift}")

    required_comparison_values = {
        "methods": ["none", "hard", "soft", "safety_union"],
        "shared_score_policy": "all_methods_start_from_the_same_author_written_synthetic_scores_and_same_base_top_2",
        "none_policy": "rank_base_candidates_by_raw_score",
        "hard_policy": "remove_base_candidates_with_level_greater_than_synthetic_vital_ceiling",
        "soft_policy": "keep_base_candidates_and_add_vital_bonus_without_adding_or_deleting_candidates",
        "safety_union_policy": "rank_the_deduplicated_union_with_additive_source_bonuses",
        "tie_break_policy": "smaller_level_number_first",
        "bonuses": {
            "complaint_priority": 0.25,
            "vital_membership": 0.15,
            "reserve_membership": 0.1,
        },
    }
    comparison_drift = {
        key: {"actual": comparison.get(key), "expected": expected}
        for key, expected in required_comparison_values.items()
        if comparison.get(key) != expected
    }
    if comparison_drift:
        raise SafetyUnionContractError(f"公平比較契約漂移：{comparison_drift}")

    levels = [int(candidate["level"]) for candidate in candidates]
    if levels != list(VALID_LEVELS):
        raise SafetyUnionContractError("候選必須是排序後的第一至第五級公開定義")
    if day23_config.get("synthetic_threshold_policy", {}).get("clinical_use") != "prohibited":
        raise SafetyUnionContractError("Day 23 合成門檻不得改成臨床用途")
    if day23_public_result.get("checks", {}).get("statistics_eligible") is not False:
        raise SafetyUnionContractError("Day 23 合成結果不得變成統計估計")

    cases = merge_safety_union_cases(day23_fixture, day24_fixture)
    evaluation = config.get("evaluation", {})
    if len(cases) != evaluation.get("expected_case_count"):
        raise SafetyUnionContractError("Day 24 合成案例數與契約不同")
    if len(day24_fixture["reused_day23_cases"]) != evaluation.get(
        "expected_reused_day23_case_count"
    ):
        raise SafetyUnionContractError("Day 23 沿用案例數與契約不同")
    if len(day24_fixture["new_cases"]) != evaluation.get(
        "expected_new_cost_case_count"
    ):
        raise SafetyUnionContractError("Day 24 新增代價案例數與契約不同")
    _assert_no_forbidden_fields(cases)

    public_by_id = {
        str(case["case_id"]): case
        for case in day23_public_result.get("case_results", [])
    }
    for case in cases:
        case_id = str(case.get("case_id"))
        scores = case.get("candidate_scores")
        if not isinstance(scores, Mapping) or set(scores) != {
            str(level) for level in VALID_LEVELS
        }:
            raise SafetyUnionContractError(f"{case_id} 必須提供五級共同分數")
        if not all(
            _is_number(value) and 0 <= float(value) <= 1
            for value in scores.values()
        ):
            raise SafetyUnionContractError(f"{case_id} 分數必須介於 0 與 1")
        if len({float(value) for value in scores.values()}) != len(VALID_LEVELS):
            raise SafetyUnionContractError(f"{case_id} 分數不可同分")
        if case.get("synthetic_reference_level") not in VALID_LEVELS:
            raise SafetyUnionContractError(f"{case_id} 合成參考級數非法")
        route = case.get("complaint_route")
        if not isinstance(route, Mapping):
            raise SafetyUnionContractError(f"{case_id} 缺少 complaint_route")
        route_levels = route.get("candidate_levels")
        if not isinstance(route_levels, list) or len(route_levels) != len(set(route_levels)):
            raise SafetyUnionContractError(f"{case_id} 主訴候選必須是無重複陣列")
        if any(level not in VALID_LEVELS for level in route_levels):
            raise SafetyUnionContractError(f"{case_id} 主訴候選級數非法")
        priority = route.get("priority_level")
        if priority is not None and priority not in route_levels:
            raise SafetyUnionContractError(f"{case_id} 主訴優先級必須位於 C 集合")
        if route_levels and priority is None:
            raise SafetyUnionContractError(f"{case_id} 有 C 集合但沒有主訴優先級")
        if not route_levels and priority is not None:
            raise SafetyUnionContractError(f"{case_id} 空 C 集合不得有主訴優先級")
        if case["origin"] == "reused_day23_case":
            public = public_by_id.get(case_id)
            if public is None:
                raise SafetyUnionContractError(f"Day 23 公開結果缺少 {case_id}")
            if public["baseline"]["top1_level"] != case["expected"]["top1_by_method"]["none"]:
                raise SafetyUnionContractError(f"{case_id} 無門控回歸結果漂移")
            if public["hard_gate"]["top1_level"] != case["expected"]["top1_by_method"]["hard"]:
                raise SafetyUnionContractError(f"{case_id} Hard Gate 回歸結果漂移")

    return {
        "scope_is_synthetic_only": True,
        "candidate_levels": levels,
        "case_count": len(cases),
        "reused_day23_case_count": len(day24_fixture["reused_day23_cases"]),
        "new_cost_case_count": len(day24_fixture["new_cases"]),
        "methods": list(comparison["methods"]),
        "base_top_k": int(candidate_contract["base_top_k"]),
        "reserve_levels": list(candidate_contract["reserve_levels"]),
        "forbidden_patient_or_label_fields_absent": True,
        "day23_thresholds_remain_synthetic_and_nonclinical": True,
        "statistics_eligible": False,
    }


def _rank_levels(
    levels: Sequence[int],
    candidates: Sequence[Mapping[str, Any]],
    scores: Mapping[str, Any],
    *,
    complaint_priority: int | None = None,
    vital_levels: Sequence[int] = (),
    reserve_levels: Sequence[int] = (),
    bonuses: Mapping[str, Any] | None = None,
) -> List[JsonObject]:
    catalog = {int(candidate["level"]): candidate for candidate in candidates}
    vital_set = set(vital_levels)
    reserve_set = set(reserve_levels)
    bonus_values = bonuses or {}
    ranked: List[JsonObject] = []
    for level in levels:
        base_score = float(scores[str(level)])
        complaint_bonus = (
            float(bonus_values.get("complaint_priority", 0.0))
            if complaint_priority == level
            else 0.0
        )
        vital_bonus = (
            float(bonus_values.get("vital_membership", 0.0))
            if level in vital_set
            else 0.0
        )
        reserve_bonus = (
            float(bonus_values.get("reserve_membership", 0.0))
            if level in reserve_set
            else 0.0
        )
        final_score = base_score + complaint_bonus + vital_bonus + reserve_bonus
        ranked.append(
            {
                "level": level,
                "rule_id": catalog[level]["rule_id"],
                "base_score": round(base_score, 6),
                "bonuses": {
                    "complaint_priority": round(complaint_bonus, 6),
                    "vital_membership": round(vital_bonus, 6),
                    "reserve_membership": round(reserve_bonus, 6),
                },
                "final_score": round(final_score, 6),
                "source_url": catalog[level]["source_url"],
            }
        )
    ranked.sort(key=lambda item: (-item["final_score"], item["level"]))
    for rank, item in enumerate(ranked, start=1):
        item["rank"] = rank
    return ranked


def build_safety_union_sets(
    base_levels: Sequence[int],
    complaint_levels: Sequence[int],
    vital_levels: Sequence[int],
    reserve_levels: Sequence[int],
) -> JsonObject:
    """Build B, C, V, H, and the de-duplicated union U."""

    sets = {
        "B": sorted(set(base_levels)),
        "C": sorted(set(complaint_levels)),
        "V": sorted(set(vital_levels)),
        "H": sorted(set(reserve_levels)),
    }
    for name, levels in sets.items():
        if any(level not in VALID_LEVELS for level in levels):
            raise SafetyUnionContractError(f"{name} 含非法級數")
    union_levels = sorted(set().union(*[set(levels) for levels in sets.values()]))
    if not set(sets["B"]).issubset(union_levels):
        raise SafetyUnionContractError("Safety Union 不得刪除 Base 候選")
    return {**sets, "U": union_levels, "base_preserved": True}


def _derive_safety_outcome(
    none_direction: str,
    hard_direction: str,
    safety_direction: str,
    *,
    hard_reference_retained: bool,
    route_status: str,
) -> str:
    if not hard_reference_retained and safety_direction == "exact":
        return "restored_hard_removed_reference"
    if none_direction == "undertriage" and safety_direction == "exact":
        return "rescued_undertriage"
    if none_direction == "exact" and safety_direction == "overtriage":
        if route_status == "unroutable_missing_complaint":
            return "reserve_cost_overtriage"
        return "complaint_route_cost_overtriage"
    if none_direction == safety_direction:
        return "unchanged"
    return f"{none_direction}_to_{safety_direction}_after_{hard_direction}"


def run_safety_union_case(
    case: Mapping[str, Any],
    candidates: Sequence[Mapping[str, Any]],
    vital_rules: Sequence[Mapping[str, Any]],
    contract: Mapping[str, Any],
    *,
    top_k: int,
) -> JsonObject:
    """Run all four methods on one shared synthetic case and emit a trace."""

    scores = case["candidate_scores"]
    base_top_k = int(contract["candidate_contract"]["base_top_k"])
    all_raw = _rank_levels(list(VALID_LEVELS), candidates, scores)
    base_levels = [int(item["level"]) for item in all_raw[:base_top_k]]
    gate = evaluate_vital_gate(case["inputs"], vital_rules)
    ceiling = gate["most_severe_level"]
    vital_levels = list(range(1, int(ceiling) + 1)) if ceiling is not None else []
    complaint_route = case["complaint_route"]
    complaint_levels = [int(level) for level in complaint_route["candidate_levels"]]
    priority = complaint_route["priority_level"]
    reserve_levels = [int(level) for level in contract["candidate_contract"]["reserve_levels"]]
    sets = build_safety_union_sets(
        base_levels, complaint_levels, vital_levels, reserve_levels
    )

    none_ranking = _rank_levels(base_levels, candidates, scores)
    hard_levels = [
        level for level in base_levels if ceiling is None or level <= int(ceiling)
    ]
    if not hard_levels:
        raise SafetyUnionContractError("Hard Gate 不得產生空候選集合")
    hard_ranking = _rank_levels(hard_levels, candidates, scores)
    bonuses = contract["comparison_contract"]["bonuses"]
    soft_ranking = _rank_levels(
        base_levels,
        candidates,
        scores,
        vital_levels=vital_levels,
        bonuses={"vital_membership": bonuses["vital_membership"]},
    )
    union_ranking = _rank_levels(
        sets["U"],
        candidates,
        scores,
        complaint_priority=priority,
        vital_levels=vital_levels,
        reserve_levels=reserve_levels,
        bonuses=bonuses,
    )

    reference = int(case["synthetic_reference_level"])
    rankings = {
        "none": none_ranking,
        "hard": hard_ranking,
        "soft": soft_ranking,
        "safety_union": union_ranking,
    }
    top1 = {name: int(ranking[0]["level"]) for name, ranking in rankings.items()}
    directions = {
        name: direction_against_synthetic_reference(level, reference)
        for name, level in top1.items()
    }
    hard_reference_retained = reference in hard_levels
    safety_reference_retained = reference in sets["U"]
    outcome = _derive_safety_outcome(
        directions["none"],
        directions["hard"],
        directions["safety_union"],
        hard_reference_retained=hard_reference_retained,
        route_status=str(complaint_route["status"]),
    )
    actual_expected = {
        "base_levels": base_levels,
        "vital_levels": vital_levels,
        "hard_levels": sorted(hard_levels),
        "union_levels": sets["U"],
        "top1_by_method": top1,
        "direction_by_method": directions,
        "hard_reference_retained": hard_reference_retained,
        "safety_reference_retained": safety_reference_retained,
        "safety_outcome": outcome,
    }
    if actual_expected != case["expected"]:
        raise SafetyUnionContractError(
            f"{case['case_id']} 與鎖定預期不同：actual={actual_expected}，"
            f"expected={case['expected']}"
        )

    method_traces = {}
    for name, ranking in rankings.items():
        method_traces[name] = {
            "candidate_count": len(ranking),
            "candidate_levels": sorted(int(item["level"]) for item in ranking),
            "reference_retained": reference
            in {int(item["level"]) for item in ranking},
            "top1_level": top1[name],
            "direction": directions[name],
            "top_k": ranking[:top_k],
        }
    return {
        "case_id": case["case_id"],
        "origin": case["origin"],
        "scenario": case["scenario"],
        "inputs": dict(case["inputs"]),
        "synthetic_reference": {
            "level": reference,
            "role": "author_written_engineering_reference_not_a_clinical_label",
        },
        "gate": gate,
        "complaint_route": copy.deepcopy(complaint_route),
        "candidate_sets": sets,
        "methods": method_traces,
        "safety_outcome": outcome,
        "locked_expectation_passed": True,
    }


def summarize_safety_union_results(
    cases: Sequence[Mapping[str, Any]],
) -> JsonObject:
    """Aggregate descriptive counts without treating them as clinical rates."""

    method_names = ("none", "hard", "soft", "safety_union")
    directions = {
        method: Counter(str(case["methods"][method]["direction"]) for case in cases)
        for method in method_names
    }
    return {
        "case_count": len(cases),
        "direction_counts": {
            method: {
                direction: directions[method].get(direction, 0)
                for direction in ("exact", "undertriage", "overtriage")
            }
            for method in method_names
        },
        "reference_removed_counts": {
            method: sum(
                not bool(case["methods"][method]["reference_retained"])
                for case in cases
            )
            for method in ("hard", "soft", "safety_union")
        },
        "base_preserved_by_union_count": sum(
            bool(case["candidate_sets"]["base_preserved"]) for case in cases
        ),
        "outcome_counts": dict(
            sorted(Counter(str(case["safety_outcome"]) for case in cases).items())
        ),
        "all_locked_expectations_passed": all(
            bool(case["locked_expectation_passed"]) for case in cases
        ),
        "claim_boundary": (
            "這些是八筆作者合成契約案例的描述性計數，不是病患分類率、"
            "檢傷不足率、檢傷過度率或臨床安全估計。"
        ),
    }

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 5：建立 scripts/run_day24_safety_union_gate.py

核對 Day 17 與 Day 23 上游雜湊，配對執行八筆四路徑案例，最後寫出公開摘要、完整 trace 與 run manifest。

請在文字編輯器建立 scripts/run_day24_safety_union_gate.py，貼入以下完整內容並儲存：

#!/usr/bin/env python3
"""Run the Day 24 four-way synthetic Safety Union Gate microbenchmark."""

from __future__ import annotations

import argparse
import platform
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Mapping

from triage_rag.knowledge_base.builder import read_jsonl
from triage_rag.rag import (
    merge_safety_union_cases,
    run_safety_union_case,
    select_level_candidates,
    summarize_safety_union_results,
    validate_safety_union_contract,
)
from triage_rag.reproducibility import (
    file_record,
    git_state,
    load_json,
    resolve_project_path,
    sha256_file,
    write_json,
)


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = "configs/rag/day-24-safety-union-gate.json"
EXPECTED_SCOPE = (
    "author_written_synthetic_safety_union_microbenchmark_"
    "not_patient_triage_or_clinical_evaluation"
)
IMPLEMENTATION_PATHS = (
    "src/triage_rag/rag/__init__.py",
    "src/triage_rag/rag/safety_union.py",
    "scripts/run_day24_safety_union_gate.py",
    "tests/test_safety_union_gate.py",
)
JsonObject = Dict[str, Any]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "以八筆作者合成案例公平比較無門控、Hard Gate、Soft Gate 與 "
            "Safety Union Gate；不讀取病患資料或臨床標籤。"
        )
    )
    parser.add_argument("--config", default=DEFAULT_CONFIG)
    return parser.parse_args()


def _verify_sha256(path: Path, expected: str, label: str) -> None:
    actual = sha256_file(path)
    if actual != expected:
        raise ValueError(
            f"{label} SHA-256 不同；預期 {expected}，實際 {actual}。"
            "請先重建上游產物，或建立新版 Day 24 契約。"
        )


def _load_and_verify_sources(
    config: Mapping[str, Any],
) -> tuple[List[Path], JsonObject]:
    sources = config["sources"]
    source_pairs = (
        ("flat_chunks_path", "flat_chunks_sha256", "Day 17 flat chunks"),
        ("day23_config_path", "day23_config_sha256", "Day 23 config"),
        ("day23_cases_path", "day23_cases_sha256", "Day 23 cases"),
        (
            "day23_public_result_path",
            "day23_public_result_sha256",
            "Day 23 public result",
        ),
        ("day24_cases_path", "day24_cases_sha256", "Day 24 cases"),
    )
    paths: List[Path] = []
    for path_key, hash_key, label in source_pairs:
        path = resolve_project_path(PROJECT_ROOT, str(sources[path_key]))
        _verify_sha256(path, str(sources[hash_key]), label)
        paths.append(path)
    loaded = {
        "day23_config": load_json(paths[1]),
        "day23_fixture": load_json(paths[2]),
        "day23_public": load_json(paths[3]),
        "day24_fixture": load_json(paths[4]),
    }
    return paths, loaded


def _validate_aggregate(
    aggregate: Mapping[str, Any], evaluation: Mapping[str, Any]
) -> None:
    checks = {
        "case_count": (aggregate["case_count"], evaluation["expected_case_count"]),
        "direction_counts": (
            aggregate["direction_counts"],
            evaluation["expected_direction_counts"],
        ),
        "reference_removed_counts": (
            aggregate["reference_removed_counts"],
            evaluation["expected_reference_removed_counts"],
        ),
        "base_preserved_by_union_count": (
            aggregate["base_preserved_by_union_count"],
            evaluation["expected_base_preserved_by_union_count"],
        ),
    }
    mismatches = {
        key: {"actual": actual, "expected": expected}
        for key, (actual, expected) in checks.items()
        if actual != expected
    }
    if mismatches:
        raise ValueError(f"Day 24 聚合結果與鎖定預期不同：{mismatches}")
    if not aggregate["all_locked_expectations_passed"]:
        raise ValueError("至少一筆 Day 24 案例未通過鎖定預期")


def main() -> int:
    args = parse_args()
    started_at = datetime.now(timezone.utc)
    config_path = resolve_project_path(PROJECT_ROOT, args.config)
    config = load_json(config_path)
    if config.get("scope") != EXPECTED_SCOPE:
        raise ValueError("Day 24 scope 不允許病患或臨床評估")

    print("步驟 1/4：核對上游雜湊、合成範圍與公平比較契約。", flush=True)
    source_paths, loaded = _load_and_verify_sources(config)
    candidates = select_level_candidates(read_jsonl(source_paths[0]))
    input_contract = validate_safety_union_contract(
        config,
        loaded["day23_config"],
        loaded["day23_fixture"],
        loaded["day23_public"],
        loaded["day24_fixture"],
        candidates,
    )
    cases = merge_safety_union_cases(
        loaded["day23_fixture"], loaded["day24_fixture"]
    )

    print("步驟 2/4：以共同 Base Top-2 配對執行四條門控路徑。", flush=True)
    vital_rules = loaded["day23_config"]["synthetic_threshold_policy"]["rules"]
    top_k = int(config["evaluation"]["top_k_trace"])
    case_results = []
    for index, case in enumerate(cases, start=1):
        trace = run_safety_union_case(
            case,
            candidates,
            vital_rules,
            config,
            top_k=top_k,
        )
        case_results.append(trace)
        print(
            f"  {index:02d}/08 {trace['case_id']}：{trace['safety_outcome']}",
            flush=True,
        )

    print("步驟 3/4：彙整方向、候選保留與安全聯集代價。", flush=True)
    aggregate = summarize_safety_union_results(case_results)
    _validate_aggregate(aggregate, config["evaluation"])
    public_summary = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "candidate_contract": config["candidate_contract"],
        "comparison_contract": config["comparison_contract"],
        "evaluation_contract": config["evaluation"],
        "input_contract": input_contract,
        "candidate_catalog": [
            {
                "level": candidate["level"],
                "rule_id": candidate["rule_id"],
                "source_url": candidate["source_url"],
            }
            for candidate in candidates
        ],
        "aggregate": aggregate,
        "case_results": case_results,
        "checks": {
            "source_hashes_match": True,
            "six_day23_cases_reused_without_input_score_or_reference_changes": True,
            "all_methods_share_the_same_scores_and_base_top_2": True,
            "safety_union_preserves_every_base_candidate": True,
            "day23_thresholds_remain_explicitly_synthetic_and_nonclinical": True,
            "missing_values_never_imputed_or_treated_as_normal": True,
            "patient_rows_not_read": True,
            "nurse_and_expert_labels_not_read": True,
            "statistics_eligible": False,
        },
        "limitations": config["limitations"],
        "interpretation": (
            "八筆作者合成案例顯示：Safety Union 在此鎖定契約中保留全部 Base "
            "候選，並消除五筆原始合成檢傷不足；但來源加權也使兩筆新增代價"
            "案例產生合成檢傷過度。這是程式行為壓力測試，不是臨床效能或"
            "安全結論。"
        ),
    }

    run_id = (
        f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-"
        f"{sha256_file(config_path)[:8]}"
    )
    run_root = resolve_project_path(PROJECT_ROOT, config["outputs"]["run_output_root"])
    run_directory = run_root / run_id
    run_directory.mkdir(parents=True, exist_ok=False)
    full_result_path = run_directory / config["outputs"]["result_filename"]
    write_json(full_result_path, public_summary)
    public_path = resolve_project_path(
        PROJECT_ROOT, config["outputs"]["public_summary_path"]
    )
    write_json(public_path, public_summary)

    manifest = {
        "manifest_schema_version": 1,
        "run_id": run_id,
        "experiment_id": config["experiment_id"],
        "started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
        "command": [
            sys.executable,
            str(Path(__file__).relative_to(PROJECT_ROOT)),
            "--config",
            args.config,
        ],
        "git": git_state(PROJECT_ROOT),
        "runtime": {"python": platform.python_version()},
        "parameters": {
            "methods": config["comparison_contract"]["methods"],
            "base_top_k": config["candidate_contract"]["base_top_k"],
            "reserve_levels": config["candidate_contract"]["reserve_levels"],
            "bonuses": config["comparison_contract"]["bonuses"],
            "top_k_trace": top_k,
        },
        "inputs": [
            file_record(PROJECT_ROOT, args.config),
            *[
                file_record(PROJECT_ROOT, str(path.relative_to(PROJECT_ROOT)))
                for path in source_paths
            ],
            *[file_record(PROJECT_ROOT, path) for path in IMPLEMENTATION_PATHS],
        ],
        "outputs": [
            file_record(
                PROJECT_ROOT, str(full_result_path.relative_to(PROJECT_ROOT))
            ),
            file_record(PROJECT_ROOT, str(public_path.relative_to(PROJECT_ROOT))),
        ],
        "scope": config["scope"],
    }
    manifest_path = run_directory / "run-manifest.json"
    write_json(manifest_path, manifest)

    print("步驟 4/4：寫出公開摘要、完整結果與 run manifest。", flush=True)
    print(f"公開摘要：{public_path.relative_to(PROJECT_ROOT)}", flush=True)
    print(f"完整結果：{full_result_path.relative_to(PROJECT_ROOT)}", flush=True)
    print(f"執行紀錄：{manifest_path.relative_to(PROJECT_ROOT)}", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 6：建立 tests/test_safety_union_gate.py

以十六項離線測試驗證合成範圍、跨日重用、禁止標籤、集合去重、Base 保留、三種門控差異、錯誤修正與檢傷過度代價。

請在文字編輯器建立 tests/test_safety_union_gate.py，貼入以下完整內容並儲存：

from __future__ import annotations

import copy
import unittest
from pathlib import Path

from triage_rag.knowledge_base.builder import read_jsonl
from triage_rag.rag import (
    SafetyUnionContractError,
    build_safety_union_sets,
    merge_safety_union_cases,
    run_safety_union_case,
    select_level_candidates,
    summarize_safety_union_results,
    validate_safety_union_contract,
)
from triage_rag.reproducibility import load_json


ROOT = Path(__file__).resolve().parents[1]
CONFIG_PATH = ROOT / "configs" / "rag" / "day-24-safety-union-gate.json"


class Day24SafetyUnionGateTests(unittest.TestCase):
    def setUp(self) -> None:
        self.config = load_json(CONFIG_PATH)
        sources = self.config["sources"]
        self.day23_config = load_json(ROOT / sources["day23_config_path"])
        self.day23_fixture = load_json(ROOT / sources["day23_cases_path"])
        self.day23_public = load_json(ROOT / sources["day23_public_result_path"])
        self.day24_fixture = load_json(ROOT / sources["day24_cases_path"])
        documents = read_jsonl(ROOT / sources["flat_chunks_path"])
        self.candidates = select_level_candidates(documents)
        self.cases = merge_safety_union_cases(
            self.day23_fixture, self.day24_fixture
        )
        self.rules = self.day23_config["synthetic_threshold_policy"]["rules"]

    def _trace(self, case_id: str):
        case = next(case for case in self.cases if case["case_id"] == case_id)
        return run_safety_union_case(
            case,
            self.candidates,
            self.rules,
            self.config,
            top_k=3,
        )

    def test_full_synthetic_contract_passes(self) -> None:
        summary = validate_safety_union_contract(
            self.config,
            self