Phase A3 Implementation Plan — 共用 matcher 抽取（parties / org-units）

grep -n "def reconcile\|def __init__\|def _reconcile_person\|def _reconcile_organization" \
  domain/oscal/service/party_reconciliation_service.py

grep -nE "reconcile|reconciliation" app/oscal/service/ssp_excel_import_app_service.py | head -20

grep -nA 20 "^class ParsedParty" domain/oscal/parser/ssp_intermediate.py

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

# domain/oscal/service/reconciliation/match_method.py
from enum import StrEnum


class MatchMethod(StrEnum):
    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.EXACT)"

# tests/test_a3_reconciliation_normalizers.py
import pytest
from domain.oscal.service.reconciliation._normalizers import (
    _strip_plus_alias, _normalize_name, _strip_org_suffix
)


class TestStripPlusAlias:
    def test_simple_plus_alias(self):
        assert _strip_plus_alias("alice+work@acme.com") == "alice@acme.com"

    def test_no_plus(self):
        assert _strip_plus_alias("alice@acme.com") == "alice@acme.com"

    def test_no_at(self):
        assert _strip_plus_alias("invalid") == "invalid"

    def test_plus_in_domain_not_local(self):
        # +在 domain side 不該 strip
        assert _strip_plus_alias("alice@a+b.com") == "alice@a+b.com"

    def test_multiple_at(self):
        # split('@', 1) 取第一個 @
        assert _strip_plus_alias("alice+work@b@c.com") == "alice@b@c.com"


class TestNormalizeName:
    def test_full_width_space(self):
        assert _normalize_name("陳　小明") == "陳 小明"

    def test_collapse_spaces(self):
        assert _normalize_name("陳   小明") == "陳 小明"

    def test_strip_outer(self):
        assert _normalize_name("  Alice  ") == "Alice"

    def test_none(self):
        assert _normalize_name(None) == ""

    def test_empty(self):
        assert _normalize_name("") == ""


class TestStripOrgSuffix:
    def test_股份有限公司(self):
        assert _strip_org_suffix("ACME 股份有限公司") == "ACME"

    def test_有限公司_not_eaten_first(self):
        # 順序檢查：「股份有限公司」優先於「有限公司」
        assert _strip_org_suffix("台積電股份有限公司") == "台積電"

    def test_inc(self):
        assert _strip_org_suffix("Foo Inc.") == "Foo"

    def test_no_suffix(self):
        assert _strip_org_suffix("ACME") == "ACME"

    def test_empty(self):
        assert _strip_org_suffix("") == ""

pytest tests/test_a3_reconciliation_normalizers.py -v

# domain/oscal/service/reconciliation/_normalizers.py
from typing import Optional


def _strip_plus_alias(email: str) -> str:
    if '@' not in email:
        return email
    local, domain = email.split('@', 1)
    if '+' not in local:
        return email
    return f"{local.split('+', 1)[0]}@{domain}"


def _normalize_name(name: Optional[str]) -> str:
    if not name:
        return ""
    s = name.replace('　', ' ')
    s = ' '.join(s.split())
    return s.strip()


ORG_SUFFIXES = [
    "股份有限公司",
    "有限公司",
    "公司",
    "Co., Ltd.", "Co., Ltd",
    "Corp.", "Corp",
    "L.L.C.", "LLC",
    "Ltd.", "Ltd",
    "Inc.", "Inc",
]


def _strip_org_suffix(name: str) -> str:
    if not name:
        return name
    for suffix in ORG_SUFFIXES:
        if name.endswith(suffix):
            return name[:-len(suffix)].strip()
    return name

pytest tests/test_a3_reconciliation_normalizers.py -v

# tests/test_a3_reconciliation_base.py
import pytest
from unittest.mock import MagicMock
from domain.oscal.service.reconciliation.base import BaseReconciliationService
from domain.oscal.service.reconciliation.match_method import MatchMethod


class _DummyReconciler(BaseReconciliationService):
    """測試用 reconciler — 5 hook 都 MagicMock-style 受控"""

    def __init__(self):
        super().__init__()
        self.exact_returns = None
        self.normalized_returns = None
        self.fuzzy_returns = (None, MatchMethod.UNMATCHED)
        self.applied_matches = []
        self.applied_unmatched = []

    def _try_exact_match(self, parsed, tenant_id):
        return self.exact_returns

    def _try_normalized_match(self, parsed, tenant_id):
        return self.normalized_returns

    def _try_fuzzy_match(self, parsed, tenant_id):
        return self.fuzzy_returns

    def _apply_match(self, parsed, entity, method, confidence):
        self.applied_matches.append((parsed, entity, method, confidence))

    def _apply_unmatched(self, parsed):
        self.applied_unmatched.append(parsed)


def test_exact_hit_short_circuits():
    r = _DummyReconciler()
    entity = MagicMock(name="user_42")
    r.exact_returns = entity
    parsed = MagicMock(name="parsed_alice")
    r.reconcile([parsed], tenant_id=102)
    assert r.applied_matches == [(parsed, entity, MatchMethod.EXACT, 1.0)]
    assert r.applied_unmatched == []


def test_normalized_hit_when_exact_miss():
    r = _DummyReconciler()
    entity = MagicMock()
    r.normalized_returns = entity
    parsed = MagicMock()
    r.reconcile([parsed], tenant_id=102)
    assert r.applied_matches == [(parsed, entity, MatchMethod.NORMALIZED, 1.0)]


