"""i2i 시네마틱 변환 스테이지 (2026-08-13 #108) — 결정론 계약 시험.

사용자 확정 구조: 최종 스틸 = nb2 생성 → grok 2.0 i2i 시네마틱 변환
(하이브리드). sel 확정 직후 스테이지가 변환본(_cine)을 만들고, 최종
자산 영속은 변환본을 쓴다. 원본 _sel 은 prev 체인 앵커로 보존.

계약 요점:
  · 문안 SOT = still_recipe 팩 v19 단일 스템 `cine_transform`
    (화풍 교체 = 팩 버전 교체 + selector bump).
  · 지문 = 계약 버전 + 문안 스템 내용 해시 + 모델 + 원본 sel bytes.
  · 재사용 = 지문 일치 + applied + 파일 실재 → 유료 호출 0 + records
    무변경(#77-B 지출 탐지가 거짓 지출을 읽지 않아야 한다).
  · 실패 = 원본 fallback + 기록 (다음 방문에서 재시도).
  · 예산 초과(ImageCallBudgetExceeded)는 삼키지 않는다 — 조용한
    미변환 완주 금지.
  · OFF(default) = 스텝 config_hash byte-identical.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List, Optional

import pytest

from app.core.image_call_budget import ImageCallBudgetExceeded


# ── 시험용 스텁 (구현과 독립 — 계약 shape 만 공유) ──────────────────


class _StubRecords:
    """_Records 계약(data dict + save)만 흉내 — 실제 json 왕복."""

    def __init__(self, path: Path):
        self._path = path
        self.data: Dict[str, Any] = {}
        self.save_calls = 0
        if path.exists():
            self.data = json.loads(path.read_text(encoding="utf-8"))

    def save(self) -> None:
        self.save_calls += 1
        self._path.write_text(
            json.dumps(self.data, ensure_ascii=False, indent=1),
            encoding="utf-8",
        )


class _StubClient:
    """GrokImageClient 계약(set_context/generate_image)만 흉내."""

    def __init__(self, png: bytes = b"CINE-PNG", fail: Optional[Exception] = None):
        self.calls: List[Dict[str, Any]] = []
        self.contexts: List[Dict[str, Any]] = []
        self._png = png
        self._fail = fail

    def set_context(self, **kw):
        self.contexts.append(dict(kw))
        return self

    def generate_image(
        self,
        prompt: str,
        reference_images=None,
        aspect_ratio: str = "16:9",
        labeled_references=None,
    ):
        self.calls.append({
            "prompt": prompt,
            "labeled_references": labeled_references,
        })
        if self._fail is not None:
            raise self._fail
        return self._png, 321


def _run(tmp_path: Path, *, client: _StubClient, records=None,
         sel_bytes: bytes = b"SEL-PNG", stem_hash: str = "stemhash-1",
         model: str = "x-ai/grok-imagine-image-2.0"):
    from app.modules.pipeline.cine_transform import (
        resolve_or_run_cine_transform,
    )

    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    sel = recipe_dir / "S1sh1_sel.png"
    sel.write_bytes(sel_bytes)
    if records is None:
        records = _StubRecords(recipe_dir / "records.json")
    rec = resolve_or_run_cine_transform(
        tag="S1sh1",
        sel_path=sel,
        recipe_dir=recipe_dir,
        records=records,
        client=client,
        prompt="Rework this still.",
        model=model,
        stem_content_hash=stem_hash,
        pack="19.202608131440",
        context={
            "project_id": "p1", "episode_id": "e1",
            "operation_type": "still_cine_transform",
            "still_id": "sid-1", "multiroll_tag": "still_S1sh1_cine",
        },
    )
    return rec, records, recipe_dir


# ── 팩 스템 SOT ──────────────────────────────────────────────────────


def test_cine_pack_stem_loads_default_prompt():
    """문안 SOT = 팩 v19 cine_transform 스템 — 로드 결과가 스템 파일
    내용과 일치해야 한다(다른 스템·조립과 섞이지 않는 단일 SOT)."""
    from app.modules.pipeline.still_recipe import (
        CINE_TRANSFORM_PROMPT_VERSION,
        build_cine_transform_prompt,
        resolve_prompt_version,
    )

    resolved = resolve_prompt_version(CINE_TRANSFORM_PROMPT_VERSION)
    loaded = build_cine_transform_prompt()
    assert loaded, "빈 문안이면 변환 지시가 없다"
    from app.modules.prompt_loader import PROMPTS_BASE

    stem = PROMPTS_BASE / "still_recipe" / resolved / "cine_transform.md"
    assert loaded == stem.read_text(encoding="utf-8").strip()
    # 짧은 범용 문안 계약 — 조립 전문(수천 자)이 아니라 변환 지시 한 단락
    assert len(loaded) < 2000


# ── 지문 ────────────────────────────────────────────────────────────


def test_cine_fingerprint_sensitive_to_each_input():
    from app.modules.pipeline.cine_transform import cine_fingerprint

    base = cine_fingerprint(
        stem_content_hash="s1", model="m1", sel_bytes=b"img")
    assert base == cine_fingerprint(
        stem_content_hash="s1", model="m1", sel_bytes=b"img")
    assert base != cine_fingerprint(
        stem_content_hash="s2", model="m1", sel_bytes=b"img")
    assert base != cine_fingerprint(
        stem_content_hash="s1", model="m2", sel_bytes=b"img")
    assert base != cine_fingerprint(
        stem_content_hash="s1", model="m1", sel_bytes=b"other")


# ── resolve-or-run ──────────────────────────────────────────────────


def test_fresh_run_calls_client_writes_file_and_record(tmp_path):
    client = _StubClient()
    rec, records, recipe_dir = _run(tmp_path, client=client)

    assert rec["applied"] is True
    assert rec["file"] == "S1sh1_cine.png"
    assert (recipe_dir / "S1sh1_cine.png").read_bytes() == b"CINE-PNG"
    # 원본 sel 보존
    assert (recipe_dir / "S1sh1_sel.png").read_bytes() == b"SEL-PNG"
    # 유료 호출 1회 — 문안+원본 1장 라벨 참조
    assert len(client.calls) == 1
    assert client.calls[0]["prompt"] == "Rework this still."
    labeled = client.calls[0]["labeled_references"]
    assert len(labeled) == 1
    assert labeled[0][0] == "SOURCE STILL"
    assert labeled[0][1] == b"SEL-PNG"
    # 컨텍스트 전달(기록·Opik 경계는 클라이언트 기계가 담당)
    assert client.contexts[0]["operation_type"] == "still_cine_transform"
    assert client.contexts[0]["multiroll_tag"] == "still_S1sh1_cine"
    # records 영속 — 파일이 durable 이 된 뒤 기록
    stored = records.data["S1sh1::cine"]
    assert stored["applied"] is True
    assert stored["fingerprint"] == rec["fingerprint"]
    assert stored["pack"] == "19.202608131440"
    assert records.save_calls >= 1
    # 최종 자산 lineage 재료 — grok 이 실제 본 직접 입력의 신원
    import hashlib as _hl

    assert stored["source_file"] == "S1sh1_sel.png"
    assert stored["source_sha256"] == _hl.sha256(b"SEL-PNG").hexdigest()


def test_reuse_skips_client_and_keeps_records_stable(tmp_path):
    """지문 일치 + 파일 실재 = 유료 호출 0 + records 무변경 —
    #77-B 지출 탐지(_jit_tag_snapshot)가 거짓 지출을 읽지 않는 전제."""
    first = _StubClient()
    _run(tmp_path, client=first)

    second = _StubClient()
    records2 = _StubRecords(tmp_path / "recipe" / "records.json")
    before = json.dumps(records2.data, sort_keys=True, ensure_ascii=False)
    rec, records2, _ = _run(tmp_path, client=second, records=records2)

    assert second.calls == []
    assert rec["applied"] is True
    assert rec.get("reused") is True
    after = json.dumps(records2.data, sort_keys=True, ensure_ascii=False)
    assert before == after, "재사용 바퀴가 records 를 움직이면 거짓 지출로 읽힌다"
    assert records2.save_calls == 0
    # 반환 record 는 저장본과 동일 내용(transient reused 제외)
    stored = records2.data["S1sh1::cine"]
    assert {k: v for k, v in rec.items() if k != "reused"} == stored


