"""era_research — 시대 인지 사전 조사 (2026-08-14 사용자 확정 "무조건").

계약: 판별+원어 질의 1콜(상한 컷) / 검색→다운로드→GPT 선택 재사용 /
실패 전부 비차단(None) / OFF(default)=byte-identical.
"""
from __future__ import annotations

from typing import Any, Dict

import app.modules.pipeline.era_research as er
from app.core.config import settings


def test_flag_default_off():
    assert settings.__class__.model_fields[
        "era_research_enabled"].default is False


def test_pack_and_policy_contract():
    # v3 (2026-08-25): 대상 칸을 「한 대상 = 사진 한 장이 혼자 보여줄 수
    # 있는 것 하나」로 못박은 판 — 검색어가 흩어지던 원인이 여기였다.
    assert er.resolve_era_pack() == "3.202608251500"
    assert len(er.era_pack_content_hash()) == 16
    # v2 (#119①): 적용 범위(샷별 판·confined)가 계약에 들어간 bump.
    # v3 (2026-08-27, 감사 2-C): 캐시 신원이 자유 저작 → 정본 세 칸.
    # ★범위 낱말은 **그대로 남아야 한다** — v2 의 계약이 사라지면 안 된다.
    v = er.ERA_RESEARCH_POLICY_VERSION
    assert "v3" in v and "plate" in v and "confined" in v
    assert "canonical_scope" in v
    role = er.build_ref_role("시대 대상 X")
    assert role.startswith("PERIOD REFERENCE — 시대 대상 X")
    assert "never the composition" in role


def _subject(i: int) -> Dict[str, Any]:
    return {"subject_native": f"대상{i}",
            "search_terms_native": ["질의 하나", "질의 둘"],
            "language_lock_native": "다른 언어로 질의를 만들지 마라.",
            "reason_ko": "이유"}


def test_assess_caps_subjects(monkeypatch):
    import app.modules.llm.llm_client as llm_client

    monkeypatch.setattr(
        llm_client, "call_structured",
        lambda *a, **k: {"subjects": [_subject(i) for i in range(4)]})
    out = er.assess_subjects(
        step_tag="t", subject_text="본문", world_facts_block="시대")
    assert len(out) == er.MAX_SUBJECTS_PER_CALL


def _wire_research(monkeypatch, tmp_path, *, images, downloads_ok=True,
                   chosen=1):
    import app.modules.pipeline.search_grounded_ref as sgr
    import app.modules.llm.llm_client as llm_client

    seen: Dict[str, Any] = {"search": None, "pick_parts": None}

    def fake_search(client, *, directive_native, terms_native=None,
                    language_lock_native="", **kw):
        seen["search"] = {"directive": directive_native,
                          "terms": terms_native,
                          "lock": language_lock_native}
        return {"queries": ["나간 질의"], "images": images}

    def fake_download(url, dest, fallback_url=""):
        if not downloads_ok:
            return False
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(b"\x89PNG\r\n\x1a\n" + url.encode())
        return True

    def fake_pick(tag, sys_p, parts, schema, **kw):
        seen["pick_parts"] = parts
        return {"verdicts": [
                    {"index": i + 1, "usable": (i + 1) == chosen,
                     "score": 80 if (i + 1) == chosen else 0,
                     "reason_ko": f"후보{i + 1} 판정"}
                    for i in range(len(images))],
                "chosen_index": chosen,
                "chosen_reason_ko": "이유"}

    monkeypatch.setattr(sgr, "search_reference_images", fake_search)
    monkeypatch.setattr(sgr, "download_candidate", fake_download)
    monkeypatch.setattr(llm_client, "call_structured", fake_pick)
    return seen


def test_research_reference_happy_path(monkeypatch, tmp_path):
    imgs = [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": ""}
            for i in range(3)]
    seen = _wire_research(monkeypatch, tmp_path, images=imgs, chosen=2)
    out = tmp_path / "era.png"
    meta = er.research_reference(
        subject=_subject(1), world_facts_block="1980년대 대한민국",
        out_path=out, step_tag="t", openai_client=object())
    assert meta is not None and out.is_file()
    assert meta["picked_index"] == 2
    assert meta["picked_url"] == "https://x/1.jpg"
    assert meta["queries"] == ["나간 질의"]
    assert len(meta["sha256"]) == 64
    # 원어 잠금이 검색에 그대로 전달된다
    assert seen["search"]["lock"].startswith("다른 언어")


