SSP OSCAL Alignment — Phase 2 Implementation Plan

# 列 ssp_excel_import_app_service 內所有 parsed_result.get("...") key
grep -E 'parsed_result\.get\(' app/oscal/service/ssp_excel_import_app_service.py | sort -u
# 列 parser layer 寫進去的 key
grep -rE 'result\["[^"]+"\]\s*=' app/oscal/service/excel_parser/ | head -50

grep -E 'parsed_result\.get\(' app/oscal/service/ssp_docx_import_app_service.py | sort -u

# Phase 2 — parsed_result Shape Reference

## v1-ssp (Legacy) Shape — pre Phase 2 deploy in-flight jobs
{ "devices": [...], "info_systems": [...], "leveraged": [...],
  "controls_with_aos": [...], "parties_org": [...], "parties_person": [...],
  "metadata": {...}, "system_characteristic": {...} }

每個 row 欄位 schema：（依 source code 列）
- devices[i]: {...}
- info_systems[i]: {...}
- leveraged[i]: {...}

## v2-bundle (New) Shape — Phase 2 onward
{ "schema_version": "v2-bundle",
  "components": [...], "leveraged_authorizations": [...], "inventory_items": [...],
  "controls_with_aos": [...], "parties": [...],
  "metadata": {...}, "system_characteristic": {...},
  "import_warnings": [...] }

每個 row 欄位 schema：（對齊 ParsedComponent / ParsedLeveragedAuthorization / ParsedInventoryItem）

## v1 → v2 轉換規則（bundle_restore 用）
- v1 `devices[i]` → v2 `components[j]` with `component_type='hardware'`
- v1 `info_systems[i]` → v2 `components[j]` with `component_type='system'`
- v1 `leveraged[i]` → v2 `components[j]` with `component_type='service'` + v2 `leveraged_authorizations[k]`
  paired by title (leveraged_authorization_ref = la.title)
- v1 `parties_org` + `parties_person` → v2 `parties` (single list)

git add docs/features/FR-028-2605-ssp-oscal-alignment/phase2-parsed-result-shape.md
git commit -m "docs(ssp-oscal-alignment): Phase 2 Task 1 — parsed_result shape reference"

{ "ref": "Crowdstrike",
  "reason": "leveraged_authorization not found in bundle",
  "context": { "component_title": "AcmeCorp Backup Service" } }

# tests/test_import_warnings.py
import pytest
from domain.oscal.import_pipeline.warnings import ImportWarning, ImportWarnings


def test_import_warning_dataclass_minimal_fields():
    w = ImportWarning(
        ref="Crowdstrike",
        reason="leveraged_authorization not found in bundle",
        context={"component_title": "AcmeCorp Backup Service"},
    )
    assert w.ref == "Crowdstrike"
    assert w.reason == "leveraged_authorization not found in bundle"
    assert w.context["component_title"] == "AcmeCorp Backup Service"


def test_collector_add_then_to_dict_list_yields_jsonb_safe():
    c = ImportWarnings()
    c.add(ImportWarning(ref="A", reason="r1", context={"k": "v"}))
    c.add(ImportWarning(ref="B", reason="r2", context={}))
    out = c.to_dict_list()
    assert isinstance(out, list)
    assert out == [
        {"ref": "A", "reason": "r1", "context": {"k": "v"}},
        {"ref": "B", "reason": "r2", "context": {}},
    ]


def test_collector_extend_accepts_multiple():
    c = ImportWarnings()
    c.extend([
        ImportWarning(ref="X", reason="x", context={}),
        ImportWarning(ref="Y", reason="y", context={}),
    ])
    assert len(c.to_dict_list()) == 2


def test_collector_is_empty_when_no_add():
    c = ImportWarnings()
    assert c.to_dict_list() == []
    assert len(c) == 0


def test_collector_supports_len():
    c = ImportWarnings()
    c.add(ImportWarning(ref="A", reason="r", context={}))
    assert len(c) == 1

pytest tests/test_import_warnings.py -v
# Expected: FAIL (module not exist)

# domain/oscal/import_pipeline/warnings.py
"""Import warnings collector — shared by Excel + Docx import pipelines.

Stored into parse_jobs.parsed_result.import_warnings (JSONB).
Per design.md §2.3.1 (FK Resolve — Warn but Allow) + §2.3.2 (title conflict).
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class ImportWarning:
    """One warning entry — JSONB-serializable to parsed_result.import_warnings."""

    ref: str
    reason: str
    context: Dict[str, Any] = field(default_factory=dict)
    source_table: Optional[str] = None  # e.g. "components" / "inventory_items"

    def to_dict(self) -> Dict[str, Any]:
        out = {"ref": self.ref, "reason": self.reason, "context": self.context}
        if self.source_table is not None:
            out["source_table"] = self.source_table
        return out


class ImportWarnings:
    """Collector — adapter / normalizer 共用累積 warnings。"""

    def __init__(self) -> None:
        self._items: List[ImportWarning] = []

    def add(self, warning: ImportWarning) -> None:
        self._items.append(warning)

    def extend(self, warnings: List[ImportWarning]) -> None:
        self._items.extend(warnings)

    def to_dict_list(self) -> List[Dict[str, Any]]:
        return [w.to_dict() for w in self._items]

    def __len__(self) -> int:
        return len(self._items)

pytest tests/test_import_warnings.py -v
# Expected: 5 passed

git add domain/oscal/import_pipeline/__init__.py \
        domain/oscal/import_pipeline/warnings.py \
        tests/test_import_warnings.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 2 — ImportWarnings collector module"

# tests/test_parsed_bundle_normalizer.py
import pytest

from domain.oscal.import_pipeline.normalizer import ParsedBundleNormalizer
from domain.oscal.import_pipeline.warnings import ImportWarnings
from domain.oscal.parser.ssp_intermediate import (
    ParsedComponent,
    ParsedLeveragedAuthorization,
    ParsedInventoryItem,
    ParsedExcelEntityBundle,
)


@pytest.fixture
def normalizer():
    return ParsedBundleNormalizer()


def test_component_dedup_emits_warning(normalizer):
    bundle = ParsedExcelEntityBundle(parsed_components=[
        ParsedComponent(title="A", component_type="service"),
        ParsedComponent(title="A", component_type="service"),
    ])
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    assert len(cleaned.parsed_components) == 1
    assert len(warnings) == 1
    assert "duplicate component" in warnings.to_dict_list()[0]["reason"]


def test_la_dedup_emits_warning(normalizer):
    bundle = ParsedExcelEntityBundle(parsed_leveraged_authorizations=[
        ParsedLeveragedAuthorization(title="Crowdstrike"),
        ParsedLeveragedAuthorization(title="Crowdstrike"),
    ])
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    assert len(cleaned.parsed_leveraged_authorizations) == 1
    assert any("duplicate leveraged_authorization" in w["reason"] for w in warnings.to_dict_list())


def test_inventory_dedup_by_description(normalizer):
    bundle = ParsedExcelEntityBundle(parsed_inventory_items=[
        ParsedInventoryItem(description="Server A"),
        ParsedInventoryItem(description="Server A"),
    ])
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    assert len(cleaned.parsed_inventory_items) == 1
    assert any("duplicate inventory_item" in w["reason"] for w in warnings.to_dict_list())


def test_component_la_ref_not_found_clears_and_warns(normalizer):
    bundle = ParsedExcelEntityBundle(
        parsed_components=[
            ParsedComponent(
                title="AcmeCorp Backup",
                component_type="service",
                leveraged_authorization_ref="Crowdstrike",  # 不在 bundle.parsed_leveraged_authorizations
            ),
        ],
        parsed_leveraged_authorizations=[],
    )
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    assert cleaned.parsed_components[0].leveraged_authorization_ref is None
    assert len(warnings) == 1
    w = warnings.to_dict_list()[0]
    assert w["ref"] == "Crowdstrike"
    assert "leveraged_authorization not found" in w["reason"]
    assert w["context"]["component_title"] == "AcmeCorp Backup"


def test_component_la_ref_found_keeps_string(normalizer):
    bundle = ParsedExcelEntityBundle(
        parsed_components=[
            ParsedComponent(
                title="AcmeCorp Backup",
                component_type="service",
                leveraged_authorization_ref="Crowdstrike",
            ),
        ],
        parsed_leveraged_authorizations=[
            ParsedLeveragedAuthorization(title="Crowdstrike"),
        ],
    )
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    # ref 維持 str（normalizer 不解 uid，write strategy 自己解）
    assert cleaned.parsed_components[0].leveraged_authorization_ref == "Crowdstrike"
    assert len(warnings) == 0


def test_inventory_implemented_ref_not_found_drops_ref(normalizer):
    bundle = ParsedExcelEntityBundle(
        parsed_components=[ParsedComponent(title="Comp A", component_type="hardware")],
        parsed_inventory_items=[
            ParsedInventoryItem(
                description="Server X",
                implemented_component_refs=["Comp A", "Comp Missing"],
            ),
        ],
    )
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    assert cleaned.parsed_inventory_items[0].implemented_component_refs == ["Comp A"]
    assert any("component_ref not found" in w["reason"] for w in warnings.to_dict_list())


def test_inventory_ambiguous_component_picks_service_first(normalizer):
    bundle = ParsedExcelEntityBundle(
        parsed_components=[
            ParsedComponent(title="Crowdstrike", component_type="hardware"),
            ParsedComponent(title="Crowdstrike", component_type="service"),
        ],
        parsed_inventory_items=[
            ParsedInventoryItem(
                description="Server X",
                implemented_component_refs=["Crowdstrike"],
            ),
        ],
    )
    warnings = ImportWarnings()
    cleaned = normalizer.normalize(bundle, warnings)
    # ref 仍維持 str — 解析只 emit warning 標 ambiguous，不重寫 ref
    assert cleaned.parsed_inventory_items[0].implemented_component_refs == ["Crowdstrike"]
    assert any("ambiguous component title" in w["reason"] for w in warnings.to_dict_list())

pytest tests/test_parsed_bundle_normalizer.py -v
# Expected: FAIL (module not exist)

# domain/oscal/import_pipeline/normalizer.py
"""ParsedBundleNormalizer — dedup + ref-resolve warnings emission.

