"""최종 i2i 변환 provider 교환 (2026-08-25 사용자 지시 "grok, reve 등으로
교환 가능하게") — 계약 시험.

무엇을 지키나:
  · 신원(provider·endpoint·model)이 **지문에 접힌다** — 같은 문안·같은 원본
    이라도 다른 제공자가 만든 그림은 다른 산출이다.
  · 그런데 **완주본을 다시 사지 않는다** — 판 번호 없는 옛 record 는 지금
    선택이 정확히 옛 Grok 일 때만 지금 것으로 읽는다(dual-read).
  · 접수 기반 제공자(fal queue)는 submit 1회가 유료 작업 1건이다 — 결말을
    못 봤으면 **다시 가져오기**하지 재구매하지 않는다($0.25 를 두 번 쓰지 않는다).
  · 오류는 **정확한 유형 이름**으로 가른다 — 검열과 「결과 없음」은 다른 실패다.
"""
from __future__ import annotations

import io
import json
import urllib.error
from pathlib import Path
from typing import Any, Dict, List, Optional

import pytest

from app.core.config import settings


@pytest.fixture(autouse=True)
def _pin_baseline(monkeypatch):
    """★기계의 `.env` 를 물지 않게 기준선을 못박는다.

    오늘 같은 함정에 세 번 빠졌다. 검사·provider·팩 selector 가 켜진 기계에서
    돌리면 시험이 재는 것이 「무엇이 계약인가」가 아니라 「이 기계가 어떻게
    설정됐나」가 된다 — 실제로 `STILL_CINE_VERIFY_ENABLED=true` 를 켜자
    재사용 갈래가 판정을 사러 가며 records 를 건드려 「무쓰기」 단정이 깨졌다.

    켜는 것을 재는 시험은 **자기 안에서 다시 켠다**.
    """
    for key, value in (
        ("still_cine_verify_enabled", False),
        ("still_cine_provider", "grok"),
        ("still_cine_stage_direction_enabled", False),
    ):
        monkeypatch.setattr(settings, key, value, raising=False)


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


class _StubRecords:
    def __init__(self, path: Path):
        self._path = path
        self.data: Dict[str, Any] = {}
        self.save_calls = 0

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


# 접수 응답이 돌려주는 신원 — 프로덕션 reve 가 실제로 넘기는 칸 그대로다.
# (조회 주소가 **둘 다** 있어야 프로세스가 죽었다 살아나도 조립에 안 기댄다)
_SUBMIT_INFO = {
    "provider": "reve",
    "endpoint": "reve/2.1/edit",
    "request_id": "req-777",
    "status_url": "https://queue/x/status",
    "response_url": "https://queue/x",
}


class _QueueClient:
    """접수 기반 제공자 계약만 흉내 — submit hook + 다시 가져오기 슬롯.

    ★프로덕션 client 를 상속하지 않는다: fake 를 구현과 같은 함수로 만들면
     둘이 같이 틀려 mock 을 검증하게 된다.
    """

    def __init__(
        self,
        png: bytes = b"CINE-PNG",
        *,
        gen_fail: Optional[Exception] = None,
        fetch_fail: Optional[Exception] = None,
        submit_before_fail: bool = False,
    ):
        self.gen_calls = 0
        self.fetch_calls: List[str] = []
        self.fetch_urls: List[tuple] = []
        self._png = png
        self._gen_fail = gen_fail
        self._fetch_fail = fetch_fail
        self._submit_before_fail = submit_before_fail
        self._hook = None
        self.contexts: List[Dict[str, Any]] = []

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

    def set_submit_hook(self, fn):
        self._hook = fn
        return self

    def generate_image(self, prompt, reference_images=None,
                       aspect_ratio="16:9", labeled_references=None):
        self.gen_calls += 1
        if self._submit_before_fail and self._hook:
            self._hook(dict(_SUBMIT_INFO))
        if self._gen_fail is not None:
            raise self._gen_fail
        if self._hook:
            self._hook(dict(_SUBMIT_INFO))
        return self._png, 42

    def fetch_submitted(self, request_id, *, prompt="",
                        labeled_references=None,
                        status_url="", response_url=""):
        self.fetch_calls.append(request_id)
        self.fetch_urls.append((status_url, response_url))
        if self._fetch_fail is not None:
            raise self._fetch_fail
        return self._png, 7


def _run(tmp_path: Path, *, client, records=None, identity=None,
         model="x-ai/grok-imagine-image-2.0", sel_bytes=b"SEL-PNG",
         stem_hash="stemhash-1"):
    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,
        identity=identity, stem_content_hash=stem_hash,
        pack="23.202608251200",
        context={"project_id": "p1", "episode_id": "e1",
                 "operation_type": "still_cine_transform",
                 "still_id": "sid-1"},
    )
    return rec, records, recipe_dir


# ── 신원 ────────────────────────────────────────────────────────────


def test_default_provider_is_legacy_grok(monkeypatch):
    """**코드 기본값**이 종전 동작이어야 byte-identical 이 성립한다.

    ★실행 중 `settings` 를 그대로 보면 안 된다 — 그 값은 이 기계의 `.env`
     가 정한다(지금은 reve). 그러면 이 시험이 재는 것은 「무엇이 기본인가」가
     아니라 「이 기계가 어떻게 설정됐나」가 된다.
    """
    from app.core.config import Settings
    from app.modules.pipeline.cine_provider import (
        GROK_ENDPOINT, cine_provider_identity, is_legacy_identity,
        resolve_cine_provider,
    )

    assert Settings.model_fields["still_cine_provider"].default == "grok"
    monkeypatch.setattr(settings, "still_cine_provider", "grok",
                        raising=False)
    assert resolve_cine_provider() == "grok"
    ident = cine_provider_identity()
    assert ident["provider"] == "grok"
    assert ident["endpoint"] == GROK_ENDPOINT
    assert ident["model"] == settings.grok_image_model
    assert is_legacy_identity(ident) is True


def test_unknown_provider_fails_loudly(monkeypatch):
    """모르는 값을 조용히 기본값으로 되돌리면 운영자는 reve 로 도는 줄
    아는데 기록은 grok 으로 남는다 — 크게 실패해야 한다."""
    from app.modules.pipeline import cine_provider

    monkeypatch.setattr(settings, "still_cine_provider", "revee",
                        raising=False)
    with pytest.raises(ValueError):
        cine_provider.resolve_cine_provider()


def test_reve_identity_uses_logical_endpoint(monkeypatch):
    from app.modules.pipeline.cine_provider import (
        cine_provider_identity, is_legacy_identity,
    )

    monkeypatch.setattr(settings, "still_cine_provider", "reve",
                        raising=False)
    ident = cine_provider_identity()
    assert ident == {"provider": "reve", "endpoint": settings.reve_image_model,
                     "model": settings.reve_image_model}
    # 옛 지문이 뜻하던 것과 다르다 — 재생성이 맞다.
    assert is_legacy_identity(ident) is False


def test_build_cine_client_matches_provider(monkeypatch):
    from app.modules.llm.grok_image_client import GrokImageClient
    from app.modules.llm.reve_image_client import ReveImageClient
    from app.modules.pipeline.cine_provider import build_cine_client

    assert isinstance(build_cine_client("grok"), GrokImageClient)
    assert isinstance(build_cine_client("reve"), ReveImageClient)


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


