"""재개할 때 성공한 도면은 다시 읽지 않는다.

## 왜 이 시험이 있나 (2026-08-19 실측)

실제 VLM 제공자를 켜면 이 단계는 도면 하나마다 모델을 부르고, 답이 매번
조금씩 다르다. 그런데 도면 넷다섯이 늘 실패해서 스텝이 `partial` 로
남고, `partial` 은 재개마다 다시 불린다. 다시 불릴 때 **성공한 도면까지
전부 다시 읽으니** 기하 값이 바뀌고, 그 도면을 쓰는 장면이 「재료가
낡았다」로 판정돼 그림이 다시 만들어졌다 — 08-19 저녁 한 바퀴에 66장.

여기에 변환 검열 실패 두 장이 겹쳐 **완주 조건이 원리적으로 충족될 수
없었다**. 자동 재개의 3회 제한이 유일한 제동이었다.

지키는 계약은 넷이다.

1. 입력이 그대로면 성공한 도면은 다시 읽지 않고 값을 **1비트도 안 바꾸고**
   그대로 옮긴다.
2. 실패한 도면은 다시 읽는다 — 그것이 재개의 목적이다.
3. 입력(도면·마커 목록)이 바뀌면 다시 읽는다. 설정만 보는 config_hash
   로는 이것을 못 가른다.
4. `force` 는 처음부터 다시라는 뜻이므로 재사용하지 않는다.
"""
from __future__ import annotations

from pathlib import Path
from typing import Any, Dict
from unittest.mock import MagicMock, patch

from app.core.steps.floor_plan_geometry_readback_step import (
    FloorPlanGeometryReadbackStep,
)
from app.modules.pipeline.floor_plan_geometry_readback import (
    BASE_OPENING,
    BASE_PERSISTENT_FIXTURE,
    BASE_PERSISTENT_FURNITURE,
    BASE_STRUCTURAL_UNIT,
)


def _dossier(fp_id: str, extra_label: str = "primary unit") -> Dict[str, Any]:
    return {
        "fp_id": fp_id,
        "fp_image_path": f"/p/c/e/floor_plan_render/{fp_id}.png",
        "grid_size": [10, 10],
        "dwelling_identity": {
            "standard_of_living_band": "modest_residential"},
        "base_marker_inventory": [
            {"number": 1, "label": extra_label, "category": "area",
             "position_hint": "",
             "base_layer_decision": BASE_STRUCTURAL_UNIT},
            {"number": 2, "label": "secondary unit", "category": "area",
             "position_hint": "",
             "base_layer_decision": BASE_STRUCTURAL_UNIT},
            {"number": 3, "label": "interior opening", "category": "opening",
             "position_hint": "", "base_layer_decision": BASE_OPENING},
            {"number": 4, "label": "service counter", "category": "furniture",
             "position_hint": "",
             "base_layer_decision": BASE_PERSISTENT_FIXTURE},
            {"number": 5, "label": "anchor seating", "category": "furniture",
             "position_hint": "",
             "base_layer_decision": BASE_PERSISTENT_FURNITURE},
        ],
        "per_bg_render_facts_by_bg_id": {},
        "diagnostics": [],
    }


class _CountingProvider:
    """부른 횟수를 세고, 부를 때마다 **다른 답**을 준다.

    실측의 핵심이 이것이다 — 다시 읽으면 값이 달라진다.
    """

    def __init__(self):
        self.calls = []
        self._n = 0

    def __call__(self, *, dossier, fp_image_path, grid_size):
        self._n += 1
        self.calls.append(str(fp_image_path))
        return {
            "status": "ok",
            "fp_id": dossier["fp_id"],
            "grid_size": list(grid_size),
            "observed_markers": [
                # 부를 때마다 칸이 달라진다(확률적 산출을 흉내).
                {"number": 1, "row": 1 + self._n % 3, "col": 1,
                 "kind": "base_structural_unit"},
            ],
            "missing_markers": [],
            "extra_markers": [],
            "confidence": 0.9,
            "diagnostics": [],
        }


def _new_step(tmp_path: Path, cp_map: dict) -> FloorPlanGeometryReadbackStep:
    step = FloorPlanGeometryReadbackStep.__new__(FloorPlanGeometryReadbackStep)
    step.project_id = "p"
    step.episode_id = "e"
    step.project_config = {}
    step.build_opik_metadata = MagicMock(return_value={})
    step._load_prev_checkpoint = MagicMock(side_effect=lambda sid: cp_map.get(sid))
    return step