def test_research_reference_no_usable_returns_none(monkeypatch, tmp_path):
    imgs = [{"image_url": "https://x/0.jpg", "thumbnail_url": ""}]
    _wire_research(monkeypatch, tmp_path, images=imgs, chosen=0)
    assert er.research_reference(
        subject=_subject(1), world_facts_block="", out_path=tmp_path / "e.png",
        step_tag="t", openai_client=object()) is None


def test_verdicts_and_urls_survive_on_success(monkeypatch, tmp_path):
    """고른 것뿐 아니라 **버린 것의 이유**도 meta 에 남는다."""
    imgs = [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": ""}
            for i in range(3)]
    _wire_research(monkeypatch, tmp_path, images=imgs, chosen=2)
    meta = er.research_reference(
        subject=_subject(1), world_facts_block="",
        out_path=tmp_path / "e.png", step_tag="t", openai_client=object())
    assert meta is not None
    assert [v["index"] for v in meta["verdicts"]] == [1, 2, 3]
    assert [v["usable"] for v in meta["verdicts"]] == [False, True, False]
    assert meta["candidate_urls"] == [f"https://x/{i}.jpg" for i in range(3)]


def test_audit_records_no_usable_verdict(monkeypatch, tmp_path):
    """★「전부 부적합」은 심판 모델을 바꾼 이유 그 자체다 — 사라지면 안 된다.

    종전에는 chosen_index=0 이면 verdicts 를 만들기 전에 None 으로 떨어져
    후보별 판정도 나간 질의도 어디에도 안 남았다(Codex BLOCK-2).
    """
    imgs = [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": ""}
            for i in range(2)]
    _wire_research(monkeypatch, tmp_path, images=imgs, chosen=0)
    aud: Dict[str, Any] = {}
    assert er.research_reference(
        subject=_subject(1), world_facts_block="",
        out_path=tmp_path / "e.png", step_tag="t",
        openai_client=object(), audit=aud) is None
    assert aud["status"] == "no_usable"
    assert aud["queries"] == ["나간 질의"]
    assert len(aud["verdicts"]) == 2
    assert aud["candidate_urls"] == ["https://x/0.jpg", "https://x/1.jpg"]


def test_audit_marks_stage_that_stopped(monkeypatch, tmp_path):
    """실패 단계가 갈려야 「전부 부적합」과 「검색이 안 됐다」를 안 섞는다."""
    imgs = [{"image_url": "https://x/0.jpg", "thumbnail_url": ""}]
    _wire_research(monkeypatch, tmp_path, images=imgs, downloads_ok=False)
    aud: Dict[str, Any] = {}
    er.research_reference(
        subject=_subject(1), world_facts_block="",
        out_path=tmp_path / "e.png", step_tag="t",
        openai_client=object(), audit=aud)
    assert aud["status"] == "download_failed"


def test_research_reference_download_dead_returns_none(monkeypatch, tmp_path):
    imgs = [{"image_url": "https://x/0.jpg", "thumbnail_url": ""}]
    _wire_research(monkeypatch, tmp_path, images=imgs, downloads_ok=False)
    assert er.research_reference(
        subject=_subject(1), world_facts_block="", out_path=tmp_path / "e.png",
        step_tag="t", openai_client=object()) is None


def test_research_reference_search_error_nonblocking(monkeypatch, tmp_path):
    import app.modules.pipeline.search_grounded_ref as sgr

    def boom(*a, **k):
        raise RuntimeError("search down")

    monkeypatch.setattr(sgr, "search_reference_images", boom)
    assert er.research_reference(
        subject=_subject(1), world_facts_block="", out_path=tmp_path / "e.png",
        step_tag="t", openai_client=object()) is None