def test_fingerprint_v2_sensitive_to_provider_and_endpoint():
    from app.modules.pipeline.cine_transform import cine_fingerprint_v2

    grok = {"provider": "grok", "endpoint": "openrouter/chat-completions",
            "model": "m1"}
    kw = {"prompt": "p1", "stem_content_hash": "s1", "sel_bytes": b"img"}
    base = cine_fingerprint_v2(identity=grok, **kw)
    assert base == cine_fingerprint_v2(identity=dict(grok), **kw)
    for changed in (
        {**grok, "provider": "reve"},
        {**grok, "endpoint": "reve/2.1/edit"},
        {**grok, "model": "m2"},
    ):
        assert base != cine_fingerprint_v2(identity=changed, **kw)
    assert base != cine_fingerprint_v2(
        prompt="p1", stem_content_hash="s1", identity=grok,
        sel_bytes=b"other")
    # ★스템 해시도 그대로 지킨다 — 「팩 교체 = 전 샷 재변환」이 문안 sha 가
    #  스템에서 나온다는 전제에 얹히면, 그 전제가 깨질 때 화풍을 바꿔도 옛
    #  그림이 남는다(v1 이 막던 실패).
    assert base != cine_fingerprint_v2(
        prompt="p1", stem_content_hash="s2", identity=grok, sel_bytes=b"img")


def test_fingerprint_v2_sees_the_prompt_that_actually_goes_out():
    """연출 재료가 샷마다 다르다 — 문안이 바뀌면 지문도 바뀌어야 한다.

    ★스템 해시만 보면 카메라 지시를 바꿔도 지문이 안 움직여, **다른 지시로
     만든 그림이 「같은 조건」으로 읽힌다.**
    """
    from app.modules.pipeline.cine_transform import cine_fingerprint_v2

    ident = {"provider": "reve", "endpoint": "reve/2.1/edit",
             "model": "reve/2.1/edit"}
    a = cine_fingerprint_v2(
        prompt="Rework this still.\n\n- CAMERA: low angle",
        stem_content_hash="s1", identity=ident, sel_bytes=b"img")
    b = cine_fingerprint_v2(
        prompt="Rework this still.\n\n- CAMERA: high angle",
        stem_content_hash="s1", identity=ident, sel_bytes=b"img")
    assert a != b, "같은 스템이라도 재료가 다르면 다른 산출이다"


def test_v1_and_v2_fingerprints_differ():
    """판이 다르면 값도 달라야 한다 — 같으면 dual-read 가 뜻이 없다."""
    from app.modules.pipeline.cine_transform import (
        cine_fingerprint, cine_fingerprint_v2,
    )

    ident = {"provider": "grok", "endpoint": "openrouter/chat-completions",
             "model": "m1"}
    assert cine_fingerprint(
        stem_content_hash="s1", model="m1", sel_bytes=b"img") != (
        cine_fingerprint_v2(
            prompt="s1", stem_content_hash="s1", identity=ident,
            sel_bytes=b"img"))


# ── dual-read: 완주본을 다시 사지 않는다 ────────────────────────────


def _seed_legacy_record(records, recipe_dir, *, model, sel_bytes=b"SEL-PNG",
                        stem_hash="stemhash-1"):
    """판 번호 없는 옛 record + 산출 파일을 심는다 (완주 판 모양)."""
    from app.modules.pipeline.cine_transform import cine_fingerprint

    (recipe_dir / "S1sh1_cine.png").write_bytes(b"OLD-CINE")
    records.data["S1sh1::cine"] = {
        "applied": True, "file": "S1sh1_cine.png",
        "fingerprint": cine_fingerprint(
            stem_content_hash=stem_hash, model=model, sel_bytes=sel_bytes),
        "model": model, "pack": "19.202608131440",
    }


def test_legacy_record_reused_under_grok(tmp_path):
    """완주본은 **다시 사지 않는다** — 호출 0 + records 무쓰기.

    ★신원을 명시로 넘긴다 — 생략하면 이 기계의 `.env` 가 정하고, 지금은
     reve 라 「옛 Grok 완주본」 갈래를 아예 안 탄다.
    """
    from app.modules.pipeline.cine_provider import GROK_ENDPOINT

    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir()
    records = _StubRecords(recipe_dir / "records.json")
    _seed_legacy_record(records, recipe_dir, model=settings.grok_image_model)
    client = _QueueClient()
    grok = {"provider": "grok", "endpoint": GROK_ENDPOINT,
            "model": settings.grok_image_model}

    rec, records, _ = _run(tmp_path, client=client, records=records,
                           identity=grok, model=grok["model"])

    assert rec["reused"] is True
    assert rec["applied"] is True
    assert client.gen_calls == 0
    # ★1비트도 쓰지 않는다 — 재사용 바퀴에 record 가 움직이면 지출로 세어진다
    assert records.save_calls == 0
    assert "fingerprint_version" not in records.data["S1sh1::cine"]


def test_legacy_record_not_reused_under_reve(tmp_path):
    """제공자를 바꾸면 산출 계약이 실제로 다르다 — 소급 동등 처리 금지."""
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir()
    records = _StubRecords(recipe_dir / "records.json")
    _seed_legacy_record(records, recipe_dir, model=settings.grok_image_model)
    client = _QueueClient()
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}

    rec, records, _ = _run(tmp_path, client=client, records=records,
                           identity=reve, model=reve["model"])

    assert client.gen_calls == 1, "다른 제공자면 새로 만들어야 한다"
    assert rec["applied"] is True
    assert rec["provider"] == "reve"
    assert rec["endpoint"] == "reve/2.1/edit"
    assert rec["fingerprint_version"] == 2


def test_new_record_carries_identity_and_is_reused(tmp_path):
    """새로 쓴 record 는 v2 판이고, 다음 방문에 재사용된다."""
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    client = _QueueClient()
    rec, records, recipe_dir = _run(
        tmp_path, client=client, identity=reve, model=reve["model"])
    assert rec["fingerprint_version"] == 2
    assert client.gen_calls == 1

    saves_before = records.save_calls
    rec2, _, _ = _run(tmp_path, client=client, records=records,
                      identity=reve, model=reve["model"])
    assert rec2["reused"] is True
    assert client.gen_calls == 1, "재사용이면 유료 호출이 없다"
    assert records.save_calls == saves_before


# ── 접수 기반: 다시 가져오기하지 재구매하지 않는다 ───────────────────────────


def test_submit_is_recorded_before_output(tmp_path):
    """접수 즉시 신원이 durable 해야 크래시 뒤 다시 가져온다."""
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    client = _QueueClient()
    rec, records, _ = _run(tmp_path, client=client, identity=reve,
                           model=reve["model"])
    # 성공하면 접수 표식은 걷히고 산출이 남는다
    assert rec["applied"] is True
    assert records.data["S1sh1::cine"].get("pending") is not True
    # 접수 시점에 한 번, 결말에 한 번 — 기록이 두 번 이상 움직였다
    assert records.save_calls >= 2


def test_pending_is_recovered_not_repurchased(tmp_path):
    """접수됐는데 결말을 못 본 작업 — **다시 가져오기**한다. 다시 사지 않는다."""
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    from app.modules.pipeline.cine_transform import cine_fingerprint_v2

    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir()
    records = _StubRecords(recipe_dir / "records.json")
    records.data["S1sh1::cine"] = {
        "applied": False, "pending": True, "request_id": "req-777",
        "fingerprint": cine_fingerprint_v2(
            prompt="Rework this still.", stem_content_hash="stemhash-1",
            identity=reve, sel_bytes=b"SEL-PNG"),
        "fingerprint_version": 2,
        "provider": "reve", "endpoint": "reve/2.1/edit",
    }
    client = _QueueClient()

    rec, records, _ = _run(tmp_path, client=client, records=records,
                           identity=reve, model=reve["model"])

    assert client.fetch_calls == ["req-777"]
    assert client.gen_calls == 0, "이미 산 작업을 또 사면 안 된다"
    assert rec["applied"] is True
    assert rec["recovered"] is True
    assert rec.get("pending") is not True
    assert (recipe_dir / "S1sh1_cine.png").read_bytes() == b"CINE-PNG"


