"""#119 카나리아 육안 3결함 수리 유닛 — prev 인물 스코프·표기 문안·era 캐시.

OFF/기본값 = byte-identical 이 1급 계약: 새 인자 미전달 조립이 기존과
같은 스템을 그대로 내는지 먼저 잠근다.
"""
from pathlib import Path

import pytest

from app.modules.pipeline import era_research as er
from app.modules.pipeline import signage_author as sa
from app.modules.pipeline.still_recipe import (
    build_prev_people_rule,
    build_signage_section,
    build_still_prompt,
    build_still_refs,
)


def _refs(prev_rule: str = "") -> list:
    return build_still_refs(
        bg_only=False, plate=None, conti=None,
        prev_sel=Path("/tmp/prev.png"), char_refs=[], prop_refs=[],
        prev_people_rule=prev_rule,
    )


class TestPrevPeopleRule:
    def test_default_is_legacy_unconditional_lock(self):
        (label, _), = _refs()
        assert "each person's clothing are LOCKED" in label
        assert "{people_rule}" not in label

    def test_no_overlap_forbids_carrying_people(self):
        rule = build_prev_people_rule([], cast_known=True)
        assert "NOT in this shot" in rule
        (label, _), = _refs(rule)
        assert "never carry their face" in label
        assert "each person's clothing are LOCKED" not in label
        assert "{people_rule}" not in label

    def test_overlap_locks_only_shared_names(self):
        rule = build_prev_people_rule(["인물갑", "인물을"], cast_known=True)
        assert "인물갑, 인물을" in rule
        (label, _), = _refs(rule)
        assert "인물갑, 인물을" in label

    def test_cast_unknown_returns_empty(self):
        assert build_prev_people_rule(["누구"], cast_known=False) == ""

    def test_continues_clause_scoped_vs_legacy(self):
        common = dict(
            shot_desc="d", place_text="p.", time_of_day_en="day",
            bg_only=False, prev_used=True)
        legacy = build_still_prompt(**common)
        assert "each person's clothing and state — persists" in legacy
        scoped = build_still_prompt(
            **common,
            prev_people_rule=build_prev_people_rule([], cast_known=True))
        assert "NOT in this shot" in scoped
        assert "each person's clothing and state — persists" not in scoped
        assert "{people_rule}" not in scoped


class TestSignageSection:
    def test_empty_inscriptions_render_nothing(self):
        assert build_signage_section([]) == ""
        assert build_signage_section(
            [{"surface_native": " ", "text_native": ""}]) == ""

    def test_renders_supplied_lines_and_prompt_carries_them(self):
        sec = build_signage_section([
            {"surface_native": "행사 안내판",
             "text_native": "안내 문구", "reason_ko": "r"},
        ])
        # 2026-08-27: 머리말에서 통칭을 걷었다 (팩 v28).
        assert "WORDS TO RENDER" in sec
        assert '- 행사 안내판: "안내 문구"' in sec
        p = build_still_prompt(
            shot_desc="d", place_text="p.", time_of_day_en="day",
            bg_only=False, prev_used=False, signage_en=sec)
        assert '"안내 문구"' in p
        base = build_still_prompt(
            shot_desc="d", place_text="p.", time_of_day_en="day",
            bg_only=False, prev_used=False)
        assert "WORDS TO RENDER" not in base

    def test_author_schema_shape(self):
        # ★selector 를 **명시**한다 — 이 시험은 2026-08-14 카나리아 판의
        #  v1 계약을 잠그는 것이다. 기본값을 따라가면(2026-08-27 부터 v2)
        #  재는 대상이 바뀌어 옛 계약이 안 지켜져도 초록이 된다.
        s = sa.build_author_schema("1")
        item = s["properties"]["inscriptions"]["items"]
        assert set(item["required"]) == {
            "surface_native", "text_native", "reason_ko"}