def test_fuzzy_hit_uses_returned_method_and_low_confidence():
    r = _DummyReconciler()
    entity = MagicMock()
    r.fuzzy_returns = (entity, MatchMethod.FUZZY_EMAIL_DOMAIN)
    parsed = MagicMock()
    r.reconcile([parsed], tenant_id=102)
    assert r.applied_matches == [(parsed, entity, MatchMethod.FUZZY_EMAIL_DOMAIN, 0.7)]


def test_unmatched_when_all_stages_miss():
    r = _DummyReconciler()
    parsed = MagicMock()
    r.reconcile([parsed], tenant_id=102)
    assert r.applied_unmatched == [parsed]
    assert r.applied_matches == []


def test_batch_iterates_all_records():
    r = _DummyReconciler()
    parsed_a, parsed_b = MagicMock(), MagicMock()
    result = r.reconcile([parsed_a, parsed_b], tenant_id=102)
    assert result == [parsed_a, parsed_b]
    assert len(r.applied_unmatched) == 2


def test_cache_lazy_loads_once():
    r = _DummyReconciler()
    loader = MagicMock(return_value=[1, 2, 3])
    assert r._get_or_load_candidates(loader) == [1, 2, 3]
    assert r._get_or_load_candidates(loader) == [1, 2, 3]
    assert loader.call_count == 1


def test_cache_exception_returns_empty_list_no_persist():
    r = _DummyReconciler()
    loader = MagicMock(side_effect=RuntimeError("DB down"))
    result = r._get_or_load_candidates(loader)
    assert result == []
    assert r._candidate_cache is None  # 不持久化 exception 結果
    # 下次 call 仍 retry
    loader_ok = MagicMock(return_value=[42])
    assert r._get_or_load_candidates(loader_ok) == [42]
    assert loader_ok.call_count == 1

pytest tests/test_a3_reconciliation_base.py -v

# domain/oscal/service/reconciliation/base.py
from abc import ABC, abstractmethod
from typing import Generic, List, Optional, Tuple, TypeVar

from domain.oscal.service.reconciliation.match_method import MatchMethod

TParsed = TypeVar("TParsed")
TEntity = TypeVar("TEntity")