def test_pending_terminal_failure_clears_marker(tmp_path):
    """작업이 산출 없이 끝났으면 접수 표식을 걷어야 다음에 새로 산다."""
    from app.modules.llm.reve_image_client import ReveTerminalError
    from app.modules.pipeline.cine_transform import cine_fingerprint_v2

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir()
    records = _StubRecords(recipe_dir / "records.json")
    records.data["S1sh1::cine"] = {
        "applied": False, "pending": True, "request_id": "req-dead",
        "fingerprint": cine_fingerprint_v2(
            prompt="Rework this still.", stem_content_hash="stemhash-1",
            identity=reve, sel_bytes=b"SEL-PNG"),
        "fingerprint_version": 2,
    }
    client = _QueueClient(fetch_fail=ReveTerminalError(
        "reve 작업 실패 no_media_generated", error_type="no_media_generated"))

    rec, records, _ = _run(tmp_path, client=client, records=records,
                           identity=reve, model=reve["model"])

    assert client.fetch_calls == ["req-dead"]
    assert rec["applied"] is False
    assert rec.get("pending") is not True
    assert "request_id" not in rec
    assert records.data["S1sh1::cine"].get("pending") is not True


def test_pending_transient_failure_keeps_marker(tmp_path):
    """일시 실패는 접수 표식을 지키고 다음 방문에 다시 다시 가져온다."""
    from app.modules.pipeline.cine_transform import cine_fingerprint_v2

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir()
    records = _StubRecords(recipe_dir / "records.json")
    records.data["S1sh1::cine"] = {
        "applied": False, "pending": True, "request_id": "req-slow",
        "fingerprint": cine_fingerprint_v2(
            prompt="Rework this still.", stem_content_hash="stemhash-1",
            identity=reve, sel_bytes=b"SEL-PNG"),
        "fingerprint_version": 2,
    }
    client = _QueueClient(fetch_fail=TimeoutError("아직 안 끝났다"))

    rec, records, _ = _run(tmp_path, client=client, records=records,
                           identity=reve, model=reve["model"])

    assert rec["applied"] is False
    assert rec["pending"] is True
    assert rec["request_id"] == "req-slow"
    assert client.gen_calls == 0, "가져오기 실패에 새로 사면 두 번 산다"


def test_failure_after_submit_preserves_request_id(tmp_path):
    """접수 뒤 결말을 못 보면 다시 가져오기 신원이 record 에 남아야 한다.

    ★이 줄이 없으면 실패 기록이 접수 표식을 덮어써 **이미 산 작업을 영영
     못 찾는다** — 다음 방문이 같은 것을 다시 산다.
    """
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    client = _QueueClient(
        gen_fail=TimeoutError("poll 중단"), submit_before_fail=True)

    rec, records, _ = _run(tmp_path, client=client, identity=reve,
                           model=reve["model"])

    assert rec["applied"] is False
    assert rec["pending"] is True
    assert rec["request_id"] == "req-777"
    assert records.data["S1sh1::cine"]["request_id"] == "req-777"


def test_terminal_failure_after_submit_does_not_keep_marker(tmp_path):
    """검열처럼 결말이 난 실패는 다시 가져오기할 결과가 없다 — 표식을 안 남긴다."""
    from app.modules.llm.reve_image_client import ReveTerminalError

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit",
            "model": "reve/2.1/edit"}
    client = _QueueClient(
        gen_fail=ReveTerminalError(
            "reve 422 content_policy_violation: blocked",
            error_type="content_policy_violation"),
        submit_before_fail=True)

    rec, _records, _ = _run(tmp_path, client=client, identity=reve,
                            model=reve["model"])

    assert rec["applied"] is False
    assert rec.get("pending") is not True
    # 검열로 읽혀 포기 셈이 돈다(공용 판별이 `content_policy` 를 표식으로 쓴다)
    assert rec.get("moderation_refusals") == 1


# ── reve 운반층: 오류를 **정확한 유형 이름**으로 가른다 ─────────────


def _http_error(status: int, body: dict, headers: Optional[dict] = None):
    return urllib.error.HTTPError(
        "https://queue.fal.run/x", status, "err", headers or {},
        io.BytesIO(json.dumps(body).encode("utf-8")))


def _client():
    from app.modules.llm.reve_image_client import ReveImageClient

    return ReveImageClient(api_key="k")


def test_policy_violation_is_terminal_and_reads_as_moderation():
    from app.modules.llm.image_moderation import is_moderation_error
    from app.modules.llm.reve_image_client import ReveTerminalError

    c = _client()
    exc = _http_error(422, {"detail": [
        {"type": "content_policy_violation", "msg": "safety filter"}]})
    with pytest.raises(ReveTerminalError) as ei:
        c._raise_http(exc)
    assert ei.value.error_type == "content_policy_violation"
    # 상위 포기 셈이 이것을 검열로 읽어야 한다
    assert is_moderation_error(ei.value) is True


def test_no_media_is_terminal_but_not_moderation():
    """검열과 **다른 실패**다 — 섞으면 포기 기준이 엉뚱한 곳에서 발화한다."""
    from app.modules.llm.image_moderation import is_moderation_error
    from app.modules.llm.reve_image_client import ReveTerminalError

    c = _client()
    exc = _http_error(422, {"detail": [
        {"type": "no_media_generated", "msg": "model produced no output"}]})
    with pytest.raises(ReveTerminalError) as ei:
        c._raise_http(exc)
    assert ei.value.error_type == "no_media_generated"
    assert is_moderation_error(ei.value) is False


def test_retryable_infra_error_is_not_terminal():
    from app.modules.llm.reve_image_client import ReveTerminalError

    c = _client()
    exc = _http_error(503, {"detail": "runner gone",
                            "error_type": "runner_disconnected"})
    with pytest.raises(RuntimeError) as ei:
        c._raise_http(exc)
    assert not isinstance(ei.value, ReveTerminalError)


def test_needs_retry_header_wins_over_type_table():
    """재시도 가부는 헤더가 우선이다 (fal 공식 계약)."""
    from app.modules.llm.reve_image_client import ReveTerminalError

    c = _client()
    exc = _http_error(
        500, {"detail": [{"type": "internal_server_error", "msg": "x"}]},
        {"x-fal-needs-retry": "false"})
    with pytest.raises(ReveTerminalError):
        c._raise_http(exc)


def test_unknown_type_is_terminal_but_not_guessed_as_moderation():
    """유형이 없으면 「알 수 없는 4xx terminal」 — 검열로 추정하지 않는다."""
    from app.modules.llm.image_moderation import is_moderation_error
    from app.modules.llm.reve_image_client import ReveTerminalError

    c = _client()
    exc = _http_error(400, {"detail": "something went wrong"})
    with pytest.raises(ReveTerminalError) as ei:
        c._raise_http(exc)
    assert ei.value.error_type == ""
    assert is_moderation_error(ei.value) is False


def test_edit_endpoint_takes_exactly_one_reference():
    """`reve/2.1/edit` 의 입력 칸은 단수다 — 여러 장이 오면 조용히 버리지
    않는다(잘려 나간 참조는 그림에서만 드러난다)."""
    c = _client()
    with pytest.raises(ValueError):
        c.generate_image("p", labeled_references=[("A", b"1"), ("B", b"2")])
    with pytest.raises(ValueError):
        c.generate_image("p")