def test_sel_bytes_change_reruns(tmp_path):
    _run(tmp_path, client=_StubClient(png=b"CINE-1"))
    client2 = _StubClient(png=b"CINE-2")
    records2 = _StubRecords(tmp_path / "recipe" / "records.json")
    rec, _, recipe_dir = _run(
        tmp_path, client=client2, records=records2, sel_bytes=b"SEL-CHANGED")

    assert len(client2.calls) == 1, "원본이 바뀌면 변환도 다시"
    assert rec["applied"] is True
    assert (recipe_dir / "S1sh1_cine.png").read_bytes() == b"CINE-2"


def test_stem_hash_change_reruns(tmp_path):
    """화풍 교체 = 팩 버전 교체 → 스템 해시 변화 → 전 샷 재변환."""
    _run(tmp_path, client=_StubClient(png=b"CINE-1"))
    client2 = _StubClient(png=b"CINE-2")
    records2 = _StubRecords(tmp_path / "recipe" / "records.json")
    rec, _, _ = _run(
        tmp_path, client=client2, records=records2, stem_hash="stemhash-2")

    assert len(client2.calls) == 1
    assert rec["applied"] is True


def test_failure_falls_back_records_error(tmp_path):
    client = _StubClient(fail=RuntimeError("Grok image API error 500"))
    rec, records, recipe_dir = _run(tmp_path, client=client)

    assert rec["applied"] is False
    assert "500" in rec["error"]
    assert not (recipe_dir / "S1sh1_cine.png").exists()
    # 실패도 기록 — 다음 방문 재시도의 근거
    stored = records.data["S1sh1::cine"]
    assert stored["applied"] is False
    assert stored["fingerprint"] == rec["fingerprint"]