Per design.md §2.3.1 (FK Resolve — Warn but Allow) + §2.3.2 (title conflict).

Pure in-memory transform: adapter raw bundle → clean bundle. No DB / no DI.
WriteStrategy 之後仍負責把 `_ref` str 解成 uid（§11.6 type discipline）。
"""
from __future__ import annotations

from typing import List, Tuple

from domain.oscal.import_pipeline.warnings import ImportWarning, ImportWarnings
from domain.oscal.parser.ssp_intermediate import (
    ParsedComponent,
    ParsedExcelEntityBundle,
    ParsedInventoryItem,
    ParsedLeveragedAuthorization,
)


class ParsedBundleNormalizer:
    """Idempotent: 同一 bundle 跑兩次 normalize 結果相同（dedup 已收斂）。"""

    def normalize(
        self,
        bundle: ParsedExcelEntityBundle,
        warnings: ImportWarnings,
    ) -> ParsedExcelEntityBundle:
        components = self._dedup_components(bundle.parsed_components, warnings)
        las = self._dedup_las(bundle.parsed_leveraged_authorizations, warnings)
        inventory = self._dedup_inventory(bundle.parsed_inventory_items, warnings)

        self._resolve_component_la_refs(components, las, warnings)
        self._resolve_inventory_component_refs(inventory, components, warnings)

        # 回傳新 bundle (不 mutate 入參) — 其他欄位原樣帶過
        return ParsedExcelEntityBundle(
            parsed_system_characteristic=bundle.parsed_system_characteristic,
            parsed_parties=bundle.parsed_parties,
            parsed_components=components,
            parsed_leveraged_authorizations=las,
            parsed_inventory_items=inventory,
            parsed_controls=bundle.parsed_controls,
            parsed_metadata=bundle.parsed_metadata,
        )

    # ------------------------------------------------------------------
    # Dedup helpers
    # ------------------------------------------------------------------
    @staticmethod
    def _dedup_components(
        items: List[ParsedComponent], warnings: ImportWarnings,
    ) -> List[ParsedComponent]:
        seen: dict[Tuple[str, str], ParsedComponent] = {}
        out: List[ParsedComponent] = []
        for c in items:
            key = (c.component_type, c.title)
            if key in seen:
                warnings.add(ImportWarning(
                    ref=c.title,
                    reason="duplicate component skipped",
                    context={"component_type": c.component_type},
                    source_table="components",
                ))
                continue
            seen[key] = c
            out.append(c)
        return out

    @staticmethod
    def _dedup_las(
        items: List[ParsedLeveragedAuthorization], warnings: ImportWarnings,
    ) -> List[ParsedLeveragedAuthorization]:
        seen: dict[str, ParsedLeveragedAuthorization] = {}
        out: List[ParsedLeveragedAuthorization] = []
        for la in items:
            if la.title in seen:
                warnings.add(ImportWarning(
                    ref=la.title,
                    reason="duplicate leveraged_authorization skipped",
                    context={},
                    source_table="leveraged_authorizations",
                ))
                continue
            seen[la.title] = la
            out.append(la)
        return out

    @staticmethod
    def _dedup_inventory(
        items: List[ParsedInventoryItem], warnings: ImportWarnings,
    ) -> List[ParsedInventoryItem]:
        seen: dict[str, ParsedInventoryItem] = {}
        out: List[ParsedInventoryItem] = []
        for inv in items:
            if inv.description in seen:
                warnings.add(ImportWarning(
                    ref=inv.description,
                    reason="duplicate inventory_item skipped",
                    context={},
                    source_table="inventory_items",
                ))
                continue
            seen[inv.description] = inv
            out.append(inv)
        return out

    # ------------------------------------------------------------------
    # Ref-resolve helpers — 不轉成 uid（write strategy 之後做），只負責 warning
    # ------------------------------------------------------------------
    @staticmethod
    def _resolve_component_la_refs(
        components: List[ParsedComponent],
        las: List[ParsedLeveragedAuthorization],
        warnings: ImportWarnings,
    ) -> None:
        la_titles = {la.title for la in las}
        for c in components:
            ref = c.leveraged_authorization_ref
            if ref is None:
                continue
            if ref not in la_titles:
                warnings.add(ImportWarning(
                    ref=ref,
                    reason="leveraged_authorization not found in bundle",
                    context={"component_title": c.title},
                    source_table="components",
                ))
                c.leveraged_authorization_ref = None  # clear so write strategy 寫 NULL

    @staticmethod
    def _resolve_inventory_component_refs(
        inventory: List[ParsedInventoryItem],
        components: List[ParsedComponent],
        warnings: ImportWarnings,
    ) -> None:
        # title → list[component_type]
        title_index: dict[str, list[str]] = {}
        for c in components:
            title_index.setdefault(c.title, []).append(c.component_type)

        for inv in inventory:
            keep_refs: list[str] = []
            for ref in inv.implemented_component_refs:
                if ref not in title_index:
                    warnings.add(ImportWarning(
                        ref=ref,
                        reason="component_ref not found in bundle",
                        context={"inventory_description": inv.description},
                        source_table="inventory_items",
                    ))
                    continue
                types = title_index[ref]
                if len(types) > 1:
                    warnings.add(ImportWarning(
                        ref=ref,
                        reason="ambiguous component title, picked first service",
                        context={
                            "inventory_description": inv.description,
                            "candidate_types": types,
                        },
                        source_table="inventory_items",
                    ))
                keep_refs.append(ref)
            inv.implemented_component_refs = keep_refs

pytest tests/test_parsed_bundle_normalizer.py -v
# Expected: 7 passed

git add domain/oscal/import_pipeline/normalizer.py tests/test_parsed_bundle_normalizer.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 3 — ParsedBundleNormalizer + warnings emission"

def confirm(
    self,
    parsed_result: dict,
    *,
    ssp_id: int,
    catalog_id: Optional[int],
    user_context,
) -> dict:
    """Return: {'written_components': N, 'written_las': N, 'written_inventory': N, 'warnings_added': N}"""

# tests/test_bundle_restore.py
import pytest
from domain.oscal.import_pipeline.bundle_restore import (
    dict_to_bundle, LegacyConfirmRequired,
)
from domain.oscal.parser.ssp_intermediate import ParsedExcelEntityBundle


def test_v2_bundle_dict_restores_to_bundle():
    parsed_result = {
        "schema_version": "v2-bundle",
        "components": [
            {"title": "Comp A", "component_type": "service"},
        ],
        "leveraged_authorizations": [
            {"title": "Crowdstrike"},
        ],
        "inventory_items": [
            {"description": "Server X", "implemented_component_refs": ["Comp A"]},
        ],
    }
    bundle = dict_to_bundle(parsed_result)
    assert isinstance(bundle, ParsedExcelEntityBundle)
    assert bundle.parsed_components[0].title == "Comp A"
    assert bundle.parsed_leveraged_authorizations[0].title == "Crowdstrike"
    assert bundle.parsed_inventory_items[0].description == "Server X"


def test_v1_ssp_dict_raises_legacy_required():
    parsed_result = {
        "schema_version": "v1-ssp",
        "devices": [{"name": "D1"}],
    }
    with pytest.raises(LegacyConfirmRequired):
        dict_to_bundle(parsed_result)


def test_missing_schema_version_defaults_to_legacy():
    # in-flight job (deploy 前已存) — 視為 v1-ssp
    parsed_result = {"devices": []}
    with pytest.raises(LegacyConfirmRequired):
        dict_to_bundle(parsed_result)


def test_v2_bundle_optional_keys_empty():
    parsed_result = {"schema_version": "v2-bundle"}
    bundle = dict_to_bundle(parsed_result)
    assert bundle.parsed_components == []
    assert bundle.parsed_leveraged_authorizations == []
    assert bundle.parsed_inventory_items == []

# domain/oscal/import_pipeline/bundle_restore.py
"""dict ↔ ParsedExcelEntityBundle conversion — Phase 2 confirm path.

Only supports v2-bundle shape; v1-ssp dispatches via LegacyConfirmRequired
so caller can drop to legacy confirm path.

