"""★A4 최종 승인 조건 — **실제 DB** 로 확인하는 계보 불변 회귀.

계획 §3.2 = "1회차 씨드 입력 간선이 2회차 force 뒤에도 1회차 UUID·path·sha 를
가리킴 — 실제 DB + `ImagePathType` 왕복으로".

## 왜 fake DB 로는 부족한가

`ImageAsset.file_path` 는 `ImagePathType` 이라 **저장에서 절대→상대,
읽기에서 상대→절대**로 표현이 바뀐다(file_paths.py:90). 2026-08-01 에 이
왕복을 fake 로 흉내내다 **구현이 틀린 것과 같이 틀려** 필수 참조가 전부
막힐 뻔했다. 게다가 PG 에는 `file_path NOT LIKE '/%'` CHECK 가 있어, 루트
밖 경로는 저장 자체가 거부된다 — 픽스처를 **프로젝트 루트 안** 형태로
세워야 실제 배치와 같아진다.

Lane: ``-m pg`` 명시 시만 실행.
"""
from __future__ import annotations

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path

import pytest
from sqlalchemy import text as sql_text

from tests.core.test_form_reference_round_wiring import (  # noqa: F401
    GID,
    _Wired,
    _sha_bytes,
    stub_run_group,
)

pytestmark = pytest.mark.pg


def _seed_project(session, pid: str) -> None:
    """FK 의존 — user_account + project_registry 시드 (기존 pg 테스트 관례)."""
    uid = f"test-a4-{uuid.uuid4()}"
    session.execute(sql_text(
        "INSERT INTO user_account (id, username, display_name, password_hash, "
        "role, is_active, created_at, updated_at) VALUES "
        "(:uid, :uname, 't', 'x', 'creator', 1, '2026-01-01', '2026-01-01')"
    ), {"uid": uid, "uname": f"u_{uid}"})
    session.execute(sql_text(
        "INSERT INTO project_registry (id, name, created_by, created_at, "
        "updated_at) VALUES (:pid, 'a4-lineage', :uid, '2026-01-01', "
        "'2026-01-01')"
    ), {"pid": pid, "uid": uid})
    session.commit()


@pytest.fixture()
def pg_wired(tmp_path: Path, monkeypatch, pg_session) -> _Wired:
    """`_execute` 를 **실제 PG 세션**으로 태우는 배선."""
    from app.core.config import settings
    from app.core.steps.outdoor_structure_form_reference_step import (
        OutdoorStructureFormReferenceStep,
    )

    pid = f"a4-{uuid.uuid4()}"
    _seed_project(pg_session, pid)

    root = tmp_path / "the-road-root"
    projects = root / "projects"
    projects.mkdir(parents=True, exist_ok=True)
    monkeypatch.setattr("app.core.file_paths._resolve_root", lambda: root)
    monkeypatch.setattr(settings, "projects_dir", str(projects),
                        raising=False)
    for flag in ("outdoor_lane_pipe_enabled", "outdoor_lane_plan_enabled"):
        monkeypatch.setattr(settings, flag, True, raising=False)
    monkeypatch.setattr(settings, "outdoor_seed_all_groups_enabled", False,
                        raising=False)

    cps = {
        "outdoor_lane_plan": {"data": {"groups": {
            GID: {"status": "ok", "plan": {
                "shot_bindings": [{"lane": "structure_plate"}]}}}}},
        "outdoor_place_spec": {"data": {"groups": {
            GID: {"spec": {"items": [{"kind": "sample-fixture-item"}]}}}}},
        "background_classify": {"data": {"building_groups": []}},
        "entity_merge": {"data": {"locations": []}},
        "scene_save": {"data": {"segments": [
            {"text": "sample fixture scene text"}]}},
        "visual_world_rules": None,
    }

    step = OutdoorStructureFormReferenceStep.__new__(
        OutdoorStructureFormReferenceStep)
    step.project_id, step.episode_id = pid, "E1"
    step.db = pg_session
    step._load_prev_checkpoint = lambda sid: cps.get(sid)  # type: ignore

    w = _Wired(step, root, cps)
    step.load_checkpoint = lambda: w.cp                    # type: ignore

    monkeypatch.setattr(
        "app.modules.pipeline.outdoor_structure_seed.derive_seed_inputs",
        lambda **_kw: {"structure_desc": "sample fixture structure"})
    monkeypatch.setattr(
        "app.core.steps.shot_conti_light_step._resolve_openai_client",
        lambda: object())
    monkeypatch.setattr(
        "app.core.world_context.build_world_facts_block", lambda _cp: "")
    return w


def _record_seed_edge(w: _Wired, form_ref_asset_id: str) -> str:
    """씨드 자산이 그 참조를 입력 간선으로 물었다고 기록한다 (A3 배선과 동형)."""
    from app.models.project import ImageAsset

    seed_png = (Path(w.step._ref_dir(GID)).parent / "seed" / "seed.png")
    seed_png.parent.mkdir(parents=True, exist_ok=True)
    seed_png.write_bytes(b"sample-fixture-seed")
    seed_id = str(uuid.uuid4())
    w.step.db.add(ImageAsset(
        id=seed_id, project_id=w.step.project_id,
        episode_id=w.step.episode_id, asset_type="structure_seed",
        entity_id=GID, variant_type="structure_seed",
        file_path=str(seed_png), status="generated", is_primary=0,
        input_image_ids=json.dumps([form_ref_asset_id]),
        created_at=datetime.now(timezone.utc)))
    w.step.db.commit()
    return seed_id