def test_prior_failure_retries_on_next_visit(tmp_path):
    _run(tmp_path, client=_StubClient(fail=RuntimeError("transient")))
    client2 = _StubClient()
    records2 = _StubRecords(tmp_path / "recipe" / "records.json")
    rec, _, recipe_dir = _run(tmp_path, client=client2, records=records2)

    assert len(client2.calls) == 1, "기록된 실패는 재방문에서 재시도한다"
    assert rec["applied"] is True
    assert (recipe_dir / "S1sh1_cine.png").exists()


def test_missing_file_with_prior_success_reruns(tmp_path):
    _run(tmp_path, client=_StubClient(png=b"CINE-1"))
    (tmp_path / "recipe" / "S1sh1_cine.png").unlink()
    client2 = _StubClient(png=b"CINE-2")
    records2 = _StubRecords(tmp_path / "recipe" / "records.json")
    rec, _, recipe_dir = _run(tmp_path, client=client2, records=records2)

    assert len(client2.calls) == 1, "산출 파일이 없으면 기록만으로 재사용 금지"
    assert (recipe_dir / "S1sh1_cine.png").read_bytes() == b"CINE-2"


def test_budget_exceeded_propagates(tmp_path):
    """예산 브레이크는 fallback 으로 삼키지 않는다 — 조용한 미변환
    완주는 '변환된 최종본' 오독을 만든다(결과 오독 범주)."""
    client = _StubClient(
        fail=ImageCallBudgetExceeded(cap=1, used=1, source="test"))
    with pytest.raises(ImageCallBudgetExceeded):
        _run(tmp_path, client=client)


# ── 설정 API .env 영속 (UI 토글의 재기동 생존 계약) ──────────────────


def test_persist_env_flag_replaces_only_target_line(tmp_path):
    from app.api.v1.settings import persist_env_flag

    env = tmp_path / ".env"
    env.write_text(
        "A=1\nSTILL_CINE_TRANSFORM_ENABLED=false\nB=2\n", encoding="utf-8")
    persist_env_flag("STILL_CINE_TRANSFORM_ENABLED", "true", env_path=env)
    assert env.read_text(encoding="utf-8") == (
        "A=1\nSTILL_CINE_TRANSFORM_ENABLED=true\nB=2\n"
    ), "대상 키 한 줄만 바뀌고 다른 줄은 byte 보존"


def test_persist_env_flag_appends_when_missing(tmp_path):
    from app.api.v1.settings import persist_env_flag

    env = tmp_path / ".env"
    env.write_text("A=1", encoding="utf-8")  # 개행 없는 마지막 줄
    persist_env_flag("STILL_CINE_TRANSFORM_ENABLED", "true", env_path=env)
    assert env.read_text(encoding="utf-8") == (
        "A=1\nSTILL_CINE_TRANSFORM_ENABLED=true\n"
    )


def test_persist_env_flag_missing_file_raises(tmp_path):
    from app.api.v1.settings import persist_env_flag

    with pytest.raises(RuntimeError):
        persist_env_flag(
            "STILL_CINE_TRANSFORM_ENABLED", "true",
            env_path=tmp_path / "no_such.env")


# ── 스텝 config_hash 스탬프 (ON 일 때만 — OFF byte-identical) ────────


def test_scene_image_step_hash_stamps_cine_on_only(monkeypatch):
    from app.core.config import settings
    from app.core.steps.image_steps import SceneImagePipelineStep

    step = object.__new__(SceneImagePipelineStep)
    step.project_config = {"some": "config"}

    # 환경 핀 — 운영 .env 상태와 무관하게 이 시험의 전제를 고정
    monkeypatch.setattr(settings, "scene_image_target_scenes", "")
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "still_identity_ref_role_enabled", False)
    monkeypatch.setattr(settings, "still_confined_fp_enabled", False)
    monkeypatch.setattr(settings, "still_image_backend", "nb2")
    monkeypatch.setattr(settings, "still_recipe_mode", "v1")
    monkeypatch.setattr(settings, "still_cine_transform_enabled", False)

    h_off = SceneImagePipelineStep._config_hash(step)
    monkeypatch.setattr(settings, "still_cine_transform_enabled", True)
    h_on = SceneImagePipelineStep._config_hash(step)
    assert h_on != h_off, "ON 이 완료 스텝 clean SKIP 을 못 뚫으면 전환이 조용히 안 돈다"

    monkeypatch.setattr(settings, "still_cine_transform_enabled", False)
    assert SceneImagePipelineStep._config_hash(step) == h_off