Conversion rule reference: docs/features/FR-028-2605-ssp-oscal-alignment/phase2-parsed-result-shape.md
"""
from __future__ import annotations

from datetime import date
from typing import Any, Dict, List, Optional

from domain.oscal.parser.ssp_intermediate import (
    ParsedComponent,
    ParsedExcelEntityBundle,
    ParsedInventoryItem,
    ParsedLeveragedAuthorization,
    ParsedParty,
    ParsedSystemCharacteristic,
)


class LegacyConfirmRequired(Exception):
    """Raised by dict_to_bundle when parsed_result is v1-ssp shape.

    Caller (app service) catches this and dispatches to legacy confirm path.
    """


def dict_to_bundle(parsed_result: Dict[str, Any]) -> ParsedExcelEntityBundle:
    schema_version = parsed_result.get("schema_version", "v1-ssp")
    if schema_version != "v2-bundle":
        raise LegacyConfirmRequired(
            f"parsed_result schema_version={schema_version!r}, legacy path required"
        )

    return ParsedExcelEntityBundle(
        parsed_system_characteristic=_restore_system_characteristic(
            parsed_result.get("system_characteristic")
        ),
        parsed_parties=[_restore_party(p) for p in parsed_result.get("parties", [])],
        parsed_components=[_restore_component(c) for c in parsed_result.get("components", [])],
        parsed_leveraged_authorizations=[
            _restore_la(la) for la in parsed_result.get("leveraged_authorizations", [])
        ],
        parsed_inventory_items=[
            _restore_inventory(i) for i in parsed_result.get("inventory_items", [])
        ],
        parsed_controls=[],  # Phase 2 不動 control path — 沿用既有 controls_with_aos 流程
        parsed_metadata=parsed_result.get("metadata"),
    )


def _restore_component(d: Dict[str, Any]) -> ParsedComponent:
    return ParsedComponent(
        title=d["title"],
        component_type=d["component_type"],
        description=d.get("description"),
        purpose=d.get("purpose"),
        status=d.get("status"),
        leveraged_authorization_ref=d.get("leveraged_authorization_ref"),
        protocol=d.get("protocol"),
        port_ranges=d.get("port_ranges"),
        security_auth=d.get("security_auth"),
        matched_party_uuid=d.get("matched_party_uuid"),
    )


def _restore_la(d: Dict[str, Any]) -> ParsedLeveragedAuthorization:
    date_auth = d.get("date_authorized")
    if isinstance(date_auth, str):
        try:
            date_auth = date.fromisoformat(date_auth)
        except ValueError:
            date_auth = None
    return ParsedLeveragedAuthorization(
        title=d["title"],
        provider=d.get("provider"),
        date_authorized=date_auth,
        fedramp_package_id=d.get("fedramp_package_id"),
        impact_level=d.get("impact_level"),
        data_types=d.get("data_types"),
        nature_of_agreement=d.get("nature_of_agreement"),
        authorized_users=d.get("authorized_users"),
        remarks=d.get("remarks"),
    )


def _restore_inventory(d: Dict[str, Any]) -> ParsedInventoryItem:
    return ParsedInventoryItem(
        description=d["description"],
        asset_id=d.get("asset_id"),
        asset_tag=d.get("asset_tag"),
        ipv4_address=d.get("ipv4_address"),
        mac_address=d.get("mac_address"),
        fqdn=d.get("fqdn"),
        hostname=d.get("hostname"),
        software_name=d.get("software_name"),
        os_name=d.get("os_name"),
        implemented_component_refs=list(d.get("implemented_component_refs") or []),
    )


def _restore_party(d: Dict[str, Any]) -> Optional[ParsedParty]:
    # 既有 _dict_to_parsed_parties 邏輯不變 — 此處 v2-bundle 用 single list
    return ParsedParty(
        name=d.get("name") or "(未命名)",
        party_type=d.get("party_type", "person"),
        role=d.get("role"),
        title=d.get("title"),
        email_address=d.get("email_address"),
        telephone_number=d.get("telephone_number"),
        address=d.get("address"),
        short_name=d.get("short_name"),
        matched_user_id=d.get("matched_user_id"),
        matched_org_unit_id=d.get("matched_org_unit_id"),
    )


def _restore_system_characteristic(d: Optional[Dict[str, Any]]) -> Optional[ParsedSystemCharacteristic]:
    if not d:
        return None
    return ParsedSystemCharacteristic(
        system_name=d.get("system_name") or "",
        security_sensitivity_level=d.get("security_sensitivity_level"),
        status=d.get("status"),
        target_type=d.get("target_type"),
        scope_description=d.get("scope_description"),
        owner_login_name=d.get("owner_login_name"),
    )

pytest tests/test_bundle_restore.py -v
# Expected: 4 passed

# tests/test_ssp_import_confirm_service.py
from unittest.mock import Mock
import pytest

from domain.oscal.import_pipeline.confirm_service import SspImportConfirmService
from domain.oscal.import_pipeline.bundle_restore import LegacyConfirmRequired


@pytest.fixture
def mock_orchestrator():
    return Mock()


@pytest.fixture
def mock_leveraged_strategy():
    s = Mock(); s.write.return_value = 0
    return s


@pytest.fixture
def mock_component_strategy():
    s = Mock(); s.write.return_value = 0
    return s


@pytest.fixture
def mock_inventory_strategy():
    s = Mock(); s.write.return_value = 0
    return s


@pytest.fixture
def confirm_service(mock_orchestrator, mock_leveraged_strategy,
                    mock_component_strategy, mock_inventory_strategy):
    return SspImportConfirmService(
        orchestrator=mock_orchestrator,
        leveraged_write_strategy=mock_leveraged_strategy,
        component_write_strategy=mock_component_strategy,
        inventory_write_strategy=mock_inventory_strategy,
        normalizer=None,  # default fallback in service ctor
    )


def test_legacy_path_raises(confirm_service):
    user_ctx = Mock(login_name="alice", tenant_id=1)
    with pytest.raises(LegacyConfirmRequired):
        confirm_service.confirm(
            parsed_result={"devices": []},  # 無 schema_version → v1-ssp
            ssp_id=10,
            catalog_id=5,
            user_context=user_ctx,
        )


def test_v2_bundle_calls_strategies_in_order(confirm_service, mock_orchestrator,
                                              mock_leveraged_strategy,
                                              mock_component_strategy,
                                              mock_inventory_strategy):
    user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99)
    mock_leveraged_strategy.write.return_value = 2
    mock_component_strategy.write.return_value = 5
    mock_inventory_strategy.write.return_value = 3
    parsed_result = {
        "schema_version": "v2-bundle",
        "components": [
            {"title": "A", "component_type": "hardware"},
            {"title": "B", "component_type": "service"},
        ],
        "leveraged_authorizations": [{"title": "Crowdstrike"}],
        "inventory_items": [
            {"description": "Srv1", "implemented_component_refs": ["A"]},
        ],
    }
    out = confirm_service.confirm(
        parsed_result=parsed_result,
        ssp_id=10,
        catalog_id=5,
        user_context=user_ctx,
    )
    # 順序：LA → Component → Inventory（per design.md §2.3）
    assert mock_leveraged_strategy.write.called
    assert mock_component_strategy.write.called
    assert mock_inventory_strategy.write.called
    assert out["written_las"] == 2
    assert out["written_components"] == 5
    assert out["written_inventory"] == 3


def test_v2_bundle_warnings_threaded_back(confirm_service):
    user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99)
    parsed_result = {
        "schema_version": "v2-bundle",
        "components": [
            {"title": "Comp", "component_type": "service",
             "leveraged_authorization_ref": "MissingLA"},
        ],
        "leveraged_authorizations": [],
    }
    out = confirm_service.confirm(
        parsed_result=parsed_result,
        ssp_id=10,
        catalog_id=5,
        user_context=user_ctx,
    )
    assert out["warnings_added"] >= 1
    # warnings 也寫回 parsed_result（caller 之後 persist 到 parse_jobs.parsed_result）
    assert any("leveraged_authorization not found" in w["reason"]
               for w in parsed_result["import_warnings"])

# domain/oscal/import_pipeline/confirm_service.py
"""SspImportConfirmService — shared Excel + Docx confirm entry point.

Caller (excel / docx app service) wraps @transaction; this service does
not open a session. schema_version dispatch:
  - "v2-bundle" → bundle_restore → normalize → orchestrator → strategies
  - "v1-ssp" or missing → raise LegacyConfirmRequired (caller handles legacy)
