"""설정의 **내용 지문** — 이름이 아니라 내용으로 실험을 식별한다. 왜 필요한가: 이 엔진의 자산은 오래 이름으로만 식별돼 왔고(`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}