solar_eval/core/fingerprint.py

"""설정의 **내용 지문** — 이름이 아니라 내용으로 실험을 식별한다.

왜 필요한가: 이 엔진의 자산은 오래 이름으로만 식별돼 왔고(`prompt_dev_v23`
↔ `pipeline_dev_v24` 처럼 짝의 번호가 어긋나는 일이 잦았다), 그래서
"이름이 다른 두 run 이 사실 같은 설정" 인지를 사람이 해시를 떠서 대조해야 했다
(`insights/lab_notes_archive.md#260803_etd_regression_prod_vs_v24` 의 '이름 매핑' 절이 그 흔적이다).
run 에 지문을 찍어 두면 그 대조가 조회 한 번으로 끝난다.

지문은 **해석이 끝난 설정**을 대상으로 계산한다. `extends`/override 로 조립한
파이프라인과 같은 내용을 통째로 적어 둔 파일은 같은 지문을 가져야 하기 때문이다
(그 등가성이 파일 복제를 override 로 옮길 때의 안전망이다).

형식: `sha256:<앞 12자>` — 눈으로 비교할 수 있을 만큼 짧고, 프로젝트당 수백 개
규모에서 충돌하지 않을 만큼 길다.
"""

import hashlib
import json
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any

#: 지문 계산에서 제외하는 키 — 사람이 읽는 라벨일 뿐 실행에 영향을 주지 않는다.
#: 이걸 제외해야 "이름만 바꾼 사본"이 같은 지문으로 잡힌다.
NON_FUNCTIONAL_KEYS = frozenset({"name", "description", "comment", "extends", "overrides"})

#: 스텝 하위에서 제외하는 키 — 스텝의 `name` 은 프롬프트 키·rule 타입 조회에
#: 쓰이므로 (`pipelines/registry.py:resolve_step`) 기능적이다. 지우면 안 된다.
NON_FUNCTIONAL_STEP_KEYS = frozenset({"description", "comment"})

#: 프롬프트 디렉토리에서 지문 대상으로 삼는 확장자.
PROMPT_ASSET_SUFFIXES = frozenset({".txt", ".yaml", ".yml", ".json"})

_DIGEST_PREFIX = "sha256"
_DIGEST_LENGTH = 12


def digest(payload: Any) -> str:
    """JSON 직렬화 가능한 값의 정규 지문."""
    canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
    full = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return f"{_DIGEST_PREFIX}:{full[:_DIGEST_LENGTH]}"


def pipeline_fingerprint(config: Mapping[str, Any] | None) -> str | None:
    """해석이 끝난 파이프라인 설정의 지문.

    Args:
        config: `extends`/override 해석까지 끝난 파이프라인 dict.
            None 이거나 비어 있으면 None 을 반환한다 (single_step 처럼 파이프라인
            파일이 없는 태스크가 있다).

    Returns:
        `sha256:xxxxxxxxxxxx` 또는 None.
    """
    if not config:
        return None
    return digest(_strip_pipeline(config))


def prompt_fingerprint(path: Path) -> str | None:
    """프롬프트 자산의 지문 — 디렉토리(여러 파일)와 단일 파일 모두 지원한다.

    디렉토리면 하위 자산 파일을 상대경로로 정렬해 `{경로: 내용해시}` 맵을 만들고
    그 맵의 지문을 낸다. 파일 하나가 바뀌면 전체 지문이 바뀌고, 파일 순서나
    디렉토리 이름이 바뀌어도 내용이 같으면 지문은 유지된다.

    Args:
        path: 프롬프트 디렉토리 또는 YAML 파일 경로.

    Returns:
        `sha256:xxxxxxxxxxxx`, 경로가 없거나 읽을 수 없으면 None.
    """
    try:
        if path.is_dir():
            assets = _collect_prompt_assets(path)
            if not assets:
                return None
            return digest(assets)
        if path.is_file():
            return digest({path.name: _file_digest(path)})
    except OSError:
        return None
    return None


def _collect_prompt_assets(root: Path) -> dict[str, str]:
    """프롬프트 디렉토리의 `{상대경로: 파일지문}` 맵. 숨김 파일은 제외한다."""
    assets: dict[str, str] = {}
    for file in sorted(root.rglob("*")):
        if not file.is_file() or file.suffix.lower() not in PROMPT_ASSET_SUFFIXES:
            continue
        if any(part.startswith(".") for part in file.relative_to(root).parts):
            continue
        assets[file.relative_to(root).as_posix()] = _file_digest(file)
    return assets


def _file_digest(path: Path) -> str:
    """파일 바이트의 지문 — 텍스트 인코딩·개행 정규화를 하지 않는다(있는 그대로)."""
    return hashlib.sha256(path.read_bytes()).hexdigest()[:_DIGEST_LENGTH]


def _strip_pipeline(config: Mapping[str, Any]) -> dict[str, Any]:
    """파이프라인 dict 에서 비기능 필드를 걷어낸 사본."""
    stripped = {k: v for k, v in config.items() if k not in NON_FUNCTIONAL_KEYS}
    steps = stripped.get("steps")
    if isinstance(steps, Sequence) and not isinstance(steps, str | bytes):
        stripped["steps"] = [_strip_step(s) for s in steps]
    return stripped


def _strip_step(step: Any) -> Any:
    if not isinstance(step, Mapping):
        return step
    return {k: v for k, v in step.items() if k not in NON_FUNCTIONAL_STEP_KEYS}