"""
from __future__ import annotations

from typing import Any, Dict, Optional

from domain.oscal.import_pipeline.bundle_restore import (
    LegacyConfirmRequired,
    dict_to_bundle,
)
from domain.oscal.import_pipeline.normalizer import ParsedBundleNormalizer
from domain.oscal.import_pipeline.warnings import ImportWarnings
from domain.oscal.parser.ssp_intermediate import SspEntityReconciliationContext


class SspImportConfirmService:
    def __init__(
        self,
        orchestrator,
        leveraged_write_strategy,
        component_write_strategy,
        inventory_write_strategy,
        normalizer: Optional[ParsedBundleNormalizer] = None,
    ):
        self._orchestrator = orchestrator
        self._leveraged = leveraged_write_strategy
        self._component = component_write_strategy
        self._inventory = inventory_write_strategy
        self._normalizer = normalizer or ParsedBundleNormalizer()

    def confirm(
        self,
        parsed_result: Dict[str, Any],
        *,
        ssp_id: int,
        catalog_id: Optional[int],
        user_context,
    ) -> Dict[str, int]:
        """Raises LegacyConfirmRequired for v1-ssp shape. Mutates parsed_result
        (adds import_warnings list)."""
        # Step 1: dict → bundle (raises LegacyConfirmRequired on v1)
        bundle = dict_to_bundle(parsed_result)

        # Step 2: normalize + collect warnings
        warnings = ImportWarnings()
        bundle = self._normalizer.normalize(bundle, warnings)

        # Step 3: reconcile (orchestrator) — optional dep
        if self._orchestrator is not None and catalog_id is not None:
            ctx = SspEntityReconciliationContext(
                tenant_id=user_context.tenant_id,
                catalog_id=catalog_id,
            )
            try:
                self._orchestrator.reconcile(bundle, ctx)
            except Exception:
                pass  # reconciler 失敗 fall through 寫純文字 — caller log

        # Step 4: write 順序：LA → Component → Inventory (FK 依賴)
        written_las = self._leveraged.write(
            bundle.parsed_leveraged_authorizations, ssp_id, user_context,
        ) if bundle.parsed_leveraged_authorizations else 0

        written_components = self._component.write(
            bundle.parsed_components, ssp_id, user_context,
        ) if bundle.parsed_components else 0

        written_inventory = self._inventory.write(
            bundle.parsed_inventory_items, ssp_id, user_context,
        ) if bundle.parsed_inventory_items else 0

        # Step 5: warnings 寫回 parsed_result，caller persist 到 JSONB
        parsed_result["import_warnings"] = warnings.to_dict_list()

        return {
            "written_las": written_las,
            "written_components": written_components,
            "written_inventory": written_inventory,
            "warnings_added": len(warnings),
        }

pytest tests/test_bundle_restore.py tests/test_ssp_import_confirm_service.py -v
# Expected: 7 passed

git add domain/oscal/import_pipeline/bundle_restore.py \
        domain/oscal/import_pipeline/confirm_service.py \
        tests/test_bundle_restore.py \
        tests/test_ssp_import_confirm_service.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 4 — SspImportConfirmService + bundle_restore"

# NEW signature
def write(
    self,
    parsed_las: Iterable[ParsedLeveragedAuthorization],
    ssp_id: int,
    user_context,
) -> int:

grep -n "LeveragedWriteStrategy\|test_leveraged" tests/test_ssp_write_strategy.py | head

# tests/test_ssp_write_strategy.py — 加段
def test_leveraged_write_strategy_consumes_parsed_la():
    from domain.oscal.service.write_strategy.leveraged_write_strategy import LeveragedWriteStrategy
    from domain.oscal.parser.ssp_intermediate import ParsedLeveragedAuthorization
    la_service = Mock()
    la_service.get_all.return_value = []
    saved = Mock(); saved.uid = uuid.uuid4()
    la_service.add.return_value = saved
    strat = LeveragedWriteStrategy(la_service)  # 不再 inject component_service
    user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99)
    out = strat.write(
        [ParsedLeveragedAuthorization(title="Crowdstrike", impact_level="moderate")],
        ssp_id=10,
        user_context=user_ctx,
    )
    assert out == 1
    la_entity = la_service.add.call_args[0][0]
    assert la_entity.title == "Crowdstrike"
    # impact_level / data_types 等都放 props JSONB
    assert la_entity.props.get("impact_level") == "moderate"

# domain/oscal/service/write_strategy/leveraged_write_strategy.py（重寫）
"""LeveragedWriteStrategy v3 — consumes ParsedLeveragedAuthorization.

Phase 2 Task 5A refactor — strategy 只寫 oscal.ssp_leveraged_authorizations，
linked service Component 由 ComponentWriteStrategy 走 leveraged_authorization_ref
字串 lookup 處理。Signature 移除 dead param sys_impl_main_id。

UNMATCHED: party_uuid stays None when no matched party.
Overwrite semantics: delete existing rows for ssp_id before insert.
"""
from __future__ import annotations

import logging
import uuid
from typing import Iterable, Optional

from domain.oscal.parser.ssp_intermediate import ParsedLeveragedAuthorization

logger = logging.getLogger(__name__)


def _strip_nulls(d: dict) -> Optional[dict]:
    cleaned = {k: v for k, v in d.items() if v is not None and v != ""}
    return cleaned or None


class LeveragedWriteStrategy:
    def __init__(self, leveraged_authorization_domain_service):
        self._la = leveraged_authorization_domain_service

    def write(
        self,
        parsed_las: Iterable[ParsedLeveragedAuthorization],
        ssp_id: int,
        user_context,
    ) -> int:
        from jedi_oscal.domain.entity.base.oscal_leveraged_authorization_entity import (
            LeveragedAuthorizationEntity,
        )
        from jedi_oscal.domain.entity.base.oscal_leveraged_authorization_query_entity import (
            LeveragedAuthorizationQueryEntity,
        )

        tenant_id = getattr(user_context, "tenant_id", None)
        org_unit_id = getattr(user_context, "org_unit_id", None)
        login_name = getattr(user_context, "login_name", None)
        if not login_name:
            raise ValueError("LeveragedWriteStrategy.write: user_context.login_name required")

        # Overwrite: delete existing LA for ssp_id
        existing = self._la.get_all(LeveragedAuthorizationQueryEntity(ssp_id=ssp_id)) or []
        to_delete = [e.id for e in existing if e.id is not None]
        if to_delete:
            self._la.delete_by_ids(to_delete)

        written = 0
        for parsed in parsed_las:
            props = _strip_nulls({
                "fedramp_package_id": parsed.fedramp_package_id,
                "impact_level": parsed.impact_level,
                "data_types": parsed.data_types,
                "nature_of_agreement": parsed.nature_of_agreement,
                "authorized_users": parsed.authorized_users,
            })

            party_uuid_val: uuid.UUID | None = None
            # provider 是 CSP/vendor name — reconciler 在 bundle 階段填 matched_party_uuid
            # 之後（v3 dataclass 還沒這欄，預留 future）。Phase 2 先用 None。

            entity = LeveragedAuthorizationEntity(
                ssp_id=ssp_id,
                title=parsed.title,
                party_uuid=party_uuid_val,
                date_authorized=parsed.date_authorized,
                props=props,
                remarks=parsed.remarks,
                tenant_id=tenant_id,
                org_unit_id=org_unit_id,
                created_user=login_name,
                updated_user=login_name,
            )
            try:
                self._la.add(entity)
                written += 1
            except Exception as e:
                logger.warning(
                    "LeveragedWriteStrategy: insert failed for ssp_id=%s title=%r: %s",
                    ssp_id, parsed.title, e, exc_info=True,
                )
        return written

pytest tests/test_ssp_write_strategy.py -v -k leveraged

git add domain/oscal/service/write_strategy/leveraged_write_strategy.py tests/test_ssp_write_strategy.py
git commit -m "refactor(ssp-oscal-alignment): Phase 2 Task 5A — LeveragedWriteStrategy v3 (consume ParsedLeveragedAuthorization, remove dead param)"

# tests/test_component_write_strategy.py
import uuid
from unittest.mock import Mock

from domain.oscal.parser.ssp_intermediate import ParsedComponent
from domain.oscal.service.write_strategy.component_write_strategy import (
    ComponentWriteStrategy,
)


def _ctx():
    return Mock(login_name="alice", tenant_id=1, org_unit_id=99)


def test_component_write_no_la_ref_writes_null_la_uid():
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    la_svc = Mock(); la_svc.get_all.return_value = []
    strat = ComponentWriteStrategy(comp_svc, la_svc)
    out = strat.write(
        [ParsedComponent(title="Comp", component_type="hardware")],
        ssp_id=10, user_context=_ctx(),
    )
    assert out == 1
    entity = comp_svc.add.call_args[0][0]
    assert entity.leveraged_authorization_uid is None


def test_component_write_la_ref_resolves_via_la_lookup():
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    la_uid = uuid.uuid4()
    la_row = Mock(uid=la_uid, title="Crowdstrike")
    la_svc = Mock(); la_svc.get_all.return_value = [la_row]
    strat = ComponentWriteStrategy(comp_svc, la_svc)
    out = strat.write(
        [ParsedComponent(title="C", component_type="service",
                         leveraged_authorization_ref="Crowdstrike")],
        ssp_id=10, user_context=_ctx(),
    )
    entity = comp_svc.add.call_args[0][0]
    assert entity.leveraged_authorization_uid == la_uid


def test_component_write_la_ref_unmatched_writes_null():
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    la_svc = Mock(); la_svc.get_all.return_value = []  # 沒對應 LA
    strat = ComponentWriteStrategy(comp_svc, la_svc)
    strat.write(
        [ParsedComponent(title="C", component_type="service",
                         leveraged_authorization_ref="Missing")],
        ssp_id=10, user_context=_ctx(),
    )
    entity = comp_svc.add.call_args[0][0]
    assert entity.leveraged_authorization_uid is None


def test_overwrite_deletes_existing_then_inserts():
    existing_a = Mock(id=1); existing_b = Mock(id=2)
    comp_svc = Mock()
    comp_svc.get_all.return_value = [existing_a, existing_b]
    la_svc = Mock(); la_svc.get_all.return_value = []
    strat = ComponentWriteStrategy(comp_svc, la_svc)
    strat.write([ParsedComponent(title="C", component_type="hardware")], 10, _ctx())
    comp_svc.delete_by_ids.assert_called_once_with([1, 2])
    assert comp_svc.add.called


def test_component_type_persisted():
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    la_svc = Mock(); la_svc.get_all.return_value = []
    strat = ComponentWriteStrategy(comp_svc, la_svc)
    strat.write(
        [ParsedComponent(title="X", component_type="system",
                         description="d", purpose="p", status="operational")],
        ssp_id=10, user_context=_ctx(),
    )
    e = comp_svc.add.call_args[0][0]
    assert e.component_type == "system"
    assert e.description == "d"
    assert e.purpose == "p"
    assert e.status == "operational"

# domain/oscal/service/write_strategy/component_write_strategy.py
"""ComponentWriteStrategy — Phase 2 Task 5B.

