solar_eval/pipelines/rules.py

"""multi_step 파이프라인의 rule 스텝 구현체.

`multi_step.py` 가 518줄이라 (`.agents/rules/code-style.md` 의 400줄 권장선을
넘어) 룰 구현만 따로 뽑아냈다. vocabulary/post_process/correction_filter 는
동작을 바꾸지 않고 `MultiStepPipeline._apply_*` 메서드에서 그대로 옮겨온
것이다 — 순수 텍스트 변환 로직이라 `self` 상태가 필요 없다(`apply_vocabulary`
만 예외 — 데이터셋에서 로드한 어휘 캐시를 인자로 받고 갱신된 캐시를 되돌려준다).

`apply_judge_decisions`/`apply_change_rate_guard` 는 v24 계열 포팅 단계(§5-F)에서
추가됐다. 전자는 `EvalSample.artifacts` 를 읽어야 해서(judge 스텝의 구조화 결정,
self-consistency 의 corrections 목록) `sample` 을 인자로 받는다 — 다른 rule
함수들과 시그니처가 다른 유일한 이유다.
"""

from __future__ import annotations

import difflib
import logging
import re
from pathlib import Path
from typing import TYPE_CHECKING, Any

from solar_eval.models.sample import EvalSample
from solar_eval.pipelines.errors import KnowledgeAssetError
from solar_eval.pipelines.pipeline import ResourceRef, relative_posix

if TYPE_CHECKING:
    from solar_eval.core.dataset_loader import DatasetLoader

logger = logging.getLogger(__name__)


def apply_vocabulary(
    text: str,
    config: dict[str, Any],
    vocabulary: list[dict[str, str]] | None,
    dataset_loader: DatasetLoader | None,
    config_dir: Path | None = None,
) -> tuple[str, list[dict[str, str]] | None]:
    """`vocabulary_sub`/`vocabulary_reapply` rule_type 구현.

    호출자(`VocabularyStep`)가 run 수명 캐시를 들고 있으므로, 최초 호출 시 로드한
    어휘 목록을 반환값으로 되돌려줘 캐시로 저장하게 한다.

    Args:
        config_dir: 레포의 config 정본 디렉토리(`03-evaluation`). 치환 사전은
            데이터가 아니라 **설정**이라(바뀌면 같은 문제를 다르게 푼다) 여기를
            먼저 찾는다. 데이터 루트 기준 레거시 선언도 계속 지원한다.

    Returns:
        (치환된 text, 갱신된 vocabulary 캐시) 튜플.

    Raises:
        KnowledgeAssetError: `vocab_dataset` 을 선언했는데 못 읽을 때.
    """
    if vocabulary is None:
        vocabulary = _load_vocabulary(config, dataset_loader, config_dir)

    for row in vocabulary:
        wrong = row.get("wrong", row.get("before", ""))
        correct = row.get("correct", row.get("after", ""))
        if wrong and correct:
            text = text.replace(wrong, correct)

    return text, vocabulary


def _load_vocabulary(
    config: dict[str, Any],
    dataset_loader: DatasetLoader | None,
    config_dir: Path | None,
) -> list[dict[str, str]]:
    """치환 사전을 로드한다 — config 정본 우선, 데이터 루트 폴백.

    선언이 없으면 빈 목록(정상적인 no-op)이고, 선언했는데 못 읽으면 예외다.
    """
    vocab_path = config.get("vocab_dataset", "")
    if not vocab_path:
        return []

    searched: list[str] = []

    # 1. config 정본 (레포의 03-evaluation/) — 치환 사전은 설정이다.
    if config_dir is not None:
        candidate = _config_candidate(config_dir, vocab_path)
        searched.append(str(candidate))
        if candidate.is_file() and dataset_loader is not None:
            return dataset_loader.load_csv(str(config_dir), vocab_path, source="local")

    # 2. 레거시 선언 (repo/source 로 데이터 루트나 HF 를 가리키는 경우)
    repo = config.get("repo", "")
    source = config.get("source", "huggingface")
    if dataset_loader is not None:
        searched.append(f"{source}:{repo}/{vocab_path}" if repo else f"{source}:{vocab_path}")
        try:
            return dataset_loader.load_csv(repo, vocab_path, source=source)
        except Exception as e:  # noqa: BLE001 — 원인을 KnowledgeAssetError 에 실어 다시 던진다
            logger.debug("vocabulary load failed (%s): %s", vocab_path, e)
    else:
        searched.append("(dataset_loader 없음)")

    raise KnowledgeAssetError(vocab_path, searched)


