Phase A4 Design — devices / info_systems / leveraged / 控制項 / AO 鉤稽 + 寫入

Phase：A4（Track A：Excel 匯入第四階段） 級別：重型（brainstorm → design → plan → 開工） 狀態：design draft（brainstorm 收斂完成 2026-05-20） 前置：A1（樣板下載）+ A2（Parser + 解析 API）+ A3（共用 matcher 抽取）皆 BE shipped 依賴本 phase：A5（預覽 UI + Confirm 拍板）需要本 phase 寫入 matched_*_id 結果到 parsed_result JSONB 才能展示

1.1 既有現況（A3 ship 後）

A3 phase 抽出 BaseReconciliationService[TParsed, TEntity] 抽象基類 + 兩個 reference reconciler（Person / Organization）+ A2 confirm flow 串接 + 67 個新 test。A2 confirm flow _write_all_data 在 write_parties 前已呼叫 self._reconciliation.reconcile(...)。

A2 phase 留下的 5 個 stub：

A2 §15.3 留下的 superset flow include_controls 缺口：

• _confirm_superset_flow 內 include_controls=[] (全納入)
• 因為 parser 解出 control_id='AC-1' 字面，但 ProfileService.add_profile(include_controls=...) 接的是 catalog_control_uid (UUID)
• A4 ControlReconciler 完成後可順手做反查（D6 拍板整併）

1.2 A4 三個目標

1. 完成 5 個新 reconciler：device / info_system / leveraged / catalog_control / AO，建在 A3 BaseReconciliationService 上
2. 完成 5 個 entity 寫入 path：5 個 WriteStrategy + superset flow 建 SSP shell；update flow 走既有 SSP
3. A3 retroactive：補 USER_SELECTED stage 0 涵蓋樣板 matched_user / org_unit lookup col；同 PR ship

1.3 與其他 phase 關係

• 依賴 A0.1：ssp_system_implementation_items 主結構（含 scope_type / scope_id / device_id / system_characteristic_id / party_uuid / date_authorized）已 ship；A4 直接寫入
• 依賴 A2：Excel parser ParsedExcel 解出 5 領域 list[dict]；A4 寫 _dict_to_parsed_* 5 個轉換
• 依賴 A3：BaseReconciliationService 抽象基類 + MatchMethod enum + _normalizers.py；A4 直接擴 hook
• 解鎖 A5：A5 預覽 UI 需要 reconcile 完成後 matched_*_id / match_method 寫回 parsed_result JSONB；A4 Step 7 已含此處理

1.4 Pre-flight 帶出的最大新訊號

樣板已內建 matched_* lookup column — A1 SHEET_DEVICES.matched_device / SHEET_INFO_SYSTEMS.matched_info_system / SHEET_INFO_SYSTEMS.system_owner / SHEET_LEVERAGED.party / SHEET_PERSONS.matched_user / SHEET_ORGS（無但 SHEET_PERSONS.org_unit）user 填表時可直接從下拉選現有 entity。

A3 PersonReconciler / OrganizationReconciler 沒處理此 col（A3 design 沒看到）— A4 必補 USER_SELECTED stage 0 並同步 retroactive 補 A3 兩個 reconciler。

2.1 In Scope（A4 task arc 內）

鉤稽層（D1~D8）：

1. 新建 6 reconciler 檔（reconciliation/ 下平鋪 — D7）：
   • device_reconciler.py
   • information_system_reconciler.py
   • leveraged_reconciler.py
   • catalog_control_reconciler.py
   • assessment_objective_reconciler.py
   • ssp_entity_orchestrator.py（D1 facade）
2. MatchMethod enum 加 USER_SELECTED 值（D2）
3. _normalizers.py 補 3 個 helper：parse_user_lookup_label / parse_device_lookup_label / parse_info_system_lookup_label / _normalize_control_id
4. 5 個新 typed dataclass（在 ssp_intermediate.py）：ParsedDevice / ParsedInformationSystem / ParsedLeveraged / ParsedControl / ParsedAssessmentObjective
5. SspEntityReconciliationOrchestrator facade — enforce control → AO 順序
6. A2 _dict_to_* 5 個轉換函式（沿 A3 _dict_to_parsed_parties 樣板）
7. A2 §15.3 superset include_controls 反查（D6 — 用 ControlReconciler 結果填）

寫入層（D9 拍板）：

8. Superset flow 建 SSP shell（_confirm_superset_flow 內）：
   • create SSP placeholder 綁 MF + Profile
   • create system_characteristic（用 metadata.target_* 預填）
   • create system_implementation main（A0.1 main，scope_type='ssp'）
9. write_strategy/ 新 sub-folder（domain layer） 5 strategy：
   • ControlWriteStrategy → system_security_plan_control_implementations
   • AoWriteStrategy → ssp_control_impl_objective（依賴 control_implementation id）
   • DeviceWriteStrategy → ssp_system_implementation_items (implementation_type='hardware')
   • InformationSystemWriteStrategy → ssp_system_implementation_items (implementation_type='system')
   • LeveragedWriteStrategy → ssp_system_implementation_items (implementation_type='leveraged-authorization')
10. _write_all_data 重整 stub→實作（8 step 串接）

A3 retroactive（同 PR ship）：

11. PersonReconciler / OrganizationReconciler 加 stage 0 _try_user_selected_match override
12. ParsedParty 加 matched_user_label / matched_org_unit_label 兩個 optional 欄位
13. _dict_to_parsed_parties 把 Excel matched_user / org_unit lookup col 傳遞
14. BaseReconciliationService._reconcile_one 加 stage 0 dispatch + _try_user_selected_match default no-op hook

Infra：

15. DI Container 新增 11 Factory（5 reconciler + 1 orchestrator + 5 WriteStrategy）+ SspExcelImportAppService 加 ~11 個依賴
16. Test ~110~140 個（reconciler 60-75 + WriteStrategy 30-40 + integration 15-20 + ParsedXxx fixture 10）
17. Cucumber regression 新增 06-ssp-excel-import-entity-match.feature 含 5 scenarios（per entity 1，含 1 fuzzy）
18. Changelog YYYY-MM-DD-feat-ssp-entity-reconcile-and-write.md（type: feat / modules: oscal）

2.2 Out of Scope（列 follow-up）

2.3 不在 scope 的事（明確排除）

• 不動 jedi-oscal / jedi-device / jedi_information_system 套件（jedi-oscal 仍 path-dep，feature 整體完工才一次 bump + 推 Nexus）
• 不動 FE（A5 phase 才動預覽 UI）
• 不擴展 docx flow（D5 拍板）
• 不做 fuzzy 細分 MatchMethod 值（D8 — FUZZY_HOSTNAME / FUZZY_IP / FUZZY_VENDOR 不加）
• 不重複 ParsedExcel parser 結構（parser 仍維持 list[dict]，由 _write_all_data 內 _dict_to_* 轉 typed dataclass）

3.1 File 結構（D7 平鋪 + 新增 write_strategy/）

domain/oscal/service/
├── reconciliation/                         ← A3 既有 + A4 擴
│   ├── __init__.py                          # export base + MatchMethod
│   ├── base.py                              # A4 加 stage 0 dispatch + no-op hook
│   ├── _normalizers.py                      # A4 補 4 helper
│   ├── match_method.py                      # A4 加 USER_SELECTED
│   ├── person_reconciler.py                 # A4 retroactive: override _try_user_selected_match
│   ├── organization_reconciler.py           # A4 retroactive: override _try_user_selected_match
│   ├── device_reconciler.py                 # A4 新建
│   ├── information_system_reconciler.py     # A4 新建（接 2 個 domain service）
│   ├── leveraged_reconciler.py              # A4 新建
│   ├── catalog_control_reconciler.py        # A4 新建
│   ├── assessment_objective_reconciler.py   # A4 新建（特殊：reconcile_with_control_map）
│   └── ssp_entity_orchestrator.py           # A4 新建（D1 facade）
│
└── write_strategy/                          ← A4 新增 sub-folder
    ├── __init__.py
    ├── control_write_strategy.py
    ├── ao_write_strategy.py
    ├── device_write_strategy.py
    ├── information_system_write_strategy.py
    └── leveraged_write_strategy.py