Single strategy that writes all ParsedComponent rows (any component_type) to
oscal.ssp_components. Supersedes DeviceWriteStrategy + InformationSystemWriteStrategy
(those file are removed in Task 11 cleanup).

leveraged_authorization_ref (title string) resolution: lookup
oscal.ssp_leveraged_authorizations where (ssp_id, title) match → use uid.
Caller (SspImportConfirmService) MUST run LeveragedWriteStrategy first
so the LA rows exist for this ssp_id.

Overwrite semantics: delete all existing components for ssp_id before insert.
"""
from __future__ import annotations

import logging
from typing import Iterable, Optional

from domain.oscal.parser.ssp_intermediate import ParsedComponent

logger = logging.getLogger(__name__)


def _strip_nulls(d: dict) -> Optional[dict]:
    cleaned = {k: v for k, v in d.items() if v is not None and v != ""}
    return cleaned or None


class ComponentWriteStrategy:
    def __init__(self, component_domain_service, leveraged_authorization_domain_service):
        self._comp = component_domain_service
        self._la = leveraged_authorization_domain_service

    def write(
        self,
        parsed_components: Iterable[ParsedComponent],
        ssp_id: int,
        user_context,
    ) -> int:
        from jedi_oscal.domain.entity.base.oscal_component_entity import ComponentEntity
        from jedi_oscal.domain.entity.base.oscal_component_query_entity import (
            ComponentQueryEntity,
        )
        from jedi_oscal.domain.entity.base.oscal_leveraged_authorization_query_entity import (
            LeveragedAuthorizationQueryEntity,
        )

        tenant_id = getattr(user_context, "tenant_id", None)
        org_unit_id = getattr(user_context, "org_unit_id", None)
        login_name = getattr(user_context, "login_name", None)
        if not login_name:
            raise ValueError("ComponentWriteStrategy.write: user_context.login_name required")

        # Overwrite: 砍 ssp_id 全部 components
        existing = self._comp.get_all(ComponentQueryEntity(ssp_id=ssp_id)) or []
        to_delete = [c.id for c in existing if c.id is not None]
        if to_delete:
            self._comp.delete_by_ids(to_delete)

        # 預載 LA title → uid map（一次 query 替整批解 ref）
        existing_las = self._la.get_all(
            LeveragedAuthorizationQueryEntity(ssp_id=ssp_id)
        ) or []
        la_title_to_uid = {la.title: la.uid for la in existing_las if la.title and la.uid}

        written = 0
        for parsed in parsed_components:
            la_uid = None
            if parsed.leveraged_authorization_ref:
                la_uid = la_title_to_uid.get(parsed.leveraged_authorization_ref)
                if la_uid is None:
                    logger.info(
                        "ComponentWriteStrategy: LA ref %r unmatched for ssp_id=%s — writing NULL",
                        parsed.leveraged_authorization_ref, ssp_id,
                    )

            props = _strip_nulls({
                "protocol": parsed.protocol,
                "port_ranges": parsed.port_ranges,
                "security_auth": parsed.security_auth,
            })

            entity = ComponentEntity(
                ssp_id=ssp_id,
                component_type=parsed.component_type,
                title=parsed.title,
                description=parsed.description,
                purpose=parsed.purpose,
                status=parsed.status,
                leveraged_authorization_uid=la_uid,
                props=props,
                tenant_id=tenant_id,
                org_unit_id=org_unit_id,
                created_user=login_name,
                updated_user=login_name,
            )
            try:
                self._comp.add(entity)
                written += 1
            except Exception as e:
                logger.warning(
                    "ComponentWriteStrategy: insert failed ssp_id=%s title=%r type=%s: %s",
                    ssp_id, parsed.title, parsed.component_type, e, exc_info=True,
                )
        return written

pytest tests/test_component_write_strategy.py -v
# Expected: 5 passed

git add domain/oscal/service/write_strategy/component_write_strategy.py \
        tests/test_component_write_strategy.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 5B — ComponentWriteStrategy v3 (consume ParsedComponent, resolve LA ref)"

# tests/test_inventory_item_write_strategy.py
import uuid
from unittest.mock import Mock

from domain.oscal.parser.ssp_intermediate import ParsedInventoryItem
from domain.oscal.service.write_strategy.inventory_item_write_strategy import (
    InventoryItemWriteStrategy,
)


def _ctx():
    return Mock(login_name="alice", tenant_id=1, org_unit_id=99)


def test_write_with_no_refs():
    inv_svc = Mock(); inv_svc.get_all.return_value = []
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    strat = InventoryItemWriteStrategy(inv_svc, comp_svc)
    out = strat.write([ParsedInventoryItem(description="Server X")], 10, _ctx())
    assert out == 1
    entity = inv_svc.add.call_args[0][0]
    assert entity.description == "Server X"
    assert entity.implemented_component_uids == []


def test_write_resolves_refs_via_component_lookup():
    inv_svc = Mock(); inv_svc.get_all.return_value = []
    comp_a_uid = uuid.uuid4()
    comp_a = Mock(uid=comp_a_uid, title="CompA", component_type="hardware")
    comp_svc = Mock(); comp_svc.get_all.return_value = [comp_a]
    strat = InventoryItemWriteStrategy(inv_svc, comp_svc)
    out = strat.write(
        [ParsedInventoryItem(description="Srv", implemented_component_refs=["CompA"])],
        ssp_id=10, user_context=_ctx(),
    )
    entity = inv_svc.add.call_args[0][0]
    assert entity.implemented_component_uids == [comp_a_uid]


def test_write_ambiguous_picks_service_type_first():
    inv_svc = Mock(); inv_svc.get_all.return_value = []
    hw_uid, svc_uid = uuid.uuid4(), uuid.uuid4()
    comp_hw = Mock(uid=hw_uid, title="X", component_type="hardware")
    comp_svc_row = Mock(uid=svc_uid, title="X", component_type="service")
    comp_svc = Mock(); comp_svc.get_all.return_value = [comp_hw, comp_svc_row]
    strat = InventoryItemWriteStrategy(inv_svc, comp_svc)
    strat.write(
        [ParsedInventoryItem(description="S", implemented_component_refs=["X"])],
        ssp_id=10, user_context=_ctx(),
    )
    entity = inv_svc.add.call_args[0][0]
    assert entity.implemented_component_uids == [svc_uid]


def test_write_unmatched_ref_drops():
    inv_svc = Mock(); inv_svc.get_all.return_value = []
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    strat = InventoryItemWriteStrategy(inv_svc, comp_svc)
    strat.write(
        [ParsedInventoryItem(description="S", implemented_component_refs=["NotExist"])],
        ssp_id=10, user_context=_ctx(),
    )
    entity = inv_svc.add.call_args[0][0]
    assert entity.implemented_component_uids == []


def test_overwrite_deletes_then_inserts():
    e1 = Mock(id=1)
    inv_svc = Mock(); inv_svc.get_all.return_value = [e1]
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    strat = InventoryItemWriteStrategy(inv_svc, comp_svc)
    strat.write([ParsedInventoryItem(description="S")], 10, _ctx())
    inv_svc.delete_by_ids.assert_called_once_with([1])
    assert inv_svc.add.called


def test_props_jsonb_built_from_optional_fields():
    inv_svc = Mock(); inv_svc.get_all.return_value = []
    comp_svc = Mock(); comp_svc.get_all.return_value = []
    strat = InventoryItemWriteStrategy(inv_svc, comp_svc)
    strat.write(
        [ParsedInventoryItem(description="S", asset_id="A1", ipv4_address="10.0.0.1")],
        10, _ctx(),
    )
    entity = inv_svc.add.call_args[0][0]
    # asset_id / ipv4 等放 props 還是 first-class 欄位，依 entity 設計而定
    # — 詳見 Phase 1 InventoryItemEntity；先檢查不為 None
    assert entity is not None

# domain/oscal/service/write_strategy/inventory_item_write_strategy.py
"""InventoryItemWriteStrategy — Phase 2 Task 5C.