def _config_candidate(config_dir: Path | str, rel: str) -> Path:
    """config 정본 기준 자산 경로 — 로더(`_load_vocabulary`)와 스냅샷 해석이 공유한다.

    config 자산은 config_dir **안**에 있어야 한다: `../` 나 밖을 가리키는 심볼릭 링크로 풀리는
    파일은 커밋으로 복원되지 않고(dirty 범위 밖), 스냅샷 라벨이 기계마다 다른 절대경로가 된다.
    그런 파일이 실제로 있으면 쓰지 않고 `KnowledgeAssetError` 로 거부한다(로더·스냅샷 공통).
    없으면 경로만 돌려준다 — 호출자가 레거시 선언으로 넘어가거나 같은 오류를 낸다.

    Raises:
        KnowledgeAssetError: 후보 파일이 있는데 실제 위치가 config_dir 밖일 때.
    """
    candidate = Path(config_dir) / rel
    if candidate.is_file():
        resolved = candidate.resolve()
        root = Path(config_dir).resolve()
        if not resolved.is_relative_to(root):
            raise KnowledgeAssetError(
                rel, [f"{candidate} → {resolved} (config_dir {root} 밖 — config 자산은 그 안에)"]
            )
    return candidate


def resolve_vocabulary_path(
    config: dict[str, Any],
    dataset_loader: DatasetLoader | None,
    config_dir: Path | None,
) -> ResourceRef | None:
    """치환 사전 선언을 **읽지 않고** 어느 파일인지로 푼다 (run.json 스냅샷의 `resources`).

    `_load_vocabulary` 와 같은 순서를 따른다: config 정본 → 레거시 선언(repo/source).
    레거시 `local` 은 로더의 경로 규칙(`DatasetLoader._resolve_local_path`)을 그대로 쓰고
    라벨은 데이터 루트 상대(`data:`), 원격(`huggingface` 등)은 네트워크를 타지 않고
    선언만으로 `hf:<repo>/<path>`(path None)를 낸다.

    Returns:
        선언이 없으면 None(정상적인 no-op).

    Raises:
        KnowledgeAssetError: 선언했는데 로컬에서 찾을 수 없을 때 — 로더가 run 도중에 낼
            오류를 run 시작 전으로 앞당긴다.
    """
    vocab_path = config.get("vocab_dataset", "")
    if not vocab_path:
        return None
    searched: list[str] = []
    if config_dir is not None:
        candidate = _config_candidate(config_dir, vocab_path)
        searched.append(str(candidate))
        if candidate.is_file() and dataset_loader is not None:
            return ResourceRef("config", relative_posix(candidate, config_dir), candidate)
    if dataset_loader is None:
        searched.append("(dataset_loader 없음)")
        raise KnowledgeAssetError(vocab_path, searched)
    repo = config.get("repo", "")
    source = config.get("source", "huggingface")
    if source != "local":
        if not repo:
            # repo 없는 원격 선언은 로더가 run 도중 다운로드에 실패한다(prod_251231 의 옛 사고) —
            # 스냅샷이 null 로 넘기지 않고 시작 전에 같은 오류로 죽인다.
            searched.append(f"{source}:{vocab_path}")
            raise KnowledgeAssetError(vocab_path, searched)
        return ResourceRef("hf", f"{repo}/{vocab_path}", None)
    local = dataset_loader._resolve_local_path(repo, vocab_path)
    searched.append(str(local))
    if not local.is_file():
        raise KnowledgeAssetError(vocab_path, searched)
    return ResourceRef("data", relative_posix(local, dataset_loader.base_dir), local)