def test_scene_image_step_hash_folds_cine_contract(monkeypatch):
    """Codex R1 BLOCK-2: 계약 버전이 outer hash 에도 접혀야 완료 스텝이
    계약 bump 를 본다 — inner 지문만이면 clean SKIP 이 먼저라 영원히
    실행되지 않는다."""
    import app.modules.pipeline.cine_transform as cine_mod
    from app.core.config import settings
    from app.core.steps.image_steps import SceneImagePipelineStep

    step = object.__new__(SceneImagePipelineStep)
    step.project_config = {"some": "config"}
    monkeypatch.setattr(settings, "scene_image_target_scenes", "")
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "still_identity_ref_role_enabled", False)
    monkeypatch.setattr(settings, "still_confined_fp_enabled", False)
    monkeypatch.setattr(settings, "still_image_backend", "nb2")
    monkeypatch.setattr(settings, "still_recipe_mode", "v1")
    monkeypatch.setattr(settings, "still_cine_transform_enabled", True)

    h_v1 = SceneImagePipelineStep._config_hash(step)
    monkeypatch.setattr(cine_mod, "CINE_CONTRACT_VERSION", "cine_vNEXT")
    h_v2 = SceneImagePipelineStep._config_hash(step)
    assert h_v1 != h_v2


def test_settings_put_keeps_runtime_off_when_persist_fails(monkeypatch):
    """Codex R1 BLOCK-4: .env 영속 실패면 런타임 플래그도 안 바뀐다 —
    역순이면 UI 는 실패로 읽는데 프로세스는 ON 이라 유료 변환이 나간다."""
    import app.api.v1.settings as settings_api
    from app.core.config import settings

    monkeypatch.setattr(settings, "still_cine_transform_enabled", False)

    def _boom(key, value, env_path=None):
        raise RuntimeError("SAMPLE persist down")

    monkeypatch.setattr(settings_api, "persist_env_flag", _boom)
    body = settings_api.StillPipelineUpdate(
        still_cine_transform_enabled=True)
    with pytest.raises(RuntimeError):
        settings_api.update_still_pipeline_settings(body, _user=None)
    assert settings.still_cine_transform_enabled is False


# ── 공용 「보냈는지 모른다」 예외도 표식을 남긴다 (2026-08-26 자체 리뷰) ──

def test_공용_전송불명_예외도_다시_안_보내게_표식을_남긴다(tmp_path):
    """★이 판정이 클래스 **이름**이라 기본 경로에서 열려 있었다.

    종전: `type(exc).__name__ == "ReveSubmissionUnknown"`.
    그런데 Grok·Gemini 가 던지는 것은 이름이 다른 `ImageSubmissionUnknown`
    이고, **grok 이 기본 변환 제공자**(`config.still_cine_provider`)다.
    표식이 안 세워지면 다음 걷기가 같은 변환을 또 산다 — 이 판이 막으려던
    바로 그 중복 결제다.
    """
    from app.modules.llm.image_send_state import ImageSubmissionUnknown

    client = _StubClient(fail=ImageSubmissionUnknown(
        "응답을 못 받았다", cause="wall_timeout"))
    rec, records, _dir = _run(tmp_path, client=client)

    assert rec["applied"] is False
    assert rec["submission_unknown"] is True, (
        "공용 전송불명 예외에 표식이 안 남으면 다음 걷기가 또 산다")
    assert rec["submission_unknown_reason"] == "wall_timeout"
    # 다음 방문은 자동으로 다시 보내지 않는다.
    again = _StubClient()
    _rec2, _r2, _d2 = _run(tmp_path, client=again, records=records)
    assert len(again.calls) == 0, "표식이 있는데 또 보냈다"


def test_일반_실패는_표식_없이_다음_방문에_다시_시도한다(tmp_path):
    """전송불명이 **아닌** 실패는 종전대로 재시도 대상이다."""
    client = _StubClient(fail=RuntimeError("일시 장애"))
    rec, _records, _dir = _run(tmp_path, client=client)

    assert rec["applied"] is False
    assert "submission_unknown" not in rec