domain/oscal/parser/ssp_intermediate.py     # A3 既有 + A4 加 5 dataclass + ParsedParty 加 2 欄位
app/oscal/service/ssp_excel_import_app_service.py    # A4 改 _write_all_data + superset SSP shell
di_containers/oscal/oscal_containers.py              # A4 新增 11 Factory + 加注入

3.2 SspEntityReconciliationOrchestrator shape

@dataclass
class ParsedExcelEntityBundle:
    """5 entity typed list 集合 — 給 orchestrator 用，避免 5 個位置參數。"""
    parsed_devices: List[ParsedDevice]
    parsed_info_systems: List[ParsedInformationSystem]
    parsed_leveraged: List[ParsedLeveraged]
    parsed_controls: List[ParsedControl]   # 含 nested objectives


@dataclass
class SspEntityReconciliationContext:
    tenant_id: int
    catalog_id: int     # superset 用 fw_version.catalog_id；update 從 mf.profile.catalog_id 拿


class SspEntityReconciliationOrchestrator:
    """A4 facade — dispatch 5 reconciler，enforce control → AO 順序。

    Caller wraps in @transaction; this class never opens a session.
    """

    def __init__(
        self,
        catalog_control_reconciler,
        assessment_objective_reconciler,
        device_reconciler,
        information_system_reconciler,
        leveraged_reconciler,
    ):
        self._ctrl = catalog_control_reconciler
        self._ao = assessment_objective_reconciler
        self._device = device_reconciler
        self._info_system = information_system_reconciler
        self._leveraged = leveraged_reconciler

    def reconcile(
        self,
        parsed: ParsedExcelEntityBundle,
        ctx: SspEntityReconciliationContext,
    ) -> None:
        # 1. control 先（D3 二階段 — USER_SELECTED skip + EXACT + NORMALIZED）
        self._ctrl.reconcile(parsed.parsed_controls, ctx.catalog_id)
        # 2. AO（依賴 catalog_control_id）
        ctrl_map = {
            c.control_id: c.matched_catalog_control_id
            for c in parsed.parsed_controls
            if c.matched_catalog_control_id is not None
        }
        parsed_aos = [ao for c in parsed.parsed_controls for ao in c.objectives]
        self._ao.reconcile_with_control_map(parsed_aos, ctrl_map)
        # 3. 3 個獨立 entity (D1)
        self._device.reconcile(parsed.parsed_devices, ctx.tenant_id)
        self._info_system.reconcile(parsed.parsed_info_systems, ctx.tenant_id)
        self._leveraged.reconcile(parsed.parsed_leveraged, ctx.tenant_id)

3.3 Caller / Orchestrator / Reconciler / WriteStrategy 關係（A4 後）

SspExcelImportAppService._confirm_*_flow (A2 既有 + A4 改)
  └─ _write_all_data (A4 重整)  ← @transaction scope (caller 開)
     │
     ├─ Step 1: Party reconcile (A3 既有；A4 retroactive 加 stage 0)
     │  └─ PartyReconciliationService.reconcile(parties)
     │     ├─ PersonReconciler (A4: + _try_user_selected_match)
     │     └─ OrganizationReconciler (A4: + _try_user_selected_match)
     │
     ├─ Step 2: SSP shell — superset flow 建；update flow resolve 既有
     │  ├─ ssp_domain_service.add(SspEntity)
     │  ├─ system_characteristic_domain_service.add(SystemCharacteristic)
     │  └─ system_implementation_domain_service.add(SystemImplementation main)
     │
     ├─ Step 3: parsed dict → typed dataclass (5 個新 _dict_to_*)
     │  ├─ _dict_to_parsed_devices
     │  ├─ _dict_to_parsed_info_systems
     │  ├─ _dict_to_parsed_leveraged
     │  └─ _dict_to_parsed_controls (含 nested ParsedAssessmentObjective)
     │
     ├─ Step 4: A4 Orchestrator reconcile
     │  └─ SspEntityReconciliationOrchestrator.reconcile(bundle, ctx)
     │     ├─ CatalogControlReconciler ← 先
     │     ├─ AssessmentObjectiveReconciler ← 依賴 control_map
     │     ├─ DeviceReconciler
     │     ├─ InformationSystemReconciler (+ system_owner 反查)
     │     └─ LeveragedReconciler
     │
     ├─ Step 5: Party write (A3 既有 ModuleFrameWriteStrategy)
     │  └─ ModuleFrameWriteStrategy.write_parties(...)
     │
     ├─ Step 6: A4 entity write — 5 strategy 順序
     │  ├─ ControlWriteStrategy.write(parsed_controls, ssp_id)
     │  │   回傳 ctrl_impl_id_map (catalog_control_id → ssp_control_implementation_id)
     │  ├─ AoWriteStrategy.write(parsed_aos, ssp_id, ctrl_impl_id_map)
     │  ├─ DeviceWriteStrategy.write(parsed_devices, ssp_id, sys_impl_main_id)
     │  ├─ InformationSystemWriteStrategy.write(parsed_info_systems, ssp_id, sys_impl_main_id)
     │  └─ LeveragedWriteStrategy.write(parsed_leveraged, ssp_id, sys_impl_main_id)
     │
     ├─ Step 7: parsed_result 回填（matched_*_id / match_method 寫回 JSONB）
     │  └─ 給 A5 預覽 UI 讀
     │
     └─ Step 8: import_summary 更新到 parse_job
                                                   全在 @transaction scope 內

3.4 DDD 規範

• Reconciler / WriteStrategy 純 domain layer：不開 @transaction、不直接 import ORM model、只透過 domain service
• @transaction scope 在 caller confirm_import 開（A2 既有）；whole pipeline 同 scope。失敗整段 rollback 屬「import 一致性正確行為」— SSP shell + reconcile + write 任一失敗都應該 rollback
• DI Factory provider：每次 build 新 instance；candidate cache 隨 reconciler instance 生死
• Domain service 跨 container 引用：jedi-device / jedi_information_system / jedi-auth 等套件透過 auth_container / device_container / is_container（沿 A3 line 490/491 pattern）

4.1 MatchMethod enum 擴張

class MatchMethod(StrEnum):
    USER_SELECTED = "user_selected"          # A4 新增 — stage 0 樣板 matched_* col 反查命中
    EXACT = "exact"                           # A3
    NORMALIZED = "normalized"                 # A3
    FUZZY_EMAIL_DOMAIN = "fuzzy_email_domain" # A3 Person only
    FUZZY_NAME_PREFIX = "fuzzy_name_prefix"   # A3 Organization / A4 Leveraged
    UNMATCHED = "unmatched"

4.2 ParsedParty 補 2 個 optional 欄位（A3 retroactive）

@dataclass
class ParsedParty:
    # ... A3 既有欄位（name / party_type / email_address / role / matched_user_id / matched_org_unit_id / ...）
    match_method: MatchMethod = MatchMethod.UNMATCHED  # A3 既有
    match_confidence: float = 0.0                       # A3 既有

    # A4 retroactive 新增
    matched_user_label: Optional[str] = None      # Excel `matched_user` lookup col ('nickname <login_name>')
    matched_org_unit_label: Optional[str] = None  # Excel `org_unit` lookup col (name)