Writes oscal.ssp_inventory_items + M2M oscal.ssp_inventory_implemented_components.
Component refs (title list) resolved via (ssp_id, title) lookup; ambiguous title
picks first component_type='service' (else first match).

Caller MUST run ComponentWriteStrategy first so component rows exist.

Overwrite semantics: delete all inventory rows for ssp_id; CASCADE drops M2M.
"""
from __future__ import annotations

import logging
from typing import Iterable, Optional

from domain.oscal.parser.ssp_intermediate import ParsedInventoryItem

logger = logging.getLogger(__name__)


def _strip_nulls(d: dict) -> Optional[dict]:
    cleaned = {k: v for k, v in d.items() if v is not None and v != ""}
    return cleaned or None


class InventoryItemWriteStrategy:
    def __init__(self, inventory_item_domain_service, component_domain_service):
        self._inv = inventory_item_domain_service
        self._comp = component_domain_service

    def write(
        self,
        parsed_inventory: Iterable[ParsedInventoryItem],
        ssp_id: int,
        user_context,
    ) -> int:
        from jedi_oscal.domain.entity.base.oscal_inventory_item_entity import (
            InventoryItemEntity,
        )
        from jedi_oscal.domain.entity.base.oscal_inventory_item_query_entity import (
            InventoryItemQueryEntity,
        )
        from jedi_oscal.domain.entity.base.oscal_component_query_entity import (
            ComponentQueryEntity,
        )

        tenant_id = getattr(user_context, "tenant_id", None)
        org_unit_id = getattr(user_context, "org_unit_id", None)
        login_name = getattr(user_context, "login_name", None)
        if not login_name:
            raise ValueError("InventoryItemWriteStrategy.write: user_context.login_name required")

        # Overwrite: 砍 ssp_id 全部 inventory (CASCADE 自動清 M2M)
        existing = self._inv.get_all(InventoryItemQueryEntity(ssp_id=ssp_id)) or []
        to_delete = [e.id for e in existing if e.id is not None]
        if to_delete:
            self._inv.delete_by_ids(to_delete)

        # 預載 (title → list[(uid, component_type)]) 解 ref ambiguity
        components = self._comp.get_all(ComponentQueryEntity(ssp_id=ssp_id)) or []
        title_to_candidates: dict[str, list[tuple]] = {}
        for c in components:
            if c.title and c.uid:
                title_to_candidates.setdefault(c.title, []).append((c.uid, c.component_type))

        written = 0
        for parsed in parsed_inventory:
            resolved_uids: list = []
            for ref in parsed.implemented_component_refs:
                candidates = title_to_candidates.get(ref, [])
                if not candidates:
                    continue  # normalizer 已 emit warning，這裡只 drop
                # 取 component_type='service' 第一筆，fallback 任意第一筆
                service_match = next(
                    ((uid, _) for (uid, t) in candidates if t == "service"), None
                )
                resolved_uids.append(service_match[0] if service_match else candidates[0][0])

            props = _strip_nulls({
                "asset_id": parsed.asset_id,
                "asset_tag": parsed.asset_tag,
                "ipv4_address": parsed.ipv4_address,
                "mac_address": parsed.mac_address,
                "fqdn": parsed.fqdn,
                "hostname": parsed.hostname,
                "software_name": parsed.software_name,
                "os_name": parsed.os_name,
            })

            entity = InventoryItemEntity(
                ssp_id=ssp_id,
                description=parsed.description,
                props=props,
                implemented_component_uids=resolved_uids,
                tenant_id=tenant_id,
                org_unit_id=org_unit_id,
                created_user=login_name,
                updated_user=login_name,
            )
            try:
                self._inv.add(entity)
                written += 1
            except Exception as e:
                logger.warning(
                    "InventoryItemWriteStrategy: insert failed ssp_id=%s desc=%r: %s",
                    ssp_id, parsed.description, e, exc_info=True,
                )
        return written

pytest tests/test_inventory_item_write_strategy.py -v

git add domain/oscal/service/write_strategy/inventory_item_write_strategy.py \
        tests/test_inventory_item_write_strategy.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 5C — InventoryItemWriteStrategy v3 (consume ParsedInventoryItem, M2M write)"

# domain/oscal/service/reconciliation/ssp_entity_orchestrator.py（refactor）
"""SSP entity orchestrator v3 — read bundle.parsed_components (filtered by type)
+ parsed_leveraged_authorizations + parsed_inventory_items.

Phase 2 Task 6 refactor: signature unchanged (reconcile(parsed, ctx)), but
internally reads v3 fields and filters components by type before dispatching
to device / information_system / leveraged reconcilers.