def apply_pronoun_postprocess(text: str, replacements: dict[str, str]) -> str:
    """대명사/직함 치환 + 한국어 조사 교정.

    단어를 치환할 때(예: "이재명 대표" → "이재명 대통령") 뒤따르는 조사도
    새 단어의 받침에 맞춰 바꿔야 한다.
    """
    # 모음으로 끝나는 조사 → 자음으로 끝나는 단어 뒤에 붙는 대응 조사
    correction_map = {
        "는": "은",
        "가": "이",
        "를": "을",
        "와": "과",
        "로": "으로",
        "여": "이여",
        "라": "이라",
        "랑": "이랑",
        "다": "이다",
        "였다": "이었다",
        "라면": "이라면",
        "라서": "이라서",
        "로부터": "으로부터",
    }
    all_particles = (
        list(correction_map.keys()) + list(correction_map.values()) + ["만", "도", "께서"]
    )
    particle_pattern = "|".join(re.escape(p) for p in sorted(all_particles, key=len, reverse=True))

    for old, new in replacements.items():
        regex = re.compile(f"({re.escape(old)})({particle_pattern})?")

        def replacer(m: re.Match, new_word: str = new) -> str:
            particle = m.group(2)
            if particle:
                return new_word + correction_map.get(particle, particle)
            return new_word

        text = regex.sub(replacer, text)
    return text


def apply_currency_compact(text: str) -> str:
    """통화 표현에서 숫자와 단위 사이 공백을 제거한다."""
    _unit = r"(원|명|개|대|위안|달러|유로|엔|톤|권|건|회|장|마리|그루|송이|필)"
    _tail = r"(?=[\s.,!?()\"\'가-힣]|$)"
    # "3억 5000만 원" → "3억5000만원"
    text = re.sub(
        r"(\d+(?:\.\d+)?)(만|억|조)\s*(\d+)\s*" + _unit + _tail,
        r"\1\2\3\4",
        text,
    )
    # "3억 원" → "3억원"
    text = re.sub(
        r"(\d+(?:\.\d+)?)(만|억|조)\s*" + _unit + _tail,
        r"\1\2\3",
        text,
    )
    return text


def apply_comma_removal(text: str) -> str:
    """숫자 사이의 천단위 콤마를 제거한다 (1,000 → 1000)."""
    return re.sub(r"(?<=\d),(?=\d)", "", text)


def apply_correction_filter(text: str, config: dict[str, Any], original: str = "") -> str:
    """Filter corrections by comparing with original text.

    Keeps only 'safe' corrections (spacing changes, small character diffs).
    Reverts larger changes that are likely unnecessary expression substitutions.

    Config options:
        max_char_diff: Maximum character differences to allow (default: 2)
        allow_spacing: Keep spacing-only changes (default: true)
    """
    from solar_eval.evaluators.scoring import compute_lcs

    if not original:
        return text

    max_cd = config.get("max_char_diff", 2)
    allow_spacing = config.get("allow_spacing", True)

    orig_tokens = original.split()
    pred_tokens = text.split()

    if orig_tokens == pred_tokens:
        return text

    lcs_pairs = compute_lcs(orig_tokens, pred_tokens)

    def is_safe(o_text: str, c_text: str) -> bool:
        if allow_spacing and o_text.replace(" ", "") == c_text.replace(" ", ""):
            return True
        if abs(len(o_text) - len(c_text)) <= 1:
            d = sum(1 for a, b in zip(o_text, c_text) if a != b)
            d += abs(len(o_text) - len(c_text))
            if d <= max_cd:
                return True
        return False

    result_tokens: list[str] = []
    prev_orig = -1
    prev_pred = -1

    for oi, pi in lcs_pairs:
        orig_gap = orig_tokens[prev_orig + 1 : oi]
        pred_gap = pred_tokens[prev_pred + 1 : pi]
        if orig_gap or pred_gap:
            o_text = " ".join(orig_gap)
            p_text = " ".join(pred_gap)
            if o_text == p_text or is_safe(o_text, p_text):
                result_tokens.extend(pred_gap)
            else:
                result_tokens.extend(orig_gap)
        result_tokens.append(orig_tokens[oi])
        prev_orig = oi
        prev_pred = pi

    orig_tail = orig_tokens[prev_orig + 1 :]
    pred_tail = pred_tokens[prev_pred + 1 :]
    if orig_tail or pred_tail:
        o_text = " ".join(orig_tail)
        p_text = " ".join(pred_tail)
        if o_text == p_text or is_safe(o_text, p_text):
            result_tokens.extend(pred_tail)
        else:
            result_tokens.extend(orig_tail)

    return " ".join(result_tokens)