def test_submit_without_request_id_is_fail_closed(monkeypatch):
    """접수 여부가 모호하면 **자동 재요청 금지** — 요금이 두 번 나가는 길을 막는다.

    ★유형이 `ReveSubmissionUnknown` 이어야 한다 (2026-08-26 Codex 재리뷰
     BLOCK-2). `cine_transform` 은 이 유형일 때만 `submission_unknown` 표식을
     남기고, 그 표식이 있어야 **다음 걷기가** 다시 안 보낸다. 종전에는
     `ReveTerminalError` 라 이 호출 안에서만 멈추고 다음 걷기가 다시 보냈다.
    """
    from app.modules.llm.reve_image_client import ReveSubmissionUnknown

    c = _client()
    monkeypatch.setattr(c, "_request",
                        lambda *a, **k: (200, {"queue_position": 0}, {}))
    with pytest.raises(ReveSubmissionUnknown) as ei:
        c._submit({"prompt": "p"})
    assert ei.value.cause == "submit_ack_missing"


def test_generate_image_는_번호없는_200을_감싸지_않고_그대로_올린다(monkeypatch):
    """`generate_image` 의 일반 except 가 사유를 뭉개지 않아야 한다."""
    from app.modules.llm.reve_image_client import ReveSubmissionUnknown

    c = _client()
    monkeypatch.setattr(c, "_request",
                        lambda *a, **k: (200, {"queue_position": 0}, {}))
    with pytest.raises(ReveSubmissionUnknown) as ei:
        c.generate_image("p", labeled_references=[("SOURCE", b"x")])
    assert ei.value.cause == "submit_ack_missing"


def test_fetch_submitted_requires_request_id():
    c = _client()
    with pytest.raises(ValueError):
        c.fetch_submitted("")


# ── 기록: 「여기가 남기는 자리가 아니다」와 「안 남긴다」는 다르다 ────
#
# 운반 원시층(`_request`)은 추적 스캐너에서 면제돼 있다 — 한 유료 작업이
# 여러 왕복으로 나뉘어 거기서 남기면 한 작업이 여러 번 기록되기 때문이다.
# 면제가 「기록이 없다」로 굳지 않도록 **실제로 남는 것을 여기서 잰다**.


class _SpyTracer:
    def __init__(self):
        self.calls: List[Dict[str, Any]] = []

    def log(self, **kw):
        self.calls.append(kw)


class _FakeResp:
    def __init__(self, data: bytes):
        self._d = data

    def __enter__(self):
        return self

    def __exit__(self, *a):
        return False

    def read(self):
        return self._d


def _patch_logging(monkeypatch):
    import app.modules.llm.image_tracer as tracer_mod
    import app.modules.llm.reve_image_client as reve_mod

    log_calls: List[Dict[str, Any]] = []
    tracer = _SpyTracer()

    def _spy_log(**kw):
        log_calls.append(kw)
        return "call-1"

    monkeypatch.setattr(reve_mod, "log_llm_call", _spy_log)
    monkeypatch.setattr(tracer_mod, "get_image_tracer", lambda: tracer)
    monkeypatch.setattr(reve_mod, "capture_generated_image",
                        lambda *a, **k: None)
    monkeypatch.setattr(reve_mod, "ensure_png_bytes", lambda b, **k: b)
    return log_calls, tracer, reve_mod


def test_one_paid_job_leaves_exactly_one_record(monkeypatch):
    """poll 이 여러 번 돌아도 기록은 **한 번** — 한 작업이 한 기록이다."""
    log_calls, tracer, reve_mod = _patch_logging(monkeypatch)
    monkeypatch.setattr(reve_mod.time, "sleep", lambda s: None)
    c = _client()
    c.set_context(project_id="p1", episode_id="e1",
                  operation_type="still_cine_transform")
    seq = [
        (200, {"status": "IN_QUEUE"}, {}),
        (200, {"status": "IN_PROGRESS"}, {}),
        (200, {"status": "COMPLETED"}, {}),
        (200, {"images": [{"url": "https://x/y.png"}]}, {}),
    ]
    monkeypatch.setattr(c, "_request", lambda *a, **k: seq.pop(0))
    monkeypatch.setattr(reve_mod.urllib.request, "urlopen",
                        lambda *a, **k: _FakeResp(b"PNGBYTES"))

    png, _ms = c._finish("req-9", prompt="p", ref_image_ids=["SOURCE STILL"],
                         start_time=0.0, deadline=1e18)

    assert png == b"PNGBYTES"
    assert seq == [], "queue 왕복을 다 쓰지 않았다 — 시험이 헐겁다"
    assert len(log_calls) == 1 and log_calls[0]["status"] == "success"
    assert len(tracer.calls) == 1
    assert tracer.calls[0]["params"]["request_id"] == "req-9"


def test_failure_also_leaves_exactly_one_record(monkeypatch):
    """실패해도 기록이 남아야 한다 — 기록 없는 실패는 「살아 있나」를 볼
    근거를 통째로 없앤다(2026-08-07 사고와 같은 자리)."""
    log_calls, tracer, reve_mod = _patch_logging(monkeypatch)
    c = _client()

    def _boom(*a, **k):
        raise TimeoutError("작업 대기 마감 초과")

    monkeypatch.setattr(c, "_poll", _boom)
    with pytest.raises(TimeoutError):
        c._finish("req-x", prompt="p", ref_image_ids=[],
                  start_time=0.0, deadline=1e18)

    assert len(log_calls) == 1 and log_calls[0]["status"] == "error"
    assert len(tracer.calls) == 1 and tracer.calls[0]["status"] == "error"


# ── fal 결과 조회 주소 (2026-08-26 405 원인) ────────────────────────────


def test_status_and_result_urls_drop_the_action_segment():
    """접수 주소에는 동작 이름이 붙지만 **조회 주소에는 안 붙는다.**

    fal 실측 응답:
        접수  POST queue.fal.run/reve/2.1/edit
        상태  GET  queue.fal.run/reve/2.1/requests/{id}/status
        결과  GET  queue.fal.run/reve/2.1/requests/{id}

    종전 구현은 `{model}/{id}` 로 만들어 `reve/2.1/edit/{id}` 를 요청했고
    **405** 를 받았다. 405 는 「메서드가 틀렸다」는 코드라 원인을 몸통
    모양·인증에서 찾느라 두 판을 날렸다. 이 파일 머리말에는 올바른 계약이
    처음부터 적혀 있었다 — 문서와 구현이 갈린 것이 진짜 결함이다.
    """
    from app.modules.llm.reve_image_client import ReveImageClient

    c = ReveImageClient(model="reve/2.1/edit")
    assert c._status_url("ID1") == (
        "https://queue.fal.run/reve/2.1/requests/ID1/status")
    assert c._response_url("ID1") == (
        "https://queue.fal.run/reve/2.1/requests/ID1")
    assert "/edit/" not in c._status_url("ID1")


def test_urls_that_fal_returned_win_over_the_ones_we_build():
    """평소에는 **fal 이 알려 준 주소**를 쓴다 — 조립은 대비책일 뿐이다.

    ★단 **그 요청의 주소일 때만**이다. 종전 이 테스트는 `"무시됨"` 이라는
     다른 id 를 넘겨도 저장된 주소를 쓰는 것을 정답으로 못박고 있었는데,
     그것이 곧 앞 샷 주소로 다음 샷 결과를 가져오는 결함이었다(2026-08-26).
     의도(405 수리)는 옳았고 증명 방식이 틀렸다 — 같은 id 로 증명한다.
    """
    from app.modules.llm.reve_image_client import ReveImageClient

    c = ReveImageClient(model="reve/2.1/edit")
    c.last_request_id = "X"
    c.last_status_url = "https://queue.fal.run/reve/2.1/requests/X/status"
    c.last_response_url = "https://queue.fal.run/reve/2.1/requests/X"
    assert c._status_url("X") == c.last_status_url
    assert c._response_url("X") == c.last_response_url