class BaseReconciliationService(ABC, Generic[TParsed, TEntity]):
    """Domain-level base class for parsed-record ↔ system-entity reconciliation.

    Each subclass owns one domain pair (e.g. ParsedParty(person) ↔ User);
    subclass implements 5 hooks. The base class owns the shared batch loop,
    candidate cache, and three-state fallback (exact / normalized / fuzzy).

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

    def __init__(self):
        self._candidate_cache: Optional[List[TEntity]] = None

    def reconcile(
        self,
        parsed_list: List[TParsed],
        tenant_id: Optional[int] = None,
    ) -> List[TParsed]:
        for parsed in parsed_list:
            self._reconcile_one(parsed, tenant_id)
        return parsed_list

    def _reconcile_one(self, parsed: TParsed, tenant_id: Optional[int]) -> None:
        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)

    @abstractmethod
    def _try_exact_match(
        self, parsed: TParsed, tenant_id: Optional[int]
    ) -> Optional[TEntity]: ...

    @abstractmethod
    def _try_normalized_match(
        self, parsed: TParsed, tenant_id: Optional[int]
    ) -> Optional[TEntity]: ...

    @abstractmethod
    def _try_fuzzy_match(
        self, parsed: TParsed, tenant_id: Optional[int]
    ) -> Tuple[Optional[TEntity], MatchMethod]: ...

    @abstractmethod
    def _apply_match(
        self,
        parsed: TParsed,
        entity: TEntity,
        method: MatchMethod,
        confidence: float,
    ) -> None: ...

    @abstractmethod
    def _apply_unmatched(self, parsed: TParsed) -> None: ...

    def _get_or_load_candidates(self, loader) -> List[TEntity]:
        if self._candidate_cache is not None:
            return self._candidate_cache
        try:
            result = loader() or []
        except Exception:
            return []
        self._candidate_cache = result
        return result

pytest tests/test_a3_reconciliation_base.py -v

grep -rn "ParsedParty(" tests/ app/ domain/ | head -20

# domain/oscal/parser/ssp_intermediate.py
# 既有 import
from domain.oscal.service.reconciliation.match_method import MatchMethod

@dataclass
class ParsedParty:
    name: str
    party_type: Literal["person", "organization"]
    role: Optional[str]
    email_address: Optional[str]
    # ... 其他既有欄位
    matched_user_id: Optional[int] = None
    matched_org_unit_id: Optional[int] = None
    target_party_uid: Optional[str] = None

    # A3 新增
    match_method: MatchMethod = MatchMethod.UNMATCHED
    match_confidence: float = 0.0

pytest tests/test_party_reconciliation_service.py tests/test_ssp_docx_*.py tests/test_ssp_excel_*.py -v 2>&1 | tail -30

# tests/test_a3_reconciliation_person.py
import pytest
from unittest.mock import MagicMock
from domain.oscal.parser.ssp_intermediate import ParsedParty
from domain.oscal.service.reconciliation.person_reconciler import PersonReconciler
from domain.oscal.service.reconciliation.match_method import MatchMethod


@pytest.fixture
def user_domain():
    m = MagicMock()
    m.get_users.return_value = []
    return m


@pytest.fixture
def reconciler(user_domain):
    return PersonReconciler(user_domain_service=user_domain)


def _parsed(name="Alice", email="alice@acme.com"):
    return ParsedParty(
        name=name, party_type="person", role=None, email_address=email,
    )


class TestExactStage:
    def test_email_lowercase_trim_exact_hit(self, reconciler, user_domain):
        user = MagicMock(id=42, is_active=True, email="alice@acme.com")
        user_domain.get_users.return_value = [user]
        p = _parsed(email="  Alice@Acme.com  ")
        reconciler.reconcile([p], tenant_id=102)
        assert p.matched_user_id == 42
        assert p.match_method == MatchMethod.EXACT
        assert p.match_confidence == 1.0

    def test_inactive_user_skipped(self, reconciler, user_domain):
        user = MagicMock(id=42, is_active=False, email="alice@acme.com")
        user_domain.get_users.return_value = [user]
        p = _parsed()
        reconciler.reconcile([p], tenant_id=102)
        assert p.matched_user_id is None
        assert p.match_method == MatchMethod.UNMATCHED
        assert p.match_confidence == 0.0

    def test_empty_email_skipped(self, reconciler):
        p = _parsed(email=None)
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

    def test_exception_from_domain_swallowed_to_unmatched(self, reconciler, user_domain):
        user_domain.get_users.side_effect = RuntimeError("DB down")
        p = _parsed()
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

# domain/oscal/service/reconciliation/person_reconciler.py
from typing import List, Optional, Tuple

from domain.oscal.parser.ssp_intermediate import ParsedParty
from domain.oscal.service.reconciliation.base import BaseReconciliationService
from domain.oscal.service.reconciliation.match_method import MatchMethod


class PersonReconciler(BaseReconciliationService[ParsedParty, object]):
    def __init__(self, user_domain_service):
        super().__init__()
        self._user = user_domain_service

    def _try_exact_match(self, parsed, tenant_id):
        email = (parsed.email_address or "").strip().lower()
        if not email:
            return None
        try:
            # 對齊 既有 PartyReconciliationService Query Entity 簽章
            from jedi_auth.domain.entity.user_query_entity import UserQueryEntity
            users = self._user.get_users(UserQueryEntity(email=email))
        except Exception:
            return None
        for u in users or []:
            if getattr(u, "is_active", True):
                return u
        return None

    def _try_normalized_match(self, parsed, tenant_id):
        return None  # T4 落地

    def _try_fuzzy_match(self, parsed, tenant_id):
        return (None, MatchMethod.UNMATCHED)  # T4 落地

    def _apply_match(self, parsed, entity, method, confidence):
        parsed.matched_user_id = entity.id
        parsed.match_method = method
        parsed.match_confidence = confidence

    def _apply_unmatched(self, parsed):
        parsed.match_method = MatchMethod.UNMATCHED
        parsed.match_confidence = 0.0
        # matched_user_id 維持 None

# tests/test_a3_reconciliation_organization.py
import pytest
from unittest.mock import MagicMock
from domain.oscal.parser.ssp_intermediate import ParsedParty
from domain.oscal.service.reconciliation.organization_reconciler import OrganizationReconciler
from domain.oscal.service.reconciliation.match_method import MatchMethod


@pytest.fixture
def org_unit_domain():
    m = MagicMock()
    m.get_org_units.return_value = []
    return m


@pytest.fixture
def reconciler(org_unit_domain):
    return OrganizationReconciler(org_unit_domain_service=org_unit_domain)


def _parsed(name="ACME"):
    return ParsedParty(
        name=name, party_type="organization", role=None, email_address=None,
    )


class TestExactStage:
    def test_name_trim_exact_hit(self, reconciler, org_unit_domain):
        org = MagicMock(id=7)
        org_unit_domain.get_org_units.return_value = [org]
        p = _parsed(name="  ACME  ")
        reconciler.reconcile([p], tenant_id=102)
        assert p.matched_org_unit_id == 7
        assert p.match_method == MatchMethod.EXACT
        assert p.match_confidence == 1.0

    def test_empty_name(self, reconciler):
        p = _parsed(name="")
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

    def test_exception_swallowed(self, reconciler, org_unit_domain):
        org_unit_domain.get_org_units.side_effect = RuntimeError("DB")
        p = _parsed()
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

    def test_no_hits(self, reconciler):
        p = _parsed()
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED
        assert p.matched_org_unit_id is None

    def test_tenant_id_passed_when_provided(self, reconciler, org_unit_domain):
        p = _parsed()
        reconciler.reconcile([p], tenant_id=102)
        # 確認 query 帶 tenant_id（沿用 既有 PartyReconciliationService 行為）
        call_args = org_unit_domain.get_org_units.call_args
        assert call_args is not None
        # 看 Query Entity 內 tenant_id 是否被設（依 既有 Query Entity 介面）

# domain/oscal/service/reconciliation/organization_reconciler.py
from typing import Optional, Tuple

from domain.oscal.parser.ssp_intermediate import ParsedParty
from domain.oscal.service.reconciliation.base import BaseReconciliationService
from domain.oscal.service.reconciliation.match_method import MatchMethod


class OrganizationReconciler(BaseReconciliationService[ParsedParty, object]):
    def __init__(self, org_unit_domain_service):
        super().__init__()
        self._org = org_unit_domain_service

    def _try_exact_match(self, parsed, tenant_id):
        name = (parsed.name or "").strip()
        if not name:
            return None
        try:
            from jedi_auth.domain.entity.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)
        except Exception:
            return None
        return orgs[0] if orgs else None

    def _try_normalized_match(self, parsed, tenant_id):
        return None  # T4 落地

    def _try_fuzzy_match(self, parsed, tenant_id):
        return (None, MatchMethod.UNMATCHED)  # T4 落地

    def _apply_match(self, parsed, entity, method, confidence):
        parsed.matched_org_unit_id = entity.id
        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/__init__.py
from domain.oscal.service.reconciliation.base import BaseReconciliationService
from domain.oscal.service.reconciliation.match_method import MatchMethod
from domain.oscal.service.reconciliation.person_reconciler import PersonReconciler
from domain.oscal.service.reconciliation.organization_reconciler import OrganizationReconciler

__all__ = [
    "BaseReconciliationService",
    "MatchMethod",
    "PersonReconciler",
    "OrganizationReconciler",
]

git status --short

git add domain/oscal/service/reconciliation/__init__.py \
        domain/oscal/service/reconciliation/base.py \
        domain/oscal/service/reconciliation/match_method.py \
        domain/oscal/service/reconciliation/_normalizers.py \
        domain/oscal/service/reconciliation/person_reconciler.py \
        domain/oscal/service/reconciliation/organization_reconciler.py \
        domain/oscal/parser/ssp_intermediate.py \
        tests/test_a3_reconciliation_normalizers.py \
        tests/test_a3_reconciliation_base.py \
        tests/test_a3_reconciliation_person.py \
        tests/test_a3_reconciliation_organization.py

git commit -m "$(cat <<'EOF'
feat(oscal): A3 T1 base + reconcilers skeleton

抽 BaseReconciliationService[TParsed, TEntity] 抽象基類 + Person /
Organization Reconciler skeleton（exact stage 完整、normalized/fuzzy 留 T4）；
ParsedParty 加 match_method + match_confidence optional 欄位；
13 + 7 + 4 + 5 = 29 個 unit test 全綠。

對齊 design-A3.md §3.2 / §4 / §5.3。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"

# domain/oscal/service/party_reconciliation_service.py
from typing import List, Optional

from domain.oscal.parser.ssp_intermediate import ParsedParty


class PartyReconciliationService:
    """Facade dispatching ParsedParty list by party_type.

    Mixed-list dispatch is domain knowledge — callers (docx / A2 confirm flows)
    should not need to know how to split persons / organizations.
    """

    def __init__(self, person_reconciler, organization_reconciler):
        self._person = person_reconciler
        self._org = organization_reconciler

    def reconcile(
        self,
        parties: List[ParsedParty],
        tenant_id: Optional[int] = None,
    ) -> List[ParsedParty]:
        persons = [p for p in parties if p.party_type == "person"]
        orgs = [p for p in parties if p.party_type == "organization"]
        self._person.reconcile(persons, tenant_id)
        self._org.reconcile(orgs, tenant_id)
        return parties

pytest tests/test_party_reconciliation_service.py -v

grep -nA 5 "party_reconciliation_service\|person_reconciler\|organization_reconciler" \
  di_containers/oscal/oscal_containers.py | head -30

grep -nE "^from|^import" di_containers/oscal/oscal_containers.py | grep -i auth

# di_containers/oscal/oscal_containers.py
# 在 既有 party_reconciliation_service provider 前加：

from domain.oscal.service.reconciliation.person_reconciler import PersonReconciler
from domain.oscal.service.reconciliation.organization_reconciler import OrganizationReconciler

person_reconciler = providers.Factory(
    PersonReconciler,
    user_domain_service=auth_container.user_domain_service,
)
organization_reconciler = providers.Factory(
    OrganizationReconciler,
    org_unit_domain_service=auth_container.org_unit_domain_service,
)

# 改既有 line 488 provider：
party_reconciliation_service = providers.Factory(
    PartyReconciliationService,
    person_reconciler=person_reconciler,
    organization_reconciler=organization_reconciler,
)

lsof -ti:8000 | xargs kill -9 2>/dev/null
set -a; source .env; set +a
nohup poetry run python main_app.py > /tmp/a3-be-boot.log 2>&1 &
sleep 5
tail -50 /tmp/a3-be-boot.log
grep -E "ERROR|Traceback|Failed" /tmp/a3-be-boot.log | head -20

# tests/test_a3_reconciliation_facade.py
import pytest
from unittest.mock import MagicMock
from domain.oscal.parser.ssp_intermediate import ParsedParty
from domain.oscal.service.party_reconciliation_service import PartyReconciliationService


@pytest.fixture
def person_rec():
    m = MagicMock()
    m.reconcile.side_effect = lambda lst, tenant_id=None: lst
    return m


@pytest.fixture
def org_rec():
    m = MagicMock()
    m.reconcile.side_effect = lambda lst, tenant_id=None: lst
    return m


@pytest.fixture
def facade(person_rec, org_rec):
    return PartyReconciliationService(
        person_reconciler=person_rec,
        organization_reconciler=org_rec,
    )


def test_dispatch_by_party_type(facade, person_rec, org_rec):
    persons = [ParsedParty(name="A", party_type="person", role=None, email_address="a@x.com")]
    orgs = [ParsedParty(name="X", party_type="organization", role=None, email_address=None)]
    facade.reconcile(persons + orgs, tenant_id=102)
    person_rec.reconcile.assert_called_once_with(persons, 102)
    org_rec.reconcile.assert_called_once_with(orgs, 102)


def test_empty_list(facade, person_rec, org_rec):
    result = facade.reconcile([], tenant_id=102)
    assert result == []
    person_rec.reconcile.assert_called_once_with([], 102)
    org_rec.reconcile.assert_called_once_with([], 102)


def test_returns_same_list_reference(facade):
    parties = [ParsedParty(name="A", party_type="person", role=None, email_address="a@x.com")]
    result = facade.reconcile(parties, tenant_id=102)
    assert result is parties

git add domain/oscal/service/party_reconciliation_service.py \
        di_containers/oscal/oscal_containers.py \
        tests/test_a3_reconciliation_facade.py

git commit -m "$(cat <<'EOF'
feat(oscal): A3 T2 facade + DI wiring

PartyReconciliationService 改門面 dispatch by party_type；新增 DI Factory
person_reconciler + organization_reconciler；門面改注入兩 reconciler。
既有 8 unit test 0 行改動全綠；門面新增 3 個 dispatch test 全綠。
BE 起動 smoke 通過 — DI container build 無 wire error。

對齊 design-A3.md §3.3 / §6.4。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"

grep -n "_confirm_superset_flow\|_confirm_update_flow\|write_parties" \
  app/oscal/service/ssp_excel_import_app_service.py

def _confirm_superset_flow(self, job, parsed_result, decisions, overrides, user_context):
    # ... 既有 step 1~3（取 framework_version + catalog + 建 profile + 建 MF）

    # A3 新增：reconcile parties → 標 matched_user_id / matched_org_unit_id / match_method
    parties_combined = list(parsed_result.parties_org) + list(parsed_result.parties_person)
    self._reconciliation.reconcile(parties_combined, tenant_id=user_context.tenant_id)

    # 既有 _write_parties 呼叫不動：
    self._write_parties(parties_combined, context_type='module_frame', context_id=new_mf.id, user_context=user_context)
    # ... 既有 _write_items / _write_control_defaults / _write_ref_docs

def _confirm_update_flow(self, job, parsed_result, decisions, overrides, user_context):
    # ... 既有取 MF
    parties_combined = list(parsed_result.parties_org) + list(parsed_result.parties_person)
    self._reconciliation.reconcile(parties_combined, tenant_id=user_context.tenant_id)
    self._write_parties(parties_combined, context_type='module_frame', context_id=existing_mf.id, user_context=user_context)
    # ... 既有 upsert

# tests/test_a3_reconciliation_a2_integration.py
import pytest
from unittest.mock import MagicMock, patch
from contextlib import contextmanager

from app.oscal.service.ssp_excel_import_app_service import SspExcelImportAppService


@pytest.fixture
def patch_session_scope():
    """對齊 conftest.py 已有的 helper；若不存在則 inline 定義"""
    @contextmanager
    def _noop():
        yield
    with patch("jedi_common.session.database.db.session_scope", _noop):
        yield


@pytest.fixture
def mock_reconciliation():
    m = MagicMock()
    m.reconcile.side_effect = lambda lst, tenant_id=None: lst
    return m


@pytest.fixture
def mock_user_context():
    ctx = MagicMock()
    ctx.tenant_id = 102
    ctx.user_id = 7
    return ctx


@pytest.fixture
def app_service(mock_reconciliation):
    """Mirror A2 既有 fixture：所有 domain service 都 mock，只關心 reconciliation 被呼叫"""
    return SspExcelImportAppService(
        parse_job_domain_service=MagicMock(),
        parser=MagicMock(),
        file_upload_service=MagicMock(),
        module_frame_domain_service=MagicMock(),
        profile_service=MagicMock(),
        module_frame_service=MagicMock(),
        party_domain_service=MagicMock(),
        responsible_party_domain_service=MagicMock(),
        oscal_framework_version_domain_service=MagicMock(),
        catalog_control_domain_service=MagicMock(),
        party_reconciliation_service=mock_reconciliation,
        # ... 其他 dep 依 A2 既有 signature 補
    )

def test_superset_flow_calls_reconcile_once(app_service, mock_reconciliation, mock_user_context, patch_session_scope):
    """_confirm_superset_flow 必須在 write_parties 前 call reconcile(...) 一次"""
    job = MagicMock(uid="parse-uid", source_type="framework_version")
    parsed = MagicMock(parties_org=[MagicMock(party_type="organization")],
                       parties_person=[MagicMock(party_type="person")])
    app_service._confirm_superset_flow(job, parsed, decisions=[], overrides={}, user_context=mock_user_context)
    assert mock_reconciliation.reconcile.call_count == 1


def test_update_flow_calls_reconcile_once(app_service, mock_reconciliation, mock_user_context, patch_session_scope):
    """同上 — update flow"""
    job = MagicMock(uid="parse-uid", source_type="module_frame")
    parsed = MagicMock(parties_org=[], parties_person=[MagicMock(party_type="person")])
    app_service._confirm_update_flow(job, parsed, decisions=[], overrides={}, user_context=mock_user_context)
    assert mock_reconciliation.reconcile.call_count == 1


def test_reconcile_gets_mixed_list(app_service, mock_reconciliation, mock_user_context, patch_session_scope):
    """reconcile 拿到 (parties_org + parties_person) 合併 list"""
    org_party = MagicMock(party_type="organization", name="ACME")
    person_party = MagicMock(party_type="person", email_address="a@x.com")
    job = MagicMock(source_type="framework_version")
    parsed = MagicMock(parties_org=[org_party], parties_person=[person_party])
    app_service._confirm_superset_flow(job, parsed, decisions=[], overrides={}, user_context=mock_user_context)
    args, _ = mock_reconciliation.reconcile.call_args
    assert org_party in args[0]
    assert person_party in args[0]


def test_tenant_id_propagated_from_user_context(app_service, mock_reconciliation, mock_user_context, patch_session_scope):
    """tenant_id 從 user_context 傳到 reconcile"""
    job = MagicMock(source_type="framework_version")
    parsed = MagicMock(parties_org=[], parties_person=[])
    app_service._confirm_superset_flow(job, parsed, decisions=[], overrides={}, user_context=mock_user_context)
    _, kwargs = mock_reconciliation.reconcile.call_args
    assert kwargs.get("tenant_id") == 102

git add app/oscal/service/ssp_excel_import_app_service.py \
        tests/test_a3_reconciliation_a2_integration.py

git commit -m "$(cat <<'EOF'
feat(oscal): A3 T3 wire A2 confirm flows to reconciliation

_confirm_superset_flow + _confirm_update_flow 在 write_parties 前補
reconcile() 呼叫；A2 path 終於跟 docx path 行為一致。

A3 期間 fuzzy 在 A2 path **自動寫入**（行為等同 matched）；A5 phase
補 user 拍板路徑（design-A3.md §11 follow-up F2）。

對齊 design-A3.md §6.1 / §6.2。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"

class TestNormalizedStage:
    def test_plus_alias_normalize_hit(self, reconciler, user_domain):
        # input email: alice+work@acme.com
        # system: alice@acme.com
        def get_users(q):
            if getattr(q, "email", None) == "alice+work@acme.com":
                return []
            if getattr(q, "email", None) == "alice@acme.com":
                return [MagicMock(id=42, is_active=True, email="alice@acme.com")]
            return []
        user_domain.get_users.side_effect = get_users
        p = _parsed(email="alice+work@acme.com")
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.NORMALIZED
        assert p.matched_user_id == 42

    def test_no_plus_skips_normalized(self, reconciler, user_domain):
        # normalized = original → 跳過 normalized stage 直入 fuzzy
        ...

class TestFuzzyStage:
    def test_fuzzy_email_domain_with_nickname_match(self, reconciler, user_domain):
        # input: name='陳小明', email='unknown@acme.com'（exact / normalized 都不中）
        # candidate: User(email='alice@acme.com', nickname='陳小明', login_name='alice')
        def get_users(q):
            if getattr(q, "email", None):
                return []
            # 第三 stage 拉全 tenant list
            return [
                MagicMock(id=99, is_active=True, email="alice@acme.com",
                          nickname="陳小明", login_name="alice"),
            ]
        user_domain.get_users.side_effect = get_users
        p = _parsed(name="陳小明", email="unknown@acme.com")
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.FUZZY_EMAIL_DOMAIN
        assert p.match_confidence == 0.7
        assert p.matched_user_id == 99

    def test_fuzzy_no_email_skips(self, reconciler):
        p = _parsed(name="陳小明", email=None)
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

    def test_fuzzy_no_at_skips(self, reconciler):
        p = _parsed(name="陳小明", email="invalid")
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

    def test_fuzzy_no_candidates_in_domain(self, reconciler, user_domain):
        user_domain.get_users.return_value = []
        p = _parsed()
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

# domain/oscal/service/reconciliation/person_reconciler.py
from domain.oscal.service.reconciliation._normalizers import _strip_plus_alias, _normalize_name


class PersonReconciler(BaseReconciliationService[ParsedParty, object]):
    # ... __init__ + _try_exact_match 沿用 T1.5

    def _try_normalized_match(self, parsed, tenant_id):
        email = (parsed.email_address or "").strip().lower()
        if not email or '@' not in email:
            return None
        normalized = _strip_plus_alias(email)
        if normalized == email:
            return None
        try:
            from jedi_auth.domain.entity.user_query_entity import UserQueryEntity
            users = self._user.get_users(UserQueryEntity(email=normalized))
        except Exception:
            return None
        for u in users or []:
            if getattr(u, "is_active", True):
                return u
        return None

    def _try_fuzzy_match(self, parsed, tenant_id):
        email = (parsed.email_address or "").strip().lower()
        if not email or '@' not in email:
            return (None, MatchMethod.UNMATCHED)
        domain = email.split('@', 1)[1]
        if not domain:
            return (None, MatchMethod.UNMATCHED)

        def _loader():
            from jedi_auth.domain.entity.user_query_entity import UserQueryEntity
            return self._user.get_users(UserQueryEntity())

        candidates = self._get_or_load_candidates(_loader)
        parsed_name = _normalize_name(parsed.name)
        for u in candidates:
            if not getattr(u, "is_active", True):
                continue
            u_email = getattr(u, "email", None)
            if not u_email or '@' not in u_email:
                continue
            if u_email.split('@', 1)[1].lower() != domain:
                continue
            u_nickname = _normalize_name(getattr(u, "nickname", None))
            u_login = _normalize_name(getattr(u, "login_name", None))
            if parsed_name and (u_nickname == parsed_name or u_login == parsed_name):
                return (u, MatchMethod.FUZZY_EMAIL_DOMAIN)
        return (None, MatchMethod.UNMATCHED)

class TestNormalizedStage:
    def test_full_width_space_normalize(self, reconciler, org_unit_domain):
        # input: "ACME　Inc"（全形）；system: "ACME Inc"（半形）
        ...

class TestFuzzyStage:
    def test_strip_股份有限公司_match(self, reconciler, org_unit_domain):
        # input: "ACME 股份有限公司"；system: "ACME"
        def get_org_units(q):
            if getattr(q, "name", None):
                return []
            return [MagicMock(id=7, name="ACME")]
        org_unit_domain.get_org_units.side_effect = get_org_units
        p = _parsed(name="ACME 股份有限公司")
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.FUZZY_NAME_PREFIX
        assert p.matched_org_unit_id == 7

    def test_no_suffix_skips_fuzzy(self, reconciler):
        p = _parsed(name="ACME")
        # normalized == name，不會 strip 出新 stripped → fuzzy 直接 UNMATCHED
        reconciler.reconcile([p], tenant_id=102)
        assert p.match_method == MatchMethod.UNMATCHED

from domain.oscal.service.reconciliation._normalizers import _normalize_name, _strip_org_suffix


class OrganizationReconciler(...):
    # ... __init__ + _try_exact_match 沿用 T1.6

    def _try_normalized_match(self, parsed, tenant_id):
        name = (parsed.name or "").strip()
        if not name:
            return None
        normalized = _normalize_name(name)
        if normalized == name:
            return None
        try:
            from jedi_auth.domain.entity.org_unit_query_entity import OrgUnitQueryEntity
            q = OrgUnitQueryEntity(name=normalized)
            if tenant_id is not None:
                q.tenant_id = tenant_id
            orgs = self._org.get_org_units(q)
        except Exception:
            return None
        return orgs[0] if orgs else None

    def _try_fuzzy_match(self, parsed, tenant_id):
        name = (parsed.name or "").strip()
        if not name:
            return (None, MatchMethod.UNMATCHED)
        normalized = _normalize_name(name)
        stripped = _strip_org_suffix(normalized)
        if stripped == normalized:
            return (None, MatchMethod.UNMATCHED)

        def _loader():
            from jedi_auth.domain.entity.org_unit_query_entity import OrgUnitQueryEntity
            q = OrgUnitQueryEntity()
            if tenant_id is not None:
                q.tenant_id = tenant_id
            return self._org.get_org_units(q)

        candidates = self._get_or_load_candidates(_loader)
        for o in candidates:
            o_name = _normalize_name(getattr(o, "name", None))
            if _strip_org_suffix(o_name) == stripped:
                return (o, MatchMethod.FUZZY_NAME_PREFIX)
        return (None, MatchMethod.UNMATCHED)

git add domain/oscal/service/reconciliation/person_reconciler.py \
        domain/oscal/service/reconciliation/organization_reconciler.py \
        tests/test_a3_reconciliation_person.py \
        tests/test_a3_reconciliation_organization.py

git commit -m "$(cat <<'EOF'
feat(oscal): A3 T4 fuzzy match algorithm

PersonReconciler 補 normalized stage（+alias 移除）+ fuzzy stage
（email domain + nickname/login_name 對齊）；
OrganizationReconciler 補 normalized stage（全形半形 / collapse）+
fuzzy stage（剔除尾綴）。

Client-side filter — 拉全 tenant list cache 在 reconciler instance
lifetime；exception 不寫入 cache（next call retry）。
ParsedParty 帶 match_method + match_confidence。

對齊 design-A3.md §5.1 / §5.2 / §5.4。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"

from dataclasses import asdict
import json
from domain.oscal.parser.ssp_intermediate import ParsedParty
from domain.oscal.service.reconciliation.match_method import MatchMethod


def test_default_match_method_unmatched():
    p = ParsedParty(name="A", party_type="person", role=None, email_address=None)
    assert p.match_method == MatchMethod.UNMATCHED
    assert p.match_confidence == 0.0

def test_set_match_method_explicit():
    p = ParsedParty(name="A", party_type="person", role=None, email_address=None,
                    match_method=MatchMethod.FUZZY_EMAIL_DOMAIN, match_confidence=0.7)
    assert p.match_method == "fuzzy_email_domain"  # StrEnum value 比較

def test_asdict_serializes_match_method_as_string_value():
    """StrEnum 在 dataclasses.asdict() 出來值是 enum instance；JSON dump 後是字串值。
    A5 預覽 UI 反序列化拿到的必須是 "unmatched" 不是 "MatchMethod.UNMATCHED"。
    """
    p = ParsedParty(name="A", party_type="person", role=None, email_address=None)
    d = asdict(p)
    # asdict 行為：StrEnum 仍是 enum instance（不是字串）
    assert d["match_method"] == MatchMethod.UNMATCHED
    # JSON dump 過後變字串 — JSONB 寫入 → A5 UI 反序列化拿到的是這個
    json_str = json.dumps(d, default=str)
    parsed_back = json.loads(json_str)
    assert parsed_back["match_method"] == "unmatched"
    assert parsed_back["match_confidence"] == 0.0

def test_fuzzy_method_jsonb_value():
    """同上 — FUZZY_EMAIL_DOMAIN 序列化後是 "fuzzy_email_domain" 字串"""
    p = ParsedParty(name="A", party_type="person", role=None, email_address=None,
                    match_method=MatchMethod.FUZZY_EMAIL_DOMAIN, match_confidence=0.7)
    d = asdict(p)
    json_str = json.dumps(d, default=str)
    parsed_back = json.loads(json_str)
    assert parsed_back["match_method"] == "fuzzy_email_domain"
    assert parsed_back["match_confidence"] == 0.7

pytest tests/ -v --ignore=tests/legacy 2>&1 | tail -50

git add tests/test_a3_reconciliation_person.py \
        tests/test_a3_reconciliation_organization.py \
        tests/test_a3_parsed_party_fields.py

git commit -m "$(cat <<'EOF'
test(oscal): A3 T5 unit + integration tests

補齊 PersonReconciler / OrganizationReconciler 邊界 case；
新增 ParsedParty 加欄位 + asdict JSONB serialize 的 4 個 fixture test；
累計 A3 新增 ~51 unit + integration test 全綠（13 normalizers + 7 base
+ 12 person + 10 org + 3 facade + 4 A2 integration + 4 ParsedParty
+ ~2 matrix gap）；
既有 213 test 不破壞 → 累計 ~264 BE test 全綠。

對齊 design-A3.md §8 acceptance criteria（原寫 ~38 漏算 normalizers helper
test，§11 reconciliation 補一條說明）。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"

cd ~/Projects/Billows/Audit-Manager/compliance-manager-be
git status --short

cd ~/Projects/Billows/Audit-Manager/compliance-manager-test
git status
git checkout -b feature/a3-ssp-party-match-regression  # 或沿用既有 phase 2 branch

cd ~/Projects/Billows/Audit-Manager/compliance-manager-test
npm run cucumber -- features/regression/module-frame/05-ssp-docx-import-party-match.feature

cd ~/Projects/Billows/Audit-Manager/compliance-manager-test
git add features/regression/module-frame/05-ssp-docx-import-party-match.feature \
        features/regression/module-frame/steps/ssp-docx-import-party-match.steps.js

git commit -m "$(cat <<'EOF'
test(compliance-manager-test): A3 T6 cucumber party match regression

新增 5 scenarios cover docx import party match（4 exact/normalized + 1 fuzzy）；
A3 ship 後 docx import party match regression baseline = 這 5 個 scenarios。

對齊 compliance-manager-be design-A3.md §6.5。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"

---
type: feat
breaking: false
modules: [oscal, ssp-import]
commit: <T1 + T2 + T3 + T4 commit hash chain>
---

# feat: 共用 matcher 抽取 — parties / org-units (A3)

## 需求說明

A2 ship 後，docx + Excel 兩條 SSP 匯入 path 都需要 ParsedParty → 系統
user / org_unit 鉤稽；A2 phase 跳過串接、行為跟 docx 不一致。A3 抽出共用
`BaseReconciliationService` 並補上 A2 串接、引入 fuzzy 中間狀態。

## 變更範圍

### 新建檔案

- domain/oscal/service/reconciliation/ sub-folder（5 個檔 + 1 個 helper）
- tests/test_a3_reconciliation_*.py + tests/test_a3_parsed_party_fields.py（7 個檔，共 ~51 個 case）

### 改動既有檔案

- domain/oscal/service/party_reconciliation_service.py — 改門面（公開 method 簽章不變）
- domain/oscal/parser/ssp_intermediate.py — ParsedParty 加 match_method + match_confidence
- app/oscal/service/ssp_excel_import_app_service.py — _confirm_*_flow 補 reconcile() 呼叫
- di_containers/oscal/oscal_containers.py — 加 person_reconciler / organization_reconciler Factory

### 跨 repo

- compliance-manager-test — 新增 cucumber regression `05-ssp-docx-import-party-match.feature` 含 5 scenarios

## 行為差異

| Caller | A3 前 | A3 後 |
|--------|-------|-------|
| docx flow parse-time | exact match only | exact + normalized + fuzzy 三階段 |
| docx flow confirm-time | 同上 | 同上 |
| A2 superset flow | **不呼叫** reconcile | 呼叫 reconcile，fuzzy 自動寫入 |
| A2 update flow | **不呼叫** reconcile | 呼叫 reconcile，fuzzy 自動寫入 |

## 測試結果

- A3 新增 ~51 unit + integration test 全綠（13 normalizers + 7 base + 12 person + 10 org + 3 facade + 4 A2 integration + 4 ParsedParty + ~2 matrix gap）
- 既有 213 BE test（A1 106 + A2 99 + party 8）全綠
- 累計 ~264 BE test passed
- Cucumber regression：5 scenarios 撰寫完成（**partial ship**：env 配齊後才能跑）

## A3 ship 後 regression baseline

任何後續 phase 動到 PartyReconciliationService / PersonReconciler /
OrganizationReconciler / fuzzy 演算法時，cucumber 5 scenarios 必過 + 既有 8
party unit test 必過。

## Follow-up

- F1：docx parser role normalize map（issue 修補 B）— A3 後獨立 commit
- F2：A5 phase 補 A2 fuzzy 的 user 拍板路徑
- F4：jedi-auth Query Entity 加 _in_email_domain（prod 大 tenant 才考慮）
- T8（cucumber partial ship）：env 配齊後跑 5 scenarios

## 參考

- docs/features/FR-011.2-2605-ssp-import-export-phase2/design-A3.md（13 章）
- docs/features/FR-011.2-2605-ssp-import-export-phase2/implementation-plan-A3.md

git add docs/changelog/2026-05-XX-feat-party-matcher-shared.md \
        docs/features/FR-011.2-2605-ssp-import-export-phase2/README.md \
        docs/features/FR-011.2-2605-ssp-import-export-phase2/design-A3.md \
        docs/features/FR-011.2-2605-ssp-import-export-phase2/handoff/2026-05-XX-a3-SUMMARY.md \
        docs/conversation-history/2026-05-XX/ssp-import-export-phase2/

git commit -m "$(cat <<'EOF'
docs(ssp-import-export-phase2): A3 收尾 + SUMMARY

- changelog feat-party-matcher-shared.md
- tracker README A3 row → BE shipped + commit chain
- design-A3.md §11 implementation reality reconciliation
- handoff A3 SUMMARY（task arc 收口）
- conversation-history dump（A3 task arc 全對話原文）

A3 phase 完整收口；A4 phase 可在 BaseReconciliationService 上擴
device / info_system / leveraged / catalog control / AO 5 種新 reconciler。

對齊 design-A3.md §8 + §12 收口規範 / CLAUDE.md 做 summary 段。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"