def _run(step, tmp_path: Path, mode: str = "resume"):
    with patch("app.core.config.settings.projects_dir", str(tmp_path)), \
         patch("app.core.config.settings.background_mode", "on"), \
         patch("app.core.config.settings.floor_plan_prompt_version", "6"), \
         patch("app.core.config.settings.base_location_dossier_enabled",
               True), \
         patch("app.core.config.settings."
               "floor_plan_vlm_readback_real_provider_enabled", False), \
         patch("app.core.config.settings."
               "floor_plan_geometry_readback_enabled", True), \
         patch("app.core.config.settings."
               "floor_plan_geometry_readback_max_retries", 0):
        return step._execute(mode=mode)


def _cp(dossiers: Dict[str, Any], prior=None) -> dict:
    out = {"base_location_dossier": {"data": {"dossiers": dossiers}}}
    if prior is not None:
        out["floor_plan_geometry_readback"] = prior
    return out


# ── 계약 1: 성공분은 그대로, 값이 안 바뀐다 ─────────────────────────

def test_successful_fp_is_reused_verbatim_on_resume(tmp_path):
    dossiers = {"fp_a": _dossier("fp_a"), "fp_b": _dossier("fp_b")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    assert first["completed_count"] == 2
    assert "reused_fp_ids" not in first["data"]

    second = _run(_new_step(tmp_path, _cp(dossiers, prior=first)), tmp_path)
    assert second["data"]["reused_fp_ids"] == ["fp_a", "fp_b"]
    assert second["completed_count"] == 2
    # ★값이 1비트도 안 바뀌어야 하류가 "재료가 낡았다"고 안 읽는다.
    assert second["data"]["per_fp"] == first["data"]["per_fp"]


def test_reuse_needs_the_html_file_to_exist(tmp_path):
    """기록만 믿지 않는다 — 산출 파일이 없으면 다시 읽는다."""
    dossiers = {"fp_a": _dossier("fp_a")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    html = (tmp_path / "p" / "checkpoints" / "episodes" / "e"
            / "floor_plan_geometry_readback" / "review_html" / "fp_a.html")
    assert html.exists()
    html.unlink()

    second = _run(_new_step(tmp_path, _cp(dossiers, prior=first)), tmp_path)
    assert "reused_fp_ids" not in second["data"]
    assert html.exists(), "다시 읽었으면 산출도 다시 써야 한다"


# ── 계약 2: 실패분은 다시 읽는다 ────────────────────────────────────

def test_failed_fp_is_read_again_while_successful_one_is_reused(tmp_path):
    good = _dossier("fp_a")
    bad = _dossier("fp_b")
    bad["base_marker_inventory"] = []       # 고정 실패(빈 목록)
    dossiers = {"fp_a": good, "fp_b": bad}

    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    assert first["completed_count"] == 1 and first["failed_count"] == 1

    second = _run(_new_step(tmp_path, _cp(dossiers, prior=first)), tmp_path)
    assert second["data"]["reused_fp_ids"] == ["fp_a"]
    assert second["failed_count"] == 1, "실패분은 다시 읽고 또 실패한다"
    assert "error" in second["data"]["per_fp"]["fp_b"]


# ── 계약 3: 입력이 바뀌면 다시 읽는다 ───────────────────────────────

def test_changed_dossier_is_not_reused(tmp_path):
    dossiers = {"fp_a": _dossier("fp_a")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)

    changed = {"fp_a": _dossier("fp_a", extra_label="a different unit")}
    second = _run(
        _new_step(tmp_path, _cp(changed, prior=first)), tmp_path)
    assert "reused_fp_ids" not in second["data"]


def test_redrawn_floor_plan_image_at_the_same_path_is_not_reused(tmp_path):
    """★도면 그림이 **같은 경로에서 다시 그려지면** 다시 읽어야 한다.

    dossier 는 그림의 경로만 갖고 내용 지문이 없다
    (base_location_dossier.py:647). dossier 만 보고 재사용하면 그림이
    바뀌었는데 옛 읽기를 그대로 쓰게 된다 — 이 단계가 실제로 보는 것은
    그 그림이다.
    """
    png = tmp_path / "fp_a.png"
    png.write_bytes(b"PNG-FIRST")
    dossier = _dossier("fp_a")
    dossier["fp_image_path"] = str(png)
    dossiers = {"fp_a": dossier}

    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    again = _run(_new_step(tmp_path, _cp(dossiers, prior=first)), tmp_path)
    assert again["data"]["reused_fp_ids"] == ["fp_a"], "그림이 같으면 재사용"

    png.write_bytes(b"PNG-REDRAWN")          # 같은 경로, 다른 그림
    third = _run(_new_step(tmp_path, _cp(dossiers, prior=first)), tmp_path)
    assert "reused_fp_ids" not in third["data"]


def test_config_hash_mismatch_disables_reuse(tmp_path):
    dossiers = {"fp_a": _dossier("fp_a")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    stale = {**first, "config_hash": "0" * 16}

    second = _run(_new_step(tmp_path, _cp(dossiers, prior=stale)), tmp_path)
    assert "reused_fp_ids" not in second["data"]


def test_prior_entry_without_input_hash_is_not_reused(tmp_path):
    """지문 없는 옛 기록은 재사용 대상이 아니다 — 무엇으로 만든 값인지
    모르는 것을 그대로 쓰면 낡은 값을 붙잡는다."""
    dossiers = {"fp_a": _dossier("fp_a")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    legacy = {
        **first,
        "data": {**first["data"], "per_fp": {
            "fp_a": {k: v for k, v in first["data"]["per_fp"]["fp_a"].items()
                     if k != "input_hash"}}},
    }
    second = _run(_new_step(tmp_path, _cp(dossiers, prior=legacy)), tmp_path)
    assert "reused_fp_ids" not in second["data"]


# ── 계약 4: force 는 재사용하지 않는다 ──────────────────────────────

def test_force_mode_never_reuses(tmp_path):
    dossiers = {"fp_a": _dossier("fp_a")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)
    second = _run(
        _new_step(tmp_path, _cp(dossiers, prior=first)), tmp_path,
        mode="force")
    assert "reused_fp_ids" not in second["data"]


# ── 모델을 실제로 안 부른다는 것까지 확인 ───────────────────────────

def test_reuse_reads_the_real_checkpoint_file_on_disk(tmp_path):
    """★직전 결과를 **파일에서** 되찾는 경로까지 잠근다.

    위 시험들은 직전 결과를 흉내로 건네준다. 그러면 파일 경로나 저장
    모양이 어긋나도 통과해 버리고, 재사용은 프로덕션에서 조용히 한 번도
    안 일어난다 — 고쳤다고 보고했는데 아무 일도 안 하는 부류다. 그래서
    이 시험만은 실제 체크포인트 파일을 만들어 놓고 읽게 한다.
    """
    import json as _json

    dossiers = {"fp_a": _dossier("fp_a")}
    first = _run(_new_step(tmp_path, _cp(dossiers)), tmp_path)

    # StepRunner 가 저장하는 그 자리·그 모양으로 쓴다(실물 확인:
    # projects/<pid>/checkpoints/episodes/<eid>/<step_id>/manifest.json,
    # 최상위에 config_hash 와 data).
    cp_dir = (tmp_path / "p" / "checkpoints" / "episodes" / "e"
              / "floor_plan_geometry_readback")
    cp_dir.mkdir(parents=True, exist_ok=True)
    (cp_dir / "manifest.json").write_text(
        _json.dumps({"status": "partial", **first}, ensure_ascii=False),
        encoding="utf-8")

    # 도면 자료만 흉내로 주고, 이 단계의 직전 결과는 **파일에서** 읽게 한다.
    step = FloorPlanGeometryReadbackStep.__new__(FloorPlanGeometryReadbackStep)
    step.project_id = "p"
    step.episode_id = "e"
    step.project_config = {}
    step.build_opik_metadata = MagicMock(return_value={})
    real_loader = FloorPlanGeometryReadbackStep._load_prev_checkpoint

    def _load(step_id):
        if step_id == "base_location_dossier":
            return _cp(dossiers)["base_location_dossier"]
        return real_loader(step, step_id)

    step._load_prev_checkpoint = _load
    second = _run(step, tmp_path)

    assert second["data"]["reused_fp_ids"] == ["fp_a"]
    assert second["data"]["per_fp"] == first["data"]["per_fp"]


def test_reuse_makes_no_provider_call(tmp_path):
    dossiers = {"fp_a": _dossier("fp_a"), "fp_b": _dossier("fp_b")}
    provider = _CountingProvider()
    step1 = _new_step(tmp_path, _cp(dossiers))
    step1.set_vlm_provider_for_testing(provider)
    first = _run(step1, tmp_path)
    assert len(provider.calls) == 2
    assert first["completed_count"] == 2

    provider2 = _CountingProvider()
    step2 = _new_step(tmp_path, _cp(dossiers, prior=first))
    step2.set_vlm_provider_for_testing(provider2)
    second = _run(step2, tmp_path)

    assert provider2.calls == [], "재사용이면 모델을 부르지 않는다"
    assert second["data"]["real_vlm_call_count"] == 0
    assert second["data"]["per_fp"] == first["data"]["per_fp"]