def test_submit_body_is_flat():
    """fal 공식 예제의 `arguments={...}` 가 그대로 몸통이다.

    ★`{"input": {...}}` 로 감싸면 접수는 200 이 오지만 작업자가 인자를
     못 읽는다 — 증상이 안 바뀌어서 오진이 오래 갔다.
    """
    from app.modules.llm.reve_image_client import ReveImageClient

    c = ReveImageClient(model="reve/2.1/edit")
    sent = {}

    def fake_request(url, *, method="GET", body=None, timeout=120):
        sent["url"] = url
        sent["body"] = body
        return 200, {"request_id": "R1"}, {}

    c._request = fake_request
    c._submit({"prompt": "p", "image_url": "u"})
    assert sent["url"] == "https://queue.fal.run/reve/2.1/edit"
    assert "input" not in sent["body"]
    assert sent["body"]["prompt"] == "p"


def test_앞_샷의_접수_신원이_다음_샷으로_새지_않는다(monkeypatch):
    """★2026-08-26 실측 결함 — 클라이언트가 샷 루프 **바깥에서 한 번** 만들어져
    모든 샷이 공유하는데, 재시도용 `last_request_id` 가 호출 시작에 안 비워졌다.
    그래서 앞 샷이 성공하며 남긴 신원을 다음 샷이 물어 **앞 샷의 그림을 자기
    결과로 가져왔다** (S2sh1 → S3sh2, 3.7초 · 바이트 해시까지 동일).

    검사이 「장소·인물·사물이 다르다」로 잡아서 드러났다. 검사이 없었으면
    남의 그림이 최종본으로 확정됐다.
    """
    from app.modules.llm.reve_image_client import ReveImageClient

    client = ReveImageClient()
    submits: list[str] = []
    finishes: list[str] = []

    def _fake_submit(body):
        rid = f"req-{len(submits) + 1}"
        submits.append(body["prompt"])
        client.last_request_id = rid          # 실제 `_submit` 과 같은 부수효과
        return {"request_id": rid}

    def _fake_finish(rid, **kw):
        finishes.append(rid)
        return (f"PNG-{rid}".encode(), 100)

    monkeypatch.setattr(client, "_submit", _fake_submit)
    monkeypatch.setattr(client, "_finish", _fake_finish)

    png_a, _ = client.generate_image(
        "샷 A 변환", labeled_references=[("ref", b"A-SEL")])
    png_b, _ = client.generate_image(
        "샷 B 변환", labeled_references=[("ref", b"B-SEL")])

    assert submits == ["샷 A 변환", "샷 B 변환"], (
        f"두 번째 샷이 접수를 건너뛰었다 — submits={submits}"
    )
    assert finishes == ["req-1", "req-2"], (
        f"두 번째 샷이 앞 샷의 신원으로 결과를 가져왔다 — finishes={finishes}"
    )
    assert png_a != png_b, "두 샷이 같은 그림을 받았다"


def test_조회_주소가_요청_신원에_묶인다():
    """★앞 샷의 조회 주소로 다음 샷 결과를 가져오던 것 (2026-08-26 2차).

    `last_request_id` 하나만 비웠더니 조회 주소 두 칸이 남아 `_status_url`/
    `_response_url` 이 **넘어온 request_id 를 무시하고** 앞 샷 주소를 계속
    썼다. 신원은 세 칸이 함께 이룬다.
    """
    from app.modules.llm.reve_image_client import ReveImageClient

    client = ReveImageClient()
    client.last_request_id = "req-A"
    client.last_status_url = "https://queue.fal.run/저장된/A/status"
    client.last_response_url = "https://queue.fal.run/저장된/A"

    # 같은 신원이면 fal 이 준 주소를 쓴다 — 405 를 겪고 정한 규칙.
    assert client._status_url("req-A") == "https://queue.fal.run/저장된/A/status"
    assert client._response_url("req-A") == "https://queue.fal.run/저장된/A"

    # 다른 신원이면 그 id 로 조립한다. 저장된 주소를 쓰면 남의 결과를 가져온다.
    다른_상태 = client._status_url("req-B")
    다른_결과 = client._response_url("req-B")
    assert "req-B" in 다른_상태 and "저장된/A" not in 다른_상태
    assert "req-B" in 다른_결과 and "저장된/A" not in 다른_결과


def test_새_변환은_주소_세_칸을_모두_비운다(monkeypatch):
    from app.modules.llm.reve_image_client import ReveImageClient

    client = ReveImageClient()
    client.last_request_id = "req-앞샷"
    client.last_status_url = "https://queue.fal.run/앞샷/status"
    client.last_response_url = "https://queue.fal.run/앞샷"

    본_것 = {}

    def _fake_submit(body):
        본_것["빈_id"] = client.last_request_id
        본_것["빈_상태"] = client.last_status_url
        본_것["빈_결과"] = client.last_response_url
        client.last_request_id = "req-새샷"
        return {"request_id": "req-새샷"}

    monkeypatch.setattr(client, "_submit", _fake_submit)
    monkeypatch.setattr(client, "_finish", lambda rid, **kw: (b"PNG", 1))

    client.generate_image("새 샷", labeled_references=[("ref", b"SEL")])

    assert 본_것 == {"빈_id": "", "빈_상태": "", "빈_결과": ""}, (
        f"앞 샷 신원이 남았다 — {본_것}"
    )


# ── 요금이 두 번 나가는 것 막기 (2026-08-26 Codex 지적) ──────────────
#
# 접수 번호가 없다고 「요금이 안 나갔다」로 읽으면 안 된다. 번호가 없는 경우가
# 둘이고, 그중 하나는 **fal 이 받아서 요금이 나갔을 수 있는** 경우다.


def test_연결이_아예_안_됐으면_다시_보낸다(monkeypatch):
    """요청이 서버에 닿지 못했으면 요금이 안 나갔다 — 다시 보내는 게 맞다."""
    import socket
    from app.modules.llm.reve_image_client import ReveImageClient

    client = ReveImageClient()
    보낸_횟수 = {"n": 0}

    def _fake_submit(body):
        보낸_횟수["n"] += 1
        if 보낸_횟수["n"] == 1:
            raise ConnectionRefusedError("연결 거부")
        client.last_request_id = "req-2"
        return {"request_id": "req-2"}

    monkeypatch.setattr(client, "_submit", _fake_submit)
    monkeypatch.setattr(client, "_finish", lambda rid, **kw: (b"PNG", 1))
    monkeypatch.setattr(
        "app.modules.llm.reve_image_client._CINE_RETRY_WAIT", 0)

    png, _ = client.generate_image("샷", labeled_references=[("r", b"S")])
    assert 보낸_횟수["n"] == 2, "연결 거부는 다시 보내야 한다"
    assert png == b"PNG"

    # 이름 조회 실패도 같은 갈래다.
    from app.modules.llm.reve_image_client import classify_submit_failure
    assert classify_submit_failure(socket.gaierror("이름 못 찾음")) == "never_sent"
    assert classify_submit_failure(ConnectionRefusedError()) == "never_sent"


def test_답이_끊기면_다시_안_보낸다(monkeypatch):
    """★핵심 — 보냈는데 답이 끊긴 것은 **요금이 나갔을 수 있다.**

    종전에는 번호가 없다고 그냥 다시 보냈고, 그러면 같은 이미지에 $0.25 가
    두 번 나갔다. 조용해서 로그에도 안 보였다.
    """
    from app.modules.llm.reve_image_client import (
        ReveImageClient, ReveSubmissionUnknown, classify_submit_failure,
    )

    client = ReveImageClient()
    보낸_횟수 = {"n": 0}

    def _fake_submit(body):
        보낸_횟수["n"] += 1
        raise ConnectionResetError("답이 오다가 끊겼다")

    monkeypatch.setattr(client, "_submit", _fake_submit)
    monkeypatch.setattr(
        "app.modules.llm.reve_image_client._CINE_RETRY_WAIT", 0)

    with pytest.raises(ReveSubmissionUnknown):
        client.generate_image("샷", labeled_references=[("r", b"S")])

    assert 보낸_횟수["n"] == 1, (
        f"답이 끊겼는데 {보낸_횟수['n']}번 보냈다 — 요금이 두 번 나간다"
    )
    assert classify_submit_failure(ConnectionResetError()) == "unknown"
    assert classify_submit_failure(TimeoutError()) == "unknown"