def test_a_second_force_run_never_rewrites_the_first_rounds_lineage(
        pg_wired, monkeypatch):
    """★재검색 한 번이 과거 씨드의 입력 간선을 바꾸던 결함의 회귀."""
    from app.core.file_paths import resolve_image_path
    from app.models.project import ImageAsset

    w = pg_wired
    stub_run_group(w, monkeypatch)

    first = w.run_and_save_cp()
    e1 = first["data"]["groups"][GID]
    a1, p1, sha1 = (e1["form_ref_asset_id"], e1["form_ref_path"],
                    e1["form_ref_sha256"])
    seed_id = _record_seed_edge(w, a1)

    # ★ORM 왕복이 표현을 바꾸는지부터 확인한다 — 안 바꾸면 이 회귀가 무의미
    w.step.db.expire_all()
    stored = w.step.db.execute(
        sql_text("SELECT file_path FROM image_asset WHERE id = :i"),
        {"i": a1}).scalar()
    loaded = w.step.db.query(ImageAsset).filter_by(id=a1).first().file_path
    assert stored != loaded, "round-trip 이 표현을 바꾸지 않으면 이 회귀는 공허"
    assert not str(stored).startswith("/"), "DB 에는 상대로 저장돼야 한다"

    second = w.run_and_save_cp(mode="force")
    e2 = second["data"]["groups"][GID]
    assert e2["form_ref_asset_id"] != a1
    assert e2["round_id"] == "r002"

    # ── 최종 승인 조건 ──────────────────────────────────────────
    w.step.db.expire_all()
    row1 = w.step.db.query(ImageAsset).filter_by(id=a1).first()
    assert row1 is not None, "1회차 자산이 사라졌다"
    got = resolve_image_path(row1.file_path)
    assert str(got) == str(resolve_image_path(p1)), "1회차 경로가 바뀌었다"
    assert _sha_bytes(Path(str(got)).read_bytes()) == sha1, "1회차 bytes 변조"

    seed_row = w.step.db.query(ImageAsset).filter_by(id=seed_id).first()
    assert json.loads(seed_row.input_image_ids) == [a1]


def test_the_lineage_regression_would_catch_the_old_upsert(
        pg_wired, monkeypatch):
    """★위반 주입 — 옛 upsert 를 되살리면 위 회귀가 실제로 깨지는가.

    되살린 동작 = 그룹당 한 row 를 찾아 `file_path` 를 갱신(신규 UUID 는 row
    가 없을 때만). 이것이 A4 이전의 실제 코드였다.
    """
    from app.core.file_paths import (
        resolve_image_path,
        to_relative_image_path,
    )
    from app.models.project import ImageAsset

    w = pg_wired
    stub_run_group(w, monkeypatch)

    def _legacy_upsert(*, group_id, asset_id, abs_png_path, sha256,
                       prompt_used):
        existing = (
            w.step.db.query(ImageAsset).filter_by(
                project_id=w.step.project_id, episode_id=w.step.episode_id,
                asset_type="structure_form_ref", entity_id=group_id,
                variant_type="form_ref").first())
        if existing:
            existing.file_path = to_relative_image_path(abs_png_path)
            w.step.db.flush()
            return str(existing.id)
        row = ImageAsset(
            id=asset_id, project_id=w.step.project_id,
            episode_id=w.step.episode_id, asset_type="structure_form_ref",
            entity_id=group_id, variant_type="form_ref",
            file_path=abs_png_path, status="generated", is_primary=0,
            created_at=datetime.now(timezone.utc))
        w.step.db.add(row)
        w.step.db.flush()
        return asset_id

    monkeypatch.setattr(w.step, "_bind_ref_asset", _legacy_upsert)

    first = w.run_and_save_cp()
    e1 = first["data"]["groups"][GID]
    a1, p1 = e1["form_ref_asset_id"], e1["form_ref_path"]
    _record_seed_edge(w, a1)

    w.run_and_save_cp(mode="force")

    w.step.db.expire_all()
    row1 = w.step.db.query(ImageAsset).filter_by(id=a1).first()
    # 소급 변조가 **실제로 일어난다** — 그래서 앞 회귀가 실효성이 있다
    assert str(resolve_image_path(row1.file_path)) != str(
        resolve_image_path(p1)), (
        "옛 upsert 를 되살렸는데도 경로가 그대로다 — 앞 회귀가 공허하다")


