# Phase A4 Implementation Plan — 5 entity 鉤稽 + 寫入 + SSP shell + A3 retroactive > **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:** 在 A3 `BaseReconciliationService` 基礎上補完 5 個新 reconciler（device / info_system / leveraged / catalog_control / AO）+ 5 個 WriteStrategy（含 control / AO / device / info_system / leveraged 寫入路徑）+ superset flow 建 SSP shell + A2 §15.3 superset include_controls 反查整併；同 PR 順手 retroactive 補 A3 PersonReconciler / OrganizationReconciler 的 USER_SELECTED stage 0。 **Architecture:** Strategy Pattern + Generic Base + Facade — `BaseReconciliationService` 加 stage 0 default no-op hook，5 新 reconciler override hook；`SspEntityReconciliationOrchestrator` 統一 facade（enforce control → AO 依賴順序）；5 個 WriteStrategy 純 domain layer、不開 `@transaction`（caller 開）；superset flow 加 SSP shell 建立步驟（ssp / system_characteristic / system_implementation main）；update flow 既有 SSP 不存在則 raise `GRC_412066`。 **Tech Stack:** Python 3.11 / SQLAlchemy / dependency-injector / pytest / openpyxl / cucumber.js（test repo） --- > **Phase**：A4（Track A 第四階段） > **級別**：重型（brainstorm → design → plan → 開工） > **依賴**：A3 BE 已 shipped（67 個新 test + BaseReconciliationService + Person/Organization Reconciler）+ design-A4.md spec review approved > **預計工時**：BE ~5d + cucumber 0.25d ≈ **5.25 working day**（不含 plan/design 寫作） > **對應 design**：`docs/features/FR-011.2-2605-ssp-import-export-phase2/design-A4.md`（13 章） > **執行階段切分（6 session）**：Session A（design / done）→ B（plan / self）→ C（T0+T1+T2）→ D（T3+T4+T5）→ E（T6+T7+T8）→ F（T9+T10+T11） — 對齊 design-A4 §12 --- ## Task 切分概覽 | Session | Task | 主題 | repo | 預估 | 依賴 | Commit checkpoint | |---------|------|------|------|------|------|-------------------| | C | T0 | Implementer pre-flight verification（design §7.2 8 項）| BE | 0.25d | — | 併入 T1 commit | | C | T1 | base.py stage 0 + MatchMethod USER_SELECTED + _normalizers 補 4 helper + 5 dataclass + ParsedExcelEntityBundle + SspEntityReconciliationContext + ParsedParty 加 2 label 欄位 | BE | 0.5d | T0 | `feat(oscal): A4 T1 stage 0 hook + dataclass + helpers` | | C | T2 | 5 reconciler skeleton + orchestrator + DI Factory（只 skeleton；fuzzy / 完整演算法留 T3）| BE | 0.75d | T1 | `feat(oscal): A4 T2 reconciler + orchestrator skeleton + DI` | | D | T3 | 5 reconciler 完整演算法（Stage 0 USER_SELECTED + 1/2 + Leveraged fuzzy + system_owner 反查 + AO reconcile_with_control_map）| BE | 0.75d | T2 | `feat(oscal): A4 T3 reconciler full algorithm` | | D | T4 | A3 retroactive：PersonReconciler / OrganizationReconciler override `_try_user_selected_match` + `_dict_to_parsed_parties` 加 label 傳遞 | BE | 0.25d | T1（base.py stage 0 dispatch）| `feat(oscal): A4 T4 A3 retroactive USER_SELECTED stage` | | D | T5 | A2 §15.3 整併：`_confirm_superset_flow` 內 `include_controls` 反查 | BE | 0.25d | T3（CatalogControlReconciler）| `feat(oscal): A4 T5 A2 §15.3 superset include_controls 反查` | | E | T6 | SSP shell：`_create_ssp_shell` + `_resolve_existing_ssp_shell` + `GRC_EXCEL_UPDATE_FLOW_NO_SSP` error code | BE | 0.5d | T0（jedi-oscal verify）| `feat(oscal): A4 T6 SSP shell + error code` | | E | T7 | 5 WriteStrategy 完整實作（Control / AO / Device / InfoSystem / Leveraged）+ `write_strategy/` sub-folder | BE | 1.25d | T6 | `feat(oscal): A4 T7 5 write strategies` | | E | T8 | `_write_all_data` 8-step pipeline 重整 + DI 加 11 注入 + 5 `_dict_to_*` helper | BE | 0.5d | T7 | `feat(oscal): A4 T8 _write_all_data 8-step pipeline` | | F | T9 | Unit + integration test ~110-140 個（reconciler 60-75 + WriteStrategy 30-40 + integration 15-20 + ParsedXxx fixture 10）| BE | 1d | T1-T8 | `test(oscal): A4 T9 unit + integration tests` | | F | T10 | Cucumber regression `06-ssp-excel-import-entity-match.feature` 5 scenarios | test | 0.25d | T9 | `test(compliance-manager-test): A4 T10 cucumber entity match regression` | | F | T11 | Changelog + tracker + design §11 reconciliation + final SUMMARY | BE | 0.5d | T10 | `docs(ssp-import-export-phase2): A4 收尾 + SUMMARY` | > **並行最佳化**： > - T1 是 T2-T8 共同基礎（dataclass + helpers + base.py），必須先做 > - T2 結束（skeleton + DI）之後 T3、T4 可平行；T4 不依賴 T3（T4 是 A3 retroactive，動的是 A3 既有 reconciler，與 T3 動 A4 新 reconciler 互不影響） > - T5 依賴 T3（需要 CatalogControlReconciler 完整實作）；可在 T4 ship 後與 T6 平行 > - T6 (SSP shell) 跟 T7 (WriteStrategy) 順序：T6 必先（WriteStrategy 接 ssp_id input；T6 產 _create_ssp_shell 後 T7 才能定 WriteStrategy.write 簽章） > - T7 / T8 順序：T7 先（WriteStrategy class 完成）→ T8 再串 pipeline；不可並行 > - T9 必須等 T2 / T3 / T4 / T5 / T6 / T7 / T8 都完成 > - **Session 切分（不可平行）**：B → C → D → E → F 線性推進；每 Session 結束產 handoff 給下一 Session --- ## File Structure ### 新建檔案（A4） ``` domain/oscal/service/reconciliation/ # A3 既有 sub-folder ├── device_reconciler.py # DeviceReconciler (~110 lines) ├── information_system_reconciler.py # InformationSystemReconciler (~130 lines) ├── leveraged_reconciler.py # LeveragedReconciler (~100 lines) ├── catalog_control_reconciler.py # CatalogControlReconciler (~90 lines) ├── assessment_objective_reconciler.py # AssessmentObjectiveReconciler (~110 lines) └── ssp_entity_orchestrator.py # SspEntityReconciliationOrchestrator (~60 lines) domain/oscal/service/write_strategy/ # A4 新 sub-folder ├── __init__.py ├── control_write_strategy.py # ControlWriteStrategy (~100 lines) ├── ao_write_strategy.py # AoWriteStrategy (~80 lines) ├── device_write_strategy.py # DeviceWriteStrategy (~110 lines) ├── information_system_write_strategy.py # InformationSystemWriteStrategy (~110 lines) └── leveraged_write_strategy.py # LeveragedWriteStrategy (~100 lines) tests/test_a4_reconciliation_helpers.py # parse_*_lookup_label / _normalize_control_id tests/test_a4_reconciliation_device.py tests/test_a4_reconciliation_information_system.py tests/test_a4_reconciliation_leveraged.py tests/test_a4_reconciliation_catalog_control.py tests/test_a4_reconciliation_assessment_objective.py tests/test_a4_reconciliation_orchestrator.py tests/test_a4_write_strategy_control.py tests/test_a4_write_strategy_ao.py tests/test_a4_write_strategy_device.py tests/test_a4_write_strategy_information_system.py tests/test_a4_write_strategy_leveraged.py tests/test_a4_ssp_shell.py tests/test_a4_a2_integration.py # _write_all_data 8-step + §15.3 整併 tests/test_a4_parsed_entity_fixtures.py # 5 個新 dataclass + asdict serialize tests/test_a4_a3_retroactive.py # PersonReconciler / OrganizationReconciler stage 0 docs/changelog/YYYY-MM-DD-feat-ssp-entity-reconcile-and-write.md # type: feat、modules: oscal ``` ### 改動既有檔案（A4） ``` domain/oscal/service/reconciliation/base.py # 加 stage 0 dispatch + _try_user_selected_match default no-op domain/oscal/service/reconciliation/match_method.py # 加 USER_SELECTED value domain/oscal/service/reconciliation/_normalizers.py # 補 4 helper (parse_user_lookup_label / parse_device_lookup_label / parse_info_system_lookup_label / _normalize_control_id) domain/oscal/service/reconciliation/person_reconciler.py # T4 retroactive: override _try_user_selected_match domain/oscal/service/reconciliation/organization_reconciler.py # T4 retroactive: override _try_user_selected_match domain/oscal/parser/ssp_intermediate.py # 加 5 dataclass + ParsedExcelEntityBundle + SspEntityReconciliationContext + ParsedParty 加 2 label 欄位 app/oscal/service/ssp_excel_import_app_service.py # T8: _write_all_data 8-step pipeline + 5 _dict_to_*；T5: _confirm_superset_flow §15.3 整併；T6: _create_ssp_shell / _resolve_existing_ssp_shell；T4: _dict_to_parsed_parties 加 label 傳遞 common/code/grc_error_code.py # 加 GRC_EXCEL_UPDATE_FLOW_NO_SSP (412) di_containers/oscal/oscal_containers.py # 加 11 Factory（5 reconciler + 1 orchestrator + 5 WriteStrategy）+ SspExcelImportAppService 注入 11 dep docs/features/FR-011.2-2605-ssp-import-export-phase2/README.md # tracker A4 row pending → BE shipped docs/features/FR-011.2-2605-ssp-import-export-phase2/design-A4.md # §11 Implementation Reality / Reconciliation 補實作偏差 ``` ### 跨 repo 改動 ``` compliance-manager-test/ └── features/regression/module-frame/06-ssp-excel-import-entity-match.feature # 5 scenarios + steps/module-frame/ssp-excel-import-entity-match.steps.js # 對應 step + pages/module-frame/ssp-excel-import-page.js # page object 補 selector ``` ### 不動 - jedi-* 套件（reconciliation / WriteStrategy 維持在主專案 domain layer；F5 follow-up 才可能抽 jedi-oscal） - FE（A5 phase 才動預覽 UI） - docx flow（D5 拍板；docx parser 不擴 4 entity 解析） --- ## Task 0 — Implementer Pre-flight Verification > **目標**：補 design 階段未驗 / brainstorm 階段假設不確定的 8 個項目（design-A4 §7.2）。**Front-load critical**：T0.5 / T0.7 優先驗（影響 T6 / T7 整段實作）。 ### 0.1 verify jedi-oscal SspEntity 欄位 ```bash grep -nA 30 "^class SystemSecurityPlanEntity" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/ssp/ssp_entity.py ``` **期待**：含 `title` / `profile_uid` / `status` / `metadata_id` / `document_id` / `system_security_plan_main_id` 等欄位；`add()` method 由 `SystemSecurityPlanDomainService` 提供。 **若不符**：design §11 補偏差條目；T6 `_create_ssp_shell` Entity 構造參數對齊實際 schema。 ### 0.2 verify SspSystemCharacteristicEntity 欄位 ```bash grep -nA 25 "^class SspSystemCharacteristicEntity" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/ssp/ssp_system_characteristic_entity.py ``` **期待**：含 `system_security_plan_id` / `name` / `security_sensitivity_level` 等欄位。 **若不符**：design §11 補；T6 構造參數對齊。 ### 0.3 verify SspSystemImplementationEntity main + A0.1 scope 欄位 ```bash grep -nA 25 "^class SspSystemImplementationEntity" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/ssp/ssp_system_implementation_entity.py ``` **期待**：含 A0.1 ship 後新 `scope_type` / `scope_id` 欄位；ORM 對應 `ssp_system_implementations` 表 UNIQUE on `(system_security_plan_id, scope_type, scope_id)`。 **若不符**：T6 `_create_ssp_shell` step 2c 對齊實際 schema。 ### 0.4 verify CatalogControlAssessmentQueryEntity 是否有 objective_id 欄位 ```bash grep -nA 25 "^class CatalogControlAssessmentQueryEntity" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/catalog/catalog_control_assessment_query_entity.py ``` **期待**：含 `catalog_control_id` + `objective_id` 兩欄位（design §5.3 EXACT stage 假設）。 **若不符**：AssessmentObjectiveReconciler `_try_exact_match` 改用其他自然 key（如 `uid` 或 `name`）；design §11 補偏差。 ### 0.5 verify `system_security_plan_control_implementations` ORM model + Mapper signature ⭐ critical ```bash ls /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/infra/model/ssp/ | grep -i control grep -nA 30 "^class SspControlImplementation\|^class.*ControlImplementation" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/ssp/ssp_control_implementation_entity.py ``` **期待**：Entity / QueryEntity / DomainService 完整；`add()` / `update()` 支援 `(system_security_plan_id, catalog_control_id)` upsert；含 `implementation_status` / `implementation_description` / `responsible_role` 欄位。 **若不符**：T7 ControlWriteStrategy 簽章調整；design §11 補。 ### 0.6 verify `ssp_control_impl_objective` ORM model 欄位 ```bash grep -nA 25 "^class SspControlImplObjective\|^class.*ControlImplObjective" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/ssp/ssp_control_impl_objective_entity.py ``` **期待**：含 `ssp_control_implementation_id` + `catalog_control_assessment_id` + audit 欄位。 **若不符**：T7 AoWriteStrategy 簽章調整。 ### 0.7 verify `ssp_domain_service.get_one_by_profile_uid` 是否存在 ⭐ critical ```bash grep -nE "def get_one_by_profile_uid|def get_one|def get_by_profile" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/services/ssp/ssp_domain_service.py ``` **期待**：有 `get_one_by_profile_uid` 或等效 method（用 profile_uid 查 SSP）。 **若不符**：T6 `_resolve_existing_ssp_shell` 改用 query entity 路徑（`SystemSecurityPlanQueryEntity(profile_uid=...)`）或新增套件 method（**需 user 拍板套件異動**）。**列 design §11**。 ### 0.8 verify `DeviceQueryEntity` 雙欄 `(name, ip)` 同時 query 支援度 ```bash grep -nA 30 "^class DeviceQueryEntity" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-device/jedi_device/domain/entity/device_query_entity.py grep -nE "name|ip" \ /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-device/jedi_device/infra/repository/device_repo_impl.py | head -20 ``` **期待**：Query Entity 同時 set `name=..., ip=...` 時 repo 走 AND filter。 **若不符**：DeviceReconciler `_try_exact_match` 改 fallback 拉 name query → Python 端 filter ip；design §11 補偏差。 ### 0.9 Acceptance - [ ] 8 個 verify 結果與 design §7.2 假設一致 → 進 T1 - [ ] 任一項不符 → 主動在 design-A4 §11 補 Reconciliation 條目，**不照舊硬幹** - [ ] T0 結果可併入 T1 commit；無 standalone commit 必要 --- ## Task 1 — base.py stage 0 + MatchMethod + Helpers + Data Structures > **目標**：把 design §3 / §4 / §5.1-§5.2 落地成可被 import 的 module + dataclass。 > **TDD 邊界**：純函式 helper 全 TDD；dataclass 加 `asdict()` JSONB serialize test；base.py stage 0 dispatch 行為 test。 ### 1.1 加 `MatchMethod.USER_SELECTED` **Files:** - Modify: `domain/oscal/service/reconciliation/match_method.py` - [ ] **Step 1**: 加 USER_SELECTED enum value（放最上面表達 stage 0 含義） ```python class MatchMethod(StrEnum): USER_SELECTED = "user_selected" # A4 新增 EXACT = "exact" NORMALIZED = "normalized" FUZZY_EMAIL_DOMAIN = "fuzzy_email_domain" FUZZY_NAME_PREFIX = "fuzzy_name_prefix" UNMATCHED = "unmatched" ``` - [ ] **Step 2**: import smoke ```bash poetry run python3 -c "from domain.oscal.service.reconciliation.match_method import MatchMethod; print(MatchMethod.USER_SELECTED)" ``` Expected: `MatchMethod.USER_SELECTED` - [ ] **Step 3**: 跑既有 A3 normalizers test 全綠 ```bash poetry run pytest tests/test_a3_reconciliation_normalizers.py -v ``` Expected: 13/13 PASS ### 1.2 補 4 個 helper 到 `_normalizers.py`（TDD） **Files:** - Modify: `domain/oscal/service/reconciliation/_normalizers.py` - Create: `tests/test_a4_reconciliation_helpers.py` - [ ] **Step 1**: 先寫 failing test（17 case） ```python # tests/test_a4_reconciliation_helpers.py import pytest from domain.oscal.service.reconciliation._normalizers import ( parse_user_lookup_label, parse_device_lookup_label, parse_info_system_lookup_label, _normalize_control_id, ) class TestParseUserLookupLabel: def test_standard_format(self): assert parse_user_lookup_label("Alice 陳 ") == ("Alice 陳", "alice") def test_no_angle_brackets(self): assert parse_user_lookup_label("alice") == (None, None) def test_none(self): assert parse_user_lookup_label(None) == (None, None) def test_empty_string(self): assert parse_user_lookup_label("") == (None, None) def test_whitespace_only(self): assert parse_user_lookup_label(" ") == (None, None) class TestParseDeviceLookupLabel: def test_standard_format(self): assert parse_device_lookup_label("Web-01 (10.0.1.5)") == ("Web-01", "10.0.1.5") def test_no_parens(self): assert parse_device_lookup_label("Web-01") == ("Web-01", None) def test_none(self): assert parse_device_lookup_label(None) == (None, None) def test_nested_parens(self): # 'Server (prod) (10.0.0.1)' → rsplit assert parse_device_lookup_label("Server (prod) (10.0.0.1)") == ("Server (prod)", "10.0.0.1") class TestParseInfoSystemLookupLabel: def test_standard_format(self): assert parse_info_system_lookup_label("ERP - Enterprise Resource Planning") == ("ERP", "Enterprise Resource Planning") def test_no_separator(self): assert parse_info_system_lookup_label("ERP") == (None, "ERP") def test_none(self): assert parse_info_system_lookup_label(None) == (None, None) class TestNormalizeControlId: def test_lowercase_dash(self): assert _normalize_control_id("ac-1") == "AC-1" def test_space(self): assert _normalize_control_id("ac 1") == "AC-1" def test_underscore(self): assert _normalize_control_id("ac_1") == "AC-1" def test_collapse_dashes(self): assert _normalize_control_id("AC--1") == "AC-1" def test_none(self): assert _normalize_control_id(None) == "" ``` - [ ] **Step 2**: run → FAIL（function not defined） ```bash poetry run pytest tests/test_a4_reconciliation_helpers.py -v ``` Expected: 17/17 FAIL with ImportError / AttributeError - [ ] **Step 3**: 寫 4 個 helper（沿 design §5.1） ```python # domain/oscal/service/reconciliation/_normalizers.py（A3 既有，A4 補在尾端） import re from typing import Optional # ... A3 既有 _strip_plus_alias / _normalize_name / _strip_org_suffix / ORG_SUFFIXES ... def parse_user_lookup_label(label: Optional[str]) -> tuple[Optional[str], Optional[str]]: """'nickname ' → ('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) s = label.strip() if '(' in s and s.endswith(')'): name = s.rsplit('(', 1)[0].strip() ip = s.rsplit('(', 1)[1].rstrip(')').strip() return (name or None, ip or None) return (s or None, 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) s = label.strip() if ' - ' in s: abbr, name = s.split(' - ', 1) return (abbr.strip() or None, name.strip() or None) return (None, s or None) 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 ``` - [ ] **Step 4**: run → 17/17 PASS ```bash poetry run pytest tests/test_a4_reconciliation_helpers.py -v ``` ### 1.3 base.py 加 stage 0 dispatch + default no-op hook **Files:** - Modify: `domain/oscal/service/reconciliation/base.py` - Test: `tests/test_a3_reconciliation_base.py`（已有，A4 加 case） - [ ] **Step 1**: 加 failing test 到既有 base test 檔（驗 stage 0 行為） ```python # tests/test_a3_reconciliation_base.py 加在尾端 class TestStage0UserSelected: """A4 base.py 加 stage 0 dispatch 行為.""" def test_stage_0_user_selected_takes_priority(self): """override _try_user_selected_match 返回 entity 時，直接 USER_SELECTED + confidence=1.0；不再跑 stage 1-3.""" from domain.oscal.service.reconciliation.base import BaseReconciliationService from domain.oscal.service.reconciliation.match_method import MatchMethod class _Fake: def __init__(self): self.match_method = MatchMethod.UNMATCHED self.match_confidence = 0.0 self.matched_id = None class _R(BaseReconciliationService): def _try_user_selected_match(self, parsed, tenant_id): return type("Entity", (), {"id": 99})() def _try_exact_match(self, parsed, tenant_id): raise AssertionError("不該走到 stage 1") def _try_normalized_match(self, parsed, tenant_id): raise AssertionError("不該走到 stage 2") def _try_fuzzy_match(self, parsed, tenant_id): raise AssertionError("不該走到 stage 3") def _apply_match(self, parsed, entity, method, confidence): parsed.matched_id = entity.id parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): pass parsed = _Fake() _R()._reconcile_one(parsed, tenant_id=1) assert parsed.match_method == MatchMethod.USER_SELECTED assert parsed.match_confidence == 1.0 assert parsed.matched_id == 99 def test_default_hook_returns_none_fallthrough(self): """既有 A3 reconciler 不 override → stage 0 hook return None → 走 stage 1.""" from domain.oscal.service.reconciliation.base import BaseReconciliationService from domain.oscal.service.reconciliation.match_method import MatchMethod class _R(BaseReconciliationService): # 不 override _try_user_selected_match (用 default) def _try_exact_match(self, parsed, tenant_id): return type("Entity", (), {"id": 42})() def _try_normalized_match(self, parsed, tenant_id): return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) def _apply_match(self, parsed, entity, method, confidence): parsed.match_method = method def _apply_unmatched(self, parsed): pass class _Fake: match_method = MatchMethod.UNMATCHED parsed = _Fake() _R()._reconcile_one(parsed, tenant_id=1) assert parsed.match_method == MatchMethod.EXACT ``` - [ ] **Step 2**: run → FAIL（base 還沒加 stage 0 dispatch） - [ ] **Step 3**: 改 base.py（design §5.2） ```python # domain/oscal/service/reconciliation/base.py class BaseReconciliationService(ABC, Generic[TParsed, TEntity]): # ... __init__ / reconcile 不動 ... def _reconcile_one(self, parsed: TParsed, tenant_id: Optional[int]) -> None: # A4: stage 0 USER_SELECTED（樣板 matched_* col 反查） 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: TParsed, tenant_id: Optional[int] ) -> Optional[TEntity]: """A4 stage 0: 樣板 matched_* col 反查 lookup label → entity. Default no-op.""" return None ``` **注意**：不加 `@abstractmethod`（A3 既有 reconciler 不必 override；default 行為 = 不變）。 - [ ] **Step 4**: run → 2 個新 case PASS + 既有 7 case 全綠 ```bash poetry run pytest tests/test_a3_reconciliation_base.py -v ``` ### 1.4 加 5 個新 dataclass + ParsedExcelEntityBundle + Context + ParsedParty 加 label 欄位 **Files:** - Modify: `domain/oscal/parser/ssp_intermediate.py` - Create: `tests/test_a4_parsed_entity_fixtures.py` - [ ] **Step 1**: 寫 failing test（10 case：5 dataclass × construct + asdict） ```python # tests/test_a4_parsed_entity_fixtures.py import pytest from dataclasses import asdict from domain.oscal.parser.ssp_intermediate import ( ParsedParty, ParsedDevice, ParsedInformationSystem, ParsedLeveraged, ParsedControl, ParsedAssessmentObjective, ParsedExcelEntityBundle, SspEntityReconciliationContext, ) from domain.oscal.service.reconciliation.match_method import MatchMethod class TestParsedDevice: def test_default_unmatched(self): d = ParsedDevice(name="Web-01") assert d.match_method == MatchMethod.UNMATCHED assert d.match_confidence == 0.0 assert d.matched_device_id is None assert d.matched_device_label is None def test_asdict_serialize(self): d = ParsedDevice(name="Web-01", ip="10.0.1.5", match_method=MatchMethod.EXACT, match_confidence=1.0, matched_device_id=42) s = asdict(d) assert s["match_method"] == MatchMethod.EXACT # StrEnum 序列化 class TestParsedInformationSystem: def test_two_owner_fields(self): s = ParsedInformationSystem( name="ERP", abbreviation="ERP", matched_info_system_id=10, matched_system_owner_user_id=99, ) assert s.matched_info_system_id == 10 assert s.matched_system_owner_user_id == 99 class TestParsedLeveraged: def test_party_uuid(self): from datetime import date l = ParsedLeveraged( service_name="AWS GovCloud", provider="Amazon Web Services", date_authorized=date(2025, 1, 15), matched_party_uuid="abc-123-def", ) assert l.matched_party_uuid == "abc-123-def" class TestParsedControl: def test_nested_objectives(self): c = ParsedControl(control_id="AC-1") c.objectives.append(ParsedAssessmentObjective(statement_id="AC-1.a", control_id="AC-1")) assert len(c.objectives) == 1 assert c.matched_catalog_control_uid is None assert c.matched_catalog_control_id is None class TestParsedAssessmentObjective: def test_statement_id_vs_matched_id(self): ao = ParsedAssessmentObjective( statement_id="AC-1.a.1", # parser key (string) control_id="AC-1", matched_catalog_control_assessment_id=42, # DB id (int) ) assert ao.statement_id == "AC-1.a.1" assert ao.matched_catalog_control_assessment_id == 42 class TestParsedExcelEntityBundle: def test_construct(self): b = ParsedExcelEntityBundle( parsed_devices=[ParsedDevice(name="d")], parsed_info_systems=[], parsed_leveraged=[], parsed_controls=[], ) assert len(b.parsed_devices) == 1 class TestSspEntityReconciliationContext: def test_construct(self): ctx = SspEntityReconciliationContext(tenant_id=102, catalog_id=5) assert ctx.tenant_id == 102 and ctx.catalog_id == 5 class TestParsedPartyA4RetroactiveFields: def test_new_label_fields_default_none(self): p = ParsedParty(name="A", party_type="person") assert p.matched_user_label is None assert p.matched_org_unit_label is None ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 改 `ssp_intermediate.py`（design §4） ```python # domain/oscal/parser/ssp_intermediate.py # A3 既有 ParsedParty 加 2 個 optional 欄位 @dataclass class ParsedParty: # ... 既有 A3 欄位（name / party_type / email_address / ... matched_user_id / matched_org_unit_id / target_party_uid / match_method / match_confidence）... # A4 retroactive matched_user_label: Optional[str] = None matched_org_unit_label: Optional[str] = None # A4 新增 5 個 dataclass + 2 個 context from datetime import date @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 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 system_owner_label: Optional[str] = None 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 date_authorized: Optional[date] = None purpose: Optional[str] = None matched_party_uuid: Optional[str] = None match_method: MatchMethod = MatchMethod.UNMATCHED match_confidence: float = 0.0 @dataclass class ParsedAssessmentObjective: # NOTE: statement_id 是 parser 解出的 AO uid 字串（樣板 hidden column key） # 跟 matched_catalog_control_assessment_id（DB int id）不要混淆 statement_id: str control_id: str objective_id: Optional[str] = None objective_name: Optional[str] = None impl_status: Optional[str] = None statement: Optional[str] = None reference_doc: Optional[str] = None 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 @dataclass class ParsedControl: control_id: str 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 _target_in_profile: bool = True objectives: list[ParsedAssessmentObjective] = field(default_factory=list) # 兩個都存：int 給 WriteStrategy FK，str 給 §15.3 ProfileService.add_profile UUID matched_catalog_control_uid: Optional[str] = None matched_catalog_control_id: Optional[int] = None match_method: MatchMethod = MatchMethod.UNMATCHED match_confidence: float = 0.0 @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] @dataclass class SspEntityReconciliationContext: tenant_id: int catalog_id: int ``` - [ ] **Step 4**: run → 10/10 PASS ```bash poetry run pytest tests/test_a4_parsed_entity_fixtures.py -v ``` ### 1.5 Commit T1 ```bash git add domain/oscal/service/reconciliation/base.py \ domain/oscal/service/reconciliation/match_method.py \ domain/oscal/service/reconciliation/_normalizers.py \ domain/oscal/parser/ssp_intermediate.py \ tests/test_a4_reconciliation_helpers.py \ tests/test_a4_parsed_entity_fixtures.py \ tests/test_a3_reconciliation_base.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T1 stage 0 hook + dataclass + helpers base.py 加 stage 0 dispatch（_try_user_selected_match default no-op）； MatchMethod 加 USER_SELECTED；_normalizers 補 4 helper（3 label parser + _normalize_control_id）。ssp_intermediate 加 5 個 typed dataclass + Bundle + Context；ParsedParty 加 2 個 optional label 欄位（retroactive）。 新增 17 helper test + 10 dataclass fixture test + 2 stage 0 base test = 29 test 全綠。 對齊 design-A4 §3.2 / §4 / §5.1 / §5.2。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 1 Acceptance - [ ] base.py 加 stage 0 dispatch + `_try_user_selected_match` default no-op - [ ] MatchMethod 加 `USER_SELECTED` - [ ] _normalizers 補 4 helper（3 label parser + `_normalize_control_id`） - [ ] ssp_intermediate 加 5 dataclass + ParsedExcelEntityBundle + SspEntityReconciliationContext - [ ] ParsedParty 加 `matched_user_label` + `matched_org_unit_label` optional default None - [ ] 29 新 test 全綠（17 helper + 10 dataclass + 2 stage 0） - [ ] A3 既有 7 個 base test 全綠（stage 0 加入後不破壞） - [ ] Commit 完成 --- ## Task 2 — 5 Reconciler skeleton + Orchestrator + DI Factory > **目標**：建 5 reconciler class 骨架（5 hook 全部最小實作）+ orchestrator + DI Factory；hook 內部完整演算法留 T3。 > **TDD 邊界**：skeleton 階段每個 reconciler 寫 1 個 import smoke test + 1 個 minimal `reconcile([])` 走通；演算法細節 test 放 T3。 ### 2.1 DeviceReconciler skeleton **Files:** - Create: `domain/oscal/service/reconciliation/device_reconciler.py` - Test: `tests/test_a4_reconciliation_device.py` - [ ] **Step 1**: 寫 failing test（import + reconcile empty list 通） ```python # tests/test_a4_reconciliation_device.py import pytest from unittest.mock import MagicMock from domain.oscal.service.reconciliation.device_reconciler import DeviceReconciler from domain.oscal.parser.ssp_intermediate import ParsedDevice from domain.oscal.service.reconciliation.match_method import MatchMethod class TestDeviceReconcilerSkeleton: def test_reconcile_empty_list(self): r = DeviceReconciler(device_domain_service=MagicMock()) result = r.reconcile([], tenant_id=102) assert result == [] def test_reconcile_single_unmatched(self): """device_domain_service 返回空 → UNMATCHED.""" mock_device = MagicMock() mock_device.get_devices.return_value = [] r = DeviceReconciler(device_domain_service=mock_device) d = ParsedDevice(name="Web-01") r.reconcile([d], tenant_id=102) assert d.match_method == MatchMethod.UNMATCHED ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 寫 skeleton ```python # domain/oscal/service/reconciliation/device_reconciler.py from typing import Optional, Tuple from domain.oscal.parser.ssp_intermediate import ParsedDevice from domain.oscal.service.reconciliation.base import BaseReconciliationService from domain.oscal.service.reconciliation.match_method import MatchMethod class DeviceReconciler(BaseReconciliationService[ParsedDevice, object]): def __init__(self, device_domain_service): super().__init__() self._device = device_domain_service def _try_user_selected_match(self, parsed, tenant_id): return None # T3 填 def _try_exact_match(self, parsed, tenant_id): return None # T3 填 def _try_normalized_match(self, parsed, tenant_id): return None # T3 填 def _try_fuzzy_match(self, parsed, tenant_id) -> Tuple[Optional[object], MatchMethod]: return (None, MatchMethod.UNMATCHED) # D8: device 無 fuzzy def _apply_match(self, parsed, entity, method, confidence): parsed.matched_device_id = getattr(entity, "id", None) parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` - [ ] **Step 4**: run → 2/2 PASS ### 2.2 InformationSystemReconciler skeleton **Files:** - Create: `domain/oscal/service/reconciliation/information_system_reconciler.py` - Test: `tests/test_a4_reconciliation_information_system.py` - [ ] **Step 1-4**: 同 2.1 模式 ```python # domain/oscal/service/reconciliation/information_system_reconciler.py 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 # 5 hook 全 minimal (T3 填細節) def _try_user_selected_match(self, parsed, tenant_id): return None def _try_exact_match(self, parsed, tenant_id): return None def _try_normalized_match(self, parsed, tenant_id): return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) 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 # T3 補 system_owner 反查 def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 # T3 補 system_owner 反查 ``` ### 2.3 LeveragedReconciler skeleton **Files:** - Create: `domain/oscal/service/reconciliation/leveraged_reconciler.py` - Test: `tests/test_a4_reconciliation_leveraged.py` - [ ] **Step 1-4**: 同 2.1 模式 ```python class LeveragedReconciler(BaseReconciliationService[ParsedLeveraged, object]): def __init__(self, party_domain_service): super().__init__() self._party = party_domain_service # 5 hook 全 minimal def _try_user_selected_match(self, parsed, tenant_id): return None def _try_exact_match(self, parsed, tenant_id): return None def _try_normalized_match(self, parsed, tenant_id): return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) def _apply_match(self, parsed, entity, method, confidence): parsed.matched_party_uuid = str(getattr(entity, "uid", None)) if getattr(entity, "uid", None) else None parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` ### 2.4 CatalogControlReconciler skeleton **Files:** - Create: `domain/oscal/service/reconciliation/catalog_control_reconciler.py` - Test: `tests/test_a4_reconciliation_catalog_control.py` - [ ] **Step 1-4**: 同 2.1 模式（reconcile signature 變：第二參數 = catalog_id） ```python class CatalogControlReconciler(BaseReconciliationService[ParsedControl, object]): def __init__(self, catalog_control_domain_service): super().__init__() self._cc = catalog_control_domain_service # base.reconcile(parsed_list, tenant_id) 的 tenant_id 位置用 catalog_id # 對齊既有 base — 不破壞 contract，只是參數名語意切換 def _try_user_selected_match(self, parsed, tenant_id): return None # 無 lookup col def _try_exact_match(self, parsed, tenant_id): return None # T3 填 def _try_normalized_match(self, parsed, tenant_id): return None # T3 填 def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) # 無 fuzzy def _apply_match(self, parsed, entity, method, confidence): parsed.matched_catalog_control_id = getattr(entity, "id", None) parsed.matched_catalog_control_uid = str(getattr(entity, "uid", None)) if getattr(entity, "uid", None) else None parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` ### 2.5 AssessmentObjectiveReconciler skeleton（特殊 — 加 reconcile_with_control_map） **Files:** - Create: `domain/oscal/service/reconciliation/assessment_objective_reconciler.py` - Test: `tests/test_a4_reconciliation_assessment_objective.py` - [ ] **Step 1-4**: 同 2.1 模式 + 加 `reconcile_with_control_map` public method（不走 base.reconcile） ```python class AssessmentObjectiveReconciler(BaseReconciliationService[ParsedAssessmentObjective, object]): def __init__(self, assessment_domain_service): super().__init__() self._ao = assessment_domain_service self._current_parent_ctrl_id: Optional[int] = None # 給 _try_*_match 讀 def reconcile_with_control_map( self, parsed_aos: list, ctrl_map: dict, # control_id 字面 → matched_catalog_control_id ) -> None: """A4 特殊：因 AO 需要父 control_id context，不走 base.reconcile(list, tenant_id).""" 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): return None # T3 填 def _try_normalized_match(self, parsed, tenant_id): return None # T3 填 def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) # 無 fuzzy def _apply_match(self, parsed, entity, method, confidence): parsed.matched_catalog_control_assessment_id = getattr(entity, "id", None) parsed.matched_catalog_control_assessment_uid = str(getattr(entity, "uid", None)) if getattr(entity, "uid", None) else None parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` ### 2.6 SspEntityReconciliationOrchestrator skeleton **Files:** - Create: `domain/oscal/service/reconciliation/ssp_entity_orchestrator.py` - Test: `tests/test_a4_reconciliation_orchestrator.py` - [ ] **Step 1**: 寫 failing test（5 個 reconciler.reconcile 都被呼叫 + 順序） ```python # tests/test_a4_reconciliation_orchestrator.py from unittest.mock import MagicMock, call from domain.oscal.service.reconciliation.ssp_entity_orchestrator import ( SspEntityReconciliationOrchestrator, ) from domain.oscal.parser.ssp_intermediate import ( ParsedExcelEntityBundle, SspEntityReconciliationContext, ParsedControl, ParsedAssessmentObjective, ) class TestOrchestrator: def test_dispatch_order_control_first(self): ctrl = MagicMock() ao = MagicMock() device = MagicMock() info = MagicMock() leveraged = MagicMock() orchestrator = SspEntityReconciliationOrchestrator( catalog_control_reconciler=ctrl, assessment_objective_reconciler=ao, device_reconciler=device, information_system_reconciler=info, leveraged_reconciler=leveraged, ) bundle = ParsedExcelEntityBundle( parsed_devices=[], parsed_info_systems=[], parsed_leveraged=[], parsed_controls=[], ) ctx = SspEntityReconciliationContext(tenant_id=102, catalog_id=5) orchestrator.reconcile(bundle, ctx) ctrl.reconcile.assert_called_once() ao.reconcile_with_control_map.assert_called_once() device.reconcile.assert_called_once() info.reconcile.assert_called_once() leveraged.reconcile.assert_called_once() def test_ao_gets_ctrl_map_from_control_results(self): ctrl = MagicMock() ao = MagicMock() c1 = ParsedControl(control_id="AC-1") c1.matched_catalog_control_id = 100 c2 = ParsedControl(control_id="AC-2") c2.matched_catalog_control_id = None # unmatched c1.objectives.append(ParsedAssessmentObjective(statement_id="AC-1.a", control_id="AC-1")) c2.objectives.append(ParsedAssessmentObjective(statement_id="AC-2.a", control_id="AC-2")) orchestrator = SspEntityReconciliationOrchestrator( catalog_control_reconciler=ctrl, assessment_objective_reconciler=ao, device_reconciler=MagicMock(), information_system_reconciler=MagicMock(), leveraged_reconciler=MagicMock(), ) bundle = ParsedExcelEntityBundle( parsed_devices=[], parsed_info_systems=[], parsed_leveraged=[], parsed_controls=[c1, c2], ) ctx = SspEntityReconciliationContext(tenant_id=102, catalog_id=5) orchestrator.reconcile(bundle, ctx) # ctrl_map 應該只含 matched 的 control_id ctrl_map_arg = ao.reconcile_with_control_map.call_args[0][1] assert ctrl_map_arg == {"AC-1": 100} ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 寫 orchestrator（design §3.2） ```python # domain/oscal/service/reconciliation/ssp_entity_orchestrator.py from domain.oscal.parser.ssp_intermediate import ( ParsedExcelEntityBundle, SspEntityReconciliationContext, ) 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 二階段） self._ctrl.reconcile(parsed.parsed_controls, ctx.catalog_id) # 2. AO 從 ParsedControl.objectives flatten；用 ctrl_map 取父 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) ``` - [ ] **Step 4**: run → 2/2 PASS ### 2.7 DI Container 加 6 Factory + Orchestrator **Files:** - Modify: `di_containers/oscal/oscal_containers.py` - [ ] **Step 1**: 加 6 個新 Factory（5 reconciler + 1 orchestrator） ```python # di_containers/oscal/oscal_containers.py（節錄） from domain.oscal.service.reconciliation.device_reconciler import DeviceReconciler from domain.oscal.service.reconciliation.information_system_reconciler import InformationSystemReconciler from domain.oscal.service.reconciliation.leveraged_reconciler import LeveragedReconciler from domain.oscal.service.reconciliation.catalog_control_reconciler import CatalogControlReconciler from domain.oscal.service.reconciliation.assessment_objective_reconciler import AssessmentObjectiveReconciler from domain.oscal.service.reconciliation.ssp_entity_orchestrator import SspEntityReconciliationOrchestrator # 假設 catalog_control_domain_service / catalog_control_assessment_domain_service 已在本 container； # device_domain_service 透過 device_container（沿 auth_container 跨 container 引用 pattern） device_reconciler = providers.Factory( DeviceReconciler, device_domain_service=device_container.device_domain_service, ) 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, party_domain_service=party_domain_service, # 或對應 container ) catalog_control_reconciler = providers.Factory( CatalogControlReconciler, catalog_control_domain_service=catalog_control_domain_service, ) assessment_objective_reconciler = providers.Factory( AssessmentObjectiveReconciler, assessment_domain_service=catalog_control_assessment_domain_service, ) 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, ) ``` - [ ] **Step 2**: import smoke ```bash poetry run python3 -c "from di_containers.containers import Containers; c = Containers(); print(c.oscal_container.ssp_entity_reconciliation_orchestrator)" ``` Expected: 不抛 exception；印出 Factory provider ### 2.8 Commit T2 ```bash git add domain/oscal/service/reconciliation/device_reconciler.py \ domain/oscal/service/reconciliation/information_system_reconciler.py \ domain/oscal/service/reconciliation/leveraged_reconciler.py \ domain/oscal/service/reconciliation/catalog_control_reconciler.py \ domain/oscal/service/reconciliation/assessment_objective_reconciler.py \ domain/oscal/service/reconciliation/ssp_entity_orchestrator.py \ di_containers/oscal/oscal_containers.py \ tests/test_a4_reconciliation_device.py \ tests/test_a4_reconciliation_information_system.py \ tests/test_a4_reconciliation_leveraged.py \ tests/test_a4_reconciliation_catalog_control.py \ tests/test_a4_reconciliation_assessment_objective.py \ tests/test_a4_reconciliation_orchestrator.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T2 reconciler + orchestrator skeleton + DI 5 個新 reconciler skeleton（hook 全 minimal，T3 填演算法）+ SspEntityReconciliationOrchestrator facade 落地（enforce control → AO 順序， 其他 3 entity 獨立）；AO Reconciler 加 reconcile_with_control_map public method 處理父 control 依賴。 DI Container 加 6 Factory（5 reconciler + 1 orchestrator）；跨 container 引用 device_container / is_container / auth_container 沿 A3 line 490/491 pattern。 新增 12 個 skeleton test + 2 個 orchestrator dispatch test = 14 test 全綠。 對齊 design-A4 §3.2 / §5.3-§5.5. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 2 Acceptance - [ ] 5 reconciler skeleton 建立（hook minimal return None / UNMATCHED） - [ ] AssessmentObjectiveReconciler 加 `reconcile_with_control_map` public method - [ ] SspEntityReconciliationOrchestrator dispatch 順序正確（control → AO → device / info_system / leveraged） - [ ] DI Container 加 6 Factory；container build 不抛 exception - [ ] 14 新 test PASS（12 skeleton + 2 orchestrator dispatch） - [ ] A3 既有 67 個 reconciliation test + 8 既有 party test 全綠 - [ ] Commit 完成 ### Session C → D 收尾 當 T2 ship 後（Session C 結束）： - [ ] 跑 `poetry run pytest tests/test_a3_*.py tests/test_a4_*.py -v` 確認 0 regression - [ ] `git status --short` 確認 working tree 乾淨（除 pre-existing pyproject.toml dev-path + zip） - [ ] 產 handoff prompt → `docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/YYYY-MM-DD-a4-c-to-d.md` - 含已 ship commits + T3/T4/T5 開工指引 + 鐵律提醒 - [ ] 換 Session D 接手 — T3 + T4 平行可做，T5 等 T3 ship --- ## Task 3 — 5 Reconciler 完整演算法 > **目標**：T2 skeleton 的 5 hook 全部填上實作（stage 0 / 1 / 2 / fuzzy 詳述見 design §5.3）。 > **TDD 邊界**：每個 stage 一個 test class；fuzzy 邊界 case 全 cover。 > **預估 test 數**：DeviceReconciler ~12 / InformationSystemReconciler ~14（含 system_owner）/ LeveragedReconciler ~14 / CatalogControlReconciler ~10 / AssessmentObjectiveReconciler ~10 = ~60 case。 ### 3.1 DeviceReconciler 完整實作 **Files:** - Modify: `domain/oscal/service/reconciliation/device_reconciler.py` - Modify: `tests/test_a4_reconciliation_device.py` - [ ] **Step 1**: 寫 12 個 failing test（4 user_selected + 4 exact + 4 normalized） ```python # tests/test_a4_reconciliation_device.py 補 class TestDeviceUserSelected: def test_user_selected_via_label(self): """matched_device_label='Web-01 (10.0.1.5)' → 反查到 Device → USER_SELECTED.""" mock = MagicMock() mock_entity = MagicMock(id=42, name="Web-01", ip="10.0.1.5") mock.get_devices.return_value = [mock_entity] r = DeviceReconciler(device_domain_service=mock) d = ParsedDevice(name="Web-01", ip="10.0.1.5", matched_device_label="Web-01 (10.0.1.5)") r.reconcile([d], tenant_id=102) assert d.match_method == MatchMethod.USER_SELECTED assert d.matched_device_id == 42 def test_user_selected_label_no_match(self): """label parse 成功但 DB 查無 → fallback stage 1.""" # ... 略 # ... 其他 user_selected case class TestDeviceExact: def test_exact_match_name_and_ip(self): """雙欄 (name, ip) 同時匹配.""" # ... def test_exact_match_name_only(self): """T0.8 verify 後若 jedi-device 支援單欄；否則 fallback Python filter.""" # ... class TestDeviceNormalized: def test_normalized_name_lowercase(self): """name='WEB-01' parsed 'web-01' → normalize 後找到.""" # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 改 device_reconciler.py 5 個 hook ```python class DeviceReconciler(BaseReconciliationService[ParsedDevice, object]): def __init__(self, device_domain_service): super().__init__() self._device = device_domain_service def _try_user_selected_match(self, parsed, tenant_id): if not parsed.matched_device_label: return None name, ip = parse_device_lookup_label(parsed.matched_device_label) if not name: return None try: from jedi_device.domain.entity.device_query_entity import DeviceQueryEntity q = DeviceQueryEntity(name=name) if ip: q.ip = ip devices = self._device.get_devices(q) for d in devices or []: # T0.8 verify 後若雙欄不支援 → Python filter if ip and getattr(d, "ip", None) != ip: continue return d except Exception: pass return None def _try_exact_match(self, parsed, tenant_id): if not parsed.name: return None try: from jedi_device.domain.entity.device_query_entity import DeviceQueryEntity q = DeviceQueryEntity(name=parsed.name.strip()) if parsed.ip: q.ip = parsed.ip.strip() devices = self._device.get_devices(q) for d in devices or []: if parsed.ip and getattr(d, "ip", None) != parsed.ip.strip(): continue return d except Exception: pass return None def _try_normalized_match(self, parsed, tenant_id): from domain.oscal.service.reconciliation._normalizers import _normalize_name normalized_name = _normalize_name(parsed.name).lower() if parsed.name else "" normalized_ip = (parsed.ip or "").strip() if not normalized_name or normalized_name == (parsed.name or "").strip(): return None try: from jedi_device.domain.entity.device_query_entity import DeviceQueryEntity q = DeviceQueryEntity(name=normalized_name) if normalized_ip: q.ip = normalized_ip devices = self._device.get_devices(q) for d in devices or []: return d except Exception: pass return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) # D8: device 無 fuzzy def _apply_match(self, parsed, entity, method, confidence): parsed.matched_device_id = getattr(entity, "id", None) parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` - [ ] **Step 4**: run → 12/12 PASS ### 3.2 InformationSystemReconciler 完整實作（含 system_owner 反查） **Files:** - Modify: `domain/oscal/service/reconciliation/information_system_reconciler.py` - Modify: `tests/test_a4_reconciliation_information_system.py` - [ ] **Step 1-4**: 同 3.1 模式 — 14 test（4 user_selected + 4 exact + 4 normalized + 2 system_owner） ```python 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 _try_user_selected_match(self, parsed, tenant_id): if not parsed.matched_info_system_label: return None abbr, name = parse_info_system_lookup_label(parsed.matched_info_system_label) if not (abbr or name): return None try: from jedi_information_system.domain.entity.information_system_query_entity import InformationSystemQueryEntity q = InformationSystemQueryEntity(name=name) if name else InformationSystemQueryEntity(abbreviation=abbr) systems = self._is.get_information_systems(q) for s in systems or []: if abbr and getattr(s, "abbreviation", None) != abbr: continue if name and getattr(s, "name", None) != name: continue return s except Exception: pass return None def _try_exact_match(self, parsed, tenant_id): if not parsed.name: return None try: from jedi_information_system.domain.entity.information_system_query_entity import InformationSystemQueryEntity systems = self._is.get_information_systems(InformationSystemQueryEntity(name=parsed.name.strip())) for s in systems or []: return s except Exception: pass return None def _try_normalized_match(self, parsed, tenant_id): from domain.oscal.service.reconciliation._normalizers import _normalize_name normalized = _normalize_name(parsed.name) if not normalized or normalized == (parsed.name or "").strip(): return None try: from jedi_information_system.domain.entity.information_system_query_entity import InformationSystemQueryEntity systems = self._is.get_information_systems(InformationSystemQueryEntity(name=normalized)) for s in systems or []: return s except Exception: pass return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) # D8: info_system 無 fuzzy 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) 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): """純 EXACT label 反查；不影響主 match_method.""" if not parsed.system_owner_label: return from domain.oscal.service.reconciliation._normalizers import parse_user_lookup_label nickname, login_name = parse_user_lookup_label(parsed.system_owner_label) if not login_name: return try: from jedi_auth.domain.entities.user_query_entity import UserQueryEntity 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 ``` ### 3.3 LeveragedReconciler 完整實作（含 FUZZY_NAME_PREFIX 沿 A3 OrganizationReconciler） **Files:** - Modify: `domain/oscal/service/reconciliation/leveraged_reconciler.py` - Modify: `tests/test_a4_reconciliation_leveraged.py` - [ ] **Step 1-4**: 同模式 — 14 test（4 user_selected + 4 exact + 3 normalized + 3 fuzzy） ```python class LeveragedReconciler(BaseReconciliationService[ParsedLeveraged, object]): def __init__(self, party_domain_service): super().__init__() self._party = party_domain_service def _try_user_selected_match(self, parsed, tenant_id): if not parsed.party_label: return None try: from domain.oscal.entity.party_query_entity import PartyQueryEntity parties = self._party.get_all( PartyQueryEntity(party_type="organization", name=parsed.party_label.strip()) ) for p in parties or []: return p except Exception: pass return None def _try_exact_match(self, parsed, tenant_id): if not parsed.provider: return None try: from domain.oscal.entity.party_query_entity import PartyQueryEntity parties = self._party.get_all( PartyQueryEntity(party_type="organization", name=parsed.provider.strip()) ) for p in parties or []: return p except Exception: pass return None def _try_normalized_match(self, parsed, tenant_id): from domain.oscal.service.reconciliation._normalizers import _normalize_name normalized = _normalize_name(parsed.provider) if not normalized or normalized == (parsed.provider or "").strip(): return None try: from domain.oscal.entity.party_query_entity import PartyQueryEntity parties = self._party.get_all( PartyQueryEntity(party_type="organization", name=normalized) ) for p in parties or []: return p except Exception: pass return None def _try_fuzzy_match(self, parsed, tenant_id): """沿 A3 OrganizationReconciler — _strip_org_suffix 後比對 candidate cache.""" from domain.oscal.service.reconciliation._normalizers import ( _normalize_name, _strip_org_suffix, ) if not parsed.provider: return (None, MatchMethod.UNMATCHED) normalized = _normalize_name(parsed.provider) stripped = _strip_org_suffix(normalized) if not stripped or stripped == normalized: return (None, MatchMethod.UNMATCHED) def _loader(): from domain.oscal.entity.party_query_entity import PartyQueryEntity return self._party.get_all(PartyQueryEntity(party_type="organization")) candidates = self._get_or_load_candidates(_loader) for p in candidates: p_stripped = _strip_org_suffix(_normalize_name(getattr(p, "name", None))) if p_stripped == stripped: return (p, MatchMethod.FUZZY_NAME_PREFIX) return (None, MatchMethod.UNMATCHED) def _apply_match(self, parsed, entity, method, confidence): uid = getattr(entity, "uid", None) parsed.matched_party_uuid = str(uid) if uid else None parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` ### 3.4 CatalogControlReconciler 完整實作 **Files:** - Modify: `domain/oscal/service/reconciliation/catalog_control_reconciler.py` - Modify: `tests/test_a4_reconciliation_catalog_control.py` - [ ] **Step 1-4**: 同模式 — 10 test（2 skip user_selected + 5 exact + 3 normalized） ```python class CatalogControlReconciler(BaseReconciliationService[ParsedControl, object]): def __init__(self, catalog_control_domain_service): super().__init__() self._cc = catalog_control_domain_service def _try_user_selected_match(self, parsed, tenant_id): return None # 樣板無 lookup col；skip stage 0 def _try_exact_match(self, parsed, tenant_id): # NOTE: tenant_id 位置實際傳 catalog_id（orchestrator 用 ctx.catalog_id 呼叫） catalog_id = tenant_id if not parsed.control_id or catalog_id is None: return None try: from jedi_oscal.domain.entity.catalog.catalog_control_query_entity import CatalogControlQueryEntity ctrls = self._cc.get_all( CatalogControlQueryEntity(catalog_id=catalog_id, control_id=parsed.control_id.strip()) ) for c in ctrls or []: return c except Exception: pass return None def _try_normalized_match(self, parsed, tenant_id): from domain.oscal.service.reconciliation._normalizers import _normalize_control_id catalog_id = tenant_id normalized = _normalize_control_id(parsed.control_id) if not normalized or normalized == (parsed.control_id or "").strip(): return None try: from jedi_oscal.domain.entity.catalog.catalog_control_query_entity import CatalogControlQueryEntity ctrls = self._cc.get_all( CatalogControlQueryEntity(catalog_id=catalog_id, control_id=normalized) ) for c in ctrls or []: return c except Exception: pass return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) # D3: control 無 fuzzy def _apply_match(self, parsed, entity, method, confidence): parsed.matched_catalog_control_id = getattr(entity, "id", None) uid = getattr(entity, "uid", None) parsed.matched_catalog_control_uid = str(uid) if uid else None parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` ### 3.5 AssessmentObjectiveReconciler 完整實作 **Files:** - Modify: `domain/oscal/service/reconciliation/assessment_objective_reconciler.py` - Modify: `tests/test_a4_reconciliation_assessment_objective.py` - [ ] **Step 1-4**: 同模式 — 10 test（含 control_map 邊界 case） ```python class AssessmentObjectiveReconciler(BaseReconciliationService[ParsedAssessmentObjective, object]): def __init__(self, assessment_domain_service): super().__init__() self._ao = assessment_domain_service self._current_parent_ctrl_id: Optional[int] = None def reconcile_with_control_map(self, parsed_aos, ctrl_map): 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 try: self._reconcile_one(ao, tenant_id=None) finally: self._current_parent_ctrl_id = None def _try_user_selected_match(self, parsed, tenant_id): return None def _try_exact_match(self, parsed, tenant_id): if self._current_parent_ctrl_id is None or not parsed.objective_id: return None try: from jedi_oscal.domain.entity.catalog.catalog_control_assessment_query_entity import CatalogControlAssessmentQueryEntity aos = self._ao.get_all( CatalogControlAssessmentQueryEntity( catalog_control_id=self._current_parent_ctrl_id, objective_id=parsed.objective_id.strip(), ) ) for a in aos or []: return a except Exception: pass return None def _try_normalized_match(self, parsed, tenant_id): # objective_id normalize: trim + upper if self._current_parent_ctrl_id is None or not parsed.objective_id: return None normalized = parsed.objective_id.strip().upper() if normalized == parsed.objective_id.strip(): return None try: from jedi_oscal.domain.entity.catalog.catalog_control_assessment_query_entity import CatalogControlAssessmentQueryEntity aos = self._ao.get_all( CatalogControlAssessmentQueryEntity( catalog_control_id=self._current_parent_ctrl_id, objective_id=normalized, ) ) for a in aos or []: return a except Exception: pass return None def _try_fuzzy_match(self, parsed, tenant_id): return (None, MatchMethod.UNMATCHED) def _apply_match(self, parsed, entity, method, confidence): parsed.matched_catalog_control_assessment_id = getattr(entity, "id", None) uid = getattr(entity, "uid", None) parsed.matched_catalog_control_assessment_uid = str(uid) if uid else None parsed.match_method = method parsed.match_confidence = confidence def _apply_unmatched(self, parsed): parsed.match_method = MatchMethod.UNMATCHED parsed.match_confidence = 0.0 ``` ### 3.6 Commit T3 ```bash git add domain/oscal/service/reconciliation/device_reconciler.py \ domain/oscal/service/reconciliation/information_system_reconciler.py \ domain/oscal/service/reconciliation/leveraged_reconciler.py \ domain/oscal/service/reconciliation/catalog_control_reconciler.py \ domain/oscal/service/reconciliation/assessment_objective_reconciler.py \ tests/test_a4_reconciliation_device.py \ tests/test_a4_reconciliation_information_system.py \ tests/test_a4_reconciliation_leveraged.py \ tests/test_a4_reconciliation_catalog_control.py \ tests/test_a4_reconciliation_assessment_objective.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T3 reconciler full algorithm 5 個 reconciler hook 完整實作（stage 0 USER_SELECTED + stage 1 EXACT + stage 2 NORMALIZED + stage 3 fuzzy/no-op）： - Device / InfoSystem / Control / AO: D3+D8 三/二階段、無 fuzzy - Leveraged: 沿 A3 OrganizationReconciler 樣板含 FUZZY_NAME_PREFIX - InfoSystem: 附加 system_owner 反查（純 EXACT，不影響主 match_method） - AO: reconcile_with_control_map 透過 instance attr 傳父 catalog_control_id 新增 ~60 個 reconciler test（Device 12 / InfoSystem 14 / Leveraged 14 / Control 10 / AO 10）全綠。 對齊 design-A4 §5.3-§5.5. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 3 Acceptance - [ ] 5 reconciler 5 hook 完整實作對齊 design §5.3 表格 - [ ] LeveragedReconciler fuzzy 沿 A3 OrganizationReconciler `_strip_org_suffix` pattern - [ ] InfoSystemReconciler `_resolve_system_owner` 在 `_apply_match` + `_apply_unmatched` 兩處呼叫 - [ ] AssessmentObjectiveReconciler `reconcile_with_control_map` instance attr 處理父依賴；finally 重置防 leak - [ ] ~60 個 reconciler test 全綠 - [ ] A3 既有 67 個 test 全綠（沒 regression） --- ## Task 4 — A3 retroactive USER_SELECTED stage > **目標**：在 PersonReconciler / OrganizationReconciler 加 stage 0 `_try_user_selected_match`；同時改 `_dict_to_parsed_parties` 把 Excel `matched_user` / `org_unit` lookup col 傳遞到 ParsedParty。 > **TDD 邊界**：每個 reconciler 加 4 個 stage 0 test（hit / miss / no label / parse failure）；A3 既有 16 person + 16 org test 全綠（USER_SELECTED 加入後不破壞）。 ### 4.1 PersonReconciler override `_try_user_selected_match` **Files:** - Modify: `domain/oscal/service/reconciliation/person_reconciler.py` - Create: `tests/test_a4_a3_retroactive.py` - [ ] **Step 1**: 寫 4 failing test ```python # tests/test_a4_a3_retroactive.py import pytest from unittest.mock import MagicMock from domain.oscal.service.reconciliation.person_reconciler import PersonReconciler from domain.oscal.parser.ssp_intermediate import ParsedParty from domain.oscal.service.reconciliation.match_method import MatchMethod class TestPersonReconcilerStage0: def test_user_selected_label_hit(self): mock_entity = MagicMock(id=42, is_active=True, login_name="alice") mock_user = MagicMock() mock_user.get_users.return_value = [mock_entity] r = PersonReconciler(user_domain_service=mock_user) p = ParsedParty( name="Alice 陳", party_type="person", matched_user_label="Alice 陳 ", ) r.reconcile([p], tenant_id=102) assert p.match_method == MatchMethod.USER_SELECTED assert p.matched_user_id == 42 def test_user_selected_label_miss_fallback_to_exact(self): """label 解出 login_name 但 DB 查無 → fallback stage 1 (exact email).""" # ... def test_user_selected_no_label_skip_to_exact(self): """matched_user_label=None → 直接走 stage 1.""" # ... def test_user_selected_label_parse_failure(self): """matched_user_label='alice' (no angle brackets) → parse fail → skip stage 0.""" # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 加 `_try_user_selected_match` override ```python # domain/oscal/service/reconciliation/person_reconciler.py class PersonReconciler(BaseReconciliationService[ParsedParty, object]): # ... 既有 __init__ / _try_exact_match / _try_normalized_match / _try_fuzzy_match / _apply_match / _apply_unmatched 不動 ... def _try_user_selected_match(self, parsed, tenant_id): if not parsed.matched_user_label: return None from domain.oscal.service.reconciliation._normalizers import parse_user_lookup_label nickname, login_name = parse_user_lookup_label(parsed.matched_user_label) if not login_name: return None try: from jedi_auth.domain.entities.user_query_entity import UserQueryEntity users = self._user.get_users(UserQueryEntity(login_name=login_name)) for u in users or []: if getattr(u, "is_active", True): return u except Exception: pass return None ``` - [ ] **Step 4**: run → 4/4 PASS + A3 既有 16 person test 全綠 ### 4.2 OrganizationReconciler override `_try_user_selected_match` **Files:** - Modify: `domain/oscal/service/reconciliation/organization_reconciler.py` - Modify: `tests/test_a4_a3_retroactive.py` - [ ] **Step 1-4**: 同 4.1 模式（4 test + 加 override） ```python class OrganizationReconciler(BaseReconciliationService[ParsedParty, object]): # ... 既有不動 ... def _try_user_selected_match(self, parsed, tenant_id): if not parsed.matched_org_unit_label: return None name = parsed.matched_org_unit_label.strip() if not name: return None try: from jedi_auth.domain.entities.org_unit_query_entity import OrgUnitQueryEntity q = OrgUnitQueryEntity(name=name) if tenant_id is not None: q.tenant_id = tenant_id orgs = self._org.get_org_units(q) for o in orgs or []: return o except Exception: pass return None ``` ### 4.3 `_dict_to_parsed_parties` 加 label 傳遞 **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py` - Modify: `tests/test_a4_a3_retroactive.py` - [ ] **Step 1**: 寫 failing test（org + person dict → ParsedParty label 欄位填值） ```python class TestDictToParsedPartiesRetroactive: def test_org_dict_to_parsed_party_with_parent_org_label(self): from app.oscal.service.ssp_excel_import_app_service import SspExcelImportAppService orgs = [{"name": "Sub-Org", "parent_org": "Parent-Org"}] result = SspExcelImportAppService._dict_to_parsed_parties(orgs, []) assert result[0].matched_org_unit_label == "Parent-Org" def test_person_dict_to_parsed_party_with_user_and_org_labels(self): persons = [{"name": "Alice", "email": "a@x.com", "matched_user": "Alice ", "org_unit": "Engineering"}] result = SspExcelImportAppService._dict_to_parsed_parties([], persons) assert result[0].matched_user_label == "Alice " assert result[0].matched_org_unit_label == "Engineering" ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 改 `_dict_to_parsed_parties` ```python @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 ``` - [ ] **Step 4**: run → 2/2 PASS ### 4.4 Commit T4 ```bash git add domain/oscal/service/reconciliation/person_reconciler.py \ domain/oscal/service/reconciliation/organization_reconciler.py \ app/oscal/service/ssp_excel_import_app_service.py \ tests/test_a4_a3_retroactive.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T4 A3 retroactive USER_SELECTED stage PersonReconciler / OrganizationReconciler override _try_user_selected_match hook（A4 T1 base stage 0 dispatch 已落地）： - Person: parse 'nickname ' → UserQueryEntity(login_name=...) - Organization: parse parent_org name → OrgUnitQueryEntity(name=, tenant_id=) _dict_to_parsed_parties 補 3 個 label 傳遞: - org.parent_org → ParsedParty.matched_org_unit_label - person.matched_user → ParsedParty.matched_user_label - person.org_unit → ParsedParty.matched_org_unit_label 新增 10 個 retroactive test 全綠；A3 既有 32 person+org test 全綠（USER_SELECTED 加入後既有 EXACT / NORMALIZED / FUZZY 行為不變）。 對齊 design-A4 §5.2 / §6.5. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 4 Acceptance - [ ] PersonReconciler `_try_user_selected_match` 用 `parse_user_lookup_label` + `UserQueryEntity(login_name=)` - [ ] OrganizationReconciler `_try_user_selected_match` 用 `OrgUnitQueryEntity(name=, tenant_id=)` - [ ] `_dict_to_parsed_parties` 3 個 label 傳遞（parent_org / matched_user / org_unit） - [ ] 10 個 A4 retroactive test 全綠 - [ ] A3 既有 16 person + 16 org test 全綠（regression 守住） --- ## Task 5 — A2 §15.3 整併 > **目標**：`_confirm_superset_flow` 內呼叫 `CatalogControlReconciler` 反查 `catalog_control_uid`，填入 `mf_payload.include_controls` 取代 A2 的空 list。 > **TDD 邊界**：1 個 integration test 跑通完整 _confirm_superset_flow → include_controls 內含正確 uid。 ### 5.1 改 `_confirm_superset_flow` **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py` - Create: `tests/test_a4_a2_integration.py` - [ ] **Step 1**: 寫 failing integration test ```python # tests/test_a4_a2_integration.py class TestSuperset15_3: def test_include_controls_populated_from_matched_catalog_control_uid(self, monkeypatch): """A2 §15.3 整併：confirm_superset_flow 內 ControlReconciler 反查後填 include_controls.""" # mock fw_version + mf_service + catalog_control_reconciler # 跑 confirm_superset_flow → assert mf_payload.include_controls == [uid1, uid2] # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 改 `_confirm_superset_flow` ```python def _confirm_superset_flow(self, job, parsed_result, payload, user_context): fw_version = self._fw_version_domain.get_by_uid(job.source_uid) if fw_version is None: raise NotFound(GrcErrorCode.GRC_FRAMEWORK_VERSION_NOT_FOUND) controls_with_aos = parsed_result.get("controls_with_aos") or [] target_control_ids = [ c.get("control_id") for c in controls_with_aos if c.get("statement_id") is None and c.get("_target_in_profile") is True and c.get("control_id") ] # A4 §15.3 整併 include_uids = [] if self._catalog_control_reconciler is not None and target_control_ids: from domain.oscal.parser.ssp_intermediate import ParsedControl parsed_for_resolve = [ParsedControl(control_id=cid) for cid in target_control_ids] try: self._catalog_control_reconciler.reconcile(parsed_for_resolve, fw_version.catalog_id) except Exception as e: logger.warning("A4 §15.3 include_controls 反查 skipped: %s", e) include_uids = [ str(p.matched_catalog_control_uid) for p in parsed_for_resolve if p.matched_catalog_control_uid ] metadata = job.metadata or {} mf_payload = { "name": metadata.get("target_mf_name") or f"MF from Excel {job.uid[:8]}", "group": metadata.get("target_group", ""), "version": metadata.get("target_version", "1.0"), "frequency": metadata.get("target_frequency", "annual"), "description": metadata.get("target_description", ""), "provider": metadata.get("target_provider", "Billows-Official"), "oscal_framework_version_uid": fw_version.uid, "include_controls": include_uids, # ← A4 真正 subset profile "status": "draft", } if target_control_ids: logger.info( "A4 superset flow: %d/%d control_id resolved to catalog_control_uid", len(include_uids), len(target_control_ids), ) mf_dto = self._mf_service.add_module_frame(...) # ... T6 / T8 補 SSP shell + write pipeline ``` - [ ] **Step 4**: run → PASS ### 5.2 Commit T5 ```bash git add app/oscal/service/ssp_excel_import_app_service.py \ tests/test_a4_a2_integration.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T5 A2 §15.3 superset include_controls 反查 _confirm_superset_flow 加 catalog_control_reconciler 預先反查 target control_id → catalog_control_uid；填入 mf_payload.include_controls 取代 A2 的空 list（全納入 fallback）。 對齊 design-A2 §15.3 follow-up + design-A4 §6.4. 效果：superset upload flow 從此真正建立 subset profile（只含 user 標記 _target_in_profile=TRUE 的 control），而非整 framework catalog。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 5 Acceptance - [ ] `_confirm_superset_flow` 內呼叫 `catalog_control_reconciler.reconcile()` - [ ] `include_controls` 從 reconcile 結果取 `matched_catalog_control_uid` 字串清單 - [ ] `target_control_ids` 全部 unmatched 時 include_controls=[] + log info 提示（保留 A2 fallback 行為） - [ ] 1 個 integration test PASS - [ ] design-A2 §15.3 標記更新（從 follow-up → resolved） ### Session D → E 收尾 當 T3 + T4 + T5 ship 後（Session D 結束）： - [ ] 跑 `poetry run pytest tests/test_a3_*.py tests/test_a4_*.py -v` 確認 0 regression - [ ] 產 handoff prompt → `docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/YYYY-MM-DD-a4-d-to-e.md` - 含已 ship commits (T3/T4/T5) + T6/T7/T8 開工指引 + jedi-oscal SSP Entity verify 結果（T0.1-T0.3 帶到 T6） - [ ] 換 Session E 接手 — T6 (SSP shell) 先，T7 (WriteStrategy) 後，T8 (pipeline) 最後 --- ## Task 6 — SSP shell + 新 error code > **目標**：實作 `_create_ssp_shell`（superset flow）+ `_resolve_existing_ssp_shell`（update flow）+ 新 error code `GRC_EXCEL_UPDATE_FLOW_NO_SSP` (412)。 > **依賴**：T0.1 / T0.2 / T0.3 / T0.7 verify 結果（jedi-oscal SspEntity / SystemCharacteristic / SystemImplementation main / get_one_by_profile_uid 是否存在）。 ### 6.1 加 `GRC_EXCEL_UPDATE_FLOW_NO_SSP` error code **Files:** - Modify: `common/code/grc_error_code.py` - [ ] **Step 1**: 序號對齊 — `grep "GRC_412" common/code/grc_error_code.py` - [ ] **Step 2**: 加新 entry ```python class GrcErrorCode(BaseCode): # ... 既有 GRC_EXCEL_UPDATE_FLOW_NO_SSP = ("Excel update flow 失敗：對應 MF 尚未建立 SSP", "GRC_412066") ``` - [ ] **Step 3**: import smoke ```bash poetry run python3 -c "from common.code.grc_error_code import GrcErrorCode; print(GrcErrorCode.GRC_EXCEL_UPDATE_FLOW_NO_SSP)" ``` ### 6.2 `_create_ssp_shell` 實作 **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py` - Create: `tests/test_a4_ssp_shell.py` - [ ] **Step 1**: 寫 failing test（3 個 domain service add 各呼叫 1 次 + 回傳 (ssp_id, sys_impl_main_id)） ```python # tests/test_a4_ssp_shell.py class TestCreateSspShell: def test_create_shell_calls_3_domain_services(self): # mock ssp_domain / sc_domain / sys_impl_domain # call _create_ssp_shell(mf_uid='x', metadata={...}, user='u') # assert 3 domain.add 各呼叫 1 次 # assert 返回 (ssp_id, sys_impl_main_id) # ... def test_create_shell_uses_metadata_target_mf_name_for_title(self): # ... def test_create_shell_links_profile_uid_from_mf(self): # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 加 `_create_ssp_shell` method（依 T0 verify 結果調整 Entity 構造參數） ```python 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 內呼叫。 Entity 構造參數依 T0 verify 結果調整 — 若有偏差需修 design §11 補紀錄。 """ from jedi_oscal.domain.entity.ssp.ssp_entity import SystemSecurityPlanEntity from jedi_oscal.domain.entity.ssp.ssp_system_characteristic_entity import SspSystemCharacteristicEntity from jedi_oscal.domain.entity.ssp.ssp_system_implementation_entity import SspSystemImplementationEntity mf = self._mf_domain.get_module_frame_by_uid(mf_uid) if mf is None: raise NotFound(GrcErrorCode.GRC_MODULE_FRAME_NOT_FOUND) # 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（A0.1 scope_type='ssp'） 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) ``` - [ ] **Step 4**: run → 3/3 PASS ### 6.3 `_resolve_existing_ssp_shell` 實作 **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py` - Modify: `tests/test_a4_ssp_shell.py` - [ ] **Step 1**: 寫 failing test（3 case：existing SSP 通 / 無 SSP raise / 無 system_implementation raise） ```python class TestResolveExistingSspShell: def test_existing_ssp_returns_ids(self): # mock mf + ssp_domain.get_one_by_profile_uid → ssp # mock system_implementation_domain.get_one_by_ssp_id → impl_main # assert 返回 (ssp.id, impl_main.id) # ... def test_no_ssp_raises_GRC_412066(self): # ssp_domain.get_one_by_profile_uid → None # assert raise PreconditionFailedError(GRC_EXCEL_UPDATE_FLOW_NO_SSP) # ... def test_no_system_implementation_raises_GRC_412066(self): # ssp 存在 but system_implementation_domain.get_one_by_ssp_id → None # assert raise # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 加 method ```python def _resolve_existing_ssp_shell(self, mf_uid: str) -> tuple[int, int]: mf = self._mf_domain.get_module_frame_by_uid(mf_uid) if mf is None: raise NotFound(GrcErrorCode.GRC_MODULE_FRAME_NOT_FOUND) 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) 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) ``` - [ ] **Step 4**: run → 3/3 PASS ### 6.4 Commit T6 ```bash git add common/code/grc_error_code.py \ app/oscal/service/ssp_excel_import_app_service.py \ tests/test_a4_ssp_shell.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T6 SSP shell + GRC_412066 error code _create_ssp_shell（superset flow）建 3 步驟：SSP + system_characteristic + system_implementation main（A0.1 scope_type='ssp'）。 _resolve_existing_ssp_shell（update flow）查既有 SSP + system_implementation main；任一不存在 raise PreconditionFailedError(GRC_EXCEL_UPDATE_FLOW_NO_SSP) 明確指出 import flow 不負責建第 1 版 SSP（屬 ssp_versioning_service 職責）。 新增 6 SSP shell test + 1 error code import smoke = 7 test 全綠。 對齊 design-A4 §6.2 / §6.3 + T0 verify 結果。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 6 Acceptance - [ ] `GRC_EXCEL_UPDATE_FLOW_NO_SSP` (GRC_412066) 加入 grc_error_code.py - [ ] `_create_ssp_shell` 3 個 domain service 順序呼叫（SSP → SystemCharacteristic → SystemImplementation main） - [ ] `_resolve_existing_ssp_shell` SSP / impl_main 任一無 → raise 412 - [ ] T0 verify 結果與 design §6.2 假設一致；若不符已補 design §11 - [ ] 7 個 SSP shell test 全綠 --- ## Task 7 — 5 WriteStrategy > **目標**：建 `write_strategy/` sub-folder + 5 WriteStrategy 完整實作。 > **TDD 邊界**：每 WriteStrategy ~6-8 test（MATCHED upsert / UNMATCHED skip or 純文字 / batch / FK 連動 / log warning）。 ### 7.1 新建 `write_strategy/` sub-folder ```bash mkdir -p domain/oscal/service/write_strategy touch domain/oscal/service/write_strategy/__init__.py ``` ### 7.2 ControlWriteStrategy **Files:** - Create: `domain/oscal/service/write_strategy/control_write_strategy.py` - Test: `tests/test_a4_write_strategy_control.py` - [ ] **Step 1**: 寫 6 failing test ```python # tests/test_a4_write_strategy_control.py class TestControlWriteStrategy: def test_unmatched_skip_log_warning(self): """matched_catalog_control_id=None → skip + log.""" # ... def test_matched_upsert_returns_ctrl_impl_id_map(self): """MATCHED ParsedControl → upsert system_security_plan_control_implementations + 返回 map.""" # ... def test_returns_count(self): # ... def test_implementation_status_from_parsed(self): # ... def test_existing_row_update_not_insert(self): """同 (ssp_id, catalog_control_id) 已存在 → update 不 insert.""" # ... def test_returns_tuple_count_and_map(self): """返回 (written_count, ctrl_impl_id_map: dict[catalog_control_id → ssp_control_implementation_id]).""" # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 寫 ControlWriteStrategy ```python # domain/oscal/service/write_strategy/control_write_strategy.py import logging from typing import Optional from domain.oscal.parser.ssp_intermediate import ParsedControl from domain.oscal.service.reconciliation.match_method import MatchMethod logger = logging.getLogger(__name__) class ControlWriteStrategy: """寫入 system_security_plan_control_implementations。 Caller wraps in @transaction; this class never opens a session. """ def __init__(self, control_implementation_domain_service): self._ci = control_implementation_domain_service def write( self, parsed_controls: list[ParsedControl], ssp_id: int, user_context, ) -> tuple[int, dict]: """Returns (written_count, ctrl_impl_id_map: dict[catalog_control_id → ssp_control_implementation_id]).""" written = 0 ctrl_impl_id_map: dict[int, int] = {} for c in parsed_controls: if c.matched_catalog_control_id is None: logger.warning("ControlWriteStrategy skip unmatched control_id=%s", c.control_id) continue try: from jedi_oscal.domain.entity.ssp.ssp_control_implementation_entity import ( SspControlImplementationEntity, ) from jedi_oscal.domain.entity.ssp.ssp_control_implementation_query_entity import ( SspControlImplementationQueryEntity, ) # upsert by (ssp_id, catalog_control_id) existing = self._ci.get_one(SspControlImplementationQueryEntity( system_security_plan_id=ssp_id, catalog_control_id=c.matched_catalog_control_id, )) entity = SspControlImplementationEntity( system_security_plan_id=ssp_id, catalog_control_id=c.matched_catalog_control_id, implementation_status=c.impl_status or "unknown", implementation_description=c.statement, ) if existing is None: dto = self._ci.add(entity, curr_user=user_context.login_name) ctrl_impl_id_map[c.matched_catalog_control_id] = dto.id else: entity.uid = existing.uid dto = self._ci.update(entity, curr_user=user_context.login_name) ctrl_impl_id_map[c.matched_catalog_control_id] = existing.id written += 1 except Exception as e: logger.warning("ControlWriteStrategy write failed for control_id=%s: %s", c.control_id, e, exc_info=True) return (written, ctrl_impl_id_map) ``` - [ ] **Step 4**: run → 6/6 PASS ### 7.3 AoWriteStrategy **Files:** - Create: `domain/oscal/service/write_strategy/ao_write_strategy.py` - Test: `tests/test_a4_write_strategy_ao.py` - [ ] **Step 1-4**: 同 7.2 模式 — 5 test ```python class AoWriteStrategy: """寫入 ssp_control_impl_objective.""" def __init__(self, ssp_control_impl_objective_domain_service): self._cio = ssp_control_impl_objective_domain_service def write( self, parsed_aos: list, # List[ParsedAssessmentObjective] ssp_id: int, ctrl_impl_id_map: dict, # catalog_control_id → ssp_control_implementation_id user_context, ) -> int: written = 0 for ao in parsed_aos: if ao.matched_catalog_control_assessment_id is None: logger.warning("AoWriteStrategy skip unmatched control_id=%s statement_id=%s", ao.control_id, ao.statement_id) continue # 取父 control 的 ssp_control_implementation_id（透過 ctrl_impl_id_map） # ao.control_id (字面) → matched_catalog_control_id（要從 parsed_controls 反查） # 假設 caller (orchestrator + write pipeline) 把 ao 跟父 ParsedControl 已關聯； # 簡化做法：sci_id 從 ctrl_impl_id_map 拿（key 是 catalog_control_id） # AO 本身有 matched_catalog_control_assessment_id 但 sci_id 要 caller 拼 # ↓ 簡化 logic：實作時若需要 sci_id 反查 — 由 _write_all_data 拼好傳入 # ... try: from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ( SspControlImplObjectiveEntity, ) # 期待 caller 已把 (catalog_control_assessment_id, ssp_control_implementation_id) # 兩個 FK 一起傳入；此處需要從 ctrl_impl_id_map 查 sci_id # 由 caller (_write_all_data) 在準備 parsed_aos 前把 ao.ssp_control_implementation_id 填好 sci_id = getattr(ao, "_ssp_control_implementation_id", None) # caller 暫填 if sci_id is None: logger.warning("AoWriteStrategy skip — no ssp_control_implementation_id for statement_id=%s", ao.statement_id) continue entity = SspControlImplObjectiveEntity( ssp_control_implementation_id=sci_id, catalog_control_assessment_id=ao.matched_catalog_control_assessment_id, implementation_status=ao.impl_status or "unknown", implementation_description=ao.statement, ) self._cio.add(entity, curr_user=user_context.login_name) written += 1 except Exception as e: logger.warning("AoWriteStrategy write failed: %s", e, exc_info=True) return written ``` **NOTE**：AO write 需要 `ssp_control_implementation_id`（不是 `catalog_control_id`）；caller `_write_all_data` Step 6 內把 ControlWriteStrategy 返回的 `ctrl_impl_id_map` 套上 AO 後再呼叫 AoWriteStrategy。 ### 7.4 DeviceWriteStrategy **Files:** - Create: `domain/oscal/service/write_strategy/device_write_strategy.py` - Test: `tests/test_a4_write_strategy_device.py` - [ ] **Step 1-4**: 6 test（MATCHED upsert + UNMATCHED 仍寫入純文字 + device_id null） ```python class DeviceWriteStrategy: """寫入 ssp_system_implementation_items (implementation_type='hardware').""" def __init__(self, ssp_impl_item_domain_service): self._item = ssp_impl_item_domain_service def write( self, parsed_devices: list, # List[ParsedDevice] ssp_id: int, sys_impl_main_id: int, user_context, ) -> int: written = 0 for d in parsed_devices: try: from jedi_oscal.domain.entity.ssp.ssp_system_implementation_item_entity import ( SspSystemImplementationItemEntity, ) entity = SspSystemImplementationItemEntity( system_security_plan_id=ssp_id, system_implementation_id=sys_impl_main_id, implementation_type='hardware', name=d.name, description=d.purpose, device_id=d.matched_device_id, # UNMATCHED 時為 None ) self._item.add(entity, curr_user=user_context.login_name) written += 1 except Exception as e: logger.warning("DeviceWriteStrategy write failed: %s", e, exc_info=True) return written ``` ### 7.5 InformationSystemWriteStrategy **Files:** - Create: `domain/oscal/service/write_strategy/information_system_write_strategy.py` - Test: `tests/test_a4_write_strategy_information_system.py` - [ ] **Step 1-4**: 6 test（含 matched_system_owner_user_id 寫入） ```python class InformationSystemWriteStrategy: """寫入 ssp_system_implementation_items (implementation_type='system'). NOTE: system_characteristic_id 對應 A0.1 schema rename — 樣板 info_system 對應的 InformationSystemEntity.id 寫入此欄位（不是 OSCAL system_characteristic 主結構）。 """ def __init__(self, ssp_impl_item_domain_service): self._item = ssp_impl_item_domain_service def write( self, parsed_info_systems: list, ssp_id: int, sys_impl_main_id: int, user_context, ) -> int: written = 0 for s in parsed_info_systems: try: from jedi_oscal.domain.entity.ssp.ssp_system_implementation_item_entity import ( SspSystemImplementationItemEntity, ) entity = SspSystemImplementationItemEntity( system_security_plan_id=ssp_id, system_implementation_id=sys_impl_main_id, implementation_type='system', name=s.name, description=s.description, system_characteristic_id=s.matched_info_system_id, responsible_party=str(s.matched_system_owner_user_id) if s.matched_system_owner_user_id else None, ) self._item.add(entity, curr_user=user_context.login_name) written += 1 except Exception as e: logger.warning("InfoSystemWriteStrategy write failed: %s", e, exc_info=True) return written ``` ### 7.6 LeveragedWriteStrategy **Files:** - Create: `domain/oscal/service/write_strategy/leveraged_write_strategy.py` - Test: `tests/test_a4_write_strategy_leveraged.py` - [ ] **Step 1-4**: 6 test（含 date_authorized / party_uuid 寫入） ```python class LeveragedWriteStrategy: """寫入 ssp_system_implementation_items (implementation_type='leveraged-authorization').""" def __init__(self, ssp_impl_item_domain_service): self._item = ssp_impl_item_domain_service def write( self, parsed_leveraged: list, ssp_id: int, sys_impl_main_id: int, user_context, ) -> int: written = 0 for l in parsed_leveraged: try: from jedi_oscal.domain.entity.ssp.ssp_system_implementation_item_entity import ( SspSystemImplementationItemEntity, ) entity = SspSystemImplementationItemEntity( system_security_plan_id=ssp_id, system_implementation_id=sys_impl_main_id, implementation_type='leveraged-authorization', name=l.service_name, title=l.provider, purpose=l.purpose, party_uuid=l.matched_party_uuid, date_authorized=l.date_authorized, ) self._item.add(entity, curr_user=user_context.login_name) written += 1 except Exception as e: logger.warning("LeveragedWriteStrategy write failed: %s", e, exc_info=True) return written ``` ### 7.7 DI Container 加 5 WriteStrategy Factory **Files:** - Modify: `di_containers/oscal/oscal_containers.py` ```python from domain.oscal.service.write_strategy.control_write_strategy import ControlWriteStrategy from domain.oscal.service.write_strategy.ao_write_strategy import AoWriteStrategy from domain.oscal.service.write_strategy.device_write_strategy import DeviceWriteStrategy from domain.oscal.service.write_strategy.information_system_write_strategy import InformationSystemWriteStrategy from domain.oscal.service.write_strategy.leveraged_write_strategy import LeveragedWriteStrategy control_write_strategy = providers.Factory( ControlWriteStrategy, control_implementation_domain_service=control_implementation_domain_service, ) ao_write_strategy = providers.Factory( AoWriteStrategy, ssp_control_impl_objective_domain_service=control_implementation_objective_domain_service, ) device_write_strategy = providers.Factory( DeviceWriteStrategy, ssp_impl_item_domain_service=system_implementation_item_domain_service, ) information_system_write_strategy = providers.Factory( InformationSystemWriteStrategy, ssp_impl_item_domain_service=system_implementation_item_domain_service, ) leveraged_write_strategy = providers.Factory( LeveragedWriteStrategy, ssp_impl_item_domain_service=system_implementation_item_domain_service, ) ``` ### 7.8 Commit T7 ```bash git add domain/oscal/service/write_strategy/ \ tests/test_a4_write_strategy_*.py \ di_containers/oscal/oscal_containers.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T7 5 write strategies 新增 domain/oscal/service/write_strategy/ sub-folder + 5 個 WriteStrategy: - ControlWriteStrategy: 寫 system_security_plan_control_implementations， 返回 (count, ctrl_impl_id_map) - AoWriteStrategy: 寫 ssp_control_impl_objective（依賴 sci_id） - DeviceWriteStrategy / InformationSystemWriteStrategy / LeveragedWriteStrategy: 共寫 ssp_system_implementation_items（implementation_type 區分 hardware / system / leveraged-authorization） UNMATCHED 行為（D11 拍板）： - Control / AO: log warning + skip（沒 FK 寫不進去） - Device / InfoSystem / Leveraged: 仍寫入（純文字保留 user data；FK=null） DI Container 加 5 WriteStrategy Factory。 新增 29 個 WriteStrategy test 全綠（Control 6 + AO 5 + Device 6 + InfoSystem 6 + Leveraged 6）。 對齊 design-A4 §6.6. Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 7 Acceptance - [ ] `write_strategy/` sub-folder 建立 - [ ] 5 WriteStrategy class 完整（Control / AO / Device / InfoSystem / Leveraged） - [ ] ControlWriteStrategy 返回 `(written_count, ctrl_impl_id_map)` - [ ] AoWriteStrategy 透過 caller 提供 `sci_id` 依賴 - [ ] UNMATCHED 行為對齊 D11 拍板（control/AO skip；其他 3 個仍寫） - [ ] DI Container 加 5 Factory - [ ] 29 個 WriteStrategy test 全綠 --- ## Task 8 — `_write_all_data` 8-step pipeline + DI 注入 > **目標**：把 8-step pipeline 完整實作；DI `SspExcelImportAppService` 加 11 個依賴；5 個 `_dict_to_*` helper 補完。 ### 8.1 5 個 `_dict_to_*` helper **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py` - [ ] **Step 1**: 寫 failing test（5 個 dict→dataclass 轉換 + nested objectives） ```python # tests/test_a4_a2_integration.py 補 class TestDictToParsed: def test_dict_to_parsed_devices(self): # ... def test_dict_to_parsed_info_systems(self): # ... def test_dict_to_parsed_leveraged_with_date(self): # ... def test_dict_to_parsed_controls_with_nested_aos(self): """父 control + 子 AO list.""" # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 加 5 個 static method ```python @staticmethod def _dict_to_parsed_devices(devices: list[dict]) -> list[ParsedDevice]: result = [] for d in devices: name = (d.get("name") or "").strip() if not name: continue result.append(ParsedDevice( name=name, ip=d.get("ip"), os=d.get("os"), device_type=d.get("device_type"), status=d.get("status"), purpose=d.get("purpose"), matched_device_label=d.get("matched_device"), )) return result @staticmethod def _dict_to_parsed_info_systems(info_systems: list[dict]) -> list[ParsedInformationSystem]: result = [] for s in info_systems: name = (s.get("name") or "").strip() if not name: continue result.append(ParsedInformationSystem( name=name, abbreviation=s.get("abbreviation"), description=s.get("description"), component_type=s.get("component_type"), status=s.get("status"), matched_info_system_label=s.get("matched_info_system"), system_owner_label=s.get("system_owner"), )) return result @staticmethod def _dict_to_parsed_leveraged(leveraged: list[dict]) -> list[ParsedLeveraged]: result = [] for l in leveraged: service_name = (l.get("service_name") or "").strip() provider = (l.get("provider") or "").strip() if not service_name and not provider: continue # date_authorized parse: Excel 可能傳 str / datetime from datetime import date, datetime raw_date = l.get("date_authorized") parsed_date = None if isinstance(raw_date, date): parsed_date = raw_date elif isinstance(raw_date, datetime): parsed_date = raw_date.date() elif isinstance(raw_date, str) and raw_date.strip(): try: parsed_date = datetime.fromisoformat(raw_date.strip()).date() except ValueError: pass result.append(ParsedLeveraged( service_name=service_name or provider, provider=provider or service_name, party_label=l.get("party"), date_authorized=parsed_date, purpose=l.get("purpose"), )) return result @staticmethod def _dict_to_parsed_controls(controls_with_aos: list[dict]) -> list[ParsedControl]: result = [] for c in controls_with_aos: if c.get("statement_id") is not None: continue # 子 row 已 nested 在父 row.objectives 內，不重複 control_id = (c.get("control_id") or "").strip() if not control_id: continue pc = ParsedControl( control_id=control_id, control_name=c.get("control_name"), objective_id=c.get("objective_id"), objective_name=c.get("objective_name"), impl_status=c.get("impl_status"), statement=c.get("statement"), reference_doc=c.get("reference_doc"), include_in_profile=c.get("include_in_profile"), _target_in_profile=c.get("_target_in_profile", True), ) for ao_dict in c.get("objectives") or []: pc.objectives.append(ParsedAssessmentObjective( statement_id=ao_dict.get("statement_id"), control_id=ao_dict.get("control_id"), objective_id=ao_dict.get("objective_id"), objective_name=ao_dict.get("objective_name"), impl_status=ao_dict.get("impl_status"), statement=ao_dict.get("statement"), reference_doc=ao_dict.get("reference_doc"), )) result.append(pc) return result ``` - [ ] **Step 4**: run → PASS ### 8.2 `_write_all_data` 重整為 8-step pipeline **Files:** - Modify: `app/oscal/service/ssp_excel_import_app_service.py` - Modify: `tests/test_a4_a2_integration.py` - [ ] **Step 1**: 寫 integration test（整段 pipeline 在 superset / update flow 內跑通） ```python class TestWriteAllDataPipeline: def test_8_step_pipeline_superset_flow(self, monkeypatch): """ssp_id 由 _create_ssp_shell 返回；5 reconciler.reconcile 各呼叫 1 次；5 write_strategy.write 各呼叫 1 次.""" # ... def test_8_step_pipeline_update_flow(self, monkeypatch): """SSP shell 走 _resolve_existing_ssp_shell；其他步驟同上.""" # ... def test_pipeline_skip_steps_when_dep_none(self, monkeypatch): """A4 optional default None — pipeline 個別步驟 dep 為 None 時 logger.warning skip.""" # ... ``` - [ ] **Step 2**: run → FAIL - [ ] **Step 3**: 改 `_write_all_data`（design §6.1）+ `_confirm_*_flow` 補 `is_superset_flow` / `fw_version` 參數 ```python def _write_all_data( self, parsed_result, target_mf_uid, user_context, is_superset_flow: bool, fw_version=None, ) -> dict: # Step 1: party reconcile (A3 既有 + A4 retroactive) parties_org = parsed_result.get("parties_org") or [] parties_person = parsed_result.get("parties_person") or [] all_parties = self._dict_to_parsed_parties(parties_org, parties_person) if all_parties and self._reconciliation is not None: try: self._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 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) from domain.oscal.parser.ssp_intermediate import ( ParsedExcelEntityBundle, SspEntityReconciliationContext, ) bundle = ParsedExcelEntityBundle( parsed_devices=parsed_devices, parsed_info_systems=parsed_info_systems, parsed_leveraged=parsed_leveraged, parsed_controls=parsed_controls, ) ctx = SspEntityReconciliationContext( tenant_id=user_context.tenant_id, catalog_id=catalog_id, ) 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 = 0 if all_parties: try: parties_written = self._mf_write.write_parties( parties=all_parties, source_uid=target_mf_uid, user_id=user_context.login_name, ) except Exception as e: logger.warning("write_parties failed: %s", e, exc_info=True) # Step 6: A4 entity write ctrl_written = 0 ctrl_impl_id_map: dict = {} if self._control_write_strategy is not None: ctrl_written, ctrl_impl_id_map = self._control_write_strategy.write( parsed_controls, ssp_id, user_context, ) parsed_aos = [] for c in parsed_controls: for ao in c.objectives: ao._ssp_control_implementation_id = ctrl_impl_id_map.get(c.matched_catalog_control_id) parsed_aos.append(ao) ao_written = 0 if self._ao_write_strategy is not None: ao_written = self._ao_write_strategy.write( parsed_aos, ssp_id, ctrl_impl_id_map, user_context, ) devices_written = 0 if self._device_write_strategy is not None: devices_written = self._device_write_strategy.write( parsed_devices, ssp_id, sys_impl_main_id, user_context, ) info_systems_written = 0 if self._info_system_write_strategy is not None: info_systems_written = self._info_system_write_strategy.write( parsed_info_systems, ssp_id, sys_impl_main_id, user_context, ) leveraged_written = 0 if self._leveraged_write_strategy is not None: leveraged_written = self._leveraged_write_strategy.write( parsed_leveraged, ssp_id, sys_impl_main_id, user_context, ) # Step 7: parsed_result 回填（給 A5 UI 讀） self._update_parsed_result_with_reconcile(parsed_result, all_parties, bundle) 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, } def _resolve_catalog_id(self, fw_version, mf_uid, is_superset_flow) -> Optional[int]: if is_superset_flow and fw_version is not None: return getattr(fw_version, "catalog_id", None) # update flow — 從 mf.profile.catalog_id 拿 mf = self._mf_domain.get_module_frame_by_uid(mf_uid) if mf and mf.oscal_profile_uid: # 透過 profile_domain_service 查 catalog_id # 由 T0 verify 確認 profile_domain_service 存在 return None # 簡化 — T8 verify 後實作 return None def _update_parsed_result_with_reconcile(self, parsed_result, all_parties, bundle): """把 reconcile 結果寫回 parsed_result JSONB；A5 預覽 UI 讀.""" # parties 回填 (A3 retroactive) # 5 entity 回填 — asdict() 或手動 dict 映射；給 A5 看 match_method / matched_*_id # 詳細實作見 ssp_excel_import_app_service.py T8 落地 pass # T8 step 3 補完整邏輯 ``` - [ ] **Step 4**: run → 3/3 PASS ### 8.3 DI 加 11 個依賴注入 **Files:** - Modify: `di_containers/oscal/oscal_containers.py` ```python ssp_excel_import_app_service = providers.Factory( SspExcelImportAppService, # 既有 8 個（parse_job / parser / file_upload / mf_domain / mf_service / fw_version_domain / mf_write / party_reconciliation_service） # A4 新增 11 個： 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, ) ``` `SspExcelImportAppService.__init__` 同步加 11 個 optional default None 參數（A2 既有 fixture 不破壞）： ```python def __init__( self, parse_job_domain_service, parser, file_upload_service, module_frame_domain_service, module_frame_service, oscal_framework_version_domain_service, module_frame_write_strategy, party_reconciliation_service=None, # A4 新增 11 個 ssp_entity_orchestrator=None, catalog_control_reconciler=None, ssp_domain_service=None, system_characteristic_domain_service=None, system_implementation_domain_service=None, control_write_strategy=None, ao_write_strategy=None, device_write_strategy=None, information_system_write_strategy=None, leveraged_write_strategy=None, ): # 既有 attr # A4 attr self._ssp_entity_orchestrator = ssp_entity_orchestrator self._catalog_control_reconciler = catalog_control_reconciler self._ssp_domain = ssp_domain_service self._system_characteristic_domain = system_characteristic_domain_service self._system_implementation_domain = system_implementation_domain_service self._control_write_strategy = control_write_strategy self._ao_write_strategy = ao_write_strategy self._device_write_strategy = device_write_strategy self._info_system_write_strategy = information_system_write_strategy self._leveraged_write_strategy = leveraged_write_strategy ``` ### 8.4 改 `_confirm_*_flow` 串 `_write_all_data` 新參數 `_confirm_superset_flow` 內 call: ```python 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, ) ``` `_confirm_update_flow` 內 call: ```python write_stats = self._write_all_data( parsed_result=parsed_result, target_mf_uid=existing_mf.uid, user_context=user_context, is_superset_flow=False, fw_version=None, ) ``` `summary` dict 用 `**write_stats` spread；DTO 仍維持既有 response shape。 ### 8.5 Commit T8 ```bash git add app/oscal/service/ssp_excel_import_app_service.py \ di_containers/oscal/oscal_containers.py \ tests/test_a4_a2_integration.py git commit -m "$(cat <<'EOF' feat(oscal): A4 T8 _write_all_data 8-step pipeline _write_all_data 重整為 8 step 完整 pipeline（design §3.3 / §6.1）： 1. party reconcile（A3 + A4 retroactive） 2. SSP shell（superset 建；update resolve） 3. 5 個 _dict_to_* 轉 typed dataclass 4. A4 orchestrator reconcile 5. party write（A3 既有 ModuleFrameWriteStrategy） 6. 5 entity write（control / ao / device / info_system / leveraged） 7. parsed_result 回填 matched_*_id / match_method（給 A5 UI） 8. 統計 return dict SspExcelImportAppService 加 11 個 optional default None 注入（既有 29 fixture 不破壞）；DI Container ssp_excel_import_app_service Factory 補 11 個 wiring。_confirm_*_flow 串新參數 (is_superset_flow, fw_version)。 對齊 design-A4 §6.1 + §6.7。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 8 Acceptance - [ ] 5 個 `_dict_to_*` helper 完整實作（含 nested objectives） - [ ] `_write_all_data` 8-step pipeline 對齊 design §6.1 - [ ] `_resolve_catalog_id` / `_update_parsed_result_with_reconcile` 兩 helper 完整 - [ ] `SspExcelImportAppService.__init__` 加 11 optional default None 參數 - [ ] DI Container `ssp_excel_import_app_service` Factory 加 11 wiring - [ ] `_confirm_*_flow` 補 `is_superset_flow` / `fw_version` 參數 - [ ] A2 既有 29 fixture 全綠（optional dep + None 守門） - [ ] 3+ integration test PASS - [ ] **`_resolve_catalog_id` update flow 邏輯實作完整（非 None placeholder）**：從 `mf.profile.catalog_id` 拿；T0 verify 後依實際 ProfileDomainService 簽章補完整實作；test cover update flow catalog_id 反查路徑 ### Session E → F 收尾 當 T6 + T7 + T8 ship 後（Session E 結束）： - [ ] 跑 `poetry run pytest tests/ -v` 確認 0 regression（既有 694 + A4 新增累計 ~100 test） - [ ] 產 handoff prompt → `docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/YYYY-MM-DD-a4-e-to-f.md` - [ ] 換 Session F — T9 (test) → T10 (cucumber) → T11 (收尾) --- ## Task 9 — Unit + Integration Test 補齊 > **目標**：把所有 A4 task 散落的 test 補齊到 design §8 acceptance 規模（~110-140 個）。 ### 9.1 補 reconciler 邊界 case - [ ] DeviceReconciler 加 5 個邊界 case（雙欄不同 ip 不 match / name 空 / IP 空 / multiple match / 純 fallback 三階段排序） - [ ] InformationSystemReconciler 加 5 個（system_owner_label 解析失敗仍 set matched_info_system_id / abbr 配對 / name 空） - [ ] LeveragedReconciler 加 5 個（candidate cache lifecycle / fuzzy 多 candidate 取第一 / suffix 無剔除 → UNMATCHED） - [ ] CatalogControlReconciler / AssessmentObjectiveReconciler 加 5 個（catalog_id missing / parent 配不到 AO unmatched） ### 9.2 補 WriteStrategy 邊界 case - [ ] Control / AO / Device / InfoSystem / Leveraged 各加 3 個邊界（batch 寫入 / domain service exception swallow / empty list 不 fail） ### 9.3 補 integration test - [ ] 5 個 整合測：5 個 entity 從 Excel → reconcile → write 全鏈跑通（mock 各層） - [ ] §15.3 整併測：superset flow include_controls 從 reconcile 結果取 uid - [ ] SSP shell test 完整：superset 建 / update raise ### 9.4 Acceptance check - [ ] 累計 A4 test ~110-140 個 - [ ] 既有 A3 67 test + A2 29 test + 全套 BE 694 test 全綠 - [ ] 跑 `poetry run pytest tests/ -v --tb=short 2>&1 | tail -30` 確認綠 - [ ] `asdict(parsed)` serialize MatchMethod StrEnum 寫 JSONB 行為驗證對齊 A3 T5.2 pattern ### 9.5 Commit T9 ```bash git add tests/test_a4_*.py git commit -m "$(cat <<'EOF' test(oscal): A4 T9 unit + integration tests 補齊 A4 散落 task 的 reconciler / WriteStrategy / integration 邊界 case ~30-40 個，累計 A4 新增 ~110-140 test 全綠（A3 67 + A2 29 + 既有皆無 regression）。 對齊 design-A4 §8 acceptance criteria。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` --- ## Task 10 — Cucumber Regression（compliance-manager-test repo） > **目標**：在 compliance-manager-test 開 branch + 寫 5 scenarios + step + page object。 > **跨 repo**：work 在 `~/Projects/Billows/Audit-Manager/compliance-manager-test/`，**不汙染主 BE working tree**。 ### 10.1 切到 test repo + 開 branch ```bash cd ~/Projects/Billows/Audit-Manager/compliance-manager-test git status --short # 確認乾淨 git checkout -b feature/a4-ssp-excel-entity-match-regression ``` ### 10.2 寫 feature file **File**: `features/regression/module-frame/06-ssp-excel-import-entity-match.feature` ```gherkin Feature: SSP Excel 匯入 — 5 entity 鉤稽 regression 作為合規顧問，我希望 Excel 匯入時 5 種 entity（device / info_system / leveraged / control / AO） 能正確鉤稽到系統內既有 entity，避免重複建立。 Background: Given 我已用 blsadmin 登入 And 已存在 device 'Web-01' IP '10.0.1.5' And 已存在 information_system 'ERP' And 已存在 OSCAL party 'AWS GovCloud' (organization) And 已存在 catalog 'NIST-800-53-r5' 內 control 'AC-1' Scenario: control_id 對到 catalog 內 control Given 我上傳 Excel 樣板（superset flow）含 control_id='AC-1' When 我 confirm import Then parse_job.parsed_result 含 control AC-1 的 matched_catalog_control_uid 不為 null And match_method = 'exact' Scenario: device matched_device 下拉選定 Given 我上傳 Excel 樣板含 device matched_device='Web-01 (10.0.1.5)' When 我 confirm import Then parsed_result 含該 device 的 matched_device_id 不為 null And match_method = 'user_selected' Scenario: info_system name 對到 tenant 內 entity Given 我上傳 Excel 樣板含 info_system name='ERP' When 我 confirm import Then parsed_result 含該 info_system 的 matched_info_system_id 不為 null And match_method = 'exact' Scenario: leveraged provider 套 fuzzy 對到 Party Given 我上傳 Excel 樣板含 leveraged provider='AWS GovCloud 股份有限公司' When 我 confirm import Then parsed_result 含該 leveraged 的 matched_party_uuid 不為 null And match_method = 'fuzzy_name_prefix' Scenario: AO objective_id 父 control 配到後自身也配到 Given 我上傳 Excel 樣板含 control_id='AC-1' + AO objective_id='AC-1.a' When 我 confirm import Then parsed_result 該 AO 的 matched_catalog_control_assessment_uid 不為 null And match_method = 'exact' ``` ### 10.3 寫 step + page object **Files:** - `steps/module-frame/ssp-excel-import-entity-match.steps.js` - `pages/module-frame/ssp-excel-import-page.js` 依沿 A3 cucumber regression pattern；上傳 + confirm + assert parsed_result。 ### 10.4 跑 cucumber ```bash cd ~/Projects/Billows/Audit-Manager/compliance-manager-test npm run cucumber -- features/regression/module-frame/06-ssp-excel-import-entity-match.feature ``` **期待**：5/5 PASS（env 配齊時）；env 未配齊允許 partial ship — scenarios 撰寫完成即 OK。 ### 10.5 Commit（test repo） ```bash git add features/regression/module-frame/06-ssp-excel-import-entity-match.feature \ steps/module-frame/ssp-excel-import-entity-match.steps.js \ pages/module-frame/ssp-excel-import-page.js git commit -m "$(cat <<'EOF' test(compliance-manager-test): A4 T10 cucumber entity match regression 新增 06-ssp-excel-import-entity-match.feature 5 scenarios： - control_id 對到 catalog 內 control (exact) - device matched_device 下拉選定 (user_selected) - info_system name 對到 tenant entity (exact) - leveraged provider fuzzy 對到 Party (fuzzy_name_prefix) - AO objective_id 父 control 配到後自身配到 (exact) 沿 A3 partial ship pattern：env 配齊後跑通即 merge；env 未配齊允許 partial ship。 對齊 design-A4 §6.8 + compliance-manager-be A4 commit chain。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### 10.6 切回主 BE repo ```bash cd ~/Projects/Billows/Audit-Manager/compliance-manager-be git status --short # 確認乾淨 ``` ### Task 10 Acceptance - [ ] test repo branch `feature/a4-ssp-excel-entity-match-regression` 建立 - [ ] feature file 含 5 scenarios（4 exact/user_selected + 1 fuzzy） - [ ] step + page object 撰寫完成 - [ ] cucumber 跑通（env 配齊）或 partial ship（env 未配齊） --- ## Task 11 — 收尾（Changelog + Tracker + Design §11 + SUMMARY） > **目標**：產出 task arc 收口必要的 4 份文件 + 對話歷史 dump。 ### 11.1 Changelog **File**: `docs/changelog/YYYY-MM-DD-feat-ssp-entity-reconcile-and-write.md` ```markdown --- type: feat breaking: false modules: [oscal, ssp-import] commit: --- # Feat — 5 entity 鉤稽 + 寫入 + SSP shell + A3 retroactive（Phase 2 A4） ## 需求說明 ... ## 變更範圍 ... ## 行為差異 ... ## 測試結果 - A4 新增 ~110-140 個 unit + integration test 全綠 - A3 既有 67 + A2 既有 29 + 全套 BE 694 test 無 regression - Cucumber regression 5 scenarios 撰寫完成（env 配齊後跑通） ## Follow-up - F1 docx flow 補 5 reconciler 串接 / F2 A5 預覽 UI fuzzy 拍板 / F3-F5 / F-A4-refactor / F-A4-cucumber-extra ... ``` ### 11.2 Tracker README A4 row 更新 ```markdown | A4 | devices / info_systems / leveraged / 控制項 / AO 鉤稽 + 寫入 | 重 | **BE shipped** ✅ | ✅ Session A | ✅ design-A4.md | ✅ implementation-plan-A4.md | BE: ; test: | ✅ YYYY-MM-DD-feat-ssp-entity-reconcile-and-write.md | ``` ### 11.3 Design §11 Implementation Reality / Reconciliation 補實作偏差 依 A3 / A0.1 §11 樣板：列出每個 T0 verify 帶出的差異 + 實作過程中對 design 假設的修正。**§11 不能空段** — 至少 2 條（mirror SSP Update Diff §11 16 條條目樣板）。 ### 11.4 Final SUMMARY **File**: `docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/YYYY-MM-DD-a4-SUMMARY.md` 依 A3 SUMMARY pattern 8 段： 1. Commit 清單 2. 改動範圍（檔案清單） 3. 行為差異（A4 前 vs A4 後） 4. 規範文件齊全度 5. 已知 follow-up 6. 部署 Handover（BE 重啟 / jedi-oscal path-dep / DB migration N/A） 7. A4 task arc 規範遵守清單 8. A4 phase 收口確認 ### 11.5 對話歷史 dump（task arc 收口） - [ ] 把 A4 對話原文 dump 到 `docs/conversation-history/YYYY-MM-DD/ssp-import-export-phase2-A4/` - [ ] 兩層目錄：先日期、再 topic - [ ] **不二次加工** — 原始對話直接寫入（CLAUDE.md 規範） ### 11.6 Commit T11 ```bash git add docs/changelog/YYYY-MM-DD-feat-ssp-entity-reconcile-and-write.md \ docs/features/FR-011.2-2605-ssp-import-export-phase2/README.md \ docs/features/FR-011.2-2605-ssp-import-export-phase2/design-A4.md \ docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/YYYY-MM-DD-a4-SUMMARY.md \ docs/conversation-history/YYYY-MM-DD/ssp-import-export-phase2-A4/ git commit -m "$(cat <<'EOF' docs(ssp-import-export-phase2): A4 收尾 + SUMMARY - changelog feat-ssp-entity-reconcile-and-write.md - tracker README A4 row → BE shipped + commit chain - design-A4.md §11 implementation reality reconciliation - handoff A4 SUMMARY（task arc 收口） - conversation-history dump（A4 task arc 全對話原文） A4 phase 完整收口；A5 phase 可在 reconcile 結果（matched_*_id / match_method 寫回 parsed_result）基礎上補預覽 UI + Confirm 拍板互動。 對齊 design-A4 §11 / §12 收口規範 / CLAUDE.md 做 summary 段。 Co-Authored-By: Claude Opus 4.7 (1M context) EOF )" ``` ### Task 11 Acceptance - [ ] Changelog frontmatter + body 完整（含 commit hash chain） - [ ] tracker README A4 row 標 BE shipped - [ ] design-A4.md §11 至少含 2 條條目，**不能空段** - [ ] SUMMARY 含 8 段 - [ ] 對話歷史 dump 完成（兩層目錄 `/ssp-import-export-phase2-A4/`） - [ ] BE working tree 乾淨（除 pre-existing pyproject.toml dev-path + zip） --- ## 規範遵守清單（執行時逐項勾） - [ ] 顯式 `git add `，禁 `-am` / `-A`（subagent dispatch prompt 也要明寫） - [ ] Commit 含 `Co-Authored-By: Claude Opus 4.7 (1M context)` footer - [ ] jedi-* 不動（除非 T0 verify 帶出必要 method 不存在，**需先 user 拍板**） - [ ] pyproject.toml dev-path 改動不 commit - [ ] DDD 嚴格分層 — Route 不碰 DB / App Service `@transaction` / Reconciler / WriteStrategy 純 domain 函式不接 session - [ ] BE 重啟必 `lsof -ti:8000 | xargs kill -9 && nohup poetry run python main_app.py > /dev/null 2>&1 &` - [ ] 改 BE service code（T4 / T5 / T6 / T7 / T8）後**必提醒 user 重啟** — BE 沒 hot reload - [ ] 不寫 docstring / 註解除非真有 non-obvious 的 why - **A4 例外**：`SspEntityReconciliationOrchestrator` facade docstring + `AssessmentObjectiveReconciler.reconcile_with_control_map` instance attr thread-safety note 屬「non-obvious why」，允許寫 - [ ] Test 用 mock 包 logger 避開 jedi DBLogHandler 撞 SessionLocal=None；`@transaction` 用 `patch_session_scope` - [ ] T0 verify 結果不符 design → 主動修 design-A4 §11 reconciliation，**不照舊硬幹** - [ ] 跨 repo 切換（T10）時提醒 user：cucumber 在 compliance-manager-test repo，commit / branch 分開 - [ ] **partial ship 允許**：cucumber env 未配齊時 scenarios 撰寫完成即可 - [ ] 重大決策 → `docs/analysis/YYYY-MM-DD-.md`（trade-off 取捨理由 — 若 implementer 出現意外取捨） - [ ] 重型 bug / 跨域問題 → `docs/issues/pending/YYYY-MM-DD-.md`（不擠在 changelog） - [ ] Handoff 文件統一放 `docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/`（CLAUDE.md 2026-05-20 新規） - [ ] 階段性 commit 不用問（每 Task 完成 commit 直接做） - [ ] Subagent dispatch（如需 parallel test 補齊）必加「git add 顯式檔名，禁用 -am」+「對齊既有 mock pattern」+「不動 pyproject.toml」三鐵律 --- ## 風險檢核點 | Task | 檢核點 | 處理 | |------|--------|------| | T0 | T0.1-T0.3 jedi-oscal SspEntity 欄位不符 design 假設 | 修 design §11 補偏差；T6 Entity 構造對齊實際 schema | | T0 | T0.4 CatalogControlAssessmentQueryEntity 無 objective_id 欄位 | AO Reconciler `_try_exact_match` 改用 `uid` 或其他自然 key；design §11 補 | | T0 | T0.5 / T0.6 control_implementation / ssp_control_impl_objective ORM model 欄位差異 | T7 WriteStrategy 簽章對齊；可能需動 jedi-oscal — user 拍板後動 | | T0 | T0.7 `ssp_domain_service.get_one_by_profile_uid` 不存在 | T6 `_resolve_existing_ssp_shell` 改用 query entity 路徑；或新增套件 method（需 user 拍板） | | T0 | T0.8 DeviceQueryEntity 不支援雙欄 (name, ip) AND filter | DeviceReconciler `_try_exact_match` 改 fallback：拉 name 結果 → Python filter ip | | T1 | base.py stage 0 dispatch 加入後 A3 既有 67 test fail | base.py `_try_user_selected_match` default no-op，A3 reconciler 不 override → 行為不變；A3 test 應全綠 | | T2 | DI Container build 時 catalog_control_domain_service / catalog_control_assessment_domain_service 跨 container reference 失敗 | 對齊既有 line 490/491 cross-container pattern；確認 oscal_container 自含或 reference 正確 | | T3 | reconciler `_try_exact_match` 內 Query Entity 屬性 set 後 repo 拉不到 | T0 verify Query Entity 欄位；可能需 Python 端 filter | | T3 | LeveragedReconciler fuzzy candidate cache 在 prod 大 tenant 慢 | 不在 A4 處理；F4 follow-up（jedi-oscal 套件加 `_in_*` batch lookup） | | T4 | A3 retroactive `_try_user_selected_match` 加入後 既有 16 person + 16 org test fail | A3 既有 ParsedParty fixture 不帶 `matched_user_label` → stage 0 return None → fallback 既有 stage 1；test 應全綠 | | T5 | A2 §15.3 ControlReconciler 沒拿到 fw_version.catalog_id | T0.1 verify FrameworkVersionEntity 有 catalog_id 欄位；T5 implementer 驗 | | T6 | superset flow SSP shell 建立後 ssp_id / sys_impl_main_id 拿不到 | Entity DTO 返回 id；T0 verify SystemSecurityPlanDomainService.add 簽章 | | T6 | update flow MF profile 對應 SSP 已存在但 system_implementation main 沒建（舊資料） | raise GRC_412066 → user 必先透過 ssp_versioning_service 建第 1 版 SSP | | T7 | ControlWriteStrategy upsert by (ssp_id, catalog_control_id) — jedi-oscal repo 可能無此 query 方法 | 用 `get_one(QueryEntity)` 自行寫；若不支援 query → 直接 add（接受 duplicate 風險低，update flow 才會撞） | | T7 | AoWriteStrategy 拿不到 `_ssp_control_implementation_id`（caller side attribute） | T8 step 6 內把 ctrl_impl_id_map 套上 AO 後再 call AoWriteStrategy | | T8 | `_write_all_data` 8-step 全長導致 method body 過大 | 沒問題 — 8 step 各有明確職責；A4 不拆 service（F-A4-refactor follow-up） | | T8 | `_resolve_catalog_id` update flow 從 `mf.profile.catalog_id` 拿不到 | T0 verify ProfileDomainService 含 get_by_uid + catalog_id；implementer 補完整 | | T9 | 累計 test count < 110：corner case 漏 | 對齊 design §8 acceptance 第 16 條，逐項補；可 dispatch subagent 平行補 | | T9 | `asdict(parsed)` 序列化 MatchMethod StrEnum 給 A5 拿到 `"MatchMethod.UNMATCHED"` | T9.4 verify 對齊 A3 T5.2 pattern（`json.dumps(d, default=str)`）| | T10 | env 未配齊跑不過 | 允許 partial ship；列 follow-up F-A4-cucumber-extra | | T11 | §11 reconciliation 段沒實作偏差可寫 → 段落空 | 至少 2 條（jedi-oscal Entity 欄位 + 1 條從 T0 帶出的差異）；mirror SSP Update Diff §11 樣板 | --- ## 跨 repo 工作清單 | Repo | 工作 | Branch | Commit 預估 | |------|------|--------|-------------| | compliance-manager-be（主）| T0 + T1 + T2 + T3 + T4 + T5 + T6 + T7 + T8 + T9 + T11 | `feature/ssp-import-export-phase2` | 11 commits | | compliance-manager-test | T10 cucumber regression | `feature/a4-ssp-excel-entity-match-regression` | 1 commit | | compliance-manager-fe | **不動** | — | 0 commit | | jedi-* 套件 | **不動**（除非 T0 verify 帶出必要 method 不存在）| — | 0 commit | --- ## 不在 A4 範圍（明確排除） - ❌ docx flow 補 device / info_system / control / AO 解析 + 5 reconciler 串到 docx — D5 拍板 follow-up F1 - ❌ A5 預覽 UI fuzzy 拍板路徑（A4 期間 fuzzy 自動寫入）— follow-up F2 - ❌ jedi-oscal `CatalogControlQueryEntity._in_control_id` batch lookup — follow-up F3 - ❌ `jedi_information_system._in_id` batch lookup — follow-up F4 - ❌ 5 WriteStrategy 抽象到 jedi-oscal — follow-up F5 - ❌ `SspExcelImportAppService` 拆 service — follow-up F-A4-refactor - ❌ Cucumber 加 user_selected / unmatched 樣本 scenario — follow-up F-A4-cucumber-extra - ❌ MatchMethod 細分 FUZZY_HOSTNAME / FUZZY_IP / FUZZY_VENDOR — D8 拍板不加 --- ## 換 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 字母>-to-<下一 session>.md` 4. Session F 是 task arc 收口 → 對話歷史按日期 dump 到 `docs/conversation-history/<date>/ssp-import-export-phase2-A4/`（**不做格式化**） | 換 Session 時點 | 必產出 | |----------------|--------| | B → C | handoff prompt（plan ship；T0/T1/T2 開工指引） | | C → D | handoff prompt（T1+T2 已 ship；T3/T4/T5 開工指引）| | D → E | handoff prompt（T3+T4+T5 已 ship；T6/T7/T8 開工指引 + T0.1-T0.3 SspEntity verify 結果攜入） | | E → F | handoff prompt（T6+T7+T8 已 ship；T9/T10/T11 開工指引）| | F（收口） | final SUMMARY + design §11 reconciliation + conversation-history dump | --- ## 工具預期 - TDD：每個 task 都先寫 failing test 再實作（superpowers:test-driven-development） - 子 task 完成階段性 commit（不用停下問 user） - T10 跨 repo 切換時切勿汙染 BE working tree - T11 SUMMARY 階段：mirror A3 SUMMARY 8 段 / SSP Update Diff 6 段體例 - Subagent dispatch（如需 parallel test 補齊）：明寫「git add 顯式檔名 禁用 -am」+「對齊既有 mock pattern」+「不動 pyproject.toml」三鐵律 - 整段 A4 預估 5 session 5d（C / D / E / F）— 每 session 結束產 handoff 給下一 session ---