def _cached_calls(monkeypatch, tmp_path, subject_texts, *, subject):
    """assess 는 어떤 문장에도 같은 대상을 내고, 조사 호출 수를 센다."""
    import app.modules.llm.llm_client as llm_client

    calls = {"assess": 0, "research": 0}
    store: Dict[str, Any] = {}

    def fake_assess(*a, **k):
        calls["assess"] += 1
        return {"subjects": [subject]}

    def fake_research(**kw):
        calls["research"] += 1
        kw["out_path"].parent.mkdir(parents=True, exist_ok=True)
        kw["out_path"].write_bytes(b"\x89PNG\r\n\x1a\nX")
        import hashlib as _h
        return {"subject": subject["subject_native"],
                "sha256": _h.sha256(b"\x89PNG\r\n\x1a\nX").hexdigest(),
                "file": kw["out_path"].name}

    monkeypatch.setattr(llm_client, "call_structured", fake_assess)
    monkeypatch.setattr(er, "research_reference", fake_research)
    for txt in subject_texts:
        er.assess_and_research_cached(
            step_tag="t", subject_text=txt, world_facts_block="세계",
            out_dir=tmp_path,
            cache_get=lambda k: store.get(k),
            cache_put=lambda k, v: store.__setitem__(k, v))
    return calls, store


def test_same_subject_from_different_text_researches_once(
        monkeypatch, tmp_path):
    """★샷마다 다른 문장이 같은 대상을 내면 **조사는 1회**다 (Codex BLOCK-1).

    실측: 완주 판 두 주행의 `place_en` 12건이 전부 다른 문장이라 종전
    신원(참조 키가 판별 키를 품음)으로는 적중률이 0 이었다. 조사가 실제로
    달려 있는 것은 저작된 대상·질의이지 그것을 낳은 문장이 아니다.
    """
    texts = ["좁은 정비소 안, 작업대 곁",
             "정비소 실내 — 문간에서 본 작업 구역",
             "기름때 묻은 작업대가 있는 좁은 실내"]
    calls, store = _cached_calls(
        monkeypatch, tmp_path, texts, subject=_subject(7))
    assert calls["assess"] == 3      # 판별은 문장마다 (값싸고 샷별 판단)
    assert calls["research"] == 1    # 조사는 한 번 (검색·다운로드·심판)
    # ★이 시험이 무엇을 잠그는지 — 판별 키는 문장 수만큼 갈리는데
    #  조사 키만 하나로 모인다. 종전에는 조사 키가 판별 키를 품어 같이
    #  셋으로 갈렸고, 그것이 유료 검색을 세 번 사게 한 자리다.
    assert len([k for k in store if k.startswith("era_assess::")]) == 3
    assert len([k for k in store if k.startswith("era_ref::")]) == 1


def test_physical_model_change_moves_both_keys(monkeypatch, tmp_path):
    """★alias 는 그대로인데 **설정의 실제 모델만** 바뀐 경우 (Codex BLOCK).

    `gemini-pro` 는 모델 이름이 아니라 `gemini_text_model` 로 매핑되는
    alias 다. alias 만 신원에 넣으면 판 교체가 캐시를 못 뚫는다.
    """
    from app.core.config import settings

    calls, store = _cached_calls(
        monkeypatch, tmp_path, ["같은 문장"], subject=_subject(9))
    assert calls["research"] == 1
    monkeypatch.setattr(settings, "gemini_text_model", "다른-물리-판")
    _, store2 = _cached_calls(
        monkeypatch, tmp_path, ["같은 문장"], subject=_subject(9))
    a1 = {k for k in store if k.startswith("era_assess::")}
    a2 = {k for k in store2 if k.startswith("era_assess::")}
    r1 = {k for k in store if k.startswith("era_ref::")}
    r2 = {k for k in store2 if k.startswith("era_ref::")}
    # 판별은 gemini-flash(=gemini_flash_model)라 안 움직이고,
    # 선택은 gemini-pro(=gemini_text_model)라 움직여야 한다.
    assert a1 == a2
    assert r1 != r2


def test_physical_model_appears_in_step_config_hash(monkeypatch):
    """스텝 층 payload 에도 물리 이름이 실린다 — 완료 스텝 SKIP 방지."""
    import app.modules.pipeline.era_research as _er

    assert _er.resolve_model_physical("gemini-pro") != "gemini-pro"
    assert _er.resolve_model_physical("gemini-flash") != "gemini-flash"
    # 모르는 alias 는 그대로 — 물리 이름을 직접 쓴 경우다.
    assert _er.resolve_model_physical("직접-쓴-모델") == "직접-쓴-모델"