回溯相容性：新 default 為 None，A3 既有 caller / unit test fixture / write_strategy.write_parties 不破壞。

4.3 5 個新 Parsed* dataclass

@dataclass
class ParsedDevice:
    name: str
    ip: Optional[str] = None
    os: Optional[str] = None
    device_type: Optional[str] = None
    status: Optional[str] = None
    purpose: Optional[str] = None
    matched_device_label: Optional[str] = None    # Excel matched_device ('name (ip)')
    # reconcile 產出
    matched_device_id: Optional[int] = None
    match_method: MatchMethod = MatchMethod.UNMATCHED
    match_confidence: float = 0.0


@dataclass
class ParsedInformationSystem:
    name: str
    abbreviation: Optional[str] = None
    description: Optional[str] = None
    component_type: Optional[str] = None
    status: Optional[str] = None
    matched_info_system_label: Optional[str] = None    # Excel matched_info_system ('abbr - name')
    system_owner_label: Optional[str] = None           # Excel system_owner user lookup
    # reconcile 產出
    matched_info_system_id: Optional[int] = None
    matched_system_owner_user_id: Optional[int] = None
    match_method: MatchMethod = MatchMethod.UNMATCHED
    match_confidence: float = 0.0


@dataclass
class ParsedLeveraged:
    service_name: str
    provider: str
    party_label: Optional[str] = None             # Excel party (ORGS lookup, PartyEntity.name)
    date_authorized: Optional[date] = None
    purpose: Optional[str] = None
    # reconcile 產出 (D4 — 配 PartyEntity.uid 給 ssp_system_implementation_items.party_uuid 用)
    matched_party_uuid: Optional[str] = None
    match_method: MatchMethod = MatchMethod.UNMATCHED
    match_confidence: float = 0.0


@dataclass
class ParsedControl:
    control_id: str                              # 'AC-1' (字面 code)
    control_name: Optional[str] = None
    objective_id: Optional[str] = None
    objective_name: Optional[str] = None
    impl_status: Optional[str] = None
    statement: Optional[str] = None
    reference_doc: Optional[str] = None
    include_in_profile: Optional[str] = None     # 'TRUE'/'FALSE' (v1.2.0)
    _target_in_profile: bool = True              # parser 計算
    objectives: List["ParsedAssessmentObjective"] = field(default_factory=list)
    # reconcile 產出 (D3 二階段)
    # 兩個都存：int 給 WriteStrategy FK；str 給 A2 §15.3 superset include_controls 用
    matched_catalog_control_uid: Optional[str] = None    # UUID 字串 (給 ProfileService.add_profile)
    matched_catalog_control_id: Optional[int] = None     # int FK (給 WriteStrategy)
    match_method: MatchMethod = MatchMethod.UNMATCHED
    match_confidence: float = 0.0


@dataclass
class ParsedAssessmentObjective:
    # NOTE: statement_id 是樣板 column key (parser 解出的 AO uid 字串如 'AC-1.a.1')
    # 跟 matched_catalog_control_assessment_id (DB int id) 不要混淆
    statement_id: str                            # parser key (樣板 hidden column)
    control_id: str                              # 父 control_id 'AC-1' (給 reconcile_with_control_map 對父用)
    objective_id: Optional[str] = None
    objective_name: Optional[str] = None
    impl_status: Optional[str] = None
    statement: Optional[str] = None
    reference_doc: Optional[str] = None
    # reconcile 產出
    matched_catalog_control_assessment_uid: Optional[str] = None
    matched_catalog_control_assessment_id: Optional[int] = None
    match_method: MatchMethod = MatchMethod.UNMATCHED
    match_confidence: float = 0.0

4.4 SspEntityReconciliationContext

@dataclass
class SspEntityReconciliationContext:
    tenant_id: int
    catalog_id: int     # superset 用 fw_version.catalog_id；update 從 mf.profile.catalog_id 拿

未來加 context 不破壞 signature。

4.5 ParsedExcel 內仍維持 list[dict]，由 _dict_to_* 轉

維持 A3 ParsedParty 同樣 pattern：parser 不動（仍 list[dict] 寫入 parse_job.parsed_result JSONB），_write_all_data 內 5 個新 _dict_to_parsed_* 函式做轉換。

優點：

• Parser 不破壞
• parsed_result JSONB schema 不破壞（A5 預覽 UI 仍讀 dict 不影響）
• 既有 tests/test_ssp_excel_import_app_service.py 29 個 case fixture 不需重排

4.6 Confidence + 行為差異總表

A4 不做動態 confidence 打分；future-proof 留給 prod feedback。

5.1 共用 helper（_normalizers.py A4 補）

import re

