hyungi_document_server/scripts/verify_statute_chain.py

"""B-1 PR③ — 법령 버전 체인 검증 3술어 (plan safety-library-1).

read-only 진단 — E-1 관찰의 법령 게이트 도구로도 재사용 (반복 실행 안전).

검증 3술어 (R7-M1, B-1 단일 정본):
  ① 존재성 — watch family 각각 primary 시리즈 current 정확 1건(0건도 위반)
              + annex 시리즈당 current ≤ 1
  ② 노출 유일성 — primary current 보유 family당 primary 노출(체인+레거시 매핑 합산) 정확 1건
                  (모집단 = primary current 보유 family 한정 — R8-M2)
  ③ 고아 그물 — law_monitor in_corpus=true 레거시 중:
                (a) current 보유 family 에 매핑되는데 안 flip 된 것(flip 누락) = 0
                (b) 어느 watch family 에도 매핑 안 되는 것(제명 개정 등 매핑 구멍) = 0
                repealed family·primary current 미보유 family 의 레거시 보존은 위반 아님

repealed family 는 ①② 기대값 0 으로 면제.

실행:
  docker compose exec -T fastapi python /app/scripts/verify_statute_chain.py
종료코드: 0 = 전건 PASS, 1 = 위반 (CI/관찰 게이트 용)
"""

import asyncio
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))

from collections import defaultdict

from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import create_async_engine

from workers.statute_collector import legacy_law_name, normalize_law_name, series_suffix


async def main() -> int:
    db_url = os.getenv("DATABASE_URL", "postgresql+asyncpg://pkm:pkm@localhost:5432/pkm")
    engine = create_async_engine(db_url)
    violations: list[str] = []

    async with engine.connect() as conn:
        # ── 로드 ──
        acts = (await conn.execute(text(
            "SELECT family_id, title, repeal_detected_at IS NOT NULL AS repealed "
            "FROM legal_acts WHERE watch"))).all()
        metas = (await conn.execute(text(
            "SELECT family_id, law_doc_kind, version_key, version_status FROM legal_meta"))).all()

        act_title = {a.family_id: a.title for a in acts}
        repealed = {a.family_id for a in acts if a.repealed}
        active = [a for a in acts if not a.repealed]

        # family → primary current 수 / annex 시리즈별 current 수
        prim_current = defaultdict(int)
        annex_series_current = defaultdict(int)
        for m in metas:
            if m.version_status != "current":
                continue
            if m.law_doc_kind == "primary":
                prim_current[m.family_id] += 1
            else:
                annex_series_current[(m.family_id, series_suffix(m.version_key))] += 1

        # ── ① 존재성 ──
        for a in active:
            n = prim_current[a.family_id]
            if n != 1:
                violations.append(f"① {a.family_id} ({a.title}): primary current {n}건 (정확 1 기대)")
        for (fid, suf), n in annex_series_current.items():
            if fid not in repealed and n > 1:
                violations.append(f"① {fid} annex 시리즈 {suf}: current {n}건 (≤1 기대)")

        # ── ③ 고아 그물 (정규화 동등 매핑) ──
        # watch family 정규화명 → family_id (current 보유 여부 동반)
        norm_to_fid = {}
        for a in active:
            norm_to_fid[normalize_law_name(a.title)] = a.family_id

        legacy = (await conn.execute(text(
            "SELECT d.id, d.title, "
            "  EXISTS(SELECT 1 FROM document_chunks c WHERE c.doc_id=d.id AND c.in_corpus) AS exposed "
            "FROM documents d WHERE d.source_channel='law_monitor' AND d.deleted_at IS NULL"))).all()

        orphan_flip_miss = 0
        orphan_unmapped = 0
        unmapped_names = set()
        for row in legacy:
            if not row.exposed:
                continue  # in_corpus=false = 정상 (스윕됨 or 청크 없음)
            name = legacy_law_name(row.title or "")
            norm = normalize_law_name(name) if name else None
            fid = norm_to_fid.get(norm) if norm else None
            if fid is None:
                orphan_unmapped += 1
                if name:
                    unmapped_names.add(name)
            elif prim_current.get(fid, 0) >= 1:
                # current 보유 family 인데 레거시가 노출 중 = flip 누락
                orphan_flip_miss += 1
        if orphan_flip_miss:
            violations.append(f"③(a) flip 누락: current 보유 family 의 노출 레거시 {orphan_flip_miss}건")
        if orphan_unmapped:
            violations.append(
                f"③(b) 무매핑 노출 레거시 {orphan_unmapped}건 — 매핑 구멍(매핑 보강 신호): "
                + ", ".join(sorted(unmapped_names))[:200])

        # ── ② 노출 유일성 (primary current 보유 family 한정) ──
        # 노출 primary = 체인 primary current(=1) + 레거시 매핑 노출분.
        # ③(a)=0 이면 레거시 노출분 0 → 체인 1건만 = 정확 1. 별도 위반 추출은 ③(a)에 포함됨.
        # (annex 노출 비동기 일반화는 may — Phase 1 미적용)

        # ── 상태 요약 출력 ──
        print("=== 법령 체인 검증 (B-1 PR③ 3술어) ===")
        print(f"watch family: {len(acts)} (active {len(active)}, repealed {len(repealed)})")
        print(f"primary current 보유 family: {sum(1 for a in active if prim_current[a.family_id]==1)}/{len(active)}")
        print(f"annex current 시리즈: {len(annex_series_current)}")
        exposed_legacy = sum(1 for r in legacy if r.exposed)
        print(f"레거시 law_monitor: {len(legacy)}건 (in_corpus 노출 {exposed_legacy}건)")
        print()

    await engine.dispose()

    if violations:
        print(f"[FAIL] 위반 {len(violations)}건:")
        for v in violations:
            print("  -", v)
        return 1
    print("[PASS] 3술어 전건 통과 (존재성·노출 유일성·고아 그물)")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))