def apply_judge_decisions(text: str, config: dict[str, Any], sample: EvalSample) -> str:
    """`tool_calling_judge` 의 verdict="revert" 판정을 텍스트에 되돌린다.

    `judge_step`(기본 "tool_calling_judge")의 판정 목록과 `corrections_step`
    (기본 "basic_correction")의 corrections 목록을 `sample.artifacts` 에서 읽는다
    — 이 두 artifact 는 각각 `pipelines/judge.py::run_tool_calling_judge` 와
    `MultiStepPipeline` 의 self-consistency 경로가 채운다. 어느 한쪽이라도 비어
    있으면(judge 스텝이 없었거나 corrections 가 하나도 없었던 경우) 조용히
    원문(text)을 그대로 돌려준다 — revert 할 대상 자체가 없는 정상 상태다.

    HF 구현(`huggingface/pipelines_v24.py::apply_judge_decisions_for_bulk`)을
    포팅. bulk 단위였던 원본과 달리 여기서는 문서 전체 text 에 적용한다
    (§5-F 설계 결정 — 이 엔진은 tool_calling_judge/basic_correction 을 이미
    문서 단위로 처리하므로 bulk 정렬 문제 자체가 없다).
    """
    judge_step = config.get("judge_step", "tool_calling_judge")
    corrections_step = config.get("corrections_step", "basic_correction")
    decisions = sample.artifacts.get("judge_decisions", {}).get(judge_step, [])
    corrections = sample.artifacts.get("corrections", {}).get(corrections_step, [])
    if not decisions or not corrections:
        return text

    revert_pairs: list[tuple[str, str]] = []
    for d in decisions:
        if not isinstance(d, dict) or d.get("verdict") != "revert":
            continue
        idx = d.get("correction_index")
        if not isinstance(idx, int) or idx < 0 or idx >= len(corrections):
            continue
        entry = corrections[idx]
        if not isinstance(entry, dict):
            continue
        corrected = str(entry.get("corrected", ""))
        original = str(entry.get("original", ""))
        if corrected and original and corrected != original:
            revert_pairs.append((corrected, original))

    # 긴 것부터 되돌려야 짧은 교정이 긴 교정의 부분 문자열일 때 잘못 되돌리지 않는다.
    revert_pairs.sort(key=lambda p: -len(p[0]))
    for corrected, original in revert_pairs:
        if corrected in text:
            text = text.replace(corrected, original, 1)
    return text


def apply_change_rate_guard(text: str, original: str, config: dict[str, Any]) -> str:
    """원문 대비 토큰 변경 비율이 `max_change_rate` 를 넘으면 원문으로 되돌리는 안전망.

    v17 인사이트 문서(`insights/260504_solar-pro2_essentials.md`)가 "change_rate_guard
    (rule, bulk-level safety net, max_change_rate: 0.45)" 라고만 설명하고, 실제
    구현은 HF 아카이브(`huggingface/pipelines.py`/`pipelines_v24.py`) 어디에도 없다
    (레포 전체 검색으로 확인). 그래서 이 함수는 그 설명만 보고 새로 작성한 것이다
    — v17 수치 재현을 약속하지 않는다는 전제로 포팅 대상에 포함됐다
    (`.agents/01-plans/harness/2026-08-09-chosun-proofread-evalsample-migration.md` §9-3).

    변경 비율은 `evaluators/scoring.compute_lcs` 로 원문·현재 텍스트의 토큰 단위
    LCS 를 구해 `1 - |LCS| / |원문 토큰|` 으로 정의한다 — 이 프로젝트의 다른 모든
    교열 diff 로직(`compute_proofread_metrics`, `apply_correction_filter`)과 같은
    척도를 쓴다.
    """
    from solar_eval.evaluators.scoring import compute_lcs

    max_rate = config.get("max_change_rate", 0.45)
    orig_tokens = original.split()
    if not orig_tokens:
        return text

    text_tokens = text.split()
    lcs_pairs = compute_lcs(orig_tokens, text_tokens)
    change_rate = 1 - (len(lcs_pairs) / len(orig_tokens))

    if change_rate > max_rate:
        logger.warning(
            "[change_rate_guard] change_rate=%.3f > max=%.3f, reverting to original",
            change_rate,
            max_rate,
        )
        return original
    return text