class TestEraCachedWrapper:
    def _store(self):
        d = {}
        return d, (lambda k: d.get(k)), (lambda k, v: d.__setitem__(k, v))

    def test_negative_assess_cached_once(self, monkeypatch, tmp_path):
        calls = []
        monkeypatch.setattr(
            er, "assess_subjects",
            lambda **kw: calls.append(1) or [])
        d, get, put = self._store()
        for _ in range(2):
            out = er.assess_and_research_cached(
                step_tag="t", subject_text="평범한 방",
                world_facts_block="w", out_dir=tmp_path,
                cache_get=get, cache_put=put)
            assert out is None
        assert len(calls) == 1  # 두 번째는 캐시
        assert any(k.startswith("era_assess::") for k in d)

    def test_research_cached_and_file_verified(self, monkeypatch, tmp_path):
        subj = {"subject_native": "특정 시대 실내 공간",
                "search_terms_native": ["a", "b"],
                "language_lock_native": "잠금문 열 자 이상입니다"}
        monkeypatch.setattr(er, "assess_subjects", lambda **kw: [subj])
        researched = []

        def fake_research(**kw):
            researched.append(1)
            out = kw["out_path"]
            out.parent.mkdir(parents=True, exist_ok=True)
            out.write_bytes(b"PNGDATA")
            import hashlib
            return {"subject": subj["subject_native"],
                    "sha256": hashlib.sha256(b"PNGDATA").hexdigest(),
                    "file": out.name}

        monkeypatch.setattr(er, "research_reference", fake_research)
        d, get, put = self._store()
        m1 = er.assess_and_research_cached(
            step_tag="t", subject_text="같은 장소", world_facts_block="w",
            out_dir=tmp_path, cache_get=get, cache_put=put)
        m2 = er.assess_and_research_cached(
            step_tag="t", subject_text="같은 장소", world_facts_block="w",
            out_dir=tmp_path, cache_get=get, cache_put=put)
        assert len(researched) == 1  # 두 번째는 캐시+파일 검증
        assert m1["path"] == m2["path"]
        assert Path(m1["path"]).name.startswith("eraref_")

    def test_research_failure_memo_bounds_same_walk(
            self, monkeypatch, tmp_path):
        # Codex BLOCK-1: 같은 걷기 = 1회만 실패 구매, 다음 걷기(새 memo)
        # = 재시도. 실패 감사는 era_fail:: 로 영속되고 게이트가 아니다.
        subj = {"subject_native": "대상", "search_terms_native": ["a"],
                "language_lock_native": "잠금문 열 자 이상입니다"}
        monkeypatch.setattr(er, "assess_subjects", lambda **kw: [subj])
        n = []
        monkeypatch.setattr(
            er, "research_reference", lambda **kw: n.append(1) or None)
        d, get, put = self._store()
        walk1: set = set()
        for _ in range(3):
            assert er.assess_and_research_cached(
                step_tag="t", subject_text="s", world_facts_block="w",
                out_dir=tmp_path, cache_get=get, cache_put=put,
                failed_memo=walk1) is None
        assert len(n) == 1  # 같은 걷기 재시도 0
        walk2: set = set()
        assert er.assess_and_research_cached(
            step_tag="t", subject_text="s", world_facts_block="w",
            out_dir=tmp_path, cache_get=get, cache_put=put,
            failed_memo=walk2) is None
        assert len(n) == 2  # 새 걷기는 재시도
        assert not any(k.startswith("era_ref::") for k in d)
        assert any(k.startswith("era_fail::") for k in d)

    def test_policy_version_bumped_for_coverage(self):
        # v3 (2026-08-27, 감사 2-C): 캐시 신원 계약이 바뀌어 올렸다.
        # ★적용 범위(plate·confined)는 이름에 그대로 남는다 — 이 시험이
        #  지키는 것은 「범위가 계약에 들어 있다」이지 판 번호가 아니다.
        v = er.ERA_RESEARCH_POLICY_VERSION
        assert v == "era_research_v3_canonical_scope_plate_confined"

    # ── (era R1 BLOCK 1·2·3 반영, 2026-08-17) ──────────────────────

    def test_outcome_failed_vs_non_target(self, monkeypatch, tmp_path):
        # 실패(research None)=failed True / 비대상(subjects 빈)=False —
        # 호출측이 기존 성공 배경 보존 여부를 가르는 신호(BLOCK-1)
        subj = {"subject_native": "대상", "search_terms_native": ["a"],
                "language_lock_native": "잠금문 열 자 이상입니다"}
        monkeypatch.setattr(er, "assess_subjects", lambda **kw: [subj])
        monkeypatch.setattr(er, "research_reference", lambda **kw: None)
        d, get, put = self._store()
        oc: dict = {}
        assert er.assess_and_research_cached(
            step_tag="t", subject_text="s", world_facts_block="w",
            out_dir=tmp_path, cache_get=get, cache_put=put,
            outcome=oc) is None
        assert oc["failed"] is True
        # 비대상(판별 성공·대상 없음)은 실패가 아니다
        d2, get2, put2 = self._store()
        monkeypatch.setattr(er, "assess_subjects", lambda **kw: [])
        oc3: dict = {}
        assert er.assess_and_research_cached(
            step_tag="t", subject_text="비대상 방", world_facts_block="w",
            out_dir=tmp_path, cache_get=get2, cache_put=put2,
            outcome=oc3) is None
        assert oc3["failed"] is False

    def test_fail_record_is_monotonic_per_attempt(
            self, monkeypatch, tmp_path):
        # BLOCK-3: 반복 유료 실패가 records bytes 를 움직여야 #77 지출
        # 감지에 잡힌다 — attempts 단조 증가
        subj = {"subject_native": "대상", "search_terms_native": ["a"],
                "language_lock_native": "잠금문 열 자 이상입니다"}
        monkeypatch.setattr(er, "assess_subjects", lambda **kw: [subj])
        monkeypatch.setattr(er, "research_reference", lambda **kw: None)
        d, get, put = self._store()
        for expected in (1, 2):
            er.assess_and_research_cached(
                step_tag="t", subject_text="s", world_facts_block="w",
                out_dir=tmp_path, cache_get=get, cache_put=put,
                failed_memo=set())  # 걷기마다 새 memo = 재시도 허용
            fail = next(v for k, v in d.items()
                        if k.startswith("era_fail::"))
            assert fail["attempts"] == expected

    def test_era_preserve_truth_table(self):
        # (era R2 BLOCK-1) 보존은 "실패+기존 era 배경 정상"일 때만
        from app.modules.pipeline.still_recipe import era_preserve_applies

        base = dict(era_on=True, era_meta=None, outcome={"failed": True},
                    file_ok=True, prev_meta={"era_research_sha": "x"})
        assert era_preserve_applies(**base)
        for k, v in [("era_on", False), ("era_meta", {"sha256": "y"}),
                     ("outcome", {"failed": False}), ("outcome", None),
                     ("file_ok", False), ("prev_meta", {}),
                     ("prev_meta", None)]:
            assert not era_preserve_applies(**{**base, k: v}), (k, v)

    def test_preserve_incomplete_is_step_unseal_exception(self):
        from app.services.still_recipe_service import (
            StillEraPreserveIncomplete,
        )

        assert issubclass(StillEraPreserveIncomplete, RuntimeError)

    def test_era_on_outer_hash_folds_search_contract(self, monkeypatch):
        # (era R2 BLOCK-2) picker 팩 계약이 스텝 층 config_hash 에 접힌다
        # — 없으면 완료 스텝이 outer SKIP 돼 r_sha 재계산 자체가 안 돈다
        from app.core.config import settings
        from app.core.steps.image_steps import SceneImagePipelineStep
        import app.modules.pipeline.search_grounded_ref as sgr

        step = object.__new__(SceneImagePipelineStep)
        step.project_config = {"some": "config"}
        monkeypatch.setattr(settings, "scene_image_target_scenes", "")
        monkeypatch.setattr(settings, "still_recipe_mode", "v1")
        monkeypatch.setattr(settings, "era_research_enabled", True,
                            raising=False)
        monkeypatch.setattr(sgr, "search_contract_sha",
                            lambda *a, **k: "CONTRACT_A")
        h1 = SceneImagePipelineStep._config_hash(step)
        monkeypatch.setattr(sgr, "search_contract_sha",
                            lambda *a, **k: "CONTRACT_B")
        h2 = SceneImagePipelineStep._config_hash(step)
        assert h1 != h2
        # era OFF 면 이 계약은 스탬프에 안 실린다(키 부재 동형)
        monkeypatch.setattr(settings, "era_research_enabled", False,
                            raising=False)
        h3 = SceneImagePipelineStep._config_hash(step)
        monkeypatch.setattr(sgr, "search_contract_sha",
                            lambda *a, **k: "CONTRACT_C")
        assert SceneImagePipelineStep._config_hash(step) == h3

    def test_ref_identity_folds_search_contract(
            self, monkeypatch, tmp_path):
        # BLOCK-2: picker 팩 계약(search_contract_sha)이 참조 신원에
        # 접힌다 — 계약이 바뀌면 캐시 적중이 깨져 재조사한다
        subj = {"subject_native": "대상", "search_terms_native": ["a"],
                "language_lock_native": "잠금문 열 자 이상입니다"}
        monkeypatch.setattr(er, "assess_subjects", lambda **kw: [subj])
        researched = []

        def fake_research(**kw):
            researched.append(1)
            kw["out_path"].write_bytes(b"PNGDATA")
            import hashlib
            return {"subject": subj["subject_native"],
                    "sha256": hashlib.sha256(b"PNGDATA").hexdigest(),
                    "file": kw["out_path"].name}

        monkeypatch.setattr(er, "research_reference", fake_research)
        import app.modules.pipeline.search_grounded_ref as sgr
        d, get, put = self._store()
        monkeypatch.setattr(sgr, "search_contract_sha",
                            lambda *a, **k: "CONTRACT_A")
        m1 = er.assess_and_research_cached(
            step_tag="t", subject_text="s", world_facts_block="w",
            out_dir=tmp_path, cache_get=get, cache_put=put)
        monkeypatch.setattr(sgr, "search_contract_sha",
                            lambda *a, **k: "CONTRACT_B")
        m2 = er.assess_and_research_cached(
            step_tag="t", subject_text="s", world_facts_block="w",
            out_dir=tmp_path, cache_get=get, cache_put=put)
        assert len(researched) == 2  # 계약 변경 = 캐시 우회, 재조사
        assert m1["path"] != m2["path"]


