"""★정확한 hash adoption (Codex BLOCK 2026-09-03 · 실측 f7cc45c576c0 entity_detail): tuple 이 정확히 맞을 때만 옛 CP 에 새 hash 를
입힌다 — data 바이트 불변 · 백업 · append-only 사건 · runner 어긋남 0. 한 축이라도 다르면 **아무것도 쓰기 전에** 선다."""
from __future__ import annotations

import hashlib
import json
from pathlib import Path

import pytest

from tools.grounding_audit import canary_hash_adoption as ha


def _run(tmp_path, monkeypatch, *, old="old" * 20, data=None):
    from tools.grounding_audit import canary_pipeline as cp, canary_run as cr
    ep = tmp_path / "projects" / "p" / "checkpoints" / "episodes" / "e"
    (ep / "entity_detail").mkdir(parents=True)
    m = {"status": "completed", "config_hash": old, "run_id": "sr1", "updated_at": "2026-09-02T21:49:49+00:00",
         "schema_version": 1, "data": data or {"entity_details": {"C01": {"description": "학생"}}, "identity_contract": "IC"}}
    (ep / "entity_detail" / "manifest.json").write_text(json.dumps(m, ensure_ascii=False), encoding="utf-8")
    (tmp_path / "pipeline_attempts.json").write_text(json.dumps([
        {"attempt_id": "att1", "started_kst": "2026-09-03T06:48:38+09:00", "finished_kst": "2026-09-03T06:50:33+09:00", "status": "crashed"}]),
        encoding="utf-8")
    (tmp_path / "canary_run.json").write_text(json.dumps({"started_kst": "2026-09-03T06:21:00+09:00", "code": {"tip": "c" * 40}}), encoding="utf-8")
    monkeypatch.setattr(ha, "PACK_FOLDS", {"entity_detail": {"module": "m", "stems": (("s", "prompt"),), "pack_dir_glob": "prompts/{dir}",
                                                             "identity_constant": ("tests.grounding.test_config_hash_adoption_is_exact", "IC_NOW")}})
    monkeypatch.setattr(ha, "pack_raw_hashes_of", lambda module, stems: {"s": "rawhash", "s@dir": "15.x"})
    monkeypatch.setattr(ha, "_git", lambda *a, cwd=None: ("prompts/15.x" if a and a[0] == "ls-tree" else ""))
    monkeypatch.setattr(cp, "current_config_hash_of", lambda sid, pid, epi, cfg: "new" * 20)
    monkeypatch.setattr(cp, "contract_drift_of", lambda sid, pid, epi, cfg: None)
    monkeypatch.setattr(cp, "open_attempts", lambda root: [])
    monkeypatch.setattr(cr, "git_tip", lambda: {"tip": "d" * 40, "clean": True, "dirty_files": []})
    from tools.grounding_audit import canary_isolation as ci
    monkeypatch.setattr(ci, "root_dir", lambda rid: tmp_path)
    return ep / "entity_detail" / "manifest.json"


IC_NOW = "IC"


class TestTheProducingTipComesFromTheLedger:
    def test_a_transition_before_the_attempt_wins_over_the_first_run(self, tmp_path, monkeypatch):
        _run(tmp_path, monkeypatch)
        rows = json.loads((tmp_path / "pipeline_attempts.json").read_text(encoding="utf-8"))
        rows.append({"kind": "code_transition", "event_id": "t1", "recorded_kst": "2026-09-03T06:40:00+09:00", "to_tip": "g" * 40})
        rows.append({"kind": "code_transition", "event_id": "t2", "recorded_kst": "2026-09-03T06:55:00+09:00", "to_tip": "h" * 40})
        (tmp_path / "pipeline_attempts.json").write_text(json.dumps(rows), encoding="utf-8")
        t = ha.adoption_tuple("r", "entity_detail", {})
        assert t["produced_at_tip"] == "g" * 40, "★attempt 시작(06:48) 전 마지막 전이(06:40)의 to_tip — 뒤의 06:55 는 아니다"

    def test_a_pack_dir_that_did_not_exist_at_the_producing_tip_refuses(self, tmp_path, monkeypatch):
        _run(tmp_path, monkeypatch)
        monkeypatch.setattr(ha, "_git", lambda *a, cwd=None: "")
        t = ha.adoption_tuple("r", "entity_detail", {})
        assert t["pack_dir_unchanged_since_produced"]["ok"] is False
        with pytest.raises(ha.AdoptionRefused):
            ha.adopt("r", "entity_detail", {}, expected=t, why="x")