reconcilers' .reconcile() signature unchanged — caller now passes ParsedComponent
list filtered by type instead of ParsedDevice / ParsedInformationSystem lists.
"""

from domain.oscal.parser.ssp_intermediate import (
    ParsedExcelEntityBundle,
    SspEntityReconciliationContext,
)


class SspEntityReconciliationOrchestrator:
    def __init__(
        self,
        catalog_control_reconciler,
        assessment_objective_reconciler,
        device_reconciler,
        information_system_reconciler,
        leveraged_reconciler,
        system_characteristic_reconciler=None,
    ):
        self._ctrl = catalog_control_reconciler
        self._ao = assessment_objective_reconciler
        self._device = device_reconciler
        self._info_system = information_system_reconciler
        self._leveraged = leveraged_reconciler
        self._system_characteristic = system_characteristic_reconciler

    def reconcile(
        self,
        parsed: ParsedExcelEntityBundle,
        ctx: SspEntityReconciliationContext,
    ) -> None:
        # 1. controls
        self._ctrl.reconcile(parsed.parsed_controls, ctx.catalog_id)

        # 2. AO
        ctrl_map = {
            c.control_id: c.matched_catalog_control_id
            for c in parsed.parsed_controls
            if c.matched_catalog_control_id is not None
        }
        parsed_aos = [ao for c in parsed.parsed_controls for ao in c.objectives]
        self._ao.reconcile_with_control_map(parsed_aos, ctrl_map)

        # 3. components — filter by type then dispatch
        hardware_components = [
            c for c in parsed.parsed_components if c.component_type == "hardware"
        ]
        system_components = [
            c for c in parsed.parsed_components if c.component_type == "system"
        ]
        # device_reconciler now consumes List[ParsedComponent] (was ParsedDevice)
        self._device.reconcile(hardware_components, ctx.tenant_id)
        # information_system_reconciler now consumes List[ParsedComponent]
        self._info_system.reconcile(system_components, ctx.tenant_id)

        # 4. leveraged_authorizations (was: ParsedLeveraged list)
        self._leveraged.reconcile(parsed.parsed_leveraged_authorizations, ctx.tenant_id)

        # 5. system_characteristic
        if (
            self._system_characteristic is not None
            and parsed.parsed_system_characteristic is not None
        ):
            self._system_characteristic.reconcile_single(
                parsed.parsed_system_characteristic, ctx.tenant_id,
            )

# 看每個 reconciler 內 access 的 attr
grep -nE 'parsed\.\w+|p\.\w+' domain/oscal/service/reconciliation/device_reconciler.py
grep -nE 'parsed\.\w+|p\.\w+' domain/oscal/service/reconciliation/information_system_reconciler.py
grep -nE 'parsed\.\w+|p\.\w+' domain/oscal/service/reconciliation/leveraged_reconciler.py

pytest tests/test_a4_reconciliation_orchestrator.py -v
pytest tests/ -k "device_reconciler or information_system_reconciler or leveraged_reconciler" -v

git add domain/oscal/service/reconciliation/ tests/test_a4_reconciliation_orchestrator.py
git commit -m "refactor(ssp-oscal-alignment): Phase 2 Task 6 — SspEntityReconciliationOrchestrator + 3 reconcilers read v3 bundle (ParsedComponent / ParsedLeveragedAuthorization)"

# 在 _write_all_data 內，Step 3 起改：

schema_version = parsed_result.get("schema_version", "v1-ssp")

if schema_version == "v2-bundle":
    # 新 path
    if ssp_id is None or self._confirm_service is None:
        logger.warning("v2-bundle confirm skipped: ssp_id=%s confirm_service=%s",
                       ssp_id, self._confirm_service is not None)
        write_stats = {"warnings_added": 0, "written_components": 0,
                        "written_las": 0, "written_inventory": 0}
    else:
        write_stats = self._confirm_service.confirm(
            parsed_result=parsed_result,
            ssp_id=ssp_id,
            catalog_id=catalog_id,
            user_context=user_context,
        )
    # parsed_controls / parsed_sc 維持既有 path（control/AO/system_char write 不換）
    self._write_controls_and_sc(parsed_result, ssp_id, sys_impl_main_id, user_context, ...)
else:
    # legacy v1-ssp path — 既有 Step 3-6 邏輯保留（不經 orchestrator 拼 bundle）
    ...

# ssp_excel_import_app_service.py:76 __init__ 加參數
def __init__(
    self,
    ...,
    confirm_service=None,                 # ← 新加
    ...,
):
    ...
    self._confirm_service = confirm_service

# di_containers/oscal/oscal_container.py — 新增
from domain.oscal.import_pipeline.confirm_service import SspImportConfirmService
from domain.oscal.service.write_strategy.component_write_strategy import ComponentWriteStrategy
from domain.oscal.service.write_strategy.inventory_item_write_strategy import InventoryItemWriteStrategy

component_write_strategy = providers.Singleton(
    ComponentWriteStrategy,
    component_domain_service=oscal_container.component_domain_service,
    leveraged_authorization_domain_service=oscal_container.leveraged_authorization_domain_service,
)
inventory_item_write_strategy = providers.Singleton(
    InventoryItemWriteStrategy,
    inventory_item_domain_service=oscal_container.inventory_item_domain_service,
    component_domain_service=oscal_container.component_domain_service,
)
ssp_import_confirm_service = providers.Singleton(
    SspImportConfirmService,
    orchestrator=ssp_entity_orchestrator,
    leveraged_write_strategy=leveraged_write_strategy,
    component_write_strategy=component_write_strategy,
    inventory_write_strategy=inventory_item_write_strategy,
)

ssp_excel_import_app_service = providers.Singleton(
    SspExcelImportAppService,
    ...,
    confirm_service=ssp_import_confirm_service,
    ...,
)

pytest tests/test_ssp_excel_import_app_service.py tests/test_ssp_excel_import_ssp_source.py tests/test_ssp_excel_import_system_characteristic_e2e.py -v

git add app/oscal/service/ssp_excel_import_app_service.py \
        di_containers/oscal/oscal_container.py \
        tests/test_ssp_excel_import_app_service.py \
        tests/test_ssp_excel_import_ssp_source.py \
        tests/test_ssp_excel_import_system_characteristic_e2e.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 7 — Excel app service wire SspImportConfirmService (v2-bundle path)"

# domain/oscal/adapter/cmmc_ssp_adapter.py 加 method
def adapt_to_bundle(
    self,
    parsed_docx: ParsedDocx,
    structure: Optional[Any],
    candidates: List[CandidateControl],
    tenant_id: int = None,
    doc: Optional[Any] = None,
) -> ParsedExcelEntityBundle:
    """Phase 2: Output v3 ParsedExcelEntityBundle (replaces adapt → ParsedSsp).
    Existing adapt() preserved for back-compat (Stage 1 callers); will be
    deprecated after all docx callers migrate.
    """
    from domain.oscal.parser.docx_section_extractors import (
        extract_leveraged_category_table,
        extract_leveraged_csp_table,
        extract_metadata_table,
        extract_party_tables,
        extract_revision_history_table,
        extract_system_characteristic_from_metadata,
    )
    from domain.oscal.parser.ssp_intermediate import (
        ParsedComponent,
        ParsedExcelEntityBundle,
        ParsedLeveragedAuthorization,
        ParsedParty,
        ParsedSystemCharacteristic,
    )

    parsed_las: list[ParsedLeveragedAuthorization] = []
    parsed_components: list[ParsedComponent] = []
    parsed_parties: list[ParsedParty] = []
    parsed_sc: Optional[ParsedSystemCharacteristic] = None
    parsed_metadata: Optional[dict] = None

    if doc is not None:
        # --- Metadata ---
        meta = extract_metadata_table(doc) or {}
        parsed_metadata = {k: v for k, v in meta.items() if v}

        # --- System characteristic ---
        sc_dict = extract_system_characteristic_from_metadata(doc)
        if sc_dict and (sc_dict.get("system_name") or meta.get("title")):
            parsed_sc = ParsedSystemCharacteristic(
                system_name=sc_dict.get("system_name") or meta.get("title") or "",
                security_sensitivity_level=sc_dict.get("security_sensitivity_level"),
            )

        # --- Parties ---
        for pd in extract_party_tables(doc):
            name = (pd.get("name") or "").strip()
            if not name and pd.get("_placeholder"):
                continue
            parsed_parties.append(ParsedParty(
                name=name or "(未命名)",
                party_type=pd.get("party_type", "person"),
                title=pd.get("title"),
                email_address=pd.get("email"),
                telephone_number=pd.get("telephone"),
                address=pd.get("address"),
            ))

        # --- Table #6 CSP/CSO (FedRAMP) ---
        for row in extract_leveraged_csp_table(doc):
            title = row.get("service_name") or row.get("provider") or ""
            if not title:
                continue
            la = ParsedLeveragedAuthorization(
                title=title,
                provider=row.get("provider"),
                fedramp_package_id=row.get("fedramp_package_id"),
                impact_level=_normalize_impact(row.get("impact_level")),
                data_types=row.get("data_types"),
                nature_of_agreement=row.get("nature_of_agreement"),
                authorized_users=row.get("authorized_users"),
            )
            parsed_las.append(la)
            # Paired service component pointing back to LA
            parsed_components.append(ParsedComponent(
                title=title,
                component_type="service",
                description=row.get("purpose"),
                leveraged_authorization_ref=la.title,
            ))

        # --- Table #7 Category ---
        for row in extract_leveraged_category_table(doc):
            title = row.get("name_description") or ""
            if not title:
                continue
            comp_type = _cat_zh_to_oscal(row.get("category"))
            parsed_components.append(ParsedComponent(
                title=title,
                component_type=comp_type,
                description=row.get("purpose"),
                protocol=row.get("protocol"),
                security_auth=row.get("security_auth"),
            ))

    return ParsedExcelEntityBundle(
        parsed_system_characteristic=parsed_sc,
        parsed_parties=parsed_parties,
        parsed_components=parsed_components,
        parsed_leveraged_authorizations=parsed_las,
        parsed_inventory_items=[],   # Phase 2: docx 樣板還沒 inventory 表（Phase 3 Excel template 加）
        parsed_controls=[],          # control 仍走 matched_controls 既有 path
        parsed_metadata=parsed_metadata,
    )


# Module-level helpers
_CAT_ZH_TO_OSCAL = {
    "外部服務": "service",
    "互連": "interconnection",
    "API": "service",
    "CLI": "service",
}

def _cat_zh_to_oscal(zh: Optional[str]) -> str:
    return _CAT_ZH_TO_OSCAL.get((zh or "").strip(), "service")


def _normalize_impact(raw: Optional[str]) -> Optional[str]:
    if not raw:
        return None
    m = {"Low": "low", "Moderate": "moderate", "High": "high", "LI-SaaS": "li-saas"}
    return m.get(raw.strip(), raw.lower())

# tests/test_cmmc_ssp_adapter_v3_bundle.py
from unittest.mock import Mock, patch
from domain.oscal.adapter.cmmc_ssp_adapter import CmmcSspAdapter


def test_adapt_to_bundle_t6_creates_la_and_paired_component():
    adapter = CmmcSspAdapter()
    parsed_docx = Mock(matched_controls=[], unmatched_paragraphs=[], summary={})
    fake_doc = Mock()
    with patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_csp_table",
               return_value=[{
                   "service_name": "AcmeCorp Backup",
                   "provider": "AcmeCorp",
                   "fedramp_package_id": "FR-001",
                   "impact_level": "Moderate",
               }]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_category_table",
               return_value=[]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_metadata_table", return_value={}), \
         patch("domain.oscal.parser.docx_section_extractors.extract_party_tables", return_value=[]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_revision_history_table", return_value=[]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_system_characteristic_from_metadata",
               return_value={}):
        bundle = adapter.adapt_to_bundle(parsed_docx, None, [], doc=fake_doc)

    assert len(bundle.parsed_leveraged_authorizations) == 1
    la = bundle.parsed_leveraged_authorizations[0]
    assert la.title == "AcmeCorp Backup"
    assert la.fedramp_package_id == "FR-001"
    assert la.impact_level == "moderate"

    assert len(bundle.parsed_components) == 1
    comp = bundle.parsed_components[0]
    assert comp.component_type == "service"
    assert comp.leveraged_authorization_ref == "AcmeCorp Backup"


def test_adapt_to_bundle_t7_category_creates_component_only():
    adapter = CmmcSspAdapter()
    parsed_docx = Mock(matched_controls=[], unmatched_paragraphs=[], summary={})
    fake_doc = Mock()
    with patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_csp_table", return_value=[]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_leveraged_category_table",
               return_value=[{
                   "name_description": "GitLab",
                   "category": "外部服務",
                   "protocol": "HTTPS",
                   "security_auth": "OAuth",
               }]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_metadata_table", return_value={}), \
         patch("domain.oscal.parser.docx_section_extractors.extract_party_tables", return_value=[]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_revision_history_table", return_value=[]), \
         patch("domain.oscal.parser.docx_section_extractors.extract_system_characteristic_from_metadata",
               return_value={}):
        bundle = adapter.adapt_to_bundle(parsed_docx, None, [], doc=fake_doc)

    assert len(bundle.parsed_leveraged_authorizations) == 0
    assert len(bundle.parsed_components) == 1
    comp = bundle.parsed_components[0]
    assert comp.title == "GitLab"
    assert comp.component_type == "service"
    assert comp.protocol == "HTTPS"
    assert comp.security_auth == "OAuth"

# domain/oscal/adapter/i_ssp_docx_adapter.py
@abstractmethod
def adapt_to_bundle(
    self, parsed_docx, structure, candidates, tenant_id=None, doc=None,
) -> "ParsedExcelEntityBundle":
    ...

pytest tests/test_cmmc_ssp_adapter_v3_bundle.py tests/test_cmmc_ssp_adapter.py -v

git add domain/oscal/adapter/cmmc_ssp_adapter.py \
        domain/oscal/adapter/i_ssp_docx_adapter.py \
        tests/test_cmmc_ssp_adapter_v3_bundle.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 8 — CmmcSspAdapter.adapt_to_bundle outputs v3 bundle"

grep -n "def " app/oscal/service/ssp_docx_import_app_service.py | head -30

# 在 upload_and_parse_docx 內，bundle 寫入 parsed_result 前加：
bundle = self._adapter.adapt_to_bundle(parsed_docx, structure, candidates, doc=doc)
parsed_result = self._bundle_to_dict(bundle)
parsed_result["schema_version"] = "v2-bundle"
# 沿用既有 parse_jobs.write_parsed_result(uid, parsed_result, ...)

# confirm_docx_import 內，原本走 _adapter.adapt → ParsedSsp → write strategies 那段：
schema_version = parsed_result.get("schema_version", "v1-ssp")
if schema_version == "v2-bundle":
    write_stats = self._confirm_service.confirm(
        parsed_result=parsed_result,
        ssp_id=ssp_id,
        catalog_id=catalog_id,
        user_context=user_context,
    )
else:
    # 原本 legacy 路徑
    write_stats = self._legacy_confirm(...)

ssp_docx_import_app_service = providers.Singleton(
    SspDocxImportAppService,
    ...,
    confirm_service=ssp_import_confirm_service,
    ...,
)

pytest tests/test_ssp_docx_import_app_service.py tests/test_ssp_docx_import_e2e.py tests/test_ssp_docx_import_create_no_diff.py -v

git add app/oscal/service/ssp_docx_import_app_service.py \
        di_containers/oscal/oscal_container.py \
        tests/test_ssp_docx_import_*.py
git commit -m "feat(ssp-oscal-alignment): Phase 2 Task 9 — Docx app service wire SspImportConfirmService (v2-bundle path)"

# tests/test_ssp_import_pipeline_smoke.py
"""Phase 2 smoke — confirm pipeline end-to-end with mock domain services.