def test_pick_model_change_invalidates_reference(monkeypatch, tmp_path):
    """심판 모델을 바꾸면 옛 선택본을 재사용하지 않는다 (Codex BLOCK-3)."""
    calls, store = _cached_calls(
        monkeypatch, tmp_path, ["같은 문장"], subject=_subject(8))
    assert calls["research"] == 1
    monkeypatch.setattr(er, "PICK_MODEL", "다른-심판-모델")
    calls2, store2 = _cached_calls(
        monkeypatch, tmp_path, ["같은 문장"], subject=_subject(8))
    assert calls2["research"] == 1
    assert set(store) != set(store2)


# ══════════ 정본 신원 (2026-08-27, 감사 2-C) ══════════
#
# 실측이 이 절의 근거다 — records 전수 19파일에서 저작 대상 179건 중
# 서로 다른 문자열이 177종, **적중 1.1%**. 3씬 시나리오(물리 장소 2곳)에서
# 조사가 6번 돌았고 자전거 수리점 하나가 네 번이었다:
#
#     한국의 동네 자전거 수리점 (자전거포) 내부 작업 공간
#     대한민국의 동네 자전거 수리점 내부 (2020년대)
#     현대 한국의 동네 자전거 수리점 내부 작업 공간
#     대한민국의 현대식 동네 자전거 수리점 내부
#
# 뿌리는 캐시 신원이 **LLM 자유 저작**에 매여 있던 것이다.


def _scoped(monkeypatch, tmp_path, runs, *, subject_of=None):
    """정본 신원을 주며 여러 번 부른다. runs = [(text, sid, role, sha), …]

    ★대상 저작은 부를 때마다 **다르게** 낸다 — 그것이 실물이다. 저작이
     같아서 합쳐지는 것은 이 판이 재려는 것이 아니다.
    """
    import app.modules.llm.llm_client as llm_client

    calls = {"assess": 0, "research": 0}
    store: Dict[str, Any] = {}
    outcomes = []

    def fake_assess(*a, **k):
        calls["assess"] += 1
        n = calls["assess"]
        return {"subjects": [(subject_of or (lambda i: {
            "subject_native": f"저작이 매번 다른 이름 {i}",
            "search_terms_native": [f"질의{i}a", f"질의{i}b"],
            "language_lock_native": f"원어로만 답하라 {i}",
            "reason_ko": "근거"}))(n)]}

    def fake_research(**kw):
        calls["research"] += 1
        kw["out_path"].parent.mkdir(parents=True, exist_ok=True)
        kw["out_path"].write_bytes(b"\x89PNG\r\n\x1a\nX")
        import hashlib as _h
        return {"subject": "x",
                "sha256": _h.sha256(b"\x89PNG\r\n\x1a\nX").hexdigest(),
                "file": kw["out_path"].name}

    monkeypatch.setattr(llm_client, "call_structured", fake_assess)
    monkeypatch.setattr(er, "research_reference", fake_research)
    for txt, sid, role, sha in runs:
        oc: Dict[str, Any] = {}
        er.assess_and_research_cached(
            step_tag="t", subject_text=txt, world_facts_block="세계",
            out_dir=tmp_path,
            cache_get=lambda k: store.get(k),
            cache_put=lambda k, v: store.__setitem__(k, v),
            outcome=oc,
            canonical_scope_id=sid, canonical_scope_role=role,
            canonical_scope_sha=sha)
        outcomes.append(oc)
    return calls, store, outcomes


IN, EX = er.SCOPE_ROLE_INTERIOR, er.SCOPE_ROLE_EXTERIOR


def test_one_location_many_wordings_costs_once(monkeypatch, tmp_path):
    """★같은 장소의 여러 문안이 **판별 1회·조사 1회**로 합쳐진다.

    저작은 부를 때마다 다르다 — 그래도 신원이 정본이라 합쳐져야 한다.
    이것이 실측 6회를 1회로 만드는 자리다.
    """
    calls, store, _ = _scoped(monkeypatch, tmp_path, [
        ("좁은 정비소 안, 작업대 곁", "L01", IN, "canon1"),
        ("정비소 실내 — 문간에서 본 작업 구역", "L01", IN, "canon1"),
        ("기름때 묻은 작업대가 있는 좁은 실내", "L01", IN, "canon1"),
        ("자전거포 안쪽", "L01", IN, "canon1"),
    ])
    assert calls["assess"] == 1, "판별이 문안마다 다시 돌았다"
    assert calls["research"] == 1, "조사가 문안마다 다시 돌았다"
    assert len([k for k in store if k.startswith("era_assess::")]) == 1
    assert len([k for k in store if k.startswith("era_ref::")]) == 1