def test_URLError_안에_감싸인_원인도_가른다():
    """`urllib` 은 원인을 `URLError.reason` 으로 감싼다 — 겉만 보면 못 가른다."""
    import socket
    import urllib.error
    from app.modules.llm.reve_image_client import classify_submit_failure

    감싼_이름실패 = urllib.error.URLError(socket.gaierror("이름 못 찾음"))
    감싼_끊김 = urllib.error.URLError(ConnectionResetError("끊김"))
    assert classify_submit_failure(감싼_이름실패) == "never_sent"
    assert classify_submit_failure(감싼_끊김) == "unknown"


# ── 2026-08-26 Codex 재리뷰 — 요금이 두 번 나가는 길 ────────────────


def test_접수_여부_모르면_다음_걷기도_다시_안_보낸다(tmp_path):
    """★한 호출 안에서 안 보내는 것만으로는 모자라다.

    Codex 재리뷰 BLOCK-2 의 핵심: 종전 시험은 `_submit` 한 번만 봤고,
    **다음 걷기**를 안 봤다. 표식이 안 남으면 그 다음에 같은 지문으로 또
    나간다. 그래서 여기서는 같은 입력으로 **두 번** 걷고 요청 수를 센다.
    """
    from app.modules.llm.reve_image_client import ReveSubmissionUnknown

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    client = _QueueClient(gen_fail=ReveSubmissionUnknown(
        "번호가 없다", cause="submit_ack_missing"))

    rec1, records, _ = _run(tmp_path, client=client, identity=reve)
    assert rec1.get("submission_unknown") is True
    assert client.gen_calls == 1

    rec2, _, _ = _run(tmp_path, client=client, records=records, identity=reve)
    assert client.gen_calls == 1, (
        "같은 입력에 두 번째 요청이 나갔다 — 같은 그림에 요금이 두 번 나간다")
    assert rec2.get("reused") is True
    assert rec2.get("submission_unknown") is True


def test_지문이_바뀌면_다시_보내되_옛_번호를_잃지_않는다(tmp_path):
    """★잠그지는 않는다. 원본이 다시 만들어졌으면 그건 다른 그림이다.

    다만 옛 「접수 여부 모름」 기록이 같은 칸에 덮여 사라지면, 사람이 fal
    대시보드에서 요금을 대조할 실마리가 없어진다 (Codex 재리뷰 BLOCK-3).
    """
    from app.modules.llm.reve_image_client import ReveSubmissionUnknown

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    bad = _QueueClient(gen_fail=ReveSubmissionUnknown(
        "번호가 없다", cause="submit_ack_missing"))
    _rec1, records, _ = _run(tmp_path, client=bad, identity=reve,
                             sel_bytes=b"SEL-A")

    good = _QueueClient()
    rec2, _, _ = _run(tmp_path, client=good, records=records, identity=reve,
                      sel_bytes=b"SEL-B")          # 원본이 다시 만들어졌다
    assert good.gen_calls == 1, "지문이 바뀌었는데 안 보냈다 — 샷이 영영 안 나온다"
    assert rec2.get("applied") is True

    orphans = records.data.get("S1sh1::cine::unresolved_log")
    assert isinstance(orphans, list) and len(orphans) == 1
    assert orphans[0]["reason"] == "submission_unknown"
    assert orphans[0]["detail"] == "submit_ack_missing"


def test_확정_직전에_멈춰도_다음_걷기가_다시_안_보낸다(tmp_path):
    """★요금이 나간 그림은 파일로 남았는데 기록이 없어 다시 보내던 자리.

    Codex 재리뷰 BLOCK-5. 접수 hook 이 없는 제공자에도 통해야 하므로
    hook 을 안 부르는 스텁으로 잰다.
    """
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check,
    )

    class _PlainClient:
        """접수 hook 도 다시 가져오기 슬롯도 **아예 없는** 동기 왕복 제공자.

        상속으로 만들지 않는다 — 메서드를 물려받으면 `hasattr` 가 True 라
        「hook 없는 제공자」를 흉내 내지 못한다.
        """

        def __init__(self) -> None:
            self.gen_calls = 0

        def set_context(self, **kw):
            return self

        def generate_image(self, prompt, reference_images=None,
                           aspect_ratio="16:9", labeled_references=None):
            self.gen_calls += 1
            return b"CINE-PNG", 42

    client = _PlainClient()
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")

    def _stop():
        raise AppError(code="step.cancelled", message="정지", status_code=409)

    install_stop_check(_stop)
    try:
        with pytest.raises(AppError) as ei:
            _run(tmp_path, client=client, records=records)
        assert ei.value.code == "step.cancelled"
    finally:
        uninstall_stop_check()

    assert client.gen_calls == 1
    assert records.data["S1sh1::cine"].get("staged") is True
    assert records.data["S1sh1::cine"].get("applied") is not True

    rec2, _, _ = _run(tmp_path, client=client, records=records)
    assert client.gen_calls == 1, (
        "요금이 나간 그림이 디스크에 있는데 다시 보냈다")
    assert rec2.get("applied") is True
    assert rec2.get("recovered") is True


def test_남겨둔_파일이_다른_그림이면_되찾지_않는다(tmp_path):
    """staged 표식만 믿지 않는다 — 바이트가 그때 그 그림이어야 한다."""
    client = _QueueClient(png=b"FIRST-PNG")
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")
    _run(tmp_path, client=client, records=records)

    records.data["S1sh1::cine"]["staged"] = True
    records.data["S1sh1::cine"].pop("applied", None)
    from app.modules.pipeline.cine_transform import cine_output_path
    cine_output_path(recipe_dir, "S1sh1").write_bytes(b"SOMEONE-ELSE")

    _run(tmp_path, client=client, records=records)
    assert client.gen_calls == 2, "다른 그림을 그때 그 결과로 확정했다"


def test_기록에_남긴_조회주소를_넘기면_그것을_쓴다(monkeypatch):
    """프로세스가 죽었다 살아나도 조립에 기대지 않는다 (405 재발 방지)."""
    c = _client()
    monkeypatch.setattr(c, "_finish", lambda rid, **kw: (b"png", 1))

    c.fetch_submitted("req-9", status_url="https://fal/saved/status",
                      response_url="https://fal/saved/resp")
    assert c._status_url("req-9") == "https://fal/saved/status"
    assert c._response_url("req-9") == "https://fal/saved/resp"
    # 신원이 다르면 저장 주소를 쓰지 않는다 — 앞 샷 결과를 가져오던 그 결함.
    assert "/requests/req-8" in c._status_url("req-8")


# ── 2026-08-26 Codex 2차 재리뷰 ──────────────────────────────────────


def _reload_records(records):
    """새 프로세스를 흉내 낸다 — 기록을 디스크에서 다시 읽는다.

    ★같은 dict 를 그대로 물려주면 「메모리에 남아 있어서 됐다」와 「기록에
     제대로 적혀서 됐다」를 못 가른다.
    """
    fresh = _StubRecords(records._path)
    fresh.data = json.loads(records._path.read_text(encoding="utf-8"))
    return fresh


