# SSP OSCAL Alignment — Phase 2 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 把 SSP Excel + Docx 兩條 import path 都 wire 到統一的 import_pipeline 模組（warnings → normalizer → confirm service → orchestrator → v3 strategies），讓未來欄位 / parser 異動只需 1 處改動。 **Architecture:** 新建 `domain/oscal/import_pipeline/` 共用模組（`warnings.py` / `normalizer.py` / `confirm_service.py` / `bundle_restore.py`），Excel + Docx adapters 都輸出 `ParsedExcelEntityBundle` v3，confirm_service 按 `schema_version` 分流（`v2-bundle` 走新 path、`v1-ssp` legacy in-flight job 仍走舊 `_legacy_confirm`）。Phase 1 已 ship 的 strategies + orchestrator 還在消費 v2 dataclasses（`ParsedDevice`/`ParsedInformationSystem`/`ParsedLeveraged`），Phase 2 一併 refactor 到 v3 entity dataclasses（`ParsedComponent`/`ParsedLeveragedAuthorization`/`ParsedInventoryItem`）。 **Tech Stack:** Python 3 / Flask / SQLAlchemy 2.0 (`Mapped[]`) / dependency-injector / pytest / Poetry path-dep / PostgreSQL RLS / `jedi-oscal` (path mode, 0.0.18) / `jedi-common` `@transaction`. **Deploy Unit:** Phase 1 + Phase 2 同一 deploy（per `design.md` §3）。Phase 2 ship 時 jedi-oscal `0.0.18 → 0.1.0` bump + push Nexus（Task 13 Phase 1 延後到此 phase 收尾才做）。 **Pre-flight (controller 必先驗，per `design.md` §11.11)：** 1. 兩 repo 同 branch：`feature/ssp-oscal-alignment` 2. Dev DB Phase 1 final state：la=12, comp=341, inv=0, old_main_exists=f 3. `pyproject.toml` 主專案 `M` 是 path-dep（intentional，永不 stage） 4. pytest 主專案 102 / jedi-oscal 84（per Phase 1 mini-handoff） **錨點文件（每個 Task 開工先讀）：** - `docs/features/FR-028-2605-ssp-oscal-alignment/design.md` §11.10 §11.11 — Phase 1 reality bake-in - `docs/features/FR-028-2605-ssp-oscal-alignment/handoff/2026-05-24-phase1-COMPLETE-SUMMARY.md` §6 — Phase 1 follow-ups - 本文件對應 Task 段 --- ## File Structure ### 新建檔案 ``` domain/oscal/import_pipeline/ # ← 新模組（共用層） ├── __init__.py ├── warnings.py # ImportWarnings collector ├── normalizer.py # ParsedBundleNormalizer ├── bundle_restore.py # dict → ParsedExcelEntityBundle 轉換 └── confirm_service.py # schema_version 分流 + 共用 confirm 入口 domain/oscal/service/write_strategy/ ├── component_write_strategy.py # ← 新（取代 device + information_system） └── inventory_item_write_strategy.py # ← 新（含 M2M 寫入） tests/ ├── test_import_warnings.py # ← 新 ├── test_parsed_bundle_normalizer.py # ← 新 ├── test_bundle_restore.py # ← 新 ├── test_ssp_import_confirm_service.py # ← 新 ├── test_component_write_strategy.py # ← 新 ├── test_inventory_item_write_strategy.py # ← 新 ├── test_cmmc_ssp_adapter_v3_bundle.py # ← 新（adapter v3 output） └── test_ssp_import_pipeline_smoke.py # ← 新（end-to-end smoke） ``` ### 修改檔案 ``` domain/oscal/parser/ssp_intermediate.py # 加 ImportWarning dataclass; 留註解標 deprecate domain/oscal/adapter/cmmc_ssp_adapter.py # 加 adapt_to_bundle() domain/oscal/service/reconciliation/ssp_entity_orchestrator.py # reconcile() 讀 v3 bundle fields domain/oscal/service/write_strategy/leveraged_write_strategy.py # consume ParsedLeveragedAuthorization domain/oscal/service/write_strategy/device_write_strategy.py # 刪除（被 ComponentWriteStrategy 取代） domain/oscal/service/write_strategy/information_system_write_strategy.py # 同上 domain/oscal/service/write_strategy/base.py # AbstractSspComponentWriteStrategy `sys_impl_main_id` 參數移除 app/oscal/service/ssp_excel_import_app_service.py # 切到 confirm_service + 新 bundle restore app/oscal/service/ssp_docx_import_app_service.py # 切到 confirm_service + bundle adapter di_containers/oscal/oscal_container.py # wire 新 services + 移除舊 device/info_system strategy providers ``` ### 整批刪除（Phase 2 末段，搬完所有 caller 後） ``` domain/oscal/service/write_strategy/device_write_strategy.py domain/oscal/service/write_strategy/information_system_write_strategy.py # (ssp_intermediate.py 內 ParsedDevice / ParsedInformationSystem / ParsedLeveraged # 3 dataclass 視 caller 清乾淨後一併移除) ``` --- ## Task 1: Pre-flight — 確認 `parsed_result` dict shape + bundle restore 策略 **為何先做這個**：`_write_all_data` 目前讀 `parsed_result["devices"]` / `["info_systems"]` / `["leveraged"]`（v2 shape）。Phase 2 confirm_service 要支援 v1-ssp (legacy v2 shape) + v2-bundle (新 v3 shape) 兩種 dict。本 task **read-only 調查**，產出 1 份 markdown 給其他 task 引用。 **Files:** - Read: `app/oscal/service/excel_parser/*.py`（看 parser 怎麼填 parsed_result） - Read: `app/oscal/service/ssp_excel_import_app_service.py:1979-2070`（`_dict_to_parsed_*` 助手） - Read: `domain/oscal/parser/ssp_intermediate.py`（v3 dataclass 欄位） - Write: `docs/features/FR-028-2605-ssp-oscal-alignment/phase2-parsed-result-shape.md` **Approach:** 跑 grep + read，產出 1 份 markdown 表，列： - v1-ssp shape (legacy)：parsed_result 內 top-level keys（`devices` / `info_systems` / `leveraged` / `controls_with_aos` / `parties_org` / `parties_person` / `metadata` / `system_characteristic`） - v2-bundle shape（決定）：top-level keys 改 `components` / `leveraged_authorizations` / `inventory_items` 還是繼續沿用舊 keys 但 row schema 改？ - dict-to-bundle 轉換規則（每個 row 的欄位 mapping） **決策原則：** - 新解析直接寫 v2-bundle shape（**不沿用舊 keys**）—— 跟 v3 dataclasses 命名一致，正向變化乾淨 - v1-ssp legacy 不轉換，原樣存原樣讀，`_legacy_confirm()` 不受影響 - [ ] **Step 1: Map 既有 parser 輸出 keys** ```bash # 列 ssp_excel_import_app_service 內所有 parsed_result.get("...") key grep -E 'parsed_result\.get\(' app/oscal/service/ssp_excel_import_app_service.py | sort -u # 列 parser layer 寫進去的 key grep -rE 'result\["[^"]+"\]\s*=' app/oscal/service/excel_parser/ | head -50 ``` - [ ] **Step 2: Map 既有 docx import 路徑寫入 dict keys** ```bash grep -E 'parsed_result\.get\(' app/oscal/service/ssp_docx_import_app_service.py | sort -u ``` - [ ] **Step 3: 寫 shape 對照表 markdown** Output `docs/features/FR-028-2605-ssp-oscal-alignment/phase2-parsed-result-shape.md`，內含： ```markdown # Phase 2 — parsed_result Shape Reference ## v1-ssp (Legacy) Shape — pre Phase 2 deploy in-flight jobs { "devices": [...], "info_systems": [...], "leveraged": [...], "controls_with_aos": [...], "parties_org": [...], "parties_person": [...], "metadata": {...}, "system_characteristic": {...} } 每個 row 欄位 schema：（依 source code 列） - devices[i]: {...} - info_systems[i]: {...} - leveraged[i]: {...} ## v2-bundle (New) Shape — Phase 2 onward { "schema_version": "v2-bundle", "components": [...], "leveraged_authorizations": [...], "inventory_items": [...], "controls_with_aos": [...], "parties": [...], "metadata": {...}, "system_characteristic": {...}, "import_warnings": [...] } 每個 row 欄位 schema：（對齊 ParsedComponent / ParsedLeveragedAuthorization / ParsedInventoryItem） ## v1 → v2 轉換規則（bundle_restore 用） - v1 `devices[i]` → v2 `components[j]` with `component_type='hardware'` - v1 `info_systems[i]` → v2 `components[j]` with `component_type='system'` - v1 `leveraged[i]` → v2 `components[j]` with `component_type='service'` + v2 `leveraged_authorizations[k]` paired by title (leveraged_authorization_ref = la.title) - v1 `parties_org` + `parties_person` → v2 `parties` (single list) ``` - [ ] **Step 4: 確認決策 + commit** ```bash git add docs/features/FR-028-2605-ssp-oscal-alignment/phase2-parsed-result-shape.md git commit -m "docs(ssp-oscal-alignment): Phase 2 Task 1 — parsed_result shape reference" ``` --- ## Task 2: `ImportWarnings` collector module **Files:** - Create: `domain/oscal/import_pipeline/__init__.py`（空） - Create: `domain/oscal/import_pipeline/warnings.py` - Test: `tests/test_import_warnings.py` **Approach:** 純 dataclass + collector helper class，**no DB / no DI**。提供： - `ImportWarning` dataclass: `{ref, reason, context_key, context_value, source_table?}` - `ImportWarnings` collector: `add(warning)` / `extend(warnings_list)` / `to_dict_list()` for JSONB storage design.md §2.3.1 範例： ```json { "ref": "Crowdstrike", "reason": "leveraged_authorization not found in bundle", "context": { "component_title": "AcmeCorp Backup Service" } } ``` - [ ] **Step 1: 寫 failing tests** ```python # tests/test_import_warnings.py import pytest from domain.oscal.import_pipeline.warnings import ImportWarning, ImportWarnings def test_import_warning_dataclass_minimal_fields(): w = ImportWarning( ref="Crowdstrike", reason="leveraged_authorization not found in bundle", context={"component_title": "AcmeCorp Backup Service"}, ) assert w.ref == "Crowdstrike" assert w.reason == "leveraged_authorization not found in bundle" assert w.context["component_title"] == "AcmeCorp Backup Service" def test_collector_add_then_to_dict_list_yields_jsonb_safe(): c = ImportWarnings() c.add(ImportWarning(ref="A", reason="r1", context={"k": "v"})) c.add(ImportWarning(ref="B", reason="r2", context={})) out = c.to_dict_list() assert isinstance(out, list) assert out == [ {"ref": "A", "reason": "r1", "context": {"k": "v"}}, {"ref": "B", "reason": "r2", "context": {}}, ] def test_collector_extend_accepts_multiple(): c = ImportWarnings() c.extend([ ImportWarning(ref="X", reason="x", context={}), ImportWarning(ref="Y", reason="y", context={}), ]) assert len(c.to_dict_list()) == 2 def test_collector_is_empty_when_no_add(): c = ImportWarnings() assert c.to_dict_list() == [] assert len(c) == 0 def test_collector_supports_len(): c = ImportWarnings() c.add(ImportWarning(ref="A", reason="r", context={})) assert len(c) == 1 ``` - [ ] **Step 2: 跑 test 確認 FAIL** ```bash pytest tests/test_import_warnings.py -v # Expected: FAIL (module not exist) ``` - [ ] **Step 3: 寫 minimal impl** ```python # domain/oscal/import_pipeline/warnings.py """Import warnings collector — shared by Excel + Docx import pipelines. Stored into parse_jobs.parsed_result.import_warnings (JSONB). Per design.md §2.3.1 (FK Resolve — Warn but Allow) + §2.3.2 (title conflict). """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List, Optional @dataclass class ImportWarning: """One warning entry — JSONB-serializable to parsed_result.import_warnings.""" ref: str reason: str context: Dict[str, Any] = field(default_factory=dict) source_table: Optional[str] = None # e.g. "components" / "inventory_items" def to_dict(self) -> Dict[str, Any]: out = {"ref": self.ref, "reason": self.reason, "context": self.context} if self.source_table is not None: out["source_table"] = self.source_table return out class ImportWarnings: """Collector — adapter / normalizer 共用累積 warnings。""" def __init__(self) -> None: self._items: List[ImportWarning] = [] def add(self, warning: ImportWarning) -> None: self._items.append(warning) def extend(self, warnings: List[ImportWarning]) -> None: self._items.extend(warnings) def to_dict_list(self) -> List[Dict[str, Any]]: return [w.to_dict() for w in self._items] def __len__(self) -> int: return len(self._items) ``` - [ ] **Step 4: 跑 test 確認 PASS** ```bash pytest tests/test_import_warnings.py -v # Expected: 5 passed ``` - [ ] **Step 5: Commit** ```bash git add domain/oscal/import_pipeline/__init__.py \ domain/oscal/import_pipeline/warnings.py \ tests/test_import_warnings.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 2 — ImportWarnings collector module" ``` --- ## Task 3: `ParsedBundleNormalizer` — dedup + ref resolve + warnings **Files:** - Create: `domain/oscal/import_pipeline/normalizer.py` - Test: `tests/test_parsed_bundle_normalizer.py` **Approach:** 純 in-memory bundle 處理（no DB / no DI），把 adapter 輸出的 raw bundle 變成 clean bundle，emit warnings 到 `ImportWarnings` collector。處理 design.md §2.3.1 + §2.3.2 兩段規則。 **Normalize 步驟（按順序）：** 1. **Components dedup by `(component_type, title)`**：完全相同 → 取第一筆，第二筆 emit warning `duplicate component skipped` 2. **LeveragedAuthorization dedup by `title`**：同上 pattern 3. **InventoryItem dedup by `description`**：完全相同 → 取第一筆，第二筆 emit warning 4. **`ParsedComponent.leveraged_authorization_ref` resolve**：bundle 內無對應 LA title → 該欄位設 `None`，emit warning `leveraged_authorization not found in bundle`；有對應但 LA 是 dedup 第二筆 → resolve 到第一筆 5. **`ParsedInventoryItem.implemented_component_refs` resolve**： - 無對應 component title → 該 ref 從 list 移除，emit warning `component_ref not found in bundle` - 多筆同 title component → 取 `component_type='service'` 的第一筆，emit warning `ambiguous component title, picked first service` - 完全無 service component 對應 → 取任意第一筆 + emit warning **Type discipline**：ref 在 parse 階段是 `str`（per design.md §11.6），normalize 階段 resolve 後仍維持 `str`（不轉 `uuid.UUID`）。Write strategy 自己負責 lookup uid。 - [ ] **Step 1: 寫 failing tests** ```python # tests/test_parsed_bundle_normalizer.py import pytest from domain.oscal.import_pipeline.normalizer import ParsedBundleNormalizer from domain.oscal.import_pipeline.warnings import ImportWarnings from domain.oscal.parser.ssp_intermediate import ( ParsedComponent, ParsedLeveragedAuthorization, ParsedInventoryItem, ParsedExcelEntityBundle, ) @pytest.fixture def normalizer(): return ParsedBundleNormalizer() def test_component_dedup_emits_warning(normalizer): bundle = ParsedExcelEntityBundle(parsed_components=[ ParsedComponent(title="A", component_type="service"), ParsedComponent(title="A", component_type="service"), ]) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) assert len(cleaned.parsed_components) == 1 assert len(warnings) == 1 assert "duplicate component" in warnings.to_dict_list()[0]["reason"] def test_la_dedup_emits_warning(normalizer): bundle = ParsedExcelEntityBundle(parsed_leveraged_authorizations=[ ParsedLeveragedAuthorization(title="Crowdstrike"), ParsedLeveragedAuthorization(title="Crowdstrike"), ]) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) assert len(cleaned.parsed_leveraged_authorizations) == 1 assert any("duplicate leveraged_authorization" in w["reason"] for w in warnings.to_dict_list()) def test_inventory_dedup_by_description(normalizer): bundle = ParsedExcelEntityBundle(parsed_inventory_items=[ ParsedInventoryItem(description="Server A"), ParsedInventoryItem(description="Server A"), ]) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) assert len(cleaned.parsed_inventory_items) == 1 assert any("duplicate inventory_item" in w["reason"] for w in warnings.to_dict_list()) def test_component_la_ref_not_found_clears_and_warns(normalizer): bundle = ParsedExcelEntityBundle( parsed_components=[ ParsedComponent( title="AcmeCorp Backup", component_type="service", leveraged_authorization_ref="Crowdstrike", # 不在 bundle.parsed_leveraged_authorizations ), ], parsed_leveraged_authorizations=[], ) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) assert cleaned.parsed_components[0].leveraged_authorization_ref is None assert len(warnings) == 1 w = warnings.to_dict_list()[0] assert w["ref"] == "Crowdstrike" assert "leveraged_authorization not found" in w["reason"] assert w["context"]["component_title"] == "AcmeCorp Backup" def test_component_la_ref_found_keeps_string(normalizer): bundle = ParsedExcelEntityBundle( parsed_components=[ ParsedComponent( title="AcmeCorp Backup", component_type="service", leveraged_authorization_ref="Crowdstrike", ), ], parsed_leveraged_authorizations=[ ParsedLeveragedAuthorization(title="Crowdstrike"), ], ) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) # ref 維持 str（normalizer 不解 uid，write strategy 自己解） assert cleaned.parsed_components[0].leveraged_authorization_ref == "Crowdstrike" assert len(warnings) == 0 def test_inventory_implemented_ref_not_found_drops_ref(normalizer): bundle = ParsedExcelEntityBundle( parsed_components=[ParsedComponent(title="Comp A", component_type="hardware")], parsed_inventory_items=[ ParsedInventoryItem( description="Server X", implemented_component_refs=["Comp A", "Comp Missing"], ), ], ) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) assert cleaned.parsed_inventory_items[0].implemented_component_refs == ["Comp A"] assert any("component_ref not found" in w["reason"] for w in warnings.to_dict_list()) def test_inventory_ambiguous_component_picks_service_first(normalizer): bundle = ParsedExcelEntityBundle( parsed_components=[ ParsedComponent(title="Crowdstrike", component_type="hardware"), ParsedComponent(title="Crowdstrike", component_type="service"), ], parsed_inventory_items=[ ParsedInventoryItem( description="Server X", implemented_component_refs=["Crowdstrike"], ), ], ) warnings = ImportWarnings() cleaned = normalizer.normalize(bundle, warnings) # ref 仍維持 str — 解析只 emit warning 標 ambiguous，不重寫 ref assert cleaned.parsed_inventory_items[0].implemented_component_refs == ["Crowdstrike"] assert any("ambiguous component title" in w["reason"] for w in warnings.to_dict_list()) ``` - [ ] **Step 2: 跑 test 確認 FAIL** ```bash pytest tests/test_parsed_bundle_normalizer.py -v # Expected: FAIL (module not exist) ``` - [ ] **Step 3: 寫 impl** ```python # domain/oscal/import_pipeline/normalizer.py """ParsedBundleNormalizer — dedup + ref-resolve warnings emission. Per design.md §2.3.1 (FK Resolve — Warn but Allow) + §2.3.2 (title conflict). Pure in-memory transform: adapter raw bundle → clean bundle. No DB / no DI. WriteStrategy 之後仍負責把 `_ref` str 解成 uid（§11.6 type discipline）。 """ from __future__ import annotations from typing import List, Tuple from domain.oscal.import_pipeline.warnings import ImportWarning, ImportWarnings from domain.oscal.parser.ssp_intermediate import ( ParsedComponent, ParsedExcelEntityBundle, ParsedInventoryItem, ParsedLeveragedAuthorization, ) class ParsedBundleNormalizer: """Idempotent: 同一 bundle 跑兩次 normalize 結果相同（dedup 已收斂）。""" def normalize( self, bundle: ParsedExcelEntityBundle, warnings: ImportWarnings, ) -> ParsedExcelEntityBundle: components = self._dedup_components(bundle.parsed_components, warnings) las = self._dedup_las(bundle.parsed_leveraged_authorizations, warnings) inventory = self._dedup_inventory(bundle.parsed_inventory_items, warnings) self._resolve_component_la_refs(components, las, warnings) self._resolve_inventory_component_refs(inventory, components, warnings) # 回傳新 bundle (不 mutate 入參) — 其他欄位原樣帶過 return ParsedExcelEntityBundle( parsed_system_characteristic=bundle.parsed_system_characteristic, parsed_parties=bundle.parsed_parties, parsed_components=components, parsed_leveraged_authorizations=las, parsed_inventory_items=inventory, parsed_controls=bundle.parsed_controls, parsed_metadata=bundle.parsed_metadata, ) # ------------------------------------------------------------------ # Dedup helpers # ------------------------------------------------------------------ @staticmethod def _dedup_components( items: List[ParsedComponent], warnings: ImportWarnings, ) -> List[ParsedComponent]: seen: dict[Tuple[str, str], ParsedComponent] = {} out: List[ParsedComponent] = [] for c in items: key = (c.component_type, c.title) if key in seen: warnings.add(ImportWarning( ref=c.title, reason="duplicate component skipped", context={"component_type": c.component_type}, source_table="components", )) continue seen[key] = c out.append(c) return out @staticmethod def _dedup_las( items: List[ParsedLeveragedAuthorization], warnings: ImportWarnings, ) -> List[ParsedLeveragedAuthorization]: seen: dict[str, ParsedLeveragedAuthorization] = {} out: List[ParsedLeveragedAuthorization] = [] for la in items: if la.title in seen: warnings.add(ImportWarning( ref=la.title, reason="duplicate leveraged_authorization skipped", context={}, source_table="leveraged_authorizations", )) continue seen[la.title] = la out.append(la) return out @staticmethod def _dedup_inventory( items: List[ParsedInventoryItem], warnings: ImportWarnings, ) -> List[ParsedInventoryItem]: seen: dict[str, ParsedInventoryItem] = {} out: List[ParsedInventoryItem] = [] for inv in items: if inv.description in seen: warnings.add(ImportWarning( ref=inv.description, reason="duplicate inventory_item skipped", context={}, source_table="inventory_items", )) continue seen[inv.description] = inv out.append(inv) return out # ------------------------------------------------------------------ # Ref-resolve helpers — 不轉成 uid（write strategy 之後做），只負責 warning # ------------------------------------------------------------------ @staticmethod def _resolve_component_la_refs( components: List[ParsedComponent], las: List[ParsedLeveragedAuthorization], warnings: ImportWarnings, ) -> None: la_titles = {la.title for la in las} for c in components: ref = c.leveraged_authorization_ref if ref is None: continue if ref not in la_titles: warnings.add(ImportWarning( ref=ref, reason="leveraged_authorization not found in bundle", context={"component_title": c.title}, source_table="components", )) c.leveraged_authorization_ref = None # clear so write strategy 寫 NULL @staticmethod def _resolve_inventory_component_refs( inventory: List[ParsedInventoryItem], components: List[ParsedComponent], warnings: ImportWarnings, ) -> None: # title → list[component_type] title_index: dict[str, list[str]] = {} for c in components: title_index.setdefault(c.title, []).append(c.component_type) for inv in inventory: keep_refs: list[str] = [] for ref in inv.implemented_component_refs: if ref not in title_index: warnings.add(ImportWarning( ref=ref, reason="component_ref not found in bundle", context={"inventory_description": inv.description}, source_table="inventory_items", )) continue types = title_index[ref] if len(types) > 1: warnings.add(ImportWarning( ref=ref, reason="ambiguous component title, picked first service", context={ "inventory_description": inv.description, "candidate_types": types, }, source_table="inventory_items", )) keep_refs.append(ref) inv.implemented_component_refs = keep_refs ``` - [ ] **Step 4: 跑 test 確認 PASS** ```bash pytest tests/test_parsed_bundle_normalizer.py -v # Expected: 7 passed ``` - [ ] **Step 5: Commit** ```bash git add domain/oscal/import_pipeline/normalizer.py tests/test_parsed_bundle_normalizer.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 3 — ParsedBundleNormalizer + warnings emission" ``` --- ## Task 4: `SspImportConfirmService` — schema_version 分流 + bundle restore **Files:** - Create: `domain/oscal/import_pipeline/bundle_restore.py` - Create: `domain/oscal/import_pipeline/confirm_service.py` - Test: `tests/test_bundle_restore.py` - Test: `tests/test_ssp_import_confirm_service.py` **Approach:** confirm_service 是 Excel + Docx app service 的共用入口，**不**直接寫 DB（保持 caller 的 `@transaction` 為 owner，per CLAUDE.md DDD 規範）。 本 Task 兩件事： 1. `bundle_restore.py` — `dict → ParsedExcelEntityBundle`（**只支援 v2-bundle shape**，v1-ssp 走 caller 既有 legacy path） 2. `confirm_service.py` — `confirm(parsed_result, schema_version, ...)` 入口：v2-bundle → bundle_restore → normalize → orchestrator → strategies；v1-ssp → 拋 `LegacyConfirmRequired` 例外讓 caller 處理 **`SspImportConfirmService` 簽名（dispatch 入口）：** ```python def confirm( self, parsed_result: dict, *, ssp_id: int, catalog_id: Optional[int], user_context, ) -> dict: """Return: {'written_components': N, 'written_las': N, 'written_inventory': N, 'warnings_added': N}""" ``` Caller 是 app service `@transaction` scope 內呼叫。schema_version 從 `parsed_result.get("schema_version", "v1-ssp")` 取，預設 v1-ssp（向後相容 in-flight job）。 - [ ] **Step 1: 寫 bundle_restore failing tests** ```python # tests/test_bundle_restore.py import pytest from domain.oscal.import_pipeline.bundle_restore import ( dict_to_bundle, LegacyConfirmRequired, ) from domain.oscal.parser.ssp_intermediate import ParsedExcelEntityBundle def test_v2_bundle_dict_restores_to_bundle(): parsed_result = { "schema_version": "v2-bundle", "components": [ {"title": "Comp A", "component_type": "service"}, ], "leveraged_authorizations": [ {"title": "Crowdstrike"}, ], "inventory_items": [ {"description": "Server X", "implemented_component_refs": ["Comp A"]}, ], } bundle = dict_to_bundle(parsed_result) assert isinstance(bundle, ParsedExcelEntityBundle) assert bundle.parsed_components[0].title == "Comp A" assert bundle.parsed_leveraged_authorizations[0].title == "Crowdstrike" assert bundle.parsed_inventory_items[0].description == "Server X" def test_v1_ssp_dict_raises_legacy_required(): parsed_result = { "schema_version": "v1-ssp", "devices": [{"name": "D1"}], } with pytest.raises(LegacyConfirmRequired): dict_to_bundle(parsed_result) def test_missing_schema_version_defaults_to_legacy(): # in-flight job (deploy 前已存) — 視為 v1-ssp parsed_result = {"devices": []} with pytest.raises(LegacyConfirmRequired): dict_to_bundle(parsed_result) def test_v2_bundle_optional_keys_empty(): parsed_result = {"schema_version": "v2-bundle"} bundle = dict_to_bundle(parsed_result) assert bundle.parsed_components == [] assert bundle.parsed_leveraged_authorizations == [] assert bundle.parsed_inventory_items == [] ``` - [ ] **Step 2: 寫 bundle_restore impl** ```python # domain/oscal/import_pipeline/bundle_restore.py """dict ↔ ParsedExcelEntityBundle conversion — Phase 2 confirm path. Only supports v2-bundle shape; v1-ssp dispatches via LegacyConfirmRequired so caller can drop to legacy confirm path. Conversion rule reference: docs/features/FR-028-2605-ssp-oscal-alignment/phase2-parsed-result-shape.md """ from __future__ import annotations from datetime import date from typing import Any, Dict, List, Optional from domain.oscal.parser.ssp_intermediate import ( ParsedComponent, ParsedExcelEntityBundle, ParsedInventoryItem, ParsedLeveragedAuthorization, ParsedParty, ParsedSystemCharacteristic, ) class LegacyConfirmRequired(Exception): """Raised by dict_to_bundle when parsed_result is v1-ssp shape. Caller (app service) catches this and dispatches to legacy confirm path. """ def dict_to_bundle(parsed_result: Dict[str, Any]) -> ParsedExcelEntityBundle: schema_version = parsed_result.get("schema_version", "v1-ssp") if schema_version != "v2-bundle": raise LegacyConfirmRequired( f"parsed_result schema_version={schema_version!r}, legacy path required" ) return ParsedExcelEntityBundle( parsed_system_characteristic=_restore_system_characteristic( parsed_result.get("system_characteristic") ), parsed_parties=[_restore_party(p) for p in parsed_result.get("parties", [])], parsed_components=[_restore_component(c) for c in parsed_result.get("components", [])], parsed_leveraged_authorizations=[ _restore_la(la) for la in parsed_result.get("leveraged_authorizations", []) ], parsed_inventory_items=[ _restore_inventory(i) for i in parsed_result.get("inventory_items", []) ], parsed_controls=[], # Phase 2 不動 control path — 沿用既有 controls_with_aos 流程 parsed_metadata=parsed_result.get("metadata"), ) def _restore_component(d: Dict[str, Any]) -> ParsedComponent: return ParsedComponent( title=d["title"], component_type=d["component_type"], description=d.get("description"), purpose=d.get("purpose"), status=d.get("status"), leveraged_authorization_ref=d.get("leveraged_authorization_ref"), protocol=d.get("protocol"), port_ranges=d.get("port_ranges"), security_auth=d.get("security_auth"), matched_party_uuid=d.get("matched_party_uuid"), ) def _restore_la(d: Dict[str, Any]) -> ParsedLeveragedAuthorization: date_auth = d.get("date_authorized") if isinstance(date_auth, str): try: date_auth = date.fromisoformat(date_auth) except ValueError: date_auth = None return ParsedLeveragedAuthorization( title=d["title"], provider=d.get("provider"), date_authorized=date_auth, fedramp_package_id=d.get("fedramp_package_id"), impact_level=d.get("impact_level"), data_types=d.get("data_types"), nature_of_agreement=d.get("nature_of_agreement"), authorized_users=d.get("authorized_users"), remarks=d.get("remarks"), ) def _restore_inventory(d: Dict[str, Any]) -> ParsedInventoryItem: return ParsedInventoryItem( description=d["description"], asset_id=d.get("asset_id"), asset_tag=d.get("asset_tag"), ipv4_address=d.get("ipv4_address"), mac_address=d.get("mac_address"), fqdn=d.get("fqdn"), hostname=d.get("hostname"), software_name=d.get("software_name"), os_name=d.get("os_name"), implemented_component_refs=list(d.get("implemented_component_refs") or []), ) def _restore_party(d: Dict[str, Any]) -> Optional[ParsedParty]: # 既有 _dict_to_parsed_parties 邏輯不變 — 此處 v2-bundle 用 single list return ParsedParty( name=d.get("name") or "(未命名)", party_type=d.get("party_type", "person"), role=d.get("role"), title=d.get("title"), email_address=d.get("email_address"), telephone_number=d.get("telephone_number"), address=d.get("address"), short_name=d.get("short_name"), matched_user_id=d.get("matched_user_id"), matched_org_unit_id=d.get("matched_org_unit_id"), ) def _restore_system_characteristic(d: Optional[Dict[str, Any]]) -> Optional[ParsedSystemCharacteristic]: if not d: return None return ParsedSystemCharacteristic( system_name=d.get("system_name") or "", security_sensitivity_level=d.get("security_sensitivity_level"), status=d.get("status"), target_type=d.get("target_type"), scope_description=d.get("scope_description"), owner_login_name=d.get("owner_login_name"), ) ``` - [ ] **Step 3: 跑 bundle_restore test 確認 PASS** ```bash pytest tests/test_bundle_restore.py -v # Expected: 4 passed ``` - [ ] **Step 4: 寫 confirm_service failing tests** ```python # tests/test_ssp_import_confirm_service.py from unittest.mock import Mock import pytest from domain.oscal.import_pipeline.confirm_service import SspImportConfirmService from domain.oscal.import_pipeline.bundle_restore import LegacyConfirmRequired @pytest.fixture def mock_orchestrator(): return Mock() @pytest.fixture def mock_leveraged_strategy(): s = Mock(); s.write.return_value = 0 return s @pytest.fixture def mock_component_strategy(): s = Mock(); s.write.return_value = 0 return s @pytest.fixture def mock_inventory_strategy(): s = Mock(); s.write.return_value = 0 return s @pytest.fixture def confirm_service(mock_orchestrator, mock_leveraged_strategy, mock_component_strategy, mock_inventory_strategy): return SspImportConfirmService( orchestrator=mock_orchestrator, leveraged_write_strategy=mock_leveraged_strategy, component_write_strategy=mock_component_strategy, inventory_write_strategy=mock_inventory_strategy, normalizer=None, # default fallback in service ctor ) def test_legacy_path_raises(confirm_service): user_ctx = Mock(login_name="alice", tenant_id=1) with pytest.raises(LegacyConfirmRequired): confirm_service.confirm( parsed_result={"devices": []}, # 無 schema_version → v1-ssp ssp_id=10, catalog_id=5, user_context=user_ctx, ) def test_v2_bundle_calls_strategies_in_order(confirm_service, mock_orchestrator, mock_leveraged_strategy, mock_component_strategy, mock_inventory_strategy): user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99) mock_leveraged_strategy.write.return_value = 2 mock_component_strategy.write.return_value = 5 mock_inventory_strategy.write.return_value = 3 parsed_result = { "schema_version": "v2-bundle", "components": [ {"title": "A", "component_type": "hardware"}, {"title": "B", "component_type": "service"}, ], "leveraged_authorizations": [{"title": "Crowdstrike"}], "inventory_items": [ {"description": "Srv1", "implemented_component_refs": ["A"]}, ], } out = confirm_service.confirm( parsed_result=parsed_result, ssp_id=10, catalog_id=5, user_context=user_ctx, ) # 順序：LA → Component → Inventory（per design.md §2.3） assert mock_leveraged_strategy.write.called assert mock_component_strategy.write.called assert mock_inventory_strategy.write.called assert out["written_las"] == 2 assert out["written_components"] == 5 assert out["written_inventory"] == 3 def test_v2_bundle_warnings_threaded_back(confirm_service): user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99) parsed_result = { "schema_version": "v2-bundle", "components": [ {"title": "Comp", "component_type": "service", "leveraged_authorization_ref": "MissingLA"}, ], "leveraged_authorizations": [], } out = confirm_service.confirm( parsed_result=parsed_result, ssp_id=10, catalog_id=5, user_context=user_ctx, ) assert out["warnings_added"] >= 1 # warnings 也寫回 parsed_result（caller 之後 persist 到 parse_jobs.parsed_result） assert any("leveraged_authorization not found" in w["reason"] for w in parsed_result["import_warnings"]) ``` - [ ] **Step 5: 寫 confirm_service impl** ```python # domain/oscal/import_pipeline/confirm_service.py """SspImportConfirmService — shared Excel + Docx confirm entry point. Caller (excel / docx app service) wraps @transaction; this service does not open a session. schema_version dispatch: - "v2-bundle" → bundle_restore → normalize → orchestrator → strategies - "v1-ssp" or missing → raise LegacyConfirmRequired (caller handles legacy) """ from __future__ import annotations from typing import Any, Dict, Optional from domain.oscal.import_pipeline.bundle_restore import ( LegacyConfirmRequired, dict_to_bundle, ) from domain.oscal.import_pipeline.normalizer import ParsedBundleNormalizer from domain.oscal.import_pipeline.warnings import ImportWarnings from domain.oscal.parser.ssp_intermediate import SspEntityReconciliationContext class SspImportConfirmService: def __init__( self, orchestrator, leveraged_write_strategy, component_write_strategy, inventory_write_strategy, normalizer: Optional[ParsedBundleNormalizer] = None, ): self._orchestrator = orchestrator self._leveraged = leveraged_write_strategy self._component = component_write_strategy self._inventory = inventory_write_strategy self._normalizer = normalizer or ParsedBundleNormalizer() def confirm( self, parsed_result: Dict[str, Any], *, ssp_id: int, catalog_id: Optional[int], user_context, ) -> Dict[str, int]: """Raises LegacyConfirmRequired for v1-ssp shape. Mutates parsed_result (adds import_warnings list).""" # Step 1: dict → bundle (raises LegacyConfirmRequired on v1) bundle = dict_to_bundle(parsed_result) # Step 2: normalize + collect warnings warnings = ImportWarnings() bundle = self._normalizer.normalize(bundle, warnings) # Step 3: reconcile (orchestrator) — optional dep if self._orchestrator is not None and catalog_id is not None: ctx = SspEntityReconciliationContext( tenant_id=user_context.tenant_id, catalog_id=catalog_id, ) try: self._orchestrator.reconcile(bundle, ctx) except Exception: pass # reconciler 失敗 fall through 寫純文字 — caller log # Step 4: write 順序：LA → Component → Inventory (FK 依賴) written_las = self._leveraged.write( bundle.parsed_leveraged_authorizations, ssp_id, user_context, ) if bundle.parsed_leveraged_authorizations else 0 written_components = self._component.write( bundle.parsed_components, ssp_id, user_context, ) if bundle.parsed_components else 0 written_inventory = self._inventory.write( bundle.parsed_inventory_items, ssp_id, user_context, ) if bundle.parsed_inventory_items else 0 # Step 5: warnings 寫回 parsed_result，caller persist 到 JSONB parsed_result["import_warnings"] = warnings.to_dict_list() return { "written_las": written_las, "written_components": written_components, "written_inventory": written_inventory, "warnings_added": len(warnings), } ``` - [ ] **Step 6: 跑 test 確認 PASS** ```bash pytest tests/test_bundle_restore.py tests/test_ssp_import_confirm_service.py -v # Expected: 7 passed ``` - [ ] **Step 7: Commit** ```bash git add domain/oscal/import_pipeline/bundle_restore.py \ domain/oscal/import_pipeline/confirm_service.py \ tests/test_bundle_restore.py \ tests/test_ssp_import_confirm_service.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 4 — SspImportConfirmService + bundle_restore" ``` --- ## Task 5: Refactor write strategies to v3 entity shape ### Task 5.A: Refactor `LeveragedWriteStrategy` to consume `ParsedLeveragedAuthorization` **Files:** - Modify: `domain/oscal/service/write_strategy/leveraged_write_strategy.py` - Modify: `tests/test_ssp_write_strategy.py` (find tests for leveraged) + 新加 v3 case - Read: design.md §2.3 ordering（LA 寫完拿 uid map → Component 寫 leveraged_authorization_uid FK） **Current**：strategy 消費 `ParsedLeveraged` (`service_name`, `purpose`, etc.)，**且自己 build linked service-Component**。 **New**：strategy 消費 `ParsedLeveragedAuthorization`（只欄位對應 LA 表），**不再** build linked Component（Component 由 `ComponentWriteStrategy` 處理）。 Service Component 跟 LA 的關聯由 `ParsedComponent.leveraged_authorization_ref` 字串解 — 在 `ComponentWriteStrategy.write()` 內走 lookup by title → LA uid。 **Signature change：移除 `sys_impl_main_id` dead param**（Phase 1 follow-up §6.4）。 ```python # NEW signature def write( self, parsed_las: Iterable[ParsedLeveragedAuthorization], ssp_id: int, user_context, ) -> int: ``` - [ ] **Step 1: 看現有 test_ssp_write_strategy.py，新加 v3 case** ```bash grep -n "LeveragedWriteStrategy\|test_leveraged" tests/test_ssp_write_strategy.py | head ``` - [ ] **Step 2: 寫 failing test** ```python # tests/test_ssp_write_strategy.py — 加段 def test_leveraged_write_strategy_consumes_parsed_la(): from domain.oscal.service.write_strategy.leveraged_write_strategy import LeveragedWriteStrategy from domain.oscal.parser.ssp_intermediate import ParsedLeveragedAuthorization la_service = Mock() la_service.get_all.return_value = [] saved = Mock(); saved.uid = uuid.uuid4() la_service.add.return_value = saved strat = LeveragedWriteStrategy(la_service) # 不再 inject component_service user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99) out = strat.write( [ParsedLeveragedAuthorization(title="Crowdstrike", impact_level="moderate")], ssp_id=10, user_context=user_ctx, ) assert out == 1 la_entity = la_service.add.call_args[0][0] assert la_entity.title == "Crowdstrike" # impact_level / data_types 等都放 props JSONB assert la_entity.props.get("impact_level") == "moderate" ``` - [ ] **Step 3: 寫 impl（refactor）** 關鍵異動： - `__init__(self, leveraged_authorization_domain_service)` — 移除 `component_domain_service` - `write(parsed_las, ssp_id, user_context)` — 移除 `sys_impl_main_id` - 不再 build linked Component；`props` 從 `ParsedLeveragedAuthorization` 五個 props 欄位 `jsonb_strip_nulls`-style 組成 - `party_uuid` 來自 ref-resolve（write strategy 用 `provider` lookup `oscal.parties` by name — 或 `matched_party_uuid` 若 reconciler 已填） - overwrite semantics 保留：先刪 existing LA rows for ssp_id ```python # domain/oscal/service/write_strategy/leveraged_write_strategy.py（重寫） """LeveragedWriteStrategy v3 — consumes ParsedLeveragedAuthorization. Phase 2 Task 5A refactor — strategy 只寫 oscal.ssp_leveraged_authorizations， linked service Component 由 ComponentWriteStrategy 走 leveraged_authorization_ref 字串 lookup 處理。Signature 移除 dead param sys_impl_main_id。 UNMATCHED: party_uuid stays None when no matched party. Overwrite semantics: delete existing rows for ssp_id before insert. """ from __future__ import annotations import logging import uuid from typing import Iterable, Optional from domain.oscal.parser.ssp_intermediate import ParsedLeveragedAuthorization logger = logging.getLogger(__name__) def _strip_nulls(d: dict) -> Optional[dict]: cleaned = {k: v for k, v in d.items() if v is not None and v != ""} return cleaned or None class LeveragedWriteStrategy: def __init__(self, leveraged_authorization_domain_service): self._la = leveraged_authorization_domain_service def write( self, parsed_las: Iterable[ParsedLeveragedAuthorization], ssp_id: int, user_context, ) -> int: from jedi_oscal.domain.entity.base.oscal_leveraged_authorization_entity import ( LeveragedAuthorizationEntity, ) from jedi_oscal.domain.entity.base.oscal_leveraged_authorization_query_entity import ( LeveragedAuthorizationQueryEntity, ) tenant_id = getattr(user_context, "tenant_id", None) org_unit_id = getattr(user_context, "org_unit_id", None) login_name = getattr(user_context, "login_name", None) if not login_name: raise ValueError("LeveragedWriteStrategy.write: user_context.login_name required") # Overwrite: delete existing LA for ssp_id existing = self._la.get_all(LeveragedAuthorizationQueryEntity(ssp_id=ssp_id)) or [] to_delete = [e.id for e in existing if e.id is not None] if to_delete: self._la.delete_by_ids(to_delete) written = 0 for parsed in parsed_las: props = _strip_nulls({ "fedramp_package_id": parsed.fedramp_package_id, "impact_level": parsed.impact_level, "data_types": parsed.data_types, "nature_of_agreement": parsed.nature_of_agreement, "authorized_users": parsed.authorized_users, }) party_uuid_val: uuid.UUID | None = None # provider 是 CSP/vendor name — reconciler 在 bundle 階段填 matched_party_uuid # 之後（v3 dataclass 還沒這欄，預留 future）。Phase 2 先用 None。 entity = LeveragedAuthorizationEntity( ssp_id=ssp_id, title=parsed.title, party_uuid=party_uuid_val, date_authorized=parsed.date_authorized, props=props, remarks=parsed.remarks, tenant_id=tenant_id, org_unit_id=org_unit_id, created_user=login_name, updated_user=login_name, ) try: self._la.add(entity) written += 1 except Exception as e: logger.warning( "LeveragedWriteStrategy: insert failed for ssp_id=%s title=%r: %s", ssp_id, parsed.title, e, exc_info=True, ) return written ``` - [ ] **Step 4: 跑 test 確認 PASS** ```bash pytest tests/test_ssp_write_strategy.py -v -k leveraged ``` - [ ] **Step 5: Commit** ```bash git add domain/oscal/service/write_strategy/leveraged_write_strategy.py tests/test_ssp_write_strategy.py git commit -m "refactor(ssp-oscal-alignment): Phase 2 Task 5A — LeveragedWriteStrategy v3 (consume ParsedLeveragedAuthorization, remove dead param)" ``` ### Task 5.B: New `ComponentWriteStrategy`（取代 device + information_system） **Files:** - Create: `domain/oscal/service/write_strategy/component_write_strategy.py` - Modify: `domain/oscal/service/write_strategy/base.py`（清除 `sys_impl_main_id` 入參） - Delete (defer to Task 11): `device_write_strategy.py`, `information_system_write_strategy.py` - Test: `tests/test_component_write_strategy.py` **Approach:** 單一 `ComponentWriteStrategy.write(parsed_components, ssp_id, user_context)`，內部 dispatch by `component_type`： - 寫 row 到 `oscal.ssp_components` - 若 `leveraged_authorization_ref is not None`：lookup `ssp_leveraged_authorizations.title == ref AND ssp_id == ssp_id` → 拿 `uid` 寫進 `leveraged_authorization_uid` Overwrite semantics：刪除 `ssp_id` 全部 existing components 後 batch insert（不再按 type 分批；type 是 row attribute 不是 partition key）。 - [ ] **Step 1: 寫 failing tests** ```python # tests/test_component_write_strategy.py import uuid from unittest.mock import Mock from domain.oscal.parser.ssp_intermediate import ParsedComponent from domain.oscal.service.write_strategy.component_write_strategy import ( ComponentWriteStrategy, ) def _ctx(): return Mock(login_name="alice", tenant_id=1, org_unit_id=99) def test_component_write_no_la_ref_writes_null_la_uid(): comp_svc = Mock(); comp_svc.get_all.return_value = [] la_svc = Mock(); la_svc.get_all.return_value = [] strat = ComponentWriteStrategy(comp_svc, la_svc) out = strat.write( [ParsedComponent(title="Comp", component_type="hardware")], ssp_id=10, user_context=_ctx(), ) assert out == 1 entity = comp_svc.add.call_args[0][0] assert entity.leveraged_authorization_uid is None def test_component_write_la_ref_resolves_via_la_lookup(): comp_svc = Mock(); comp_svc.get_all.return_value = [] la_uid = uuid.uuid4() la_row = Mock(uid=la_uid, title="Crowdstrike") la_svc = Mock(); la_svc.get_all.return_value = [la_row] strat = ComponentWriteStrategy(comp_svc, la_svc) out = strat.write( [ParsedComponent(title="C", component_type="service", leveraged_authorization_ref="Crowdstrike")], ssp_id=10, user_context=_ctx(), ) entity = comp_svc.add.call_args[0][0] assert entity.leveraged_authorization_uid == la_uid def test_component_write_la_ref_unmatched_writes_null(): comp_svc = Mock(); comp_svc.get_all.return_value = [] la_svc = Mock(); la_svc.get_all.return_value = [] # 沒對應 LA strat = ComponentWriteStrategy(comp_svc, la_svc) strat.write( [ParsedComponent(title="C", component_type="service", leveraged_authorization_ref="Missing")], ssp_id=10, user_context=_ctx(), ) entity = comp_svc.add.call_args[0][0] assert entity.leveraged_authorization_uid is None def test_overwrite_deletes_existing_then_inserts(): existing_a = Mock(id=1); existing_b = Mock(id=2) comp_svc = Mock() comp_svc.get_all.return_value = [existing_a, existing_b] la_svc = Mock(); la_svc.get_all.return_value = [] strat = ComponentWriteStrategy(comp_svc, la_svc) strat.write([ParsedComponent(title="C", component_type="hardware")], 10, _ctx()) comp_svc.delete_by_ids.assert_called_once_with([1, 2]) assert comp_svc.add.called def test_component_type_persisted(): comp_svc = Mock(); comp_svc.get_all.return_value = [] la_svc = Mock(); la_svc.get_all.return_value = [] strat = ComponentWriteStrategy(comp_svc, la_svc) strat.write( [ParsedComponent(title="X", component_type="system", description="d", purpose="p", status="operational")], ssp_id=10, user_context=_ctx(), ) e = comp_svc.add.call_args[0][0] assert e.component_type == "system" assert e.description == "d" assert e.purpose == "p" assert e.status == "operational" ``` - [ ] **Step 2: 寫 impl** ```python # domain/oscal/service/write_strategy/component_write_strategy.py """ComponentWriteStrategy — Phase 2 Task 5B. Single strategy that writes all ParsedComponent rows (any component_type) to oscal.ssp_components. Supersedes DeviceWriteStrategy + InformationSystemWriteStrategy (those file are removed in Task 11 cleanup). leveraged_authorization_ref (title string) resolution: lookup oscal.ssp_leveraged_authorizations where (ssp_id, title) match → use uid. Caller (SspImportConfirmService) MUST run LeveragedWriteStrategy first so the LA rows exist for this ssp_id. Overwrite semantics: delete all existing components for ssp_id before insert. """ from __future__ import annotations import logging from typing import Iterable, Optional from domain.oscal.parser.ssp_intermediate import ParsedComponent logger = logging.getLogger(__name__) def _strip_nulls(d: dict) -> Optional[dict]: cleaned = {k: v for k, v in d.items() if v is not None and v != ""} return cleaned or None class ComponentWriteStrategy: def __init__(self, component_domain_service, leveraged_authorization_domain_service): self._comp = component_domain_service self._la = leveraged_authorization_domain_service def write( self, parsed_components: Iterable[ParsedComponent], ssp_id: int, user_context, ) -> int: from jedi_oscal.domain.entity.base.oscal_component_entity import ComponentEntity from jedi_oscal.domain.entity.base.oscal_component_query_entity import ( ComponentQueryEntity, ) from jedi_oscal.domain.entity.base.oscal_leveraged_authorization_query_entity import ( LeveragedAuthorizationQueryEntity, ) tenant_id = getattr(user_context, "tenant_id", None) org_unit_id = getattr(user_context, "org_unit_id", None) login_name = getattr(user_context, "login_name", None) if not login_name: raise ValueError("ComponentWriteStrategy.write: user_context.login_name required") # Overwrite: 砍 ssp_id 全部 components existing = self._comp.get_all(ComponentQueryEntity(ssp_id=ssp_id)) or [] to_delete = [c.id for c in existing if c.id is not None] if to_delete: self._comp.delete_by_ids(to_delete) # 預載 LA title → uid map（一次 query 替整批解 ref） existing_las = self._la.get_all( LeveragedAuthorizationQueryEntity(ssp_id=ssp_id) ) or [] la_title_to_uid = {la.title: la.uid for la in existing_las if la.title and la.uid} written = 0 for parsed in parsed_components: la_uid = None if parsed.leveraged_authorization_ref: la_uid = la_title_to_uid.get(parsed.leveraged_authorization_ref) if la_uid is None: logger.info( "ComponentWriteStrategy: LA ref %r unmatched for ssp_id=%s — writing NULL", parsed.leveraged_authorization_ref, ssp_id, ) props = _strip_nulls({ "protocol": parsed.protocol, "port_ranges": parsed.port_ranges, "security_auth": parsed.security_auth, }) entity = ComponentEntity( ssp_id=ssp_id, component_type=parsed.component_type, title=parsed.title, description=parsed.description, purpose=parsed.purpose, status=parsed.status, leveraged_authorization_uid=la_uid, props=props, tenant_id=tenant_id, org_unit_id=org_unit_id, created_user=login_name, updated_user=login_name, ) try: self._comp.add(entity) written += 1 except Exception as e: logger.warning( "ComponentWriteStrategy: insert failed ssp_id=%s title=%r type=%s: %s", ssp_id, parsed.title, parsed.component_type, e, exc_info=True, ) return written ``` - [ ] **Step 3: 跑 test 確認 PASS** ```bash pytest tests/test_component_write_strategy.py -v # Expected: 5 passed ``` - [ ] **Step 4: Commit** ```bash git add domain/oscal/service/write_strategy/component_write_strategy.py \ tests/test_component_write_strategy.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 5B — ComponentWriteStrategy v3 (consume ParsedComponent, resolve LA ref)" ``` ### Task 5.C: New `InventoryItemWriteStrategy`（含 M2M） **Files:** - Create: `domain/oscal/service/write_strategy/inventory_item_write_strategy.py` - Test: `tests/test_inventory_item_write_strategy.py` **Approach:** 寫 `oscal.ssp_inventory_items` row + M2M `oscal.ssp_inventory_implemented_components`。 - 對每個 `ParsedInventoryItem`，resolve `implemented_component_refs` (title list) → `ssp_components.uid` list（**lookup by (ssp_id, title) 取第一筆 service component，fallback 任意 first**） - 透過 `InventoryItemDomainService.add()`（Phase 1 已 ship，§11.8 已 hardened M2M observability） Overwrite semantics：刪 ssp_id 全部 inventory rows 後 insert（CASCADE 自動清 M2M join）。 - [ ] **Step 1: 寫 failing tests** ```python # tests/test_inventory_item_write_strategy.py import uuid from unittest.mock import Mock from domain.oscal.parser.ssp_intermediate import ParsedInventoryItem from domain.oscal.service.write_strategy.inventory_item_write_strategy import ( InventoryItemWriteStrategy, ) def _ctx(): return Mock(login_name="alice", tenant_id=1, org_unit_id=99) def test_write_with_no_refs(): inv_svc = Mock(); inv_svc.get_all.return_value = [] comp_svc = Mock(); comp_svc.get_all.return_value = [] strat = InventoryItemWriteStrategy(inv_svc, comp_svc) out = strat.write([ParsedInventoryItem(description="Server X")], 10, _ctx()) assert out == 1 entity = inv_svc.add.call_args[0][0] assert entity.description == "Server X" assert entity.implemented_component_uids == [] def test_write_resolves_refs_via_component_lookup(): inv_svc = Mock(); inv_svc.get_all.return_value = [] comp_a_uid = uuid.uuid4() comp_a = Mock(uid=comp_a_uid, title="CompA", component_type="hardware") comp_svc = Mock(); comp_svc.get_all.return_value = [comp_a] strat = InventoryItemWriteStrategy(inv_svc, comp_svc) out = strat.write( [ParsedInventoryItem(description="Srv", implemented_component_refs=["CompA"])], ssp_id=10, user_context=_ctx(), ) entity = inv_svc.add.call_args[0][0] assert entity.implemented_component_uids == [comp_a_uid] def test_write_ambiguous_picks_service_type_first(): inv_svc = Mock(); inv_svc.get_all.return_value = [] hw_uid, svc_uid = uuid.uuid4(), uuid.uuid4() comp_hw = Mock(uid=hw_uid, title="X", component_type="hardware") comp_svc_row = Mock(uid=svc_uid, title="X", component_type="service") comp_svc = Mock(); comp_svc.get_all.return_value = [comp_hw, comp_svc_row] strat = InventoryItemWriteStrategy(inv_svc, comp_svc) strat.write( [ParsedInventoryItem(description="S", implemented_component_refs=["X"])], ssp_id=10, user_context=_ctx(), ) entity = inv_svc.add.call_args[0][0] assert entity.implemented_component_uids == [svc_uid] def test_write_unmatched_ref_drops(): inv_svc = Mock(); inv_svc.get_all.return_value = [] comp_svc = Mock(); comp_svc.get_all.return_value = [] strat = InventoryItemWriteStrategy(inv_svc, comp_svc) strat.write( [ParsedInventoryItem(description="S", implemented_component_refs=["NotExist"])], ssp_id=10, user_context=_ctx(), ) entity = inv_svc.add.call_args[0][0] assert entity.implemented_component_uids == [] def test_overwrite_deletes_then_inserts(): e1 = Mock(id=1) inv_svc = Mock(); inv_svc.get_all.return_value = [e1] comp_svc = Mock(); comp_svc.get_all.return_value = [] strat = InventoryItemWriteStrategy(inv_svc, comp_svc) strat.write([ParsedInventoryItem(description="S")], 10, _ctx()) inv_svc.delete_by_ids.assert_called_once_with([1]) assert inv_svc.add.called def test_props_jsonb_built_from_optional_fields(): inv_svc = Mock(); inv_svc.get_all.return_value = [] comp_svc = Mock(); comp_svc.get_all.return_value = [] strat = InventoryItemWriteStrategy(inv_svc, comp_svc) strat.write( [ParsedInventoryItem(description="S", asset_id="A1", ipv4_address="10.0.0.1")], 10, _ctx(), ) entity = inv_svc.add.call_args[0][0] # asset_id / ipv4 等放 props 還是 first-class 欄位，依 entity 設計而定 # — 詳見 Phase 1 InventoryItemEntity；先檢查不為 None assert entity is not None ``` - [ ] **Step 2: 寫 impl** ```python # domain/oscal/service/write_strategy/inventory_item_write_strategy.py """InventoryItemWriteStrategy — Phase 2 Task 5C. Writes oscal.ssp_inventory_items + M2M oscal.ssp_inventory_implemented_components. Component refs (title list) resolved via (ssp_id, title) lookup; ambiguous title picks first component_type='service' (else first match). Caller MUST run ComponentWriteStrategy first so component rows exist. Overwrite semantics: delete all inventory rows for ssp_id; CASCADE drops M2M. """ from __future__ import annotations import logging from typing import Iterable, Optional from domain.oscal.parser.ssp_intermediate import ParsedInventoryItem logger = logging.getLogger(__name__) def _strip_nulls(d: dict) -> Optional[dict]: cleaned = {k: v for k, v in d.items() if v is not None and v != ""} return cleaned or None class InventoryItemWriteStrategy: def __init__(self, inventory_item_domain_service, component_domain_service): self._inv = inventory_item_domain_service self._comp = component_domain_service def write( self, parsed_inventory: Iterable[ParsedInventoryItem], ssp_id: int, user_context, ) -> int: from jedi_oscal.domain.entity.base.oscal_inventory_item_entity import ( InventoryItemEntity, ) from jedi_oscal.domain.entity.base.oscal_inventory_item_query_entity import ( InventoryItemQueryEntity, ) from jedi_oscal.domain.entity.base.oscal_component_query_entity import ( ComponentQueryEntity, ) tenant_id = getattr(user_context, "tenant_id", None) org_unit_id = getattr(user_context, "org_unit_id", None) login_name = getattr(user_context, "login_name", None) if not login_name: raise ValueError("InventoryItemWriteStrategy.write: user_context.login_name required") # Overwrite: 砍 ssp_id 全部 inventory (CASCADE 自動清 M2M) existing = self._inv.get_all(InventoryItemQueryEntity(ssp_id=ssp_id)) or [] to_delete = [e.id for e in existing if e.id is not None] if to_delete: self._inv.delete_by_ids(to_delete) # 預載 (title → list[(uid, component_type)]) 解 ref ambiguity components = self._comp.get_all(ComponentQueryEntity(ssp_id=ssp_id)) or [] title_to_candidates: dict[str, list[tuple]] = {} for c in components: if c.title and c.uid: title_to_candidates.setdefault(c.title, []).append((c.uid, c.component_type)) written = 0 for parsed in parsed_inventory: resolved_uids: list = [] for ref in parsed.implemented_component_refs: candidates = title_to_candidates.get(ref, []) if not candidates: continue # normalizer 已 emit warning，這裡只 drop # 取 component_type='service' 第一筆，fallback 任意第一筆 service_match = next( ((uid, _) for (uid, t) in candidates if t == "service"), None ) resolved_uids.append(service_match[0] if service_match else candidates[0][0]) props = _strip_nulls({ "asset_id": parsed.asset_id, "asset_tag": parsed.asset_tag, "ipv4_address": parsed.ipv4_address, "mac_address": parsed.mac_address, "fqdn": parsed.fqdn, "hostname": parsed.hostname, "software_name": parsed.software_name, "os_name": parsed.os_name, }) entity = InventoryItemEntity( ssp_id=ssp_id, description=parsed.description, props=props, implemented_component_uids=resolved_uids, tenant_id=tenant_id, org_unit_id=org_unit_id, created_user=login_name, updated_user=login_name, ) try: self._inv.add(entity) written += 1 except Exception as e: logger.warning( "InventoryItemWriteStrategy: insert failed ssp_id=%s desc=%r: %s", ssp_id, parsed.description, e, exc_info=True, ) return written ``` - [ ] **Step 3: 跑 test 確認 PASS** ```bash pytest tests/test_inventory_item_write_strategy.py -v ``` - [ ] **Step 4: Commit** ```bash git add domain/oscal/service/write_strategy/inventory_item_write_strategy.py \ tests/test_inventory_item_write_strategy.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 5C — InventoryItemWriteStrategy v3 (consume ParsedInventoryItem, M2M write)" ``` --- ## Task 6: Refactor `SspEntityReconciliationOrchestrator` 讀 v3 bundle **Files:** - Modify: `domain/oscal/service/reconciliation/ssp_entity_orchestrator.py` - Modify: `tests/test_a4_reconciliation_orchestrator.py` **Current**：line 52-54 讀 `parsed.parsed_devices` / `parsed_info_systems` / `parsed_leveraged`（v2 shape，在 v3 bundle 上 AttributeError）。 **New**：讀 v3 bundle 欄位，dispatch 既有 reconciler： - `device_reconciler.reconcile(parsed_components_filter_by_hw, tenant_id)` — hardware type - `information_system_reconciler.reconcile(parsed_components_filter_by_system, tenant_id)` — system type - `leveraged_reconciler.reconcile(parsed_leveraged_authorizations, tenant_id)` 或者改 reconcilers signature 直接吃 v3 dataclasses。本 Task 採 **wrap 過濾後傳給既有 reconciler**（最低破壞性），reconciler 內部仍消費 ParsedDevice / ParsedInformationSystem，由 orchestrator 構造 throwaway wrapper objects 餵進去 — 或更乾淨：reconciler 改 signature 吃 `ParsedComponent` filtered list（**選 B**）。 實作 ：reconciler 改吃 `List[ParsedComponent]` filtered by `component_type`。 ⚠️ 此處 reconcilers 也要動到（dataclass 對應改名）。詳見 `device_reconciler.py`/`information_system_reconciler.py`/`leveraged_reconciler.py` 內既有匹配邏輯。Phase 2 範圍把這三檔 dataclass type hint + 屬性 access 跟著改。 - [ ] **Step 1: 改 orchestrator + 既有 reconcilers** ```python # domain/oscal/service/reconciliation/ssp_entity_orchestrator.py（refactor） """SSP entity orchestrator v3 — read bundle.parsed_components (filtered by type) + parsed_leveraged_authorizations + parsed_inventory_items. Phase 2 Task 6 refactor: signature unchanged (reconcile(parsed, ctx)), but internally reads v3 fields and filters components by type before dispatching to device / information_system / leveraged reconcilers. reconcilers' .reconcile() signature unchanged — caller now passes ParsedComponent list filtered by type instead of ParsedDevice / ParsedInformationSystem lists. """ from domain.oscal.parser.ssp_intermediate import ( ParsedExcelEntityBundle, SspEntityReconciliationContext, ) class SspEntityReconciliationOrchestrator: def __init__( self, catalog_control_reconciler, assessment_objective_reconciler, device_reconciler, information_system_reconciler, leveraged_reconciler, system_characteristic_reconciler=None, ): self._ctrl = catalog_control_reconciler self._ao = assessment_objective_reconciler self._device = device_reconciler self._info_system = information_system_reconciler self._leveraged = leveraged_reconciler self._system_characteristic = system_characteristic_reconciler def reconcile( self, parsed: ParsedExcelEntityBundle, ctx: SspEntityReconciliationContext, ) -> None: # 1. controls self._ctrl.reconcile(parsed.parsed_controls, ctx.catalog_id) # 2. AO 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. components — filter by type then dispatch hardware_components = [ c for c in parsed.parsed_components if c.component_type == "hardware" ] system_components = [ c for c in parsed.parsed_components if c.component_type == "system" ] # device_reconciler now consumes List[ParsedComponent] (was ParsedDevice) self._device.reconcile(hardware_components, ctx.tenant_id) # information_system_reconciler now consumes List[ParsedComponent] self._info_system.reconcile(system_components, ctx.tenant_id) # 4. leveraged_authorizations (was: ParsedLeveraged list) self._leveraged.reconcile(parsed.parsed_leveraged_authorizations, ctx.tenant_id) # 5. system_characteristic if ( self._system_characteristic is not None and parsed.parsed_system_characteristic is not None ): self._system_characteristic.reconcile_single( parsed.parsed_system_characteristic, ctx.tenant_id, ) ``` - [ ] **Step 2: 更新 3 個 reconcilers signature** `device_reconciler.py` / `information_system_reconciler.py` / `leveraged_reconciler.py` — 把屬性 access 從 v2 dataclass 改 v3： ```bash # 看每個 reconciler 內 access 的 attr grep -nE 'parsed\.\w+|p\.\w+' domain/oscal/service/reconciliation/device_reconciler.py grep -nE 'parsed\.\w+|p\.\w+' domain/oscal/service/reconciliation/information_system_reconciler.py grep -nE 'parsed\.\w+|p\.\w+' domain/oscal/service/reconciliation/leveraged_reconciler.py ``` 對映： - `ParsedDevice.name` → `ParsedComponent.title` - `ParsedDevice.purpose` → `ParsedComponent.description` 或 `.purpose` - `ParsedDevice.matched_device_id` → 移除（component 沒 device FK） - `ParsedInformationSystem.name` → `ParsedComponent.title` - `ParsedInformationSystem.description` → `ParsedComponent.description` - `ParsedInformationSystem.matched_info_system_id` → 移除（同上） - `ParsedLeveraged.service_name` → `ParsedLeveragedAuthorization.title` - `ParsedLeveraged.provider` → `ParsedLeveragedAuthorization.provider` - `ParsedLeveraged.matched_party_uuid` → 暫無對應 attr（v3 LA 沒這欄）→ reconciler 改成 emit warning 或 skip 實作策略：每個 reconciler 內把 attr name 對齊 v3，刪除 v3 不存在的 matched_*_id 設值邏輯（這些 lookup 已不在 ComponentEntity 內，移到 Component table 的 lookup 由 write strategy 處理）。 詳見每個 reconciler 既有 code，邊改邊跑 test。 - [ ] **Step 3: 跑 orchestrator + reconciler test** ```bash pytest tests/test_a4_reconciliation_orchestrator.py -v pytest tests/ -k "device_reconciler or information_system_reconciler or leveraged_reconciler" -v ``` - [ ] **Step 4: Commit** ```bash git add domain/oscal/service/reconciliation/ tests/test_a4_reconciliation_orchestrator.py git commit -m "refactor(ssp-oscal-alignment): Phase 2 Task 6 — SspEntityReconciliationOrchestrator + 3 reconcilers read v3 bundle (ParsedComponent / ParsedLeveragedAuthorization)" ``` --- ## Task 7: Wire `ssp_excel_import_app_service` through new pipeline **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py`（`_write_all_data` + 3 個 flow） - Modify: `di_containers/oscal/oscal_container.py`（wire `SspImportConfirmService` + 移除舊 device/info_system strategy provider；改 leveraged strategy ctor） - Modify: `tests/test_ssp_excel_import_app_service.py`（既有 test fixture 跟著對齊） **Approach:** 在 `_write_all_data` 內偵測 `parsed_result.get("schema_version")`： - `v2-bundle` → 走 `confirm_service.confirm(parsed_result, ssp_id=..., catalog_id=..., user_context=...)` 一條 line - 其他（v1-ssp / None）→ 走既有 `_dict_to_parsed_devices` + orchestrator + 3 strategies 邏輯（**legacy path**，原樣保留） ⚠️ 注意：legacy path 依舊用 `ParsedExcelEntityBundle(parsed_devices=...)` — 但 v3 bundle 已不含此欄位。Legacy path 需要包成「**不用 bundle，直接 reconciler/strategy 餵 raw list**」的形式，或者另存 `_LegacyParsedBundle` namespace class（簡單 SimpleNamespace 即可）讓 orchestrator 走得通。 **簡化選擇**：legacy path 改成「直接呼叫 reconcilers + strategies」，不再經 orchestrator。orchestrator 只負責新 path。 → 既有 a2/a3/a4 test 都用 mock orchestrator/strategy dep，這個改不影響它們。 → legacy path 退化成「dict → ParsedDevice 等 v2 dataclass → strategies.write」直線跑（無 reconcile，這跟 Phase 1 前 a4 ship 行為相同）。 實作： ```python # 在 _write_all_data 內，Step 3 起改： schema_version = parsed_result.get("schema_version", "v1-ssp") if schema_version == "v2-bundle": # 新 path if ssp_id is None or self._confirm_service is None: logger.warning("v2-bundle confirm skipped: ssp_id=%s confirm_service=%s", ssp_id, self._confirm_service is not None) write_stats = {"warnings_added": 0, "written_components": 0, "written_las": 0, "written_inventory": 0} else: write_stats = self._confirm_service.confirm( parsed_result=parsed_result, ssp_id=ssp_id, catalog_id=catalog_id, user_context=user_context, ) # parsed_controls / parsed_sc 維持既有 path（control/AO/system_char write 不換） self._write_controls_and_sc(parsed_result, ssp_id, sys_impl_main_id, user_context, ...) else: # legacy v1-ssp path — 既有 Step 3-6 邏輯保留（不經 orchestrator 拼 bundle） ... ``` ⚠️ 規格要點：`_write_all_data` 簽名不變，內部依 schema_version 分流。Caller (`_confirm_*_flow`) 不動。 - [ ] **Step 1: 加 `_confirm_service` dep + ctor** ```python # ssp_excel_import_app_service.py:76 __init__ 加參數 def __init__( self, ..., confirm_service=None, # ← 新加 ..., ): ... self._confirm_service = confirm_service ``` - [ ] **Step 2: refactor `_write_all_data` 拆 v2-bundle / legacy 兩路徑** 依上面 sketch，把 Step 3-6 內舊邏輯包成 `_legacy_write_path(parsed_result, ssp_id, sys_impl_main_id, user_context, fw_version, catalog_id)`，新 path 包成 `_v2_bundle_write_path(...)`，外層 if/else dispatch。 - [ ] **Step 3: DI wire** ```python # di_containers/oscal/oscal_container.py — 新增 from domain.oscal.import_pipeline.confirm_service import SspImportConfirmService from domain.oscal.service.write_strategy.component_write_strategy import ComponentWriteStrategy from domain.oscal.service.write_strategy.inventory_item_write_strategy import InventoryItemWriteStrategy component_write_strategy = providers.Singleton( ComponentWriteStrategy, component_domain_service=oscal_container.component_domain_service, leveraged_authorization_domain_service=oscal_container.leveraged_authorization_domain_service, ) inventory_item_write_strategy = providers.Singleton( InventoryItemWriteStrategy, inventory_item_domain_service=oscal_container.inventory_item_domain_service, component_domain_service=oscal_container.component_domain_service, ) ssp_import_confirm_service = providers.Singleton( SspImportConfirmService, orchestrator=ssp_entity_orchestrator, leveraged_write_strategy=leveraged_write_strategy, component_write_strategy=component_write_strategy, inventory_write_strategy=inventory_item_write_strategy, ) ssp_excel_import_app_service = providers.Singleton( SspExcelImportAppService, ..., confirm_service=ssp_import_confirm_service, ..., ) ``` 同時： - `leveraged_write_strategy` Provider 移除 `component_domain_service` 參數（Task 5A signature 改了） - 既有 `device_write_strategy` / `information_system_write_strategy` Provider 暫保留（legacy path 還用），Task 11 cleanup 才刪 - [ ] **Step 4: 既有 excel app service test 對齊** 跑 `pytest tests/test_ssp_excel_import_app_service.py -v` 看哪些 fixture 需更新（多半是 mock service 簽名跟著改）。 - [ ] **Step 5: 跑 full excel test suite** ```bash pytest tests/test_ssp_excel_import_app_service.py tests/test_ssp_excel_import_ssp_source.py tests/test_ssp_excel_import_system_characteristic_e2e.py -v ``` - [ ] **Step 6: Commit** ```bash git add app/oscal/service/ssp_excel_import_app_service.py \ di_containers/oscal/oscal_container.py \ tests/test_ssp_excel_import_app_service.py \ tests/test_ssp_excel_import_ssp_source.py \ tests/test_ssp_excel_import_system_characteristic_e2e.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 7 — Excel app service wire SspImportConfirmService (v2-bundle path)" ``` --- ## Task 8: `CmmcSspAdapter` 加 `adapt_to_bundle()` 輸出 v3 **Files:** - Modify: `domain/oscal/adapter/cmmc_ssp_adapter.py`（加新 method，保留既有 `adapt()` 不動 — docx_import_parity Stage 1 caller 仍用） - Modify: `domain/oscal/adapter/i_ssp_docx_adapter.py`（加 abstract `adapt_to_bundle`） - Test: `tests/test_cmmc_ssp_adapter_v3_bundle.py` **Approach:** 加 `adapt_to_bundle(parsed_docx, structure, candidates, doc) → ParsedExcelEntityBundle`，內部 call 既有 `_enrich_with_section_extractors` logic 但 mapping 到 v3 dataclass： - Docx Table #6 (CSP/CSO 7-col, FedRAMP) → `ParsedLeveragedAuthorization` + paired `ParsedComponent(component_type='service', leveraged_authorization_ref=la.title)` - Docx Table #7 (Category 5-col) → `ParsedComponent` only（component_type 由 `cat_zh_to_oscal` 對映） - Parties / metadata / system_characteristic 等不變（既有 v2.5 `ParsedLeveragedService` enrich 不動），先抽 docx tables → bundle, parties 沿用既有 ```python # domain/oscal/adapter/cmmc_ssp_adapter.py 加 method def adapt_to_bundle( self, parsed_docx: ParsedDocx, structure: Optional[Any], candidates: List[CandidateControl], tenant_id: int = None, doc: Optional[Any] = None, ) -> ParsedExcelEntityBundle: """Phase 2: Output v3 ParsedExcelEntityBundle (replaces adapt → ParsedSsp). Existing adapt() preserved for back-compat (Stage 1 callers); will be deprecated after all docx callers migrate. """ from domain.oscal.parser.docx_section_extractors import ( extract_leveraged_category_table, extract_leveraged_csp_table, extract_metadata_table, extract_party_tables, extract_revision_history_table, extract_system_characteristic_from_metadata, ) from domain.oscal.parser.ssp_intermediate import ( ParsedComponent, ParsedExcelEntityBundle, ParsedLeveragedAuthorization, ParsedParty, ParsedSystemCharacteristic, ) parsed_las: list[ParsedLeveragedAuthorization] = [] parsed_components: list[ParsedComponent] = [] parsed_parties: list[ParsedParty] = [] parsed_sc: Optional[ParsedSystemCharacteristic] = None parsed_metadata: Optional[dict] = None if doc is not None: # --- Metadata --- meta = extract_metadata_table(doc) or {} parsed_metadata = {k: v for k, v in meta.items() if v} # --- System characteristic --- sc_dict = extract_system_characteristic_from_metadata(doc) if sc_dict and (sc_dict.get("system_name") or meta.get("title")): parsed_sc = ParsedSystemCharacteristic( system_name=sc_dict.get("system_name") or meta.get("title") or "", security_sensitivity_level=sc_dict.get("security_sensitivity_level"), ) # --- Parties --- for pd in extract_party_tables(doc): name = (pd.get("name") or "").strip() if not name and pd.get("_placeholder"): continue parsed_parties.append(ParsedParty( name=name or "(未命名)", party_type=pd.get("party_type", "person"), title=pd.get("title"), email_address=pd.get("email"), telephone_number=pd.get("telephone"), address=pd.get("address"), )) # --- Table #6 CSP/CSO (FedRAMP) --- for row in extract_leveraged_csp_table(doc): title = row.get("service_name") or row.get("provider") or "" if not title: continue la = ParsedLeveragedAuthorization( title=title, provider=row.get("provider"), fedramp_package_id=row.get("fedramp_package_id"), impact_level=_normalize_impact(row.get("impact_level")), data_types=row.get("data_types"), nature_of_agreement=row.get("nature_of_agreement"), authorized_users=row.get("authorized_users"), ) parsed_las.append(la) # Paired service component pointing back to LA parsed_components.append(ParsedComponent( title=title, component_type="service", description=row.get("purpose"), leveraged_authorization_ref=la.title, )) # --- Table #7 Category --- for row in extract_leveraged_category_table(doc): title = row.get("name_description") or "" if not title: continue comp_type = _cat_zh_to_oscal(row.get("category")) parsed_components.append(ParsedComponent( title=title, component_type=comp_type, description=row.get("purpose"), protocol=row.get("protocol"), security_auth=row.get("security_auth"), )) return ParsedExcelEntityBundle( parsed_system_characteristic=parsed_sc, parsed_parties=parsed_parties, parsed_components=parsed_components, parsed_leveraged_authorizations=parsed_las, parsed_inventory_items=[], # Phase 2: docx 樣板還沒 inventory 表（Phase 3 Excel template 加） parsed_controls=[], # control 仍走 matched_controls 既有 path parsed_metadata=parsed_metadata, ) # Module-level helpers _CAT_ZH_TO_OSCAL = { "外部服務": "service", "互連": "interconnection", "API": "service", "CLI": "service", } def _cat_zh_to_oscal(zh: Optional[str]) -> str: return _CAT_ZH_TO_OSCAL.get((zh or "").strip(), "service") def _normalize_impact(raw: Optional[str]) -> Optional[str]: if not raw: return None m = {"Low": "low", "Moderate": "moderate", "High": "high", "LI-SaaS": "li-saas"} return m.get(raw.strip(), raw.lower()) ``` - [ ] **Step 1: 寫 failing tests** ```python # tests/test_cmmc_ssp_adapter_v3_bundle.py from unittest.mock import Mock, patch from domain.oscal.adapter.cmmc_ssp_adapter import CmmcSspAdapter def test_adapt_to_bundle_t6_creates_la_and_paired_component(): adapter = CmmcSspAdapter() parsed_docx = Mock(matched_controls=[], unmatched_paragraphs=[], summary={}) fake_doc = Mock() with patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_csp_table", return_value=[{ "service_name": "AcmeCorp Backup", "provider": "AcmeCorp", "fedramp_package_id": "FR-001", "impact_level": "Moderate", }]), \ patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_category_table", return_value=[]), \ patch("domain.oscal.parser.docx_section_extractors.extract_metadata_table", return_value={}), \ patch("domain.oscal.parser.docx_section_extractors.extract_party_tables", return_value=[]), \ patch("domain.oscal.parser.docx_section_extractors.extract_revision_history_table", return_value=[]), \ patch("domain.oscal.parser.docx_section_extractors.extract_system_characteristic_from_metadata", return_value={}): bundle = adapter.adapt_to_bundle(parsed_docx, None, [], doc=fake_doc) assert len(bundle.parsed_leveraged_authorizations) == 1 la = bundle.parsed_leveraged_authorizations[0] assert la.title == "AcmeCorp Backup" assert la.fedramp_package_id == "FR-001" assert la.impact_level == "moderate" assert len(bundle.parsed_components) == 1 comp = bundle.parsed_components[0] assert comp.component_type == "service" assert comp.leveraged_authorization_ref == "AcmeCorp Backup" def test_adapt_to_bundle_t7_category_creates_component_only(): adapter = CmmcSspAdapter() parsed_docx = Mock(matched_controls=[], unmatched_paragraphs=[], summary={}) fake_doc = Mock() with patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_csp_table", return_value=[]), \ patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_category_table", return_value=[{ "name_description": "GitLab", "category": "外部服務", "protocol": "HTTPS", "security_auth": "OAuth", }]), \ patch("domain.oscal.parser.docx_section_extractors.extract_metadata_table", return_value={}), \ patch("domain.oscal.parser.docx_section_extractors.extract_party_tables", return_value=[]), \ patch("domain.oscal.parser.docx_section_extractors.extract_revision_history_table", return_value=[]), \ patch("domain.oscal.parser.docx_section_extractors.extract_system_characteristic_from_metadata", return_value={}): bundle = adapter.adapt_to_bundle(parsed_docx, None, [], doc=fake_doc) assert len(bundle.parsed_leveraged_authorizations) == 0 assert len(bundle.parsed_components) == 1 comp = bundle.parsed_components[0] assert comp.title == "GitLab" assert comp.component_type == "service" assert comp.protocol == "HTTPS" assert comp.security_auth == "OAuth" ``` - [ ] **Step 2: 寫 impl + 既有 interface 加 abstract method** ```python # domain/oscal/adapter/i_ssp_docx_adapter.py @abstractmethod def adapt_to_bundle( self, parsed_docx, structure, candidates, tenant_id=None, doc=None, ) -> "ParsedExcelEntityBundle": ... ``` - [ ] **Step 3: 跑 test 確認 PASS** ```bash pytest tests/test_cmmc_ssp_adapter_v3_bundle.py tests/test_cmmc_ssp_adapter.py -v ``` - [ ] **Step 4: Commit** ```bash git add domain/oscal/adapter/cmmc_ssp_adapter.py \ domain/oscal/adapter/i_ssp_docx_adapter.py \ tests/test_cmmc_ssp_adapter_v3_bundle.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 8 — CmmcSspAdapter.adapt_to_bundle outputs v3 bundle" ``` --- ## Task 9: Wire `ssp_docx_import_app_service` through new pipeline **Files:** - Modify: `app/oscal/service/ssp_docx_import_app_service.py` - Modify: `di_containers/oscal/oscal_container.py`（docx app service inject `confirm_service`） - Modify: `tests/test_ssp_docx_import_app_service.py` **Approach:** Confirm path 內： 1. parse 階段把 `adapter.adapt_to_bundle(...)` 輸出 dump 為 dict 存 `parse_jobs.parsed_result`，**標 `schema_version="v2-bundle"`** 2. confirm 階段 call `confirm_service.confirm(parsed_result, ssp_id, catalog_id, user_context)` — 跟 excel 共用同一 service 3. legacy v1-ssp 路徑（已 in-flight 的舊 docx parse_jobs）：原 `adapt() → ParsedSsp` path 保留不動 - [ ] **Step 1: 讀現有 docx import app service** ```bash grep -n "def " app/oscal/service/ssp_docx_import_app_service.py | head -30 ``` - [ ] **Step 2: 加 `confirm_service` dep + parse 階段標 schema_version** ```python # 在 upload_and_parse_docx 內，bundle 寫入 parsed_result 前加： bundle = self._adapter.adapt_to_bundle(parsed_docx, structure, candidates, doc=doc) parsed_result = self._bundle_to_dict(bundle) parsed_result["schema_version"] = "v2-bundle" # 沿用既有 parse_jobs.write_parsed_result(uid, parsed_result, ...) ``` 需新增 helper `_bundle_to_dict(bundle) → dict`：跟 `bundle_restore.dict_to_bundle` 互逆，每 dataclass → asdict。可直接 `dataclasses.asdict(bundle)` + serialize date/UUID。 - [ ] **Step 3: confirm 階段切到 confirm_service** ```python # confirm_docx_import 內，原本走 _adapter.adapt → ParsedSsp → write strategies 那段： schema_version = parsed_result.get("schema_version", "v1-ssp") if schema_version == "v2-bundle": write_stats = self._confirm_service.confirm( parsed_result=parsed_result, ssp_id=ssp_id, catalog_id=catalog_id, user_context=user_context, ) else: # 原本 legacy 路徑 write_stats = self._legacy_confirm(...) ``` - [ ] **Step 4: DI wire docx app service** ```python ssp_docx_import_app_service = providers.Singleton( SspDocxImportAppService, ..., confirm_service=ssp_import_confirm_service, ..., ) ``` - [ ] **Step 5: 既有 docx test 對齊** ```bash pytest tests/test_ssp_docx_import_app_service.py tests/test_ssp_docx_import_e2e.py tests/test_ssp_docx_import_create_no_diff.py -v ``` - [ ] **Step 6: Commit** ```bash git add app/oscal/service/ssp_docx_import_app_service.py \ di_containers/oscal/oscal_container.py \ tests/test_ssp_docx_import_*.py git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 9 — Docx app service wire SspImportConfirmService (v2-bundle path)" ``` --- ## Task 10: Pre-flight verification + smoke (end-to-end) **Files:** - Create: `tests/test_ssp_import_pipeline_smoke.py` - Read: `log/app.log` 啟動驗證 **Approach:** 不需 e2e HTTP test，但需 1 個 in-memory smoke 確認整條 chain 通： - Excel app service confirm with `schema_version=v2-bundle` → normalizer → orchestrator → 3 strategies → mock domain services → 統計回傳正確 - [ ] **Step 1: 寫 smoke test** ```python # tests/test_ssp_import_pipeline_smoke.py """Phase 2 smoke — confirm pipeline end-to-end with mock domain services. Verifies: - v2-bundle parsed_result → bundle_restore → normalize → strategies write - warnings_added 反映 normalizer 結果 - 寫 stats 累積到 confirm_service return dict """ from unittest.mock import Mock from domain.oscal.import_pipeline.confirm_service import SspImportConfirmService def test_smoke_v2_bundle_full_chain(): # Mock domain services (avoid real DB) la_svc = Mock(); la_svc.get_all.return_value = [] fake_saved_la = Mock(); fake_saved_la.uid = "la-uuid-1"; fake_saved_la.title = "Crowdstrike" la_svc.add.return_value = fake_saved_la comp_svc = Mock(); comp_svc.get_all.return_value = [] inv_svc = Mock(); inv_svc.get_all.return_value = [] from domain.oscal.service.write_strategy.leveraged_write_strategy import ( LeveragedWriteStrategy, ) from domain.oscal.service.write_strategy.component_write_strategy import ( ComponentWriteStrategy, ) from domain.oscal.service.write_strategy.inventory_item_write_strategy import ( InventoryItemWriteStrategy, ) confirm_service = SspImportConfirmService( orchestrator=None, # smoke 跳過 reconcile leveraged_write_strategy=LeveragedWriteStrategy(la_svc), component_write_strategy=ComponentWriteStrategy(comp_svc, la_svc), inventory_write_strategy=InventoryItemWriteStrategy(inv_svc, comp_svc), ) parsed_result = { "schema_version": "v2-bundle", "components": [ {"title": "AcmeCorp Backup", "component_type": "service", "leveraged_authorization_ref": "Crowdstrike"}, {"title": "Web Server", "component_type": "hardware"}, ], "leveraged_authorizations": [ {"title": "Crowdstrike", "fedramp_package_id": "FR-001"}, ], "inventory_items": [ {"description": "Srv 1", "implemented_component_refs": ["Web Server"]}, ], } user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99) stats = confirm_service.confirm( parsed_result=parsed_result, ssp_id=10, catalog_id=None, user_context=user_ctx, ) assert stats["written_las"] == 1 assert stats["written_components"] == 2 assert stats["written_inventory"] == 1 assert stats["warnings_added"] == 0 assert parsed_result["import_warnings"] == [] def test_smoke_v1_ssp_dispatch_raises(): from domain.oscal.import_pipeline.bundle_restore import LegacyConfirmRequired import pytest confirm_service = SspImportConfirmService( orchestrator=None, leveraged_write_strategy=Mock(), component_write_strategy=Mock(), inventory_write_strategy=Mock(), ) parsed_result = {"devices": [], "info_systems": [], "leveraged": []} with pytest.raises(LegacyConfirmRequired): confirm_service.confirm( parsed_result=parsed_result, ssp_id=10, catalog_id=None, user_context=Mock(login_name="x", tenant_id=1), ) ``` - [ ] **Step 2: 跑 smoke** ```bash pytest tests/test_ssp_import_pipeline_smoke.py -v ``` - [ ] **Step 3: 跑全套 BE pytest 確認 0 regression** ```bash pytest tests/ -x --ignore=tests/test_.py # Expected: ~102+N passed (N = Phase 2 新加 test 數，預估 30+)，0 failed ``` - [ ] **Step 4: 啟 BE 跑一輪 smoke** ```bash # kill 舊 process lsof -i :8000 -t | xargs -r kill -9 # 啟動 nohup poetry run python main_app.py > log/app.log 2>&1 & sleep 3 tail -50 log/app.log | grep -E "ERROR|Traceback|started|listening" # Expected: 看到 listen on 8000 + 無 startup ERROR ``` - [ ] **Step 5: Commit** ```bash git add tests/test_ssp_import_pipeline_smoke.py git commit -m "test(ssp-oscal-alignment): Phase 2 Task 10 — confirm pipeline end-to-end smoke" ``` --- ## Task 11: Cleanup — 刪舊 strategies + dead param + `ParsedLeveraged`/`ParsedDevice`/`ParsedInformationSystem` 三 dataclass **Files:** - Delete: `domain/oscal/service/write_strategy/device_write_strategy.py` - Delete: `domain/oscal/service/write_strategy/information_system_write_strategy.py` - Modify: `domain/oscal/service/write_strategy/base.py`（拿掉 `sys_impl_main_id` 入參、或整檔刪） - Modify: `domain/oscal/parser/ssp_intermediate.py`（刪 `ParsedDevice` / `ParsedInformationSystem` / `ParsedLeveraged` 3 個 v2 dataclass） - Modify: `app/oscal/service/ssp_excel_import_app_service.py`（刪 `_dict_to_parsed_devices` / `_dict_to_parsed_info_systems` / `_dict_to_parsed_leveraged` 3 helper + legacy path） - Modify: `di_containers/oscal/oscal_container.py`（移除舊 device/info_system strategy provider） - Modify: 既有 tests 引用 v2 dataclass 的全部更新 **前提：** Task 7-10 都過了，legacy path 已不被 prod 觸發（既有 in-flight parse_jobs 都 confirm 完）。**user 拍板才做**。 - [ ] **Step 1: Grep 殘留 caller** ```bash grep -rn "ParsedDevice\|ParsedInformationSystem\|ParsedLeveraged\b" --include="*.py" . grep -rn "DeviceWriteStrategy\|InformationSystemWriteStrategy" --include="*.py" . grep -rn "sys_impl_main_id" --include="*.py" . ``` - [ ] **Step 2: 刪檔 + 更新 import** 按 grep 結果逐處修。重點： - `ssp_excel_import_app_service.py` legacy `_v1_write_path` 整段刪（schema_version 必填） - `tests/test_ssp_write_strategy.py` 內 ParsedDevice/ParsedInformationSystem fixtures 刪或改 ParsedComponent - `base.py` 內 abstract method `_parsed_to_component_payload` 沒人 inherit 後整檔刪 - [ ] **Step 3: 跑全測試** ```bash pytest tests/ -x ``` - [ ] **Step 4: Commit** ```bash git add -A # ⚠️ 多個檔案，但 dispatch subagent 時要顯式列檔（per feedback_subagent_explicit_git_add） git commit -m "refactor(ssp-oscal-alignment): Phase 2 Task 11 — cleanup v2 strategies + ParsedDevice/ParsedInformationSystem/ParsedLeveraged + dead sys_impl_main_id" ``` --- ## Task 12: API contract 驗證（Phase 1 follow-up §6.1）+ Changelog batch 撰寫 **Files:** - Read: Excel template downstream consumers — grep `"implementation_type"` key - Verify: `SspImportTemplateAppService._build_info_systems` dict key 已 Phase 1 改為 `component_type`，**確認沒人依賴舊 key** - Create: `docs/changelog/2026-05-XX-feat-ssp-oscal-alignment-phase1-phase2.md` - Create: `docs/features/FR-028-2605-ssp-oscal-alignment/handoff/-phase1-phase2-FINAL-SUMMARY.md` **Approach:** Phase 1 + Phase 2 同 deploy unit，收尾時一次 batch 寫 changelog（per CLAUDE.md「收尾才 batch 寫」），列兩 phase 33+ commits + 行為差異。 - [ ] **Step 1: API contract grep** ```bash grep -rn '"implementation_type"' --include="*.py" --include="*.vue" --include="*.js" \ ~/Projects/Billows/Audit-Manager/compliance-manager-be \ ~/Projects/Billows/Audit-Manager/compliance-manager-fe \ ~/Projects/Billows/Audit-Manager/compliance-manager-test # Expected: 零殘留（Phase 1 已改）；若有，標 Phase 2 內補 patch ``` - [ ] **Step 2: 寫 changelog** ```markdown --- type: feat breaking: true modules: [oscal, ssp-oscal-alignment, jedi-oscal, ssp-excel-import, ssp-docx-import] commit: --- # SSP OSCAL Alignment — Phase 1 + Phase 2 ## 需求說明 （兩 phase 概述） ## 變更範圍 - jedi-oscal v0.0.18 → v0.1.0（新 entity Component / LeveragedAuthorization / InventoryItem） - 主專案 import_pipeline 新模組 + 4 個新 oscal 表 + RLS - ... ## API / 行為差異 - ... ## 測試 - 主專案 pytest 102 + Phase 2 新增 N 個 test - jedi-oscal pytest 84 - BE smoke 通 ## 參考 - design.md §11 (11 條 reconciliation) - handoff Phase 1 SUMMARY + Phase 2 SUMMARY ``` - [ ] **Step 3: 寫 final summary** `docs/features/FR-028-2605-ssp-oscal-alignment/handoff/-phase1-phase2-FINAL-SUMMARY.md` — 兩 phase 全收口報告，commits 清單 / 改動範圍 / 行為差異 / 已知 follow-up / 部署 handover。 - [ ] **Step 4: 對話紀錄 extract** ```bash poetry run python scripts/extract_claude_sessions.py \ --date 2026-05-25 --topic ssp-oscal-alignment-phase2 --auto # 跟 Phase 1 2026-05-24 那批分開存 ``` - [ ] **Step 5: jedi-oscal bump + Nexus push（user 拍板執行）** ⚠️ **Per CLAUDE.md「絕對禁止：自動執行套件發版 / 自動推 Nexus」— 等 user 明確說「進版」才執行：** ```bash cd ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal # bump pyproject.toml version: 0.0.18 → 0.1.0 # commit + push + 推 Nexus（per CLAUDE.md jedi-package SOP） ``` - [ ] **Step 6: 主專案 pin 還原 + commit** ```bash # pyproject.toml：path dep 註解回去，pin Nexus 0.1.0 poetry update jedi-oscal # smoke 確認 import OK poetry run python -c "import jedi_oscal; print(jedi_oscal.__version__)" git add pyproject.toml poetry.lock docs/changelog/* docs/features/FR-028-2605-ssp-oscal-alignment/handoff/* git commit -m "release(ssp-oscal-alignment): Phase 1+2 ship — jedi-oscal 0.1.0 + changelog + SUMMARY" ``` --- ## Phase 2 完成 Definition of Done - [ ] 新 `domain/oscal/import_pipeline/` 模組 4 檔（warnings / normalizer / bundle_restore / confirm_service）+ tests 全綠 - [ ] `ComponentWriteStrategy` + `InventoryItemWriteStrategy` 新增 + tests 全綠 - [ ] `LeveragedWriteStrategy` 重寫消費 `ParsedLeveragedAuthorization`（移除 dead `sys_impl_main_id`） - [ ] `SspEntityReconciliationOrchestrator` 讀 v3 bundle；3 個 reconciler signature 對齊 - [ ] `CmmcSspAdapter.adapt_to_bundle()` 落地（既有 `adapt()` 保留） - [ ] `ssp_excel_import_app_service` + `ssp_docx_import_app_service` 共用 `SspImportConfirmService` - [ ] DI container wire 完成 - [ ] 主專案 pytest 0 failed（102 + 30+ 新 test） - [ ] jedi-oscal pytest 0 failed（84，本 phase 不動套件） - [ ] BE startup smoke 0 ERROR - [ ] 兩 repo commits 全 local；user 拍板才 push + jedi-oscal bump 0.1.0 - [ ] changelog + final SUMMARY + conversation extract 三件 archive --- ## 反悔條件 / Risk register - **Legacy v1-ssp in-flight job 沒測到** → Phase 2 deploy 後 14 天觀察，發現踩坑可緊急加 `_legacy_confirm` patch（confirm_service catch LegacyConfirmRequired，caller fallback） - **`_dict_to_parsed_*` legacy helpers 被未發現 caller 依賴** → Task 11 grep 沒抓到，runtime 才炸 → 加 deprecation log 1 個 release 後再 delete - **Reconciler signature 改動破壞既有 a4 tests** → Task 6 大改，逐個 reconciler test 確認 → 若 a4 tests 大量需重寫，回頭討論是否分 2 commit - **CmmcSspAdapter docx Table #8/#9 inventory items** → Phase 2 沒抓 inventory（docx 樣板無對應表）；inventory 主要從 Excel 04 sheet 來，Phase 3 樣板更新時補 - **Bundle restore date format edge case** → 既有 parser 寫 `date_authorized` 為 string，restore 用 `date.fromisoformat` — 若 parser 偶有非 ISO 格式（如 "2024/01/15"）→ catch ValueError → 設 None + emit warning --- ## 已知 follow-up（Phase 3 接續） Phase 2 完成後 Phase 3 (Excel / Docx 樣板重設計) 處理： 1. **Excel template v2.6 → v3.0 重設計** — 04 改名「資產清冊」、05 刪、06 拆 06a/06b（per design.md §3.1） 2. **Docx 樣板加 components + inventory 表** — H2 System Components / Hardware Maintenance 兩新 section（per design.md §3.2） 3. **Field registry 抽取** — `import_pipeline/field_registry.py`，Excel ColumnDef + Docx mapping 都 reference（per Phase 2 brainstorm 建議「分階段做，Phase 3 自然觸發點」） 4. **新增 docx adapter 處理 inventory section extractor**（樣板加 inventory_table 後才有對應抽取） Phase 3 完成後再進 Phase 4 (FE 預覽 UI 重設計)。 (Phase 5 OSCAL Export 已 deferred — user 2026-05-25 拍板等全部定案後再做)