Verifies:
- v2-bundle parsed_result → bundle_restore → normalize → strategies write
- warnings_added 反映 normalizer 結果
- 寫 stats 累積到 confirm_service return dict
"""
from unittest.mock import Mock

from domain.oscal.import_pipeline.confirm_service import SspImportConfirmService


def test_smoke_v2_bundle_full_chain():
    # Mock domain services (avoid real DB)
    la_svc = Mock(); la_svc.get_all.return_value = []
    fake_saved_la = Mock(); fake_saved_la.uid = "la-uuid-1"; fake_saved_la.title = "Crowdstrike"
    la_svc.add.return_value = fake_saved_la

    comp_svc = Mock(); comp_svc.get_all.return_value = []
    inv_svc = Mock(); inv_svc.get_all.return_value = []

    from domain.oscal.service.write_strategy.leveraged_write_strategy import (
        LeveragedWriteStrategy,
    )
    from domain.oscal.service.write_strategy.component_write_strategy import (
        ComponentWriteStrategy,
    )
    from domain.oscal.service.write_strategy.inventory_item_write_strategy import (
        InventoryItemWriteStrategy,
    )

    confirm_service = SspImportConfirmService(
        orchestrator=None,  # smoke 跳過 reconcile
        leveraged_write_strategy=LeveragedWriteStrategy(la_svc),
        component_write_strategy=ComponentWriteStrategy(comp_svc, la_svc),
        inventory_write_strategy=InventoryItemWriteStrategy(inv_svc, comp_svc),
    )

    parsed_result = {
        "schema_version": "v2-bundle",
        "components": [
            {"title": "AcmeCorp Backup", "component_type": "service",
             "leveraged_authorization_ref": "Crowdstrike"},
            {"title": "Web Server", "component_type": "hardware"},
        ],
        "leveraged_authorizations": [
            {"title": "Crowdstrike", "fedramp_package_id": "FR-001"},
        ],
        "inventory_items": [
            {"description": "Srv 1", "implemented_component_refs": ["Web Server"]},
        ],
    }
    user_ctx = Mock(login_name="alice", tenant_id=1, org_unit_id=99)
    stats = confirm_service.confirm(
        parsed_result=parsed_result,
        ssp_id=10,
        catalog_id=None,
        user_context=user_ctx,
    )
    assert stats["written_las"] == 1
    assert stats["written_components"] == 2
    assert stats["written_inventory"] == 1
    assert stats["warnings_added"] == 0
    assert parsed_result["import_warnings"] == []


def test_smoke_v1_ssp_dispatch_raises():
    from domain.oscal.import_pipeline.bundle_restore import LegacyConfirmRequired
    import pytest

    confirm_service = SspImportConfirmService(
        orchestrator=None,
        leveraged_write_strategy=Mock(),
        component_write_strategy=Mock(),
        inventory_write_strategy=Mock(),
    )
    parsed_result = {"devices": [], "info_systems": [], "leveraged": []}
    with pytest.raises(LegacyConfirmRequired):
        confirm_service.confirm(
            parsed_result=parsed_result,
            ssp_id=10, catalog_id=None,
            user_context=Mock(login_name="x", tenant_id=1),
        )

pytest tests/test_ssp_import_pipeline_smoke.py -v

pytest tests/ -x --ignore=tests/test_<known-skipped>.py
# Expected: ~102+N passed (N = Phase 2 新加 test 數，預估 30+)，0 failed

# kill 舊 process
lsof -i :8000 -t | xargs -r kill -9
# 啟動
nohup poetry run python main_app.py > log/app.log 2>&1 &
sleep 3
tail -50 log/app.log | grep -E "ERROR|Traceback|started|listening"
# Expected: 看到 listen on 8000 + 無 startup ERROR

git add tests/test_ssp_import_pipeline_smoke.py
git commit -m "test(ssp-oscal-alignment): Phase 2 Task 10 — confirm pipeline end-to-end smoke"

grep -rn "ParsedDevice\|ParsedInformationSystem\|ParsedLeveraged\b" --include="*.py" .
grep -rn "DeviceWriteStrategy\|InformationSystemWriteStrategy" --include="*.py" .
grep -rn "sys_impl_main_id" --include="*.py" .

pytest tests/ -x

git add -A   # ⚠️ 多個檔案，但 dispatch subagent 時要顯式列檔（per feedback_subagent_explicit_git_add）
git commit -m "refactor(ssp-oscal-alignment): Phase 2 Task 11 — cleanup v2 strategies + ParsedDevice/ParsedInformationSystem/ParsedLeveraged + dead sys_impl_main_id"

grep -rn '"implementation_type"' --include="*.py" --include="*.vue" --include="*.js" \
    ~/Projects/Billows/Audit-Manager/compliance-manager-be \
    ~/Projects/Billows/Audit-Manager/compliance-manager-fe \
    ~/Projects/Billows/Audit-Manager/compliance-manager-test
# Expected: 零殘留（Phase 1 已改）；若有，標 Phase 2 內補 patch

---
type: feat
breaking: true
modules: [oscal, ssp-oscal-alignment, jedi-oscal, ssp-excel-import, ssp-docx-import]
commit: <main-commit-hash>
---

# SSP OSCAL Alignment — Phase 1 + Phase 2

## 需求說明
（兩 phase 概述）

## 變更範圍
- jedi-oscal v0.0.18 → v0.1.0（新 entity Component / LeveragedAuthorization / InventoryItem）
- 主專案 import_pipeline 新模組 + 4 個新 oscal 表 + RLS
- ...

## API / 行為差異
- ...

## 測試
- 主專案 pytest 102 + Phase 2 新增 N 個 test
- jedi-oscal pytest 84
- BE smoke 通

## 參考
- design.md §11 (11 條 reconciliation)
- handoff Phase 1 SUMMARY + Phase 2 SUMMARY

poetry run python scripts/extract_claude_sessions.py \
    --date 2026-05-25 --topic ssp-oscal-alignment-phase2 --auto
# 跟 Phase 1 2026-05-24 那批分開存

cd ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal
# bump pyproject.toml version: 0.0.18 → 0.1.0
# commit + push + 推 Nexus（per CLAUDE.md jedi-package SOP）

# pyproject.toml：path dep 註解回去，pin Nexus 0.1.0
poetry update jedi-oscal
# smoke 確認 import OK
poetry run python -c "import jedi_oscal; print(jedi_oscal.__version__)"
git add pyproject.toml poetry.lock docs/changelog/* docs/features/FR-028-2605-ssp-oscal-alignment/handoff/*
git commit -m "release(ssp-oscal-alignment): Phase 1+2 ship — jedi-oscal 0.1.0 + changelog + SUMMARY"