_UNIT_SYMBOLS = {
    "mm²": "㎟",
    "cm²": "㎠",
    "km²": "㎢",
    "m²": "㎡",
    "mm": "㎜",
    "cm": "㎝",
    "km": "㎞",
    "mg": "㎎",
    "kg": "㎏",
    "ml": "㎖",
    "dl": "㎗",
    "cc": "㏄",
}
# 숫자(공백 하나까지 허용) 바로 뒤에 오고 뒤에 영문자가 이어지지 않는 단위만 바꾼다. 예전에는 글자열 어디서나 바꿔
# `Grammy`·`Programme`·`Baccalaureate`·`Nadler`·메일 주소의 mm·cc·dl 까지 ㎜·㏄·㎗ 로 망가뜨렸다(2026-10-20 현업 사례 비교).
_UNIT_RE = re.compile(
    r"(?<=\d)(\s?)("
    + "|".join(sorted(map(re.escape, _UNIT_SYMBOLS), key=len, reverse=True))
    + r")(?![A-Za-z])"
)


def apply_unit_unicode(text: str) -> str:
    """숫자 뒤 단위 텍스트를 유니코드 결합 문자로 변환한다(`10mm`→`10㎜`, `5 kg`→`5 ㎏`)."""
    return _UNIT_RE.sub(lambda m: m.group(1) + _UNIT_SYMBOLS[m.group(2)], text)


_QUOTE_KINDS = (("'‘’", "‘", "’"), ('"“”', "“", "”"))
_CLOSES_AFTER = ".,!?;%…’”"
# 앞 글자만으로는 여닫이를 정할 수 없는 기호: 같은 줄에 열린 따옴표가 있거나 뒤가 공백·줄 끝이면 닫는 쪽, 아니면 여는 쪽.
# `‘AA-’`·`‘송도는~’`·`‘개(犬)’` 은 닫고, 인터뷰 질문 머리 `-‘꽃’`·`[인터뷰]“딥 리딩”` 은 연다.
_AMBIGUOUS_BEFORE = "-+~)]}>」』"