def test_a_later_groups_rollback_does_not_undo_an_earlier_groups_asset(
        pg_wired, monkeypatch):
    """★두 저장소의 durability 계약 — **실제 트랜잭션**으로 확인한다.

    라운드 상태는 파일에, 자산은 DB 트랜잭션에 산다. 앞 그룹을 닫은 뒤
    commit 하지 않으면, 뒤 그룹 실패의 `db.rollback()` 이 앞 그룹 row 까지
    되돌린다 — 그런데 journal 은 FINALIZED 로 남아 **존재하지 않는 UUID 를
    가리키는 완료 라운드**가 생긴다(Codex A4 리뷰 BLOCKING-1).
    """
    from app.models.project import ImageAsset
    from app.modules.pipeline.form_ref_rounds import RoundState, load_round

    w = pg_wired
    w.add_group("g_beta")

    def _fake(*, group_id, structure_desc, world_facts_block, source_text,
              client, candidate_dir, start_index=1, narrow=False):
        w.calls.append({"group_id": group_id})
        if group_id == "g_beta":
            raise RuntimeError("sample fixture group failure")
        candidate_dir.mkdir(parents=True, exist_ok=True)
        dest = candidate_dir / f"cand_{start_index:02d}.png"
        dest.write_bytes(b"sample-fixture-alpha")
        return {"status": "ok", "group_id": group_id,
                "form_ref_path": str(dest),
                "form_ref_sha256": _sha_bytes(b"sample-fixture-alpha"),
                "chosen_prompt_used": "d", "candidate_count": 1, "audit": {}}

    monkeypatch.setattr(w.step, "_run_group", _fake)

    out = w.run()

    assert out["completed_count"] == 1 and out["failed_count"] == 1
    aid = out["data"]["groups"][GID]["form_ref_asset_id"]
    # ★rollback 을 거친 뒤에도 앞 그룹 자산이 **DB 에 실재**한다
    w.step.db.expire_all()
    row = w.step.db.query(ImageAsset).filter_by(id=aid).first()
    assert row is not None, "뒤 그룹의 rollback 이 앞 그룹 자산을 지웠다"
    # journal 과 DB 가 같은 말을 한다
    assert load_round(w.step._ref_dir(GID), "r001").state == (
        RoundState.FINALIZED)


def test_a_commit_failure_leaves_the_round_unfinished(pg_wired, monkeypatch):
    """★자산이 durable 해지지 못하면 라운드를 완료로 닫지 않는다.

    반대 순서였다면(먼저 FINALIZED, 나중 commit) 이 실패가 phantom 완료를
    남겼을 것이다.
    """
    from app.modules.pipeline.form_ref_rounds import RoundState, load_round

    w = pg_wired
    stub_run_group(w, monkeypatch)

    def _boom():
        raise RuntimeError("sample fixture commit failure")

    monkeypatch.setattr(w.step.db, "commit", _boom)

    # commit 이 끝내 실패하면 스텝이 선다(그룹별 commit 이 이미 끝난 앞
    # 그룹들은 durable 하므로 잃는 것이 없다).
    with pytest.raises(RuntimeError):
        w.run()

    state = load_round(w.step._ref_dir(GID), "r001").state
    assert state != RoundState.FINALIZED, "commit 이 실패했는데 완료로 닫혔다"
    # 선택은 durable 하다 — 재개가 유료 없이 바인딩부터 이어갈 수 있다
    assert state == RoundState.SELECTED
    assert load_round(w.step._ref_dir(GID), "r001").result


def test_consumer_picks_only_the_row_named_by_the_checkpoint(
        pg_wired, monkeypatch):
    """계획 §3.3 — 같은 그룹에 form_ref row 가 둘일 때.

    소비자(`_form_ref_asset_matches`)는 `entity_id+variant_type` 로 아무 행이나
    잡는 것이 아니라 **CP 가 지정한 asset_id 의 행**만 쓴다.
    """
    from app.core.steps.outdoor_structure_seed_step import (
        OutdoorStructureSeedStep,
    )

    w = pg_wired
    stub_run_group(w, monkeypatch)
    first = w.run_and_save_cp()
    second = w.run_and_save_cp(mode="force")
    e1 = first["data"]["groups"][GID]
    e2 = second["data"]["groups"][GID]
    w.step.db.commit()

    rows = w.step.db.execute(sql_text(
        "SELECT count(*) FROM image_asset WHERE project_id = :p "
        "AND entity_id = :g AND variant_type = 'form_ref'"),
        {"p": w.step.project_id, "g": GID}).scalar()
    assert rows == 2, f"같은 그룹에 form_ref row 가 {rows}개"

    seed = OutdoorStructureSeedStep.__new__(OutdoorStructureSeedStep)
    seed.project_id, seed.episode_id = w.step.project_id, w.step.episode_id
    seed.db = w.step.db

    # 각 CP entry 는 **자기 라운드의 행**과만 짝이 맞는다
    assert seed._form_ref_asset_matches(
        gid=GID, asset_id=e1["form_ref_asset_id"],
        path_s=e1["form_ref_path"]) is True
    assert seed._form_ref_asset_matches(
        gid=GID, asset_id=e2["form_ref_asset_id"],
        path_s=e2["form_ref_path"]) is True
    # 교차 짝(1회차 자산 + 2회차 경로)은 거부된다
    assert seed._form_ref_asset_matches(
        gid=GID, asset_id=e1["form_ref_asset_id"],
        path_s=e2["form_ref_path"]) is False