class TestEraGroupbgCachedWiring:
    """(2026-08-17) groupbg era 조사도 캐시 계약만 쓴다 — 직호출 금지.

    직호출(assess_subjects/research_reference)은 재방문마다 검색을 다시
    돌려 참조 바이트가 바뀌고, 그 sha 가 groupbg 지문에 접혀 완성된
    배경·하류 샷을 연쇄 재생성시켰다('마지막 임무' 실측). 이 시험은
    still_recipe_service 가 두 함수를 어디서도 직접 안 쓰는 것을 AST 로
    잠근다 — 새 직호출이 생기면 여기서 걸린다.
    """

    def test_no_direct_era_calls_in_still_recipe_service(self):
        import ast
        from pathlib import Path

        import app.services.still_recipe_service as svc

        tree = ast.parse(Path(svc.__file__).read_text(encoding="utf-8"))
        banned = {"assess_subjects", "research_reference"}
        hits = []
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom) and node.module and \
                    node.module.endswith("era_research"):
                for alias in node.names:
                    if alias.name in banned:
                        hits.append((node.lineno, alias.name))
        assert not hits, (
            f"era 직호출 import 발견 {hits} — 캐시 계약"
            "(assess_and_research_cached)만 허용")


class TestConfigHashGateFailClosed:
    """(era R3 잔여 HIGH) 비교 게이트의 step-local 해시 예외는 전파한다.

    이전 결함: _config_hash() 가 죽으면 project_config hash 로 조용히
    대체 — legacy CP 와 우연 일치=stale SKIP(진짜 drift 은폐), 불일치=
    drift 오진. cast 스탬프 오타(PACK_VERSION_MAP)가 실제로 이 층 뒤에
    잠복했었다. 저장 경로는 무보호 호출이라 크게 죽으므로 비교 게이트도
    대칭이어야 한다.
    """

    def _runner(self):
        from app.core.step_runner import StepRunner

        runner = StepRunner.__new__(StepRunner)
        runner.step_id = "test_step"
        runner.manifest = {"schema_version": 1}
        runner.project_config = {}
        return runner

    def test_step_local_hash_exception_propagates(self):
        runner = self._runner()

        def _boom():
            raise ImportError("cast 스탬프 오타 재현")

        runner._config_hash = _boom
        with pytest.raises(ImportError):
            runner._check_cp_mismatch(
                {"schema_version": 1, "config_hash": "deadbeef"})

    def test_without_step_local_hash_fallback_unchanged(self):
        from app.core.step_runner import compute_config_hash

        runner = self._runner()
        cp_hash = compute_config_hash(runner.project_config)
        assert runner._check_cp_mismatch(
            {"schema_version": 1, "config_hash": cp_hash}) is None