def apply_quote_direction(text: str, original: str = "") -> str:
    """둥근 따옴표의 여닫이 방향을 앞 글자와 같은 줄의 짝으로 다시 정한다.

    FT 교열 모델이 곧은 따옴표(')를 조선일보 표기인 둥근 따옴표로 바꾸면서 여는 자리에 닫는 따옴표(’)를 넣거나 한쪽만 바꾸는
    사고를 낸다(2026-10-21 FT 약점 분석: 정답 밖 오교정의 약 1/4). 둥근 따옴표가 있는 줄에서만 그 줄의 같은 종류 따옴표를
    모두 둥근 따옴표로 맞추고, 글자·숫자·닫는 문장부호 뒤면 닫는(’ ”), 그 밖(줄 처음·공백·`◇`·`▶` 같은 머리 기호·여는 괄호·
    여는 따옴표 뒤)은 여는(‘ “) 쪽으로 둔다. `-`·`+`·`~` 뒤는 같은 줄에 열린 따옴표가 있거나 뒤가 공백·줄 끝일 때만 닫는다.
    열린 작은따옴표 뒤 숫자에 붙은 프라임(′)은 FT 가 닫는 따옴표를 바꾼 것이라 ’ 로 되돌린다(`‘하야부사2′가`). 단 여는
    따옴표부터 그 프라임까지가 `original` 에 그대로 있으면 둔다 - 조선일보 원문·게재본에는 `‘춘천사이로248′` 같은 표기가
    있고 교열자도 고치지 않는다(2026-10-27 기사 셋 확인).
    둥근 따옴표가 없는 줄의 곧은 따옴표는 건드리지 않는다.
    """
    lines = []
    for line in text.split("\n"):
        chars = list(line)
        for kinds, opening, closing in _QUOTE_KINDS:
            if opening not in line and closing not in line:
                continue
            # 아직 닫히지 않은 여는 따옴표 자리(겹따옴표 `‘…(‘…’)’` 의 바깥쪽을 잃지 않도록)
            opened: list[int] = []
            for i, ch in enumerate(chars):
                prev = chars[i - 1] if i else ""
                nxt = chars[i + 1] if i + 1 < len(chars) else ""
                if ch == "′" and opening == "‘":
                    if opened and prev.isdigit() and not (nxt.isdigit() or nxt == "″"):
                        # 여는 따옴표 글자는 원문과 다를 수 있어(`’1988KY26′`) 안쪽 글자 + 프라임으로 원문을 찾는다.
                        inner = "".join(chars[opened.pop() + 1 : i + 1])
                        chars[i] = ch if inner in original else closing
                    continue
                if ch not in kinds:
                    continue
                if prev and prev in _AMBIGUOUS_BEFORE:
                    closes = bool(opened) or not nxt or nxt.isspace()
                else:
                    closes = bool(prev) and (prev.isalnum() or prev in _CLOSES_AFTER)
                chars[i] = closing if closes else opening
                if closes:
                    if opened:
                        opened.pop()
                else:
                    opened.append(i)
        lines.append("".join(chars))
    return "\n".join(lines)


# 따옴표 글자를 종류별 대표 글자로 접어, 따옴표 모양만 다른 자리를 같은 글자로 정렬한다(′ ″ 는 FT 가 닫는 따옴표 대신 넣는다).
_QUOTE_FOLD = str.maketrans({"‘": "'", "’": "'", "′": "'", "“": '"', "”": '"', "″": '"'})


def apply_quote_preserve(text: str, original: str) -> str:
    """원문에 있던 따옴표 글자를 원문 모양 그대로 되돌린다.

    조선일보 교열자는 따옴표 모양(곧은·둥근·여닫이·프라임)을 고치지 않는다 - 기사 셋 508 · 일반 문단 541 의 원문 → 정답에서
    따옴표 글자만 바뀐 자리는 0 이다. 반면 FT 는 곧은 따옴표를 둥근 따옴표로 바꾸고 방향·프라임 사고를 내서, FT+Opus 출력에는
    따옴표만 바뀐 자리가 기사 셋 508편 중 78편에서 246곳 있었다(2026-10-27). 따옴표를 종류별로 접은 글자열끼리 정렬해 원문과 맞물리는
    따옴표 자리만 원문 글자로 돌린다. 옆 글자 교정(`‘하야부사2호′`의 `호` 삭제)은 남고, 원문에 없던 따옴표는 그대로 둔다.
    글자 정렬은 줄 단위로 먼저 짝을 지은 뒤 바뀐 줄 묶음 안에서만 한다(기사 전체를 글자로 정렬하면 긴 기사에서 1초 가까이 걸린다).
    """
    if not original or text == original:
        return text
    orig_lines, text_lines = original.split("\n"), text.split("\n")
    out_lines = list(text_lines)
    matcher = difflib.SequenceMatcher(
        None,
        [line.translate(_QUOTE_FOLD) for line in orig_lines],
        [line.translate(_QUOTE_FOLD) for line in text_lines],
        autojunk=False,
    )
    for op, i1, i2, j1, j2 in matcher.get_opcodes():
        if op == "equal":
            # 접은 줄이 같으면 따옴표 글자만 다르다 - 원문 줄 그대로.
            out_lines[j1:j2] = orig_lines[i1:i2]
        elif op == "replace":
            block = _restore_quotes("\n".join(text_lines[j1:j2]), "\n".join(orig_lines[i1:i2]))
            out_lines[j1:j2] = block.split("\n")
    return "\n".join(out_lines)