def test_실패로_끝나도_조회주소_둘을_다_남긴다(tmp_path):
    """접수 hook 이 남긴 status_url 이 마무리에서 사라지면 안 된다."""
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    client = _QueueClient(gen_fail=RuntimeError("poll 이 끊겼다"),
                          submit_before_fail=True)
    _rec, records, _ = _run(tmp_path, client=client, identity=reve)

    saved = _reload_records(records).data["S1sh1::cine"]
    assert saved["pending"] is True
    assert saved["request_id"] == "req-777"
    assert saved["status_url"] == "https://queue/x/status"
    assert saved["response_url"] == "https://queue/x"
    assert saved["submitted_at"], "접수 시각이 없으면 대시보드와 못 맞춘다"


def test_남겨둔_파일이_어긋나면_요금없이_다시_가져온다(tmp_path):
    """staged 가 접수 번호를 들고 있어야 무료 다시 가져오기로 이어진다."""
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check,
    )

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    client = _QueueClient(png=b"FIRST-PNG")
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")

    # 요금은 나갔고 파일도 남겼는데 **확정을 못 하고 끝난** 실제 상황을
    # 만든다 — 손으로 record 를 고치면 재는 것이 계약이 아니라 내 편집이 된다.
    def _stop():
        raise AppError(code="step.cancelled", message="정지", status_code=409)

    install_stop_check(_stop)
    try:
        with pytest.raises(AppError):
            _run(tmp_path, client=client, records=records, identity=reve)
    finally:
        uninstall_stop_check()

    # 그 뒤 같은 자리에 다른 그림이 덮였다.
    from app.modules.pipeline.cine_transform import cine_output_path
    cine_output_path(recipe_dir, "S1sh1").write_bytes(b"SOMEONE-ELSE")

    fresh = _reload_records(records)
    assert fresh.data["S1sh1::cine"].get("request_id") == "req-777", (
        "staged 기록이 접수 번호를 버렸다 — 무료로 되찾을 손잡이가 없다")

    rec2, _, _ = _run(tmp_path, client=client, records=fresh, identity=reve)
    assert client.gen_calls == 1, "무료로 가져올 수 있는데 다시 요청했다"
    assert client.fetch_calls == ["req-777"]
    # ★기록에 남긴 조회 주소가 **실제로 전달돼야** 조립에 안 기댄다.
    assert client.fetch_urls == [("https://queue/x/status", "https://queue/x")]
    assert rec2.get("applied") is True


def test_판정이_도는_사이에_주인이_바뀌면_확정하지_않는다(tmp_path):
    """★확정 직전 확인이 판정 **뒤**에도 있어야 한다.

    판정 VLM 은 오래 돈다. 그 사이 주인이 바뀌면 옛 worker 가 남의 주행
    자리에 applied 를 적는다 (Codex 2차 재리뷰 BLOCK-2).
    """
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check,
    )
    from app.modules.pipeline import cine_transform as ct

    client = _QueueClient()
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")

    calls = {"n": 0}

    def _stop():
        # 첫 확인(결과 직후)은 통과시키고, 판정 뒤 확인에서만 멈춘다.
        calls["n"] += 1
        if calls["n"] >= 2:
            raise AppError(code="step.owner_lost", message="주인이 바뀌었다",
                           status_code=409)

    install_stop_check(_stop)
    try:
        with pytest.raises(AppError) as ei:
            _run(tmp_path, client=client, records=records)
        assert ei.value.code == "step.owner_lost"
    finally:
        uninstall_stop_check()

    assert calls["n"] >= 2, "판정 뒤 확인이 아예 없다"
    saved = _reload_records(records).data["S1sh1::cine"]
    assert saved.get("applied") is not True, "남의 자리에 확정을 적었다"
    assert saved.get("staged") is True, "요금이 나간 그림은 되찾을 수 있어야 한다"


def test_결말을_못_본_기록은_최근것만_남는다(tmp_path):
    """이름 그대로 rolling log — 덧붙이기 전용 대장이 아니다."""
    from app.modules.llm.reve_image_client import ReveSubmissionUnknown
    from app.modules.pipeline.cine_transform import _UNRESOLVED_LOG_LIMIT

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")

    bad = _QueueClient(gen_fail=ReveSubmissionUnknown(
        "번호가 없다", cause="submit_ack_missing"))
    for i in range(_UNRESOLVED_LOG_LIMIT + 3):
        _run(tmp_path, client=bad, records=records, identity=reve,
             sel_bytes=f"SEL-{i}".encode())

    log = records.data["S1sh1::cine::unresolved_log"]
    assert len(log) == _UNRESOLVED_LOG_LIMIT
    assert log[-1]["reason"] == "submission_unknown"
    assert log[-1]["attempted_at"], "시도 시각이 없으면 대시보드와 못 맞춘다"
    assert "archived_at" in log[-1]
    assert "status_url" in log[-1]


# ── 2026-08-26 Codex 3차 재리뷰 BLOCK-1 ──────────────────────────────
#
# 판정을 부르는 갈래가 다섯인데 확인을 손으로 붙이다 **둘을 빠뜨렸다.**
# 여기서는 그 둘과 「가져오기 실패」 갈래를 태운다.


@pytest.fixture
def _판정_켜고_가짜로(monkeypatch):
    """판정을 켜되 VLM 은 부르지 않는다 — 재는 것은 확정 순서다."""
    monkeypatch.setattr(settings, "still_cine_verify_enabled", True,
                        raising=False)
    from app.modules.pipeline import cine_verify

    monkeypatch.setattr(
        cine_verify, "verify_cine_result",
        lambda **kw: {"ok": True, "broken": [], "findings": {},
                      "contract": cine_verify.verify_contract_sha(2)})
    return cine_verify.verify_contract_sha(2)


def _주인이_바뀐다():
    from app.core.errors import AppError

    def _stop():
        raise AppError(code="step.owner_lost", message="주인이 바뀌었다",
                       status_code=409)
    return _stop


@pytest.mark.parametrize("갈래", ["다시 판정", "되살리기"])
def test_다시_판정하는_갈래도_확정_전에_주인을_본다(
        tmp_path, _판정_켜고_가짜로, 갈래):
    """이미 있는 그림을 새 판정 계약으로 다시 보는 두 갈래.

    그림을 다시 요청하지는 않지만 **판정 VLM 이 돈다** — 그 사이 주인이
    바뀌면 남의 자리에 applied/rejected 를 적는다.
    """
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check,
    )

    client = _QueueClient()
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")
    _run(tmp_path, client=client, records=records)      # 그림 하나 만들어 둔다

    saved = records.data["S1sh1::cine"]
    saved["verify"] = {"contract": "옛-계약"}            # 계약이 바뀐 상태
    if 갈래 == "되살리기":
        saved["applied"] = False
        saved["rejected"] = True
        saved["rejected_reason"] = "implausible:anatomy"
    records.save()

    install_stop_check(_주인이_바뀐다())
    try:
        with pytest.raises(AppError) as ei:
            _run(tmp_path, client=client, records=_reload_records(records))
        assert ei.value.code == "step.owner_lost"
    finally:
        uninstall_stop_check()

    assert client.gen_calls == 1, "판정만 다시 볼 자리인데 그림을 또 요청했다"
    다시읽음 = _reload_records(records).data["S1sh1::cine"]
    assert 다시읽음["verify"] == {"contract": "옛-계약"}, (
        "주인이 바뀌었는데 새 판정 결과를 남의 자리에 적었다")