class TestTheExactTuple:
    def test_the_tuple_is_measured_not_typed(self, tmp_path, monkeypatch):
        _run(tmp_path, monkeypatch)
        t = ha.adoption_tuple("r", "entity_detail", {})
        assert t["old_config_hash"] == "old" * 20 and t["new_config_hash"] == "new" * 20
        assert t["produced_by_attempt"] == "att1" and t["produced_at_tip"] == "c" * 40 and t["at_tip"] == "d" * 40
        assert t["identity_contract"]["same"] is True and t["pack_dir_unchanged_since_produced"]["ok"] is True

    def test_adoption_changes_only_the_hash_metadata_and_leaves_an_event(self, tmp_path, monkeypatch):
        from tools.grounding_audit import canary_pipeline as cp
        mp = _run(tmp_path, monkeypatch)
        before = json.loads(mp.read_text(encoding="utf-8"))
        t = ha.adoption_tuple("r", "entity_detail", {})
        ev = ha.adopt("r", "entity_detail", {}, expected=t, why="hash 조리법만 바뀜")
        after = json.loads(mp.read_text(encoding="utf-8"))
        assert after["config_hash"] == "new" * 20
        assert json.dumps(after["data"], sort_keys=True) == json.dumps(before["data"], sort_keys=True), "★data 는 한 바이트도"
        assert after["config_hash_adoption"]["from"] == "old" * 20
        backups = list(mp.parent.glob("manifest_pre_adoption_*.json"))
        assert len(backups) == 1 and json.loads(backups[0].read_text(encoding="utf-8")) == before
        rows = [r for r in cp.read_attempts(tmp_path) if r.get("kind") == cp.EVENT_HASH_ADOPTION]
        assert len(rows) == 1 and rows[0]["event_id"] == ev["event_id"] and rows[0]["data_digest_after"] == t["data_digest"]
        # ★백업은 완료 CP 셈(glob */manifest.json)에 안 잡힌다
        from tools.grounding_audit import canary_run as cr
        assert cr.completed_steps_of("r") == ["entity_detail"]

    @pytest.mark.parametrize("axis,value", [
        ("old_config_hash", "x" * 60), ("new_config_hash", "y" * 60), ("data_digest", "z" * 64),
        ("produced_at_tip", "e" * 40), ("at_tip", "f" * 40), ("step_id", "entity_t2i"),
        ("pack_raw_hashes", {"s": "other", "s@dir": "15.x"}), ("produced_by_attempt", "att9"),
    ])
    def test_one_axis_off_refuses_before_writing(self, tmp_path, monkeypatch, axis, value):
        mp = _run(tmp_path, monkeypatch)
        raw = mp.read_bytes()
        t = ha.adoption_tuple("r", "entity_detail", {})
        t[axis] = value
        with pytest.raises(ha.AdoptionRefused):
            ha.adopt("r", "entity_detail", {}, expected=t, why="x")
        assert mp.read_bytes() == raw and not list(mp.parent.glob("manifest_pre_adoption_*.json"))
        assert not (tmp_path / "pipeline_attempts.json").read_text(encoding="utf-8").count("config_hash_adoption")

    def test_no_drift_means_nothing_to_adopt(self, tmp_path, monkeypatch):
        from tools.grounding_audit import canary_pipeline as cp
        mp = _run(tmp_path, monkeypatch, old="new" * 20)
        t = ha.adoption_tuple("r", "entity_detail", {})
        with pytest.raises(ha.AdoptionRefused):
            ha.adopt("r", "entity_detail", {}, expected=t, why="x")

    def test_the_runner_must_agree_afterwards_or_it_is_rolled_back(self, tmp_path, monkeypatch):
        from tools.grounding_audit import canary_pipeline as cp
        mp = _run(tmp_path, monkeypatch)
        raw = mp.read_bytes()
        monkeypatch.setattr(cp, "contract_drift_of", lambda sid, pid, epi, cfg: "config_hash mismatch: still")
        t = ha.adoption_tuple("r", "entity_detail", {})
        with pytest.raises(ha.AdoptionRefused):
            ha.adopt("r", "entity_detail", {}, expected=t, why="x")
        assert mp.read_bytes() == raw