def _restore_quotes(text: str, original: str) -> str:
    out = list(text)
    matcher = difflib.SequenceMatcher(
        None, original.translate(_QUOTE_FOLD), text.translate(_QUOTE_FOLD), autojunk=False
    )
    for i, j, size in matcher.get_matching_blocks():
        for k in range(size):
            if original[i + k] != text[j + k] and original[i + k] in "'\"‘’“”′″":
                out[j + k] = original[i + k]
    return "".join(out)


def load_compound_whitelist(config: dict[str, Any], config_dir: Path | None) -> set[str]:
    """복합어 화이트리스트(탭 구분 ``표제어\\t빈도``)를 로드한다.

    선언이 없으면 빈 set(정상적인 no-op)이고, 선언했는데 못 읽으면 예외다 —
    `KnowledgeAssetError` 계약(`errors.py`)을 치환 사전과 동일하게 따른다.
    화이트리스트는 사람이 편집하는 지식 자산이라 config 정본(`03-evaluation/`)
    기준 상대경로만 지원한다.
    """
    ref = resolve_whitelist_path(config, config_dir)
    if ref is None:
        return set()
    if ref.path is None:  # config 자산만 지원하므로 오지 않는다 — 온다면 해석 규칙이 바뀐 것
        raise KnowledgeAssetError(ref.rel, [ref.label])
    entries: set[str] = set()
    for line in ref.path.read_text(encoding="utf-8").splitlines():
        form = line.split("\t")[0].strip()
        if form:
            entries.add(form)
    return entries


def resolve_whitelist_path(config: dict[str, Any], config_dir: Path | None) -> ResourceRef | None:
    """복합어 화이트리스트 선언을 파일로 푼다 — 로더(`load_compound_whitelist`)와 스냅샷이 공유.

    Returns:
        선언이 없으면 None.

    Raises:
        KnowledgeAssetError: 선언했는데 config 정본에 파일이 없을 때(또는 config_dir 이 없을 때).
    """
    wl_path = config.get("whitelist_path", "")
    if not wl_path:
        return None
    searched: list[str] = []
    if config_dir is not None:
        candidate = _config_candidate(config_dir, wl_path)
        searched.append(str(candidate))
        if candidate.is_file():
            return ResourceRef("config", relative_posix(candidate, config_dir), candidate)
    else:
        searched.append("(config_dir 없음)")
    raise KnowledgeAssetError(wl_path, searched)


def _has_whitelist_prefix(joined: str, split_at: int, whitelist: set[str]) -> bool:
    """결합형의 접두(분리점 너머 ~ 전체)가 화이트리스트 표제어인지.

    조사·어미가 붙은 어절("삼성"+"전자는")도 표제어("삼성전자")로 잡기 위해
    정확 일치가 아니라 접두 일치를 본다. 분리점(`split_at`) 안쪽에서 끝나는
    접두는 첫 어절만으로 성립하므로 후보에서 뺀다.
    """
    return any(joined[:k] in whitelist for k in range(split_at + 1, len(joined) + 1))


def apply_compound_whitelist_restore(text: str, original: str, whitelist: set[str]) -> str:
    """`compound_whitelist_restore` rule_type 구현 (v19/v32 계보의 현 엔진 포팅).

    파이프라인(FT·step1/2/3)이 갈라놓은 화이트리스트 복합어를 다시 붙인다.
    인접 어절 쌍의 결합이 (1) 화이트리스트 표제어를 접두로 갖고 — 분리점이
    표제어 안쪽 — (2) 원문에 그대로 존재할 때만 붙인다. 조건 (2) 때문에 이
    룰은 파이프라인이 만든 분리를 되돌릴 뿐, 새 교정을 만들지 않는다.

    2026-09-02 재생 실측(combo 체인 출력 위): paragraph +2.91 f1
    (redundant 408→342), ci-v0 +0.43, ci-v2/ci-v1/yul-ryul 무변화.
    """
    if not whitelist:
        return text

    restored_lines: list[str] = []
    for line in text.split("\n"):
        tokens = line.split()
        out: list[str] = []
        i = 0
        while i < len(tokens):
            if i + 1 < len(tokens):
                joined = tokens[i] + tokens[i + 1]
                if joined in original and _has_whitelist_prefix(joined, len(tokens[i]), whitelist):
                    out.append(joined)
                    i += 2
                    continue
            out.append(tokens[i])
            i += 1
        restored_lines.append(" ".join(out))
    return "\n".join(restored_lines)