def test_가져오기_실패도_남의_자리에_적지_않는다(tmp_path):
    """조회가 길게 돌다 실패했다면 그 사이 주인이 바뀌었을 수 있다."""
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check,
    )

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")

    # 접수만 되고 결말을 못 본 상태를 만든다.
    끊긴놈 = _QueueClient(gen_fail=RuntimeError("poll 이 끊겼다"),
                          submit_before_fail=True)
    _run(tmp_path, client=끊긴놈, records=records, identity=reve)
    assert records.data["S1sh1::cine"]["pending"] is True

    실패놈 = _QueueClient(fetch_fail=RuntimeError("조회도 실패"))
    install_stop_check(_주인이_바뀐다())
    try:
        with pytest.raises(AppError) as ei:
            _run(tmp_path, client=실패놈, records=_reload_records(records),
                 identity=reve)
        assert ei.value.code == "step.owner_lost"
    finally:
        uninstall_stop_check()

    다시읽음 = _reload_records(records).data["S1sh1::cine"]
    assert 다시읽음["pending"] is True, (
        "주인이 바뀌었는데 접수 표식을 걷어 버렸다 — 요금이 나간 작업을 잃는다")
    assert 실패놈.gen_calls == 0


def test_남겨둔_기록에_접수_시각도_실린다(tmp_path):
    """gate 에서 멈춘 뒤 확정하면 시각이 영영 비게 된다 (Codex 3차 IMPORTANT)."""
    from app.core.errors import AppError
    from app.core.image_call_budget import (
        install_stop_check, uninstall_stop_check,
    )

    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    client = _QueueClient()
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)
    records = _StubRecords(recipe_dir / "records.json")

    def _stop():
        raise AppError(code="step.cancelled", message="정지", status_code=409)

    install_stop_check(_stop)
    try:
        with pytest.raises(AppError):
            _run(tmp_path, client=client, records=records, identity=reve)
    finally:
        uninstall_stop_check()

    남긴것 = _reload_records(records).data["S1sh1::cine"]
    assert 남긴것["staged"] is True
    assert 남긴것.get("submitted_at"), "남겨 둔 기록에 접수 시각이 없다"

    rec2, _, _ = _run(tmp_path, client=client,
                      records=_reload_records(records), identity=reve)
    assert rec2["applied"] is True
    assert rec2.get("submitted_at"), "확정본에서 접수 시각이 사라졌다"


def test_접수_기록을_못_남기면_호출자에게_알린다(monkeypatch):
    """★요금은 이미 나갔는데 번호를 디스크에 못 남긴 상태다.

    이걸 **삼키면** 호출자는 다 잘 된 줄 안다. 그 뒤 프로세스가 죽으면 다음
    걷기는 그 번호를 몰라 **같은 그림에 요금이 또 나간다**
    (2026-08-26 Codex PR#4 BLOCK-4).

    ★그래서 이 시험은 **폴링이 성공하도록** 만들어 둔다. 그래야 「삼켰다」와
     「올렸다」가 갈린다 — 폴링이 실패하면 어느 쪽이든 예외가 나서 아무것도
     못 가른다(실제로 그렇게 만들었다가 삼켜도 초록이 나왔다).
    """
    import base64

    from app.modules.llm import reve_image_client as rev

    monkeypatch.setattr(rev, "_CINE_RETRY_WAIT", 0.0)
    c = _client()
    submit수 = {"n": 0}
    png = base64.b64decode(
        "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==")

    def _req(url, method="GET", **kw):
        if method == "POST":
            submit수["n"] += 1                 # 여기가 유료 지점이다
            return (200, dict(_SUBMIT_INFO), {})
        if url.endswith("/status"):
            return (200, {"status": "COMPLETED"}, {})
        return (200, {"images": [{"url": "https://img/x.png"}]}, {})

    class _가짜응답:
        def read(self): return png
        def __enter__(self): return self
        def __exit__(self, *a): return False

    monkeypatch.setattr(c, "_request", _req)
    # 결과 이미지 내려받기는 실제 망을 타면 안 된다.
    monkeypatch.setattr(rev.urllib.request, "urlopen",
                        lambda url, timeout=0: _가짜응답())
    c.set_submit_hook(lambda info: (_ for _ in ()).throw(OSError("디스크 꽉 참")))

    with pytest.raises(rev.ReveTerminalError) as ei:
        c.generate_image("p", labeled_references=[("SOURCE", b"x")])
    # 호출자가 **무슨 일이 있었고 번호가 뭔지** 알 수 있어야 한다.
    assert ei.value.error_type == "submit_record_failed", (
        f"기록 실패를 삼켰거나 다른 사유로 뭉갰다: {ei.value.error_type!r}")
    assert ei.value.request_id == "req-777", (
        "번호를 안 알려 주면 사람이 대시보드에서 대조할 수 없다")

    assert submit수["n"] == 1, (
        f"기록을 못 남겼다고 새로 보냈다 — 같은 그림에 요금이 "
        f"{submit수['n']}번 나갔다")
    assert c.last_request_id == "req-777"


def test_접수_기록이_한_번_실패해도_다음_걷기는_무료로_가져온다(
        tmp_path, monkeypatch):
    """★client 만 보면 안 잡히는 갈래다 (2026-08-26 Codex PR#4 재리뷰).

    `_note_submit` 의 records.save 가 실패하면 client 는 올바로 멈춘다. 그런데
    바깥 마무리가 그 예외를 **결말(terminal)** 로 분류하면 접수 신원 복사를
    통째로 건너뛰어, 번호 없는 실패 기록만 남고 다음 걷기가 **새로 보낸다**.

    ★**진짜 `ReveImageClient` 를 쓴다.** 흉내 낸 스텁으로 짰다가 hook 이
     터뜨린 `OSError` 가 그대로 올라가는 바람에, 프로덕션이 감싸는
     `ReveTerminalError(submit_record_failed)` 판정을 아예 안 태웠다 —
     고친 것을 도로 빼도 초록이었다.
    """
    from app.modules.llm import reve_image_client as rev

    monkeypatch.setattr(rev, "_CINE_RETRY_WAIT", 0.0)
    reve = {"provider": "reve", "endpoint": "reve/2.1/edit"}
    recipe_dir = tmp_path / "recipe"
    recipe_dir.mkdir(exist_ok=True)

    class _한번만터지는기록(_StubRecords):
        def __init__(self, path):
            super().__init__(path)
            self.터질까 = False
            self.터진적 = False

        def save(self):
            if self.터질까 and not self.터진적:
                self.터진적 = True
                raise OSError("디스크가 꽉 찼다")
            super().save()

    실제 = _client()                       # ★프로덕션 client 그대로
    submit수 = {"n": 0}

    def _req(url, method="GET", **kw):
        if method == "POST":
            submit수["n"] += 1
            return (200, dict(_SUBMIT_INFO), {})
        return (200, {"status": "COMPLETED"}, {})

    monkeypatch.setattr(실제, "_request", _req)

    records = _한번만터지는기록(recipe_dir / "records.json")
    records.터질까 = True
    _run(tmp_path, client=실제, records=records, identity=reve)
    records.터질까 = False
    assert records.터진적, "기록 실패를 실제로 태우지 못했다"
    assert submit수["n"] == 1

    남긴것 = _reload_records(records).data["S1sh1::cine"]
    assert 남긴것.get("request_id") == "req-777", (
        f"접수 신원이 디스크에 안 남았다 — 다음 걷기가 새로 보낸다: "
        f"{sorted(남긴것)}")
    assert 남긴것.get("pending") is True
    assert 남긴것.get("status_url") == "https://queue/x/status"

    # 다음 걷기 — 새 요청 없이 그 번호로 결과만 가져와야 한다.
    두번째 = _QueueClient()
    rec2, _, _ = _run(tmp_path, client=두번째,
                      records=_reload_records(records), identity=reve)
    assert 두번째.gen_calls == 0, (
        f"무료로 가져올 수 있는데 새로 보냈다 — 같은 그림에 요금이 두 번 "
        f"나갔다 (gen_calls={두번째.gen_calls})")
    assert 두번째.fetch_calls == ["req-777"]
    assert rec2.get("applied") is True