Phase A4 Implementation Plan — 5 entity 鉤稽 + 寫入 + SSP shell + A3 retroactive

grep -nA 30 "^class SystemSecurityPlanEntity" \
  /Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/entity/ssp/ssp_entity.py

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

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

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

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

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

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

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

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"

poetry run python3 -c "from domain.oscal.service.reconciliation.match_method import MatchMethod; print(MatchMethod.USER_SELECTED)"

poetry run pytest tests/test_a3_reconciliation_normalizers.py -v

# 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 陳", "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) == ""

poetry run pytest tests/test_a4_reconciliation_helpers.py -v

# 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 <login_name>' → ('nickname', 'login_name')；不符 fmt 回 (None, None)."""
    if not label or '<' not in label or '>' not in label:
        return (None, None)
    nickname = label.split('<', 1)[0].strip()
    login_name = label.split('<', 1)[1].rstrip('>').strip()
    return (nickname or None, login_name or None)


def parse_device_lookup_label(label: Optional[str]) -> tuple[Optional[str], Optional[str]]:
    """'name (ip)' → ('name', 'ip')；不符 fmt 回 (label, None)."""
    if not label:
        return (None, None)
    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

poetry run pytest tests/test_a4_reconciliation_helpers.py -v

# 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

# 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

poetry run pytest tests/test_a3_reconciliation_base.py -v

# 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

# 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

poetry run pytest tests/test_a4_parsed_entity_fixtures.py -v

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) <noreply@anthropic.com>
EOF
)"

# 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

# 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

# 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 反查

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

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

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

# 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}

# 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)

# 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,
)

poetry run python3 -c "from di_containers.containers import Containers; c = Containers(); print(c.oscal_container.ssp_entity_reconciliation_orchestrator)"

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) <noreply@anthropic.com>
EOF
)"

# 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 後找到."""
        # ...

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

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

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

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

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

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) <noreply@anthropic.com>
EOF
)"

# 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 陳 <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."""
        # ...

# 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

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

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 <alice>", "org_unit": "Engineering"}]
        result = SspExcelImportAppService._dict_to_parsed_parties([], persons)
        assert result[0].matched_user_label == "Alice <alice>"
        assert result[0].matched_org_unit_label == "Engineering"

@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

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 <login_name>' → 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) <noreply@anthropic.com>
EOF
)"

# 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]
        # ...

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

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) <noreply@anthropic.com>
EOF
)"

class GrcErrorCode(BaseCode):
    # ... 既有
    GRC_EXCEL_UPDATE_FLOW_NO_SSP = ("Excel update flow 失敗：對應 MF 尚未建立 SSP", "GRC_412066")

poetry run python3 -c "from common.code.grc_error_code import GrcErrorCode; print(GrcErrorCode.GRC_EXCEL_UPDATE_FLOW_NO_SSP)"

# 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):
        # ...

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)

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
        # ...

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)

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) <noreply@anthropic.com>
EOF
)"

mkdir -p domain/oscal/service/write_strategy
touch domain/oscal/service/write_strategy/__init__.py

# 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])."""
        # ...

# 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)

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

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

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

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

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,
)

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) <noreply@anthropic.com>
EOF
)"