#: 율/률 정규화가 "명사 + 조사" 문맥으로 인정하는 어절 꼬리(한글 잔여) 목록.
#: 이 목록 밖의 한글이 이어지면(예: 인명 '김율희가'의 '희가') 건드리지 않는다.
_YUL_RYUL_JOSA = frozenset(
    {
        "은",
        "는",
        "이",
        "가",
        "을",
        "를",
        "의",
        "에",
        "도",
        "만",
        "과",
        "와",
        "로",
        "으로",
        "보다",
        "부터",
        "까지",
        "조차",
        "마저",
        "밖에",
        "처럼",
        "대로",
        "만큼",
        "뿐",
        "나",
        "이나",
        "란",
        "이란",
        "라",
        "이라",
        "라고",
        "이라고",
        "라는",
        "이라는",
        "라며",
        "이라며",
        "며",
        "이며",
        "고",
        "이고",
        "다",
        "이다",
        "입니다",
        "에서",
        "에게",
        "에는",
        "에도",
        "에서는",
        "에서도",
        "로는",
        "으로는",
        "로도",
        "으로도",
        "로서",
        "으로서",
        "로써",
        "으로써",
        "와의",
        "과의",
        "과는",
        "와는",
        "은요",
        "요",
        "대비",
    }
)


def apply_yul_ryul_normalize(text: str, exceptions: tuple[str, ...] = ()) -> str:
    """`yul_ryul_normalize` rule_type 구현 — 율/률 받침 규칙의 결정론적 적용.

    한글 맞춤법 제11항: 앞 음절에 받침이 없거나 ㄴ 받침이면 '율'(비율·환율·
    전환율), 그 외 받침이면 '률'(법률·확률·출석률). 받침은 유니코드 연산으로
    판정되므로 LLM 없이 최종 텍스트에서 강제할 수 있다.

    인명 오탐 방지 2중 가드:
    1. 어절 꼬리 검사 — 율/률 뒤의 한글 잔여가 조사 목록(`_YUL_RYUL_JOSA`)에
       없으면 접미사 용법이 아닌 것으로 보고 건너뛴다 (재생 실측의 유일한 오탐
       '김율희가'→'김률희가' 가 여기서 걸러진다).
    2. `exceptions` — 어절에 포함되면 무조건 건너뛰는 문자열 목록 (율/률로
       끝나는 인명 '김석율' 류를 위한 훅. 기본 비어 있음).

    2026-09-02 재생 실측(v33/combo 출력 위, 조사 가드 포함): yul-ryul 놓침 2건
    전량 회수(+3.07), ci-v2 +0.59, ci-v1 +0.53, paragraph·ci-v0 변경 0건.
    """
    out = list(text)
    n = len(text)
    for i, ch in enumerate(text):
        if ch not in ("율", "률"):
            continue
        prev = text[i - 1] if i > 0 else ""
        if not ("가" <= prev <= "힣"):
            continue
        jong = (ord(prev) - 0xAC00) % 28
        want = "율" if jong in (0, 4) else "률"  # 받침 없음(0)/ㄴ(4) → 율
        if want == ch:
            continue

        j = i + 1
        while j < n and not text[j].isspace():
            j += 1
        residue = "".join(c for c in text[i + 1 : j] if "가" <= c <= "힣")
        if residue and residue not in _YUL_RYUL_JOSA:
            continue

        if exceptions:
            k = i
            while k > 0 and not text[k - 1].isspace():
                k -= 1
            token = text[k:j]
            if any(e in token for e in exceptions):
                continue

        out[i] = want
    return "".join(out)