"""정확한 config_hash adoption — **hash 조리법만 바뀐** 완료 CP 에 새 hash 를 입힌다. data 는 한 바이트도 안 바뀐다. ★실측 (2026-09-03, run f7cc45c576c0 dry_28~29): entity_detail 의 `_config_hash` 가 팩 지문 결속(d6f9cac4·22524359)으로 바뀌어 runner 가 어긋남으로 봤다. 팩 내용·모델·계약은 그대로다. 그런데 canary 의 force 는 하류 59 스텝 CP 를 지우고, RERUN_SELF 는 실제 LLM 을 다시 불러(entity_steps :1240) 산출이 달라질 수 있어 「하류 보존」이 계보를 끊는다(Codex BLOCK). 그래서 다시 돌지 않는다 — 옛 CP 에 **지금 코드가 계산한 새 hash** 를 입혀 어긋남 자체를 없앤다. 그 조건은 tuple 로 결속한다: step_id · stored old config_hash · current new config_hash · resolve_effective 가 고른 prompt/schema raw_content_hash · CP data 의 canonical digest · manifest 신원(step run_id · updated_at · schema_version · status=completed) · identity_contract(CP 안 값 == 지금 상수) · CP 를 만든 attempt 와 그 코드 tip · 그 tip 이후 팩 디렉토리 diff 0 · at_tip(지금) adopt(expected=…) 는 **지금 다시 잰 tuple** 과 expected 를 축마다 대조하고 하나라도 다르면 **아무것도 쓰기 전에** 선다. 맞으면: manifest 백업(`manifest_pre_adoption_.json` · glob `*/manifest.json` 에 안 잡힘) → config_hash 와 `config_hash_adoption` 메타데이터만 바꿔 씀 → 되읽어 data digest 동일 확인 → 장부에 append-only 사건 → runner 가 다시 봐도 어긋남 없음(`contract_drift_of` None) 확인. 끝점(Codex): old hash CP → adoption → mismatch 없음 → provider 0 → data digest 동일. """ from __future__ import annotations import hashlib import json import re import subprocess from pathlib import Path from typing import Any, Dict, Optional, Sequence ADOPTION_CONTRACT = "1.202609030930" _TUPLE_AXES = ("step_id", "old_config_hash", "new_config_hash", "pack_raw_hashes", "data_digest", "manifest_identity", "identity_contract", "produced_by_attempt", "produced_at_tip", "pack_dir_unchanged_since_produced", "at_tip") class AdoptionRefused(RuntimeError): """tuple 의 한 축이라도 다르면 — 아무것도 쓰지 않았다.""" def canonical_data_digest(data: Any) -> str: return hashlib.sha256(json.dumps(data, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")).hexdigest() def _git(*args: str, cwd: Path) -> str: return subprocess.run(["git", *args], cwd=str(cwd), capture_output=True, text=True).stdout.strip() def pack_raw_hashes_of(module: str, stems: Sequence[tuple]) -> Dict[str, str]: """resolve_effective 가 **지금** 고르는 내용 bytes 의 digest — 스텝의 `_config_hash` 가 접는 것과 같은 함수.""" from app.modules.prompt_loader import resolve_effective out: Dict[str, str] = {} for stem, kind in stems: r = resolve_effective(module, stem, kind=kind) out[stem] = str(r.get("raw_content_hash") or "") out[f"{stem}@dir"] = str(r.get("version") or r.get("loaded_from") or r.get("version_dir") or "") return out #: 스텝마다 — 그 스텝의 `_config_hash` 가 접는 팩 (읽어서 적는다 · 모르면 adoption 을 못 한다) PACK_FOLDS: Dict[str, Dict[str, Any]] = { "entity_detail": {"module": "entity_extractor_v2", "stems": (("turn1_7_detail_batch", "prompt"), ("turn1_7_detail_batch_schema", "schema")), "pack_dir_glob": "prompts/_base/entity_extractor_v2/{dir}", "identity_constant": ("app.core.steps.entity_steps", "ENTITY_INSTANCE_IDENTITY_CONTRACT_VERSION")}, } def _manifest_path(root: Path, step_id: str) -> Path: eps = sorted((root / "projects").glob("*/checkpoints/episodes/*")) if not eps: raise AdoptionRefused("이 run 에 에피소드 CP 디렉토리가 없다") return eps[-1] / step_id / "manifest.json" def _produced_by(root: Path, updated_at_utc: str) -> Dict[str, Any]: """CP 의 updated_at(UTC) 을 덮는 attempt 와 **그 attempt 가 돈 코드 tip**. tip 은 장부에서 잇는다 — attempt 시작 시각 이전에 적힌 **마지막 code_transition 의 to_tip**, 하나도 없으면 첫 판 (`canary_run.json`)의 code.tip. attempt 행 자체는 tip 을 안 갖고, 크래시한 attempt 의 canary_*.json 은 attempt id 를 안 담는다(실측 79e14592ab11).""" from datetime import datetime, timedelta, timezone from tools.grounding_audit import canary_pipeline as cp t = datetime.fromisoformat(updated_at_utc.replace("Z", "+00:00")) if t.tzinfo is None: t = t.replace(tzinfo=timezone.utc) kst = t.astimezone(timezone(timedelta(hours=9))).isoformat(timespec="seconds") rows = cp.read_attempts(root) att = [r for r in rows if r.get("kind") is None] hit = [r for r in att if str(r.get("started_kst") or "") <= kst <= str(r.get("finished_kst") or "9")] if len(hit) != 1: raise AdoptionRefused(f"CP 시각 {kst} 를 덮는 attempt 가 {len(hit)}개 — 어느 판이 만들었는지 못 잇는다") a = hit[0] started = str(a.get("started_kst") or "") trans = [r for r in rows if r.get("kind") == cp.EVENT_CODE_TRANSITION and str(r.get("recorded_kst") or "") <= started] if trans: tip = str(trans[-1].get("to_tip") or "") how = f"마지막 code_transition({trans[-1].get('event_id')} · {trans[-1].get('recorded_kst')}) 의 to_tip" else: try: tip = str(((json.loads((root / "canary_run.json").read_text(encoding="utf-8")) or {}).get("code") or {}).get("tip") or "") except Exception: tip = "" how = "첫 판 canary_run.json 의 code.tip" if not tip: raise AdoptionRefused(f"attempt {a.get('attempt_id')} 의 코드 tip 을 못 잇는다") return {"attempt_id": str(a.get("attempt_id")), "tip": tip, "how": how, "started_kst": started} def adoption_tuple(run_id: str, step_id: str, project_config: Optional[Dict[str, Any]]) -> Dict[str, Any]: """**지금** 잰 tuple. adopt 는 이것을 expected 와 대조한다.""" from tools.grounding_audit import canary_isolation as ci, canary_pipeline as cp, canary_run as cr fold = PACK_FOLDS.get(step_id) if fold is None: raise AdoptionRefused(f"{step_id!r} 가 접는 팩을 모른다 — PACK_FOLDS 에 읽어서 적어야 adoption 을 한다") root = ci.root_dir(run_id) mp = _manifest_path(root, step_id) if not mp.is_file(): raise AdoptionRefused(f"{step_id} 의 manifest 가 없다: {mp}") m = json.loads(mp.read_text(encoding="utf-8")) if str(m.get("status")) != "completed": raise AdoptionRefused(f"{step_id} CP 가 completed 가 아니다({m.get('status')}) — adoption 은 끝난 CP 에만") eps = mp.parents[1] pid, epi = eps.parents[2].name, eps.name new_hash = cp.current_config_hash_of(step_id, pid, epi, project_config) packs = pack_raw_hashes_of(fold["module"], fold["stems"]) mod, const = fold["identity_constant"] ident_now = str(getattr(__import__(mod, fromlist=[const]), const)) ident_cp = str(((m.get("data") or {}).get("identity_contract")) or "") prod = _produced_by(root, str(m.get("updated_at") or "")) repo = cr.BACKEND.parent pack_dirs = sorted({v for k, v in packs.items() if k.endswith("@dir") and v}) diff_lines = [] existed = {} for d in pack_dirs: rel = fold["pack_dir_glob"].format(dir=d) diff_lines += [x for x in _git("diff", "--name-only", prod["tip"], "HEAD", "--", rel, cwd=repo).splitlines() if x] # ★그 팩 디렉토리가 CP 를 만든 tip 에 **이미 있었나** — 없었으면 그 CP 는 다른 팩으로 만든 것이다 existed[d] = bool(_git("ls-tree", "-d", "--name-only", prod["tip"], "--", rel, cwd=repo)) return { "contract": ADOPTION_CONTRACT, "step_id": step_id, "old_config_hash": str(m.get("config_hash") or ""), "new_config_hash": new_hash, "pack_raw_hashes": packs, "data_digest": canonical_data_digest(m.get("data")), "manifest_identity": {"run_id": str(m.get("run_id") or ""), "updated_at": str(m.get("updated_at") or ""), "schema_version": m.get("schema_version"), "status": str(m.get("status") or "")}, "identity_contract": {"in_cp": ident_cp, "now": ident_now, "same": ident_cp == ident_now}, "produced_by_attempt": prod["attempt_id"], "produced_at_tip": prod["tip"], "pack_dir_unchanged_since_produced": {"dirs": pack_dirs, "changed_files": diff_lines, "existed_at_produced_tip": existed, "ok": (not diff_lines) and all(existed.values())}, "produced_how": prod["how"], "at_tip": cr.git_tip()["tip"], "manifest_path": str(mp), } def adopt(run_id: str, step_id: str, project_config: Optional[Dict[str, Any]], *, expected: Dict[str, Any], why: str) -> Dict[str, Any]: """tuple 이 **정확히** 맞을 때만 — 백업 · config_hash 메타데이터만 교체 · data digest 재확인 · append-only 사건 · runner 재확인.""" from tools.grounding_audit import canary_isolation as ci, canary_pipeline as cp, canary_run as cr if not str(why or "").strip(): raise AdoptionRefused("이유(why)가 없다") now = adoption_tuple(run_id, step_id, project_config) for ax in _TUPLE_AXES: if json.dumps(now.get(ax), sort_keys=True, ensure_ascii=False) != json.dumps(expected.get(ax), sort_keys=True, ensure_ascii=False): raise AdoptionRefused(f"tuple 축 {ax} 가 다르다 — 지금 {json.dumps(now.get(ax), ensure_ascii=False)[:160]} · 기대 " f"{json.dumps(expected.get(ax), ensure_ascii=False)[:160]}. 아무것도 쓰지 않았다") if now["old_config_hash"] == now["new_config_hash"]: raise AdoptionRefused("옛 hash 와 새 hash 가 같다 — 어긋남이 없어 adoption 할 것이 없다") if not now["identity_contract"]["same"] or not now["pack_dir_unchanged_since_produced"]["ok"]: raise AdoptionRefused("identity_contract 가 다르거나 팩 디렉토리가 CP 를 만든 tip 이후 바뀌었다 — 같은 bytes 가 아니다") if cr.git_tip()["tip"] != now["at_tip"] or not cr.git_tip()["clean"]: raise AdoptionRefused("트리가 더럽거나 tip 이 움직였다") root = ci.root_dir(run_id) if cp.open_attempts(root): raise AdoptionRefused("열린 attempt 가 있다") mp = Path(now["manifest_path"]) raw_before = mp.read_bytes() m = json.loads(raw_before.decode("utf-8")) # ── 사건 id 를 먼저 만들고(백업 이름에 쓴다) 장부에는 성공 뒤 적는다 import uuid event_id = uuid.uuid4().hex[:12] backup = mp.with_name(f"manifest_pre_adoption_{event_id}.json") backup.write_bytes(raw_before) m2 = dict(m) m2["config_hash"] = now["new_config_hash"] m2["config_hash_adoption"] = {"event_id": event_id, "from": now["old_config_hash"], "contract": ADOPTION_CONTRACT, "backup": backup.name, "why": why} tmp = mp.with_name("manifest.json.adopting") tmp.write_text(json.dumps(m2, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(mp) m3 = json.loads(mp.read_text(encoding="utf-8")) after_digest = canonical_data_digest(m3.get("data")) if after_digest != now["data_digest"] or str(m3.get("config_hash")) != now["new_config_hash"]: mp.write_bytes(raw_before) raise AdoptionRefused("되읽은 CP 의 data digest 나 hash 가 기대와 다르다 — 백업으로 되돌렸다") eps = mp.parents[1]; pid, epi = eps.parents[2].name, eps.name still = cp.contract_drift_of(step_id, pid, epi, project_config) if still: mp.write_bytes(raw_before) raise AdoptionRefused(f"adoption 뒤에도 runner 가 어긋남으로 본다({still}) — 되돌렸다") ev = cp.append_event(root, {"kind": cp.EVENT_HASH_ADOPTION, "event_id": event_id, "why": why, **{k: now[k] for k in _TUPLE_AXES}, "backup": str(backup), "data_digest_after": after_digest, "runner_mismatch_after": None, "★means": "옛 CP 에 지금 코드의 hash 를 입혔다 — data 는 한 바이트도 안 바뀌었다(digest 동일). 다시 돌지 않는다"}) return ev