def parse_user_lookup_label(label: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    """'nickname <login_name>' → ('nickname', 'login_name')；不符 fmt 回 (None, None)."""
    if not label or '<' not in label or '>' not in label:
        return (None, None)
    nickname = label.split('<', 1)[0].strip()
    login_name = label.split('<', 1)[1].rstrip('>').strip()
    return (nickname or None, login_name or None)


def parse_device_lookup_label(label: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    """'name (ip)' → ('name', 'ip')；不符 fmt 回 (label, None)."""
    if not label:
        return (None, None)
    if '(' in label and label.endswith(')'):
        name = label.rsplit('(', 1)[0].strip()
        ip = label.rsplit('(', 1)[1].rstrip(')').strip()
        return (name or None, ip or None)
    return (label.strip(), None)


def parse_info_system_lookup_label(label: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    """'abbr - name' → ('abbr', 'name')；無 ' - ' 退回 (None, label)."""
    if not label:
        return (None, None)
    if ' - ' in label:
        abbr, name = label.split(' - ', 1)
        return (abbr.strip() or None, name.strip() or None)
    return (None, label.strip())


def _normalize_control_id(s: Optional[str]) -> str:
    """'ac 1' / 'AC_1' / 'ac-1' → 'AC-1' (upper + space/underscore→dash + collapse)."""
    if not s:
        return ""
    s = s.strip().upper()
    s = re.sub(r'[\s_]+', '-', s)
    s = re.sub(r'-+', '-', s)
    return s

5.2 BaseReconciliationService 加 stage 0 dispatch

class BaseReconciliationService(ABC, Generic[TParsed, TEntity]):
    def _reconcile_one(self, parsed, tenant_id):
        # A4: stage 0 USER_SELECTED
        candidate = self._try_user_selected_match(parsed, tenant_id)
        if candidate is not None:
            self._apply_match(parsed, candidate, MatchMethod.USER_SELECTED, 1.0)
            return
        # 既有 stage 1-3 不動
        candidate = self._try_exact_match(parsed, tenant_id)
        if candidate is not None:
            self._apply_match(parsed, candidate, MatchMethod.EXACT, 1.0)
            return
        candidate = self._try_normalized_match(parsed, tenant_id)
        if candidate is not None:
            self._apply_match(parsed, candidate, MatchMethod.NORMALIZED, 1.0)
            return
        candidate, fuzzy_method = self._try_fuzzy_match(parsed, tenant_id)
        if candidate is not None:
            self._apply_match(parsed, candidate, fuzzy_method, 0.7)
            return
        self._apply_unmatched(parsed)

    def _try_user_selected_match(self, parsed, tenant_id) -> Optional[TEntity]:
        """Default no-op；subclass override 處理 matched_*_label 反查."""
        return None

不是 @abstractmethod — A3 既有 5 hook 的 contract 不變；A4 加的 stage 0 是 optional hook，A3 既有 _try_fuzzy_match 等不需改。

5.3 5 個 Reconciler 行為

5.4 InformationSystemReconciler 附加 system_owner 反查

InformationSystem 跟 user 是不同 domain — 但設計上 InfoSystemReconciler 接 2 個 domain service（info_system + user）；system_owner_label 用簡單 EXACT label 反查（無 fuzzy）寫入 matched_system_owner_user_id，不影響主 entity 的 match_method：

class InformationSystemReconciler(BaseReconciliationService[ParsedInformationSystem, object]):
    def __init__(self, is_domain_service, user_domain_service):
        super().__init__()
        self._is = is_domain_service
        self._user = user_domain_service

    def _apply_match(self, parsed, entity, method, confidence):
        parsed.matched_info_system_id = getattr(entity, "id", None)
        parsed.match_method = method
        parsed.match_confidence = confidence
        self._resolve_system_owner(parsed)   # 不論 main match 結果，都嘗試反查 owner

    def _apply_unmatched(self, parsed):
        parsed.match_method = MatchMethod.UNMATCHED
        parsed.match_confidence = 0.0
        self._resolve_system_owner(parsed)

    def _resolve_system_owner(self, parsed):
        if not parsed.system_owner_label:
            return
        nickname, login_name = parse_user_lookup_label(parsed.system_owner_label)
        if not login_name:
            return
        try:
            users = self._user.get_users(UserQueryEntity(login_name=login_name))
            for u in users or []:
                if getattr(u, "is_active", True):
                    parsed.matched_system_owner_user_id = u.id
                    return
        except Exception:
            pass

5.5 AO Reconciler 依賴 control 的處理

AO 鉤稽需要父 control 的 catalog_control_id（不是字面 code）— orchestrator 順序 enforce control 先 reconcile（D1）。AO Reconciler 走 public method reconcile_with_control_map：

class AssessmentObjectiveReconciler(
    BaseReconciliationService[ParsedAssessmentObjective, object]
):
    def __init__(self, assessment_domain_service):
        super().__init__()
        self._ao_service = assessment_domain_service
        self._current_parent_ctrl_id: Optional[int] = None   # 給 _try_*_match 讀

    def reconcile_with_control_map(
        self,
        parsed_aos: List[ParsedAssessmentObjective],
        ctrl_map: dict[str, int],   # control_id 字面 → matched_catalog_control_id
    ) -> None:
        for ao in parsed_aos:
            parent_ctrl_id = ctrl_map.get(ao.control_id)
            if parent_ctrl_id is None:
                self._apply_unmatched(ao)
                continue
            self._current_parent_ctrl_id = parent_ctrl_id
            self._reconcile_one(ao, tenant_id=None)
            self._current_parent_ctrl_id = None

    def _try_user_selected_match(self, parsed, tenant_id):
        return None   # 樣板無 lookup col

    def _try_exact_match(self, parsed, tenant_id):
        if self._current_parent_ctrl_id is None or not parsed.objective_id:
            return None
        try:
            aos = self._ao_service.get_assessments(
                CatalogControlAssessmentQueryEntity(
                    catalog_control_id=self._current_parent_ctrl_id,
                    objective_id=parsed.objective_id,
                )
            )
            for a in aos or []:
                return a
        except Exception:
            pass
        return None
    # ... _try_normalized_match / _try_fuzzy_match / _apply_match / _apply_unmatched

繼承 BaseReconciliationService 是為了重用三階段 fallback 機制 + candidate cache helper；public reconcile_with_control_map 是新增 method（而非 override reconcile）— 簽章不同因為 AO 需要 ctrl_map context。

5.6 Candidate Cache 行為（沿 A3）

• 範圍：單一 reconcile() call lifetime
• 觸發：第一個 fuzzy 階段才拉（A4 Device / InfoSystem / Control / AO 都不走 fuzzy 三階段，cache 不會 trigger）
• 位置：reconciler instance attribute（base class 的 self._candidate_cache）
• 生命週期：DI Factory provider，每次 build 新 instance → 等於每次 reconcile call 一個全新 cache，不跨 request

6.1 _write_all_data 重整版

def _write_all_data(
    self, parsed_result, target_mf_uid, user_context,
    is_superset_flow: bool, fw_version,
) -> dict:
    # Step 1: party reconcile (A3 既有 + A4 retroactive stage 0)
    all_parties = self._dict_to_parsed_parties(
        parsed_result.get("parties_org") or [],
        parsed_result.get("parties_person") or [],
    )
    if all_parties and self._party_reconciliation is not None:
        try:
            self._party_reconciliation.reconcile(
                all_parties, tenant_id=user_context.tenant_id,
            )
        except Exception as e:
            logger.warning("Party reconcile skipped: %s", e)

    # Step 2: SSP shell — superset flow 建；update flow 走既有
    if is_superset_flow:
        ssp_id, sys_impl_main_id = self._create_ssp_shell(
            mf_uid=target_mf_uid,
            metadata=parsed_result.get("metadata") or {},
            user=user_context.login_name,
        )
    else:
        ssp_id, sys_impl_main_id = self._resolve_existing_ssp_shell(target_mf_uid)

    # Step 3: parsed dict → typed dataclass
    parsed_devices = self._dict_to_parsed_devices(parsed_result.get("devices") or [])
    parsed_info_systems = self._dict_to_parsed_info_systems(parsed_result.get("info_systems") or [])
    parsed_leveraged = self._dict_to_parsed_leveraged(parsed_result.get("leveraged") or [])
    parsed_controls = self._dict_to_parsed_controls(parsed_result.get("controls_with_aos") or [])

    # Step 4: A4 orchestrator reconcile
    catalog_id = self._resolve_catalog_id(fw_version, target_mf_uid, is_superset_flow)
    ctx = SspEntityReconciliationContext(
        tenant_id=user_context.tenant_id, catalog_id=catalog_id,
    )
    bundle = ParsedExcelEntityBundle(
        parsed_devices=parsed_devices,
        parsed_info_systems=parsed_info_systems,
        parsed_leveraged=parsed_leveraged,
        parsed_controls=parsed_controls,
    )
    if self._ssp_entity_orchestrator is not None:
        try:
            self._ssp_entity_orchestrator.reconcile(bundle, ctx)
        except Exception as e:
            logger.warning("Entity reconcile skipped: %s", e)

    # Step 5: party write (A3 既有)
    parties_written = self._mf_write.write_parties(
        parties=all_parties, source_uid=target_mf_uid,
        user_id=user_context.login_name,
    )

    # Step 6: A4 entity write — 5 strategy 順序
    ctrl_written, ctrl_impl_id_map = self._control_write.write(
        parsed_controls, ssp_id, user_context,
    )
    parsed_aos = [ao for c in parsed_controls for ao in c.objectives]
    ao_written = self._ao_write.write(
        parsed_aos, ssp_id, ctrl_impl_id_map, user_context,
    )
    devices_written = self._device_write.write(
        parsed_devices, ssp_id, sys_impl_main_id, user_context,
    )
    info_systems_written = self._info_system_write.write(
        parsed_info_systems, ssp_id, sys_impl_main_id, user_context,
    )
    leveraged_written = self._leveraged_write.write(
        parsed_leveraged, ssp_id, sys_impl_main_id, user_context,
    )

    # Step 7: parsed_result 回填（matched_*_id / match_method 寫回 JSONB 給 A5 UI 讀）
    self._update_parsed_result_with_reconcile(
        parsed_result, all_parties, bundle,
    )

    # Step 8: 統計
    return {
        "ssp_id": ssp_id,
        "sys_impl_main_id": sys_impl_main_id,
        "parties_written": parties_written,
        "controls_written": ctrl_written,
        "aos_written": ao_written,
        "devices_written": devices_written,
        "info_systems_written": info_systems_written,
        "leveraged_written": leveraged_written,
    }

6.2 SSP shell 建立（superset flow only）

def _create_ssp_shell(self, mf_uid: str, metadata: dict, user: str) -> tuple[int, int]:
    """Superset flow 建 SSP placeholder + system_characteristic + system_implementation main.

    Caller 必須在 @transaction scope 內呼叫。
    NOTE: 本段 jedi-oscal Entity 欄位由 implementation T0 verify 後微調（design 階段假設）。
    """
    mf = self._mf_domain.get_module_frame_by_uid(mf_uid)

    # 2a — 建 SSP
    ssp_dto = self._ssp_domain.add(SystemSecurityPlanEntity(
        title=metadata.get("target_mf_name") or f"SSP for MF {mf_uid[:8]}",
        profile_uid=mf.oscal_profile_uid,
        status='draft',
    ), curr_user=user)

    # 2b — 建 system_characteristic
    self._system_characteristic_domain.add(SspSystemCharacteristicEntity(
        system_security_plan_id=ssp_dto.id,
        name=metadata.get("target_system_name") or ssp_dto.title,
    ), curr_user=user)

    # 2c — 建 system_implementation main
    impl_main_dto = self._system_implementation_domain.add(SspSystemImplementationEntity(
        system_security_plan_id=ssp_dto.id,
        scope_type='ssp', scope_id=ssp_dto.id,
    ), curr_user=user)

    return (ssp_dto.id, impl_main_dto.id)

6.3 _resolve_existing_ssp_shell（update flow）

def _resolve_existing_ssp_shell(self, mf_uid: str) -> tuple[int, int]:
    """Update flow — 既有 MF 對應 SSP 必須已存在；否則 raise.

    Update flow 假設 MF 已有 SSP（透過 ssp_versioning_service 等其他 caller 建立）。
    若沒有 — raise PreconditionFailedError，明確指出「import flow 不負責建第 1 版」。
    """
    # 透過 mf_uid → profile_uid → ssp_query 找對應 SSP
    ssp = self._ssp_domain.get_one_by_profile_uid(mf.oscal_profile_uid)
    if ssp is None:
        raise PreconditionFailedError(GrcErrorCode.GRC_EXCEL_UPDATE_FLOW_NO_SSP)
    # 找 system_implementation main
    impl_main = self._system_implementation_domain.get_one_by_ssp_id(ssp.id)
    if impl_main is None:
        raise PreconditionFailedError(GrcErrorCode.GRC_EXCEL_UPDATE_FLOW_NO_SSP)
    return (ssp.id, impl_main.id)

新增 error code：

# common/code/grc_error_code.py
GRC_EXCEL_UPDATE_FLOW_NO_SSP = (
    "Excel update flow 失敗：對應 MF 尚未建立 SSP",
    "GRC_412066",
)

6.4 A2 §15.3 整併 — superset include_controls 反查

def _confirm_superset_flow(self, job, parsed_result, payload, user_context):
    fw_version = self._fw_version_domain.get_by_uid(job.source_uid)

    target_control_ids = [
        c.get("control_id") for c in (parsed_result.get("controls_with_aos") or [])
        if c.get("statement_id") is None
        and c.get("_target_in_profile") is True
        and c.get("control_id")
    ]

    # A4 D6 整併：先反查 catalog_control_uid
    parsed_for_resolve = [ParsedControl(control_id=cid) for cid in target_control_ids]
    if self._catalog_control_reconciler is not None and parsed_for_resolve:
        self._catalog_control_reconciler.reconcile(
            parsed_for_resolve, fw_version.catalog_id,
        )
    include_uids = [
        str(p.matched_catalog_control_uid) for p in parsed_for_resolve
        if p.matched_catalog_control_uid
    ]

    mf_payload = {
        # ... 既有
        "include_controls": include_uids,   # ← A4 真正 subset profile
    }
    mf_dto = self._mf_service.add_module_frame(...)

    write_stats = self._write_all_data(
        parsed_result=parsed_result,
        target_mf_uid=mf_dto.uid,
        user_context=user_context,
        is_superset_flow=True,
        fw_version=fw_version,
    )

    # ... 收 summary 回應

6.5 A3 retroactive — _dict_to_parsed_parties 補 label 傳遞

@staticmethod
def _dict_to_parsed_parties(orgs, persons) -> list[ParsedParty]:
    result = []
    for o in orgs:
        name = (o.get("name") or "").strip()
        if not name:
            continue
        result.append(ParsedParty(
            name=name, party_type="organization",
            role=o.get("role"), short_name=o.get("short_name"),
            email_address=o.get("email"),
            matched_org_unit_label=o.get("parent_org"),    # A4 retroactive
        ))
    for p in persons:
        name = (p.get("name") or "").strip()
        email = (p.get("email") or "").strip()
        if not name and not email:
            continue
        result.append(ParsedParty(
            name=name or email, party_type="person",
            role=p.get("role"), email_address=email or None,
            matched_user_label=p.get("matched_user"),       # A4 retroactive
            matched_org_unit_label=p.get("org_unit"),       # A4 retroactive
        ))
    return result

6.6 5 個 WriteStrategy 行為

UNMATCHED 仍寫入（非 control/AO）的理由：

• 保留 user 在樣板填的純文字資料（name / ip / description）
• A5 UI 可顯示「unmatched」項目並讓 user 後補配對 / 改純文字 / 新建系統 entity
• 比 skip 安全 — skip 等於 user 資料直接遺失，UX 差
• Control/AO 例外是因為沒對應 catalog_control_id 寫進去不合 OSCAL 規範（catalog_control_id NOT NULL FK）

6.7 DI Container 變動清單

di_containers/oscal/oscal_containers.py 新增 11 個 Factory + SspExcelImportAppService 加 11 個依賴：

# 5 個 reconciler Factory
catalog_control_reconciler         = providers.Factory(CatalogControlReconciler, ...)
assessment_objective_reconciler    = providers.Factory(AssessmentObjectiveReconciler, ...)
device_reconciler                  = providers.Factory(DeviceReconciler, ...)
information_system_reconciler      = providers.Factory(
    InformationSystemReconciler,
    is_domain_service=is_container.information_system_domain_service,
    user_domain_service=auth_container.user_domain_service,
)
leveraged_reconciler               = providers.Factory(LeveragedReconciler, ...)

# Orchestrator Factory
ssp_entity_reconciliation_orchestrator = providers.Factory(
    SspEntityReconciliationOrchestrator,
    catalog_control_reconciler=catalog_control_reconciler,
    assessment_objective_reconciler=assessment_objective_reconciler,
    device_reconciler=device_reconciler,
    information_system_reconciler=information_system_reconciler,
    leveraged_reconciler=leveraged_reconciler,
)

# 5 個 WriteStrategy Factory
control_write_strategy             = providers.Factory(ControlWriteStrategy, ...)
ao_write_strategy                  = providers.Factory(AoWriteStrategy, ...)
device_write_strategy              = providers.Factory(DeviceWriteStrategy, ...)
information_system_write_strategy  = providers.Factory(InformationSystemWriteStrategy, ...)
leveraged_write_strategy           = providers.Factory(LeveragedWriteStrategy, ...)

# SspExcelImportAppService 加注入（從 9 個變 ~20 個）
ssp_excel_import_app_service = providers.Factory(
    SspExcelImportAppService,
    # ... 既有 9 個（parse_job / parser / file_upload / mf_domain / mf_service / fw_version_domain / mf_write / party_reconciliation_service）
    ssp_entity_orchestrator=ssp_entity_reconciliation_orchestrator,
    catalog_control_reconciler=catalog_control_reconciler,
    ssp_domain_service=ssp_domain_service,
    system_characteristic_domain_service=system_characteristic_domain_service,
    system_implementation_domain_service=system_implementation_domain_service,
    control_write_strategy=control_write_strategy,
    ao_write_strategy=ao_write_strategy,
    device_write_strategy=device_write_strategy,
    information_system_write_strategy=information_system_write_strategy,
    leveraged_write_strategy=leveraged_write_strategy,
)

6.8 Cucumber Regression（compliance-manager-test repo）

新建 features/regression/module-frame/06-ssp-excel-import-entity-match.feature 含 5 scenarios：

沿 A3 partial ship pattern — env 配齊後 fixture seeding + 跑通即 ship；env 未配齊允許 partial ship（follow-up F-A4-cucumber-extra）。

7.1 已驗證項目（brainstorm 收斂前 Pre-flight 完成）

7.2 待 implementation T0 verify 項目

Implementation plan T0 必補上述 verify；任一不符需 design.md §11 補偏差紀錄。Front-load critical items：#5（control_implementation Mapper signature 影響 Session E）+ #7（ssp_domain_service.get_one_by_profile_uid 可能不存在）優先驗，避免 mid-session rework。

• 新建 domain/oscal/service/reconciliation/{device,information_system,leveraged,catalog_control,assessment_objective}_reconciler.py 5 個檔
• 新建 domain/oscal/service/reconciliation/ssp_entity_orchestrator.py
• MatchMethod 加 USER_SELECTED 值
• _normalizers.py 補 parse_user_lookup_label / parse_device_lookup_label / parse_info_system_lookup_label / _normalize_control_id
• BaseReconciliationService._reconcile_one 加 stage 0 dispatch + _try_user_selected_match default no-op hook
• ssp_intermediate.py 加 5 個新 dataclass（ParsedDevice / ParsedInformationSystem / ParsedLeveraged / ParsedControl / ParsedAssessmentObjective）+ ParsedParty 加 2 optional label 欄位
• ParsedExcelEntityBundle + SspEntityReconciliationContext 新建
• SspEntityReconciliationOrchestrator reconcile 順序：control → AO（依 ctrl_map）→ device / info_system / leveraged
• 新建 domain/oscal/service/write_strategy/ sub-folder + 5 個 WriteStrategy
• _write_all_data 重整為 8 step pipeline
• _create_ssp_shell 在 superset flow 建 SSP + system_characteristic + system_implementation main（3 jedi-oscal domain service 呼叫）
• _resolve_existing_ssp_shell 在 update flow 找既有 SSP；找不到 raise GRC_EXCEL_UPDATE_FLOW_NO_SSP
• A2 §15.3 superset include_controls 反查整併（_confirm_superset_flow 內呼叫 catalog_control_reconciler）
• A3 retroactive：PersonReconciler / OrganizationReconciler override _try_user_selected_match；_dict_to_parsed_parties 補 label 傳遞
• DI Container 新增 11 Factory + SspExcelImportAppService 加 ~11 個依賴
• 新增 unit test：reconciler 60-75 + WriteStrategy 30-40 + integration 15-20 + ParsedXxx fixture 10 = ~110-140 個
• 既有 A3 67 個 reconciliation / fixture test 全綠（A4 stage 0 加入後不破壞）
• 既有 A2 29 個 test_ssp_excel_import_app_service.py fixture 全綠（A4 加新 dep optional default None，舊 fixture 不破壞）
• Cucumber regression 06-ssp-excel-import-entity-match.feature 5 scenarios（4 + 1 fuzzy）撰寫完成；partial ship 允許
• Changelog YYYY-MM-DD-feat-ssp-entity-reconcile-and-write.md（type: feat、modules: oscal）完成
• docs/features/FR-011.2-2605-ssp-import-export-phase2/README.md tracker A4 row 標 BE shipped ✅ + commit chain
• BE 既有 694 test 全綠 + A4 新增 ~110-140 test → 累計 ~804-834

（實作過程中對原始 design 的偏離 / 擴增紀錄；A4 Session 最後一個 session 收口時統一補。）

11.1 Session C — T0 pre-flight verify 偏差紀錄（2026-05-20）

T0 8 項 verify 結果與 design §7.2 假設對照：

11.1.1 jedi-oscal SspEntity 欄位偏差（T0.1 — 影響 T6/Session E）

• design §6.2 假設 SystemSecurityPlanEntity(title=..., profile_uid=..., status='draft')
• 實際 SystemSecurityPlanEntity.__init__ 欄位（jedi-oscal domain/entity/ssp/ssp_entity.py）：
  • description: str = ""（非 title）
  • profile_id: Optional[int]（非 profile_uid 字串）
  • 無 system_security_plan_main_id 欄位 — A0.1 改用 relation system_implementation_main: Optional[SspSystemImplementationEntity]
• T6 影響：_create_ssp_shell step 2a 改寫
  • 用 description=metadata.get("target_mf_name") or f"SSP for MF {mf_uid[:8]}"
  • 用 profile_id=mf.oscal_profile_id（caller 需用 profile_id 而非 profile_uid；MF 載入時取 int）
• 不影響 Session C

11.1.2 SystemCharacteristicEntity class name + 欄位（T0.2 — 影響 T6/Session E）

• design §6.2 用 SspSystemCharacteristicEntity
• 實際 class name SystemCharacteristicEntity（無 Ssp 前綴）— jedi-oscal domain/entity/ssp/ssp_system_characteristic_entity.py
• 欄位齊全：name / system_security_plan_id / description / system_identifier / security_sensitivity_level / target_type / scope_description / status / owner_uid ✅
• T6 影響：import + class name 改 SystemCharacteristicEntity
• 不影響 Session C

11.1.3 CatalogControlAssessmentQueryEntity 無 objective_id 欄位（T0.4 — ⚠️ 影響 T3/Session D，Session D 收口補正）

• design §5.3 EXACT stage 假設用 CatalogControlAssessmentQueryEntity(catalog_control_id=..., objective_id=...)
• 實際 QueryEntity 欄位（jedi-oscal catalog_control_assessment_query_entity.py）：
  • id / uid / catalog_id / catalog_control_id / control_group_id / name / version / description
  • 無 objective_id 欄位
• Session C T0 verify 當時推薦 Option B：拉 catalog_control_id 範圍 + Python filter statement_identifier
• Session D T3.5 實作時再驗證後更正（2026-05-20）：
  • CatalogControlAssessmentEntity ORM model + Entity 本身也沒有 statement_identifier 欄位
  • 但 entity 的 name 欄位本身就是 OSCAL statement-id 字面（樣板 hidden column key 如 'AC-1.a.1' 直接 stored 在 name 欄）
  • 採 Option C（最終實作）：CatalogControlAssessmentQueryEntity(catalog_control_id=parent, name=parsed.objective_id) 直接 query
  • 比 Option B 更乾淨：無需 lazy load + Python filter；直接走 QueryEntity AND filter（既有 BaseRepositoryImpl 行為）
• NORMALIZED stage 同邏輯：parsed.objective_id.strip().upper() 後二次 query；statement_id 全 upper 是 OSCAL 慣例
• 不影響 Session C（T2.5 skeleton 全 return None）；Session D T3.5 ship 後 §11.1.3 偏差收口

11.1.4 ControlImplementationEntity class name（T0.5 — 影響 T7/Session E）

• design §6.6 假設 class name SspControlImplementation / SspControlImplementationEntity
• 實際 ControlImplementationEntity（無 Ssp 前綴）— jedi-oscal domain/entity/ssp/ssp_control_implementation_entity.py
• 欄位齊全：system_security_plan_id / catalog_control_id / implementation_status / implementation_description / responsible_role / control_identifier / control_origination / remarks / objectives ✅
• DI Container 已注入 control_implementation_domain_service（oscal_containers.py:251-254）✅
• T7 影響：import + class name 改正
• 不影響 Session C

11.1.5 ControlImplementationObjectiveEntity 欄位偏差（T0.6 — 影響 T7/Session E）

• design §6.6 假設 upsert key (ssp_control_implementation_id, catalog_control_assessment_id)
• 實際 ControlImplementationObjectiveEntity 欄位（jedi-oscal ssp_control_impl_objective_entity.py）：
  • system_security_plan_id / control_implementation_id / control_identifier / statement_identifier / implementation_status / implementation_description / remarks / reference_documents
  • 無 catalog_control_assessment_id 欄位（用 statement_identifier 字串對應 statement_id）
• T7 影響：AoWriteStrategy.write upsert key 改用 (control_implementation_id, statement_identifier)；不需 catalog_control_assessment_id FK
• 連動：AO Reconciler 的 matched_catalog_control_assessment_id 在 WriteStrategy 內派不上用場（statement_identifier 字串才是 WriteStrategy 真正用的 key）— 但 Reconciler 仍寫 matched_id 給 A5 UI 顯示，無 harm
• 不影響 Session C

11.1.6 ssp_domain_service.get_one_by_profile_uid 不存在（T0.7 — ⭐ critical，影響 T6/Session E）

• design §6.3 假設 self._ssp_domain.get_one_by_profile_uid(mf.oscal_profile_uid)
• 實際 SystemSecurityPlanDomainService 只有 get_one(query_entity) / get_by_uid(uid) / get_by_id(id)
• 但 SystemSecurityPlanQueryEntity 有 profile_id 欄位 ✅
• T6 影響：_resolve_existing_ssp_shell 改用：

  ssp = self._ssp_domain.get_one(SystemSecurityPlanQueryEntity(profile_id=mf.oscal_profile_id))

• 連動：MF 需提供 oscal_profile_id（int）而非 oscal_profile_uid（uuid 字串）— caller 從 MF 載入時取 int FK
• 不影響 Session C

11.1.7 ssp_domain_service.add() 簽章偏差（補測 — 影響 T6/Session E）

• design §6.2 寫 self._ssp_domain.add(SystemSecurityPlanEntity(...), curr_user=user)
• 實際 add(_entity: SystemSecurityPlanEntity) -> SystemSecurityPlanEntity 只接 1 個 entity 參數
• T6 影響：curr_user 透過 entity 內 created_user=user 帶入；不傳 kwarg
• 不影響 Session C

11.1.8 device_container wiring（T2 補 — 影響 Session C T2.7）

• T0 額外 verify：device_container 存在於 di_containers/device/device_containers.py；containers.py:72 已註冊全域 device_container = providers.Container(DeviceContainer, ...)
• 其他 container（associations / module_frame / ai_dashboard）已透過 providers.DependenciesContainer() 接 device_container
• T2 影響：oscal_containers.py 既有 DependenciesContainer 清單需新增 device_container = providers.DependenciesContainer()；containers.py 既有 oscal_container = providers.Container(OscalContainer, ..., device_container=device_container) wiring 也要補
• Session C T2.7 正式 wire

11.1.9 T0.3 / T0.8 ✅ 符合 design 假設

• T0.3: SspSystemImplementationEntity 含 scope_type / scope_id / remarks / props_jsonb ✅
• T0.8: DeviceQueryEntity 有 name + ip 欄位；repo 繼承 BaseRepositoryImpl[DeviceEntity, DeviceQueryEntity, Device, DeviceMapper] → get_all_by_fields 走 AND filter（jedi-common 標準）✅
• T3 開工前再寫 1 個 integration smoke 驗 DeviceQueryEntity(name=..., ip=...) 實際走 AND（避免被某些客製 repo 行為打臉），但 design 假設 OK

11.1.10 Error code 序號偏差（E1 — Session E 補，影響 T6.1）

• plan §Task 6 / design §6.3 寫 GRC_412066 為 placeholder
• 實際 common/code/grc_error_code.py GRC_412 區段最大已用 GRC_412025（GRC_FLOW_BINDING_DI_MISSING）
• 落地：GRC_EXCEL_UPDATE_FLOW_NO_SSP = ("Excel update flow 失敗：對應 MF 尚未建立 SSP", "GRC_412026")
• plan 內 412066 是 spec 階段預留樣板，未來序號可能搬位；以 ErrorCode 命名規則「序號於同 HTTP 狀態碼下遞增」為準

11.1.11 SspSystemImplementationEntity link 走 scope_type/scope_id（E2 — Session E 補，影響 T6.3 + T8._resolve_catalog_id）

• design §6.3 假設 self._system_implementation_domain.get_one_by_ssp_id(ssp.id)
• 實際 SspSystemImplementationDomainService 無 get_one_by_ssp_id；entity / QueryEntity 都無 system_security_plan_id 欄位
• 真實 link 走 A0.1 ship 的設計：SspSystemImplementationEntity(scope_type='ssp', scope_id=ssp.id)（T0.3 已驗 entity 含 scope_type / scope_id ✅）
• T6 落地：

  impl_main = self._system_impl_main_domain.get_one(
      SspSystemImplementationQueryEntity(scope_type="ssp", scope_id=ssp.id)
  )

• 連動：_create_ssp_shell 同樣傳 scope_type='ssp', scope_id=ssp_dto.id 給 entity；不寫 system_security_plan_id（不存在）

11.1.12 DI 名 system_implementation_main_domain_service（E3 — Session E 補，影響 T6.4 + T8 DI wiring）

• plan §Task 8.3 / design §6.7 寫 system_implementation_domain_service
• 實際 DI Container（di_containers/oscal/oscal_containers.py:256）名為 system_implementation_main_domain_service（區別於 system_implementation_item_domain_service）
• 落地：SspExcelImportAppService.__init__ 參數名跟 attribute 都用 system_implementation_main_domain_service / self._system_impl_main_domain；DI Factory wiring 用實際名

11.1.13 ModuleFrameEntity 無 oscal_profile_id 欄位（E4 — ⭐ critical，Session E 補，影響 T6.2/6.3 + T8._resolve_catalog_id）

• plan / design §6.2 假設 mf.oscal_profile_id（int FK）
• 實際 ModuleFrameEntity 只有 oscal_profile_uid（str）+ oscal_profile（OscalProfileSummaryEntity relation；含 id / catalog_id）
• 連動：ModuleFrameRepoImpl.get_by_uid 走 base BaseRepositoryImpl.get_by_uid，不 enrich oscal_profile relation（只在 list_* / get_all paths 才走 _enrich_with_oscal）
• 落地：T6 + T8 新加 profile_domain_service 注入 SspExcelImportAppService；走兩段式：

  profile = self._profile_domain.get_by_uid(mf.oscal_profile_uid)
  profile_id = profile.id if profile else None
  catalog_id = profile.catalog_id if profile else None  # _resolve_catalog_id update flow

• 連動：design §6.7 DI 變動清單從 11 個 dep → 實際 12 個 dep（多 profile_domain_service）

11.1.14 AbstractSspImplementationItemWriteStrategy base class（E5 — Session E 擴增，plan §Task 7 未列）

• plan §Task 7 列 5 個獨立 WriteStrategy class（Device / InfoSystem / Leveraged / Control / AO 各自寫）
• Session E 開工時 user 拍板加速策略：Device / InfoSystem / Leveraged 三族都寫同一張 ssp_system_implementation_items 表，差別只在 implementation_type (kind) + parsed entity payload — 抽 AbstractSspImplementationItemWriteStrategy base class 共用 80% pattern（loop / count / log / entity 建構樣板）
• 落地：
  • domain/oscal/service/write_strategy/base.py — base class
  • 3 個 subclass 各 ~10 行：override kind property + _parsed_to_item_payload(parsed) -> dict
  • Control / AO 仍各自獨立寫 (跨兩張 table)，不繼承 base
• 效益：5 strategy 順做 ~60 min → base + 5 subclass ~30-40 min；0 整合風險

11.1.15 Step 7 _update_parsed_result_with_reconcile 走 by-key 對齊（E6 — Session E 落地，plan §8.2 stub 補完整）

• plan §Task 8.2 step 3 寫 pass # T8 step 3 補完整邏輯（A5 phase 補完整）
• Session E 落地完整實作（A5 UI 預讀 matched_*_id / match_method 需要）：
  • 新 module-level _iter_paired(dict_list, parsed_list, key) helper：by key 欄位（name / control_id / service_name / statement_id）對齊回填，避免 by-index 在 _dict_to_* skip empty row 後失準
  • devices / info_systems / leveraged / 父 controls 各自 by-key 對齊
  • AOs by statement_id 從 pc.objectives 平面查找後對齊
• 不影響 A4 ship；給 A5 UI 早做準備

11.1.16 ssp_excel_import_app_service DI dep 從 11 個 → 12 個（連動 E4）

• design §6.7 DI 變動清單列「11 個依賴」
• 實際 ship dep 數：
  • A2 既有 8 個（parse_job / parser / file_upload / mf_domain / mf_service / fw_version_domain / mf_write）+ party_reconciliation_service + catalog_control_reconciler
  • T6 新增 4 個（ssp_domain / system_characteristic_domain / system_implementation_main_domain / profile_domain_service）
  • T8 新增 6 個（ssp_entity_orchestrator + 5 WriteStrategy）
• 合計 19 個（含 8 個 A2 既有），其中 A4 phase 新加 11 個 — 跟 design §6.7 帳對得起來
• 但 design §6.7 表列「11 個 Factory + ~11 個依賴」與實際差距：profile_domain_service 額外 1 個 = 12 個 A4 dep
• 落地 attribute：self._profile_domain 給 _create_ssp_shell / _resolve_existing_ssp_shell / _resolve_catalog_id 共用

11.2 後續 session 偏差預留段落

• T1/T2 實作偏差（Session C 收尾補）✅
• T3-T5 偏差（Session D 收尾補）✅
• T6-T8 偏差（Session E 收尾補）✅ — §11.1.10-11.1.16 共 7 條
• T9-T11 偏差（Session F 收尾補）✅（概要見下方 §11.3）
• DDD 邊界調整紀錄
• WriteStrategy 內部演算法細節 vs design 假設

11.3 Bugfix Session 3 — Excel update flow 覆蓋語義修補（2026-05-21）

觸發：A5 smoke 後 user 發現 re-import Excel（update flow）時舊資料沒有被覆蓋

11.3.1 WriteStrategy.write() 只 add 不刪 — 重複 row

• design 假設：WriteStrategy 寫入 devices / info_systems / leveraged 時覆蓋語義
• 實際：AbstractSspImplementationItemWriteStrategy.write() 每次都 add()，re-import 後 row 數倍增
• 修正：write() 開頭先 get_all(ssp_id) 篩出同 implementation_type 既有 rows，delete_by_ids() 刪除後再 INSERT
• 檔案：domain/oscal/service/write_strategy/base.py

11.3.2 write_excel_control_defaults() 只 upsert 不刪

• design 假設：control defaults 整份覆蓋
• 實際：write_excel_control_defaults() 只做 upsert，取消勾選的控制項 default 永久殘留
• 修正：整份覆蓋語義 — 先 _delete_all_control_defaults_for_mf() 清空此 MF 所有 control defaults + objectives + refdoc mappings，再只寫 _target_in_profile=True 的控制項
• 檔案：domain/oscal/strategy/module_frame_write_strategy.py

11.3.3 write_parties() 只 upsert 不刪 stale links

• design 假設：parties 整份覆蓋（移除的人員應刪除 link）
• 實際：write_parties() 只 upsert，不在 Excel 內的舊 responsible_party link 永久保留
• 修正：先快照既有 links，upsert 後刪除 UUID 不在本次 Excel 清單內的 stale links
• 檔案：domain/oscal/strategy/module_frame_write_strategy.py

11.3.4 _confirm_update_flow 沒有更新 oscal_profile.include_controls

• design 假設：template-edit 頁面的控制項清單應隨 Excel 匯入更新
• 實際：template-edit 頁面的控制項清單讀自 oscal_profile.include_controls（非 module_frame_control_defaults），但 _confirm_update_flow 沒有更新 profile
• 修正：新增 _resolve_include_uids_for_update()，在 update flow 結束前呼叫 _mf_service.update_module_frame(include_controls=include_uids) 同步更新 profile
• 檔案：app/oscal/service/ssp_excel_import_app_service.py

11.3.5 _confirm_update_flow 沒有更新 MF 基本資料

• design 假設：re-import 應更新名稱 / 類別 / 描述等基本資料
• 實際：_confirm_update_flow 沒有呼叫 update_module_frame()，基本資料維持匯入前的舊值
• 修正：合併 parsed_result['metadata']（Excel 值）與 job.metadata（form 值）後，呼叫 update_module_frame()。關鍵：必須傳 locale（從 user_context.locale 取，fallback zh_Hant_TW），否則只更新 module_frames 主表而不更新 module_frame_trans 翻譯表，讀回來仍是舊的翻譯值
• 檔案：app/oscal/service/ssp_excel_import_app_service.py

11.3.6 FE 補項（不在 A4 scope 但連動）

• confirm dialog：update flow 增加「確認覆蓋現有資料」PrimeVue confirm dialog（ImportExcelPreviewPage.vue）
• parties cache 失效：import 成功後 router.push 前呼叫 invalidatePartiesCache(mfUid) 清 module-scope 永久 cache，避免導回 template-edit 顯示舊資料（useModuleFrameParties composable）

總計：6 session、~7.5d（含 design / plan / 5 reconciler + 5 WriteStrategy + retroactive + cucumber + 收尾）

換 session 收尾規範（每 session 結束時）

按 CLAUDE.md「做 summary 觸發完整收尾」段：

1. 盤點 commits + working tree 乾淨
2. 規範文件齊全度檢查（changelog / analysis / issue / design §11 reconciliation）
3. 產 handoff prompt → docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/YYYY-MM-DD-a4-<session 字母 + 階段>-next.md
4. Session F 是 task arc 收口 → 對話歷史 dump 到 docs/conversation-history/<date>/ssp-import-export-phase2-A4/

13.1 D1~D8 拍板（22 條 question 收斂結果）

13.2 13 條 sub-question 預設拍板（user §1~§5 review 時口頭授權 by C 選項）

下一步：spec review loop（dispatch spec-document-reviewer subagent）→ user review → 換 Session B invoke writing-plans 寫 implementation-plan-A4.md。