def test_indoor_and_outdoor_of_one_location_stay_apart(monkeypatch, tmp_path):
    """★같은 `L01` 이라도 **안과 밖은 다른 참조**다.

    역할을 안 나누면 실내 뷰와 실외 뷰가 한 장으로 합쳐진다
    (2026-08-27 Codex 정정).
    """
    calls, store, _ = _scoped(monkeypatch, tmp_path, [
        ("정비소 실내", "L01", IN, "canon1"),
        ("정비소 앞 길", "L01", EX, "canon1"),
    ])
    assert calls["research"] == 2, "안과 밖이 한 참조로 합쳐졌다"
    assert len([k for k in store if k.startswith("era_ref::")]) == 2


def test_different_locations_never_merge(monkeypatch, tmp_path):
    """★서로 다른 장소는 **절대** 안 합쳐진다 — 거짓 적중 양성 대조."""
    calls, store, _ = _scoped(monkeypatch, tmp_path, [
        ("정비소 실내", "L01", IN, "canon1"),
        ("교실 실내", "L02", IN, "canon2"),
    ])
    assert calls["research"] == 2
    assert len([k for k in store if k.startswith("era_ref::")]) == 2


def test_canon_content_change_moves_the_key(monkeypatch, tmp_path):
    """★정본 내용이 고쳐지면 키가 움직인다 — 옛 참조가 안 남는다."""
    calls, _, _ = _scoped(monkeypatch, tmp_path, [
        ("정비소 실내", "L01", IN, "canon1"),
        ("정비소 실내", "L01", IN, "canon2"),   # 정본 서술이 바뀌었다
    ])
    assert calls["research"] == 2, "정본이 바뀌었는데 옛 참조를 재사용했다"


def test_missing_or_unknown_scope_does_not_merge(monkeypatch, tmp_path):
    """★셋 중 하나라도 없거나 역할이 모르는 값이면 **합치지 않는다.**

    잘못 합쳐 다른 장소의 참조를 재사용하는 것이 중복 조사보다 나쁘다.
    사유는 기록에 남는다.

    ★문안을 **다르게** 준다 — 그것이 실물이다(샷마다 문장이 다르다).
     같은 문안을 주면 옛 키가 어차피 합쳐져서 「정본이 합친 것」과
     「옛 동작이 합친 것」이 안 갈린다.
    """
    calls, _, oc = _scoped(monkeypatch, tmp_path, [
        ("정비소 실내 A", "L01", IN, None),          # sha 결손
        ("정비소 실내 B", "L01", None, "canon1"),    # 역할 결손
        ("정비소 실내 C", None, IN, "canon1"),       # id 결손
        ("정비소 실내 D", "L01", "복도", "canon1"),  # 모르는 역할
    ])
    assert calls["research"] == 4, "신원이 불완전한데 합쳤다"
    assert all(o.get("identity_fallback") for o in oc)
    assert oc[3]["reason"] == "scope_role_unknown"
    assert oc[0]["reason"] == "scope_incomplete"


def test_a_valid_scope_is_recorded_as_canonical(monkeypatch, tmp_path):
    """★어느 신원으로 갔는지가 **기록에 남는다** — 나중에 셀 수 있게."""
    _, store, oc = _scoped(monkeypatch, tmp_path, [
        ("정비소 실내", "L01", IN, "canon1"),
    ])
    assert oc[0]["identity"] == "canonical"
    assert oc[0]["scope_id"] == "L01"
    a = [v for k, v in store.items() if k.startswith("era_assess::")][0]
    assert a["identity"] == "canonical" and a["scope_role"] == IN


def test_the_schema_asks_for_one_subject_not_four():
    """★소비자는 `subjects[0]` 하나만 조사한다 — 물어 봐야 안 쓴다."""
    props = er.build_assess_schema()["properties"]["subjects"]
    assert props["maxItems"] == 1