"""C2 owned_object_usage[] echo schema v1 — integrated test (G1-G7).

Spec ref: docs/superpowers/specs/2026-05-20-c2-owned-object-usage-echo-v1-design.md
Plan ref: docs/superpowers/plans/2026-05-20-c2-owned-object-usage-echo-v1-implementation.md
Codex APPROVED_FOR_W3_ENTRY_WITH_NARROW_PLAN_GAP_ABSORB.

no live LLM + NO VLM + no regex semantic. 모든 gate 는 순수 helper 함수 +
prompt-pack 파일 read 만 사용한다.
"""
import ast
import json
import pathlib

import pytest

from app.core.errors import AppError
from app.core.steps._owned_helpers import (
    OWNED_SENTINEL_SCHEMA_VERSION,
    _REQUIRED_SENTINEL_FIELDS,
    assert_close_framing_absent_echo,
    assert_owned_sentinel_shape,
    assert_owned_usage_hash_matches,
    build_owned_sentinel,
    compute_owned_usage_hash,
    has_redraw_violation,
    validate_owned_object_usage_coverage,
)


REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]


def _resolve_latest_major_pack(stem: str, major: str) -> pathlib.Path:
    """prompt-pack stem 의 major 버전 dir 를 glob 으로 resolve (UTC suffix 무관)."""
    candidates = sorted((REPO_ROOT / "prompts/_base" / stem).glob(f"{major}.*"))
    assert candidates, f"no prompt-pack found for {stem}/{major}.*"
    return candidates[-1]


SCENE_DETAIL_V29_DIR = _resolve_latest_major_pack("scene_detail", "29")
SCENE_DETAIL_V30_DIR = _resolve_latest_major_pack("scene_detail", "30")
OWNED_JUDGE_V3_DIR = _resolve_latest_major_pack("scene_detail_owned_judge", "3")
OWNED_JUDGE_V4_DIR = _resolve_latest_major_pack("scene_detail_owned_judge", "4")


# === HISTORICAL AUDIT TRAIL — v3 owned_judge closed-list residue catalog BEGIN ===
# 아래 문자열은 v3 system.md 의 closed-list 잔재 (redraw 동사 화이트리스트 /
# redraw 예시 / 판정 우선순위 / ambiguous default) 로, judge v4 echo-verify-only
# prompt 에서 0 임을 증명하는 G4 gate 의 검사 대상이다. live policy 아님 —
# audit / residue scan 이 본 catalog 를 잔재로 오인하지 않도록 marker 로 감싼다.
_V3_CLOSED_LIST_RESIDUE = (
    "redraw 동사 화이트리스트",
    "판정 우선순위",
    "ambiguous mention 기본값",
    "add a new door on the right wall",
    "render a glowing portal in the wall",
    "set up a desk by the window",
)
# === HISTORICAL AUDIT TRAIL — v3 owned_judge closed-list residue catalog END ===

_VLM_FORBIDDEN = {"vlm", "vision_lm", "vision_language", "lvm_judge", "lvm_validator"}


def _assert_no_forbidden_model_refs(tree: ast.AST, where: str) -> None:
    """AST-based import/call leak check (substring grep 보다 엄격)."""
    for node in ast.walk(tree):
        if isinstance(node, (ast.Import, ast.ImportFrom)):
            for alias in node.names:
                name_lower = (alias.name or "").lower()
                for kw in _VLM_FORBIDDEN:
                    assert kw not in name_lower, f"VLM import leak in {where}: {alias.name}"
            if isinstance(node, ast.ImportFrom):
                mod_lower = (node.module or "").lower()
                for kw in _VLM_FORBIDDEN:
                    assert kw not in mod_lower, f"VLM module leak in {where}: {node.module}"
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
            for kw in _VLM_FORBIDDEN:
                assert kw not in node.func.id.lower(), f"VLM call leak in {where}: {node.func.id}"


def _entry(token: str, kind: str = "anchor", phrase: str = "ref phrase") -> dict:
    return {
        "owned_token": token,
        "usage_kind": kind,
        "source_phrase": "" if kind == "absent" else phrase,
    }


# ── G1 — Producer schema owned_object_usage[] required + field shape ──
def test_g1_producer_schema_owned_object_usage_required():
    schema = json.loads((SCENE_DETAIL_V30_DIR / "detail_schema.json").read_text())
    item = schema["properties"]["t2i_variations"]["items"]
    assert "owned_object_usage" in item["required"]
    oou = item["properties"]["owned_object_usage"]
    assert oou["type"] == "array"
    oitem = oou["items"]
    props = oitem["properties"]
    assert set(props) == {"owned_token", "usage_kind", "source_phrase"}
    assert set(oitem["required"]) == {"owned_token", "usage_kind", "source_phrase"}
    assert oitem["additionalProperties"] is False
    assert props["owned_token"]["minLength"] == 1


def test_g1_coverage_duplicate_token_raises():
    # cardinality 일치하나 owned_token 중복 → AppError.
    dup = [_entry("door"), _entry("door")]
    with pytest.raises(AppError) as exc:
        validate_owned_object_usage_coverage(dup, ["door", "window"])
    assert exc.value.code == "step.contract_violation"
    assert "duplicate" in exc.value.message.lower()


def test_g1_coverage_missing_token_raises():
    # owned 에 door 있는데 echo 가 비어 cardinality 부족 → AppError.
    with pytest.raises(AppError) as exc:
        validate_owned_object_usage_coverage([], ["door"])
    assert exc.value.code == "step.contract_violation"


def test_g1_coverage_extra_token_raises():
    # echo 에 owned 밖 token 추가 → cardinality 초과 → AppError.
    extra = [_entry("door"), _entry("window")]
    with pytest.raises(AppError) as exc:
        validate_owned_object_usage_coverage(extra, ["door"])
    assert exc.value.code == "step.contract_violation"


def test_g1_coverage_empty_owned_ok():
    # owned 가 비면 echo 도 [] — required field 존재, no raise.
    validate_owned_object_usage_coverage([], [])


# ── G2 — usage_kind enum strict + source_phrase non-empty unless absent ──
def test_g2_usage_kind_enum_strict():
    bad = [{"owned_token": "door", "usage_kind": "future_kind", "source_phrase": "x"}]
    with pytest.raises(AppError) as exc:
        validate_owned_object_usage_coverage(bad, ["door"])
    assert exc.value.code == "step.contract_violation"


def test_g2_source_phrase_required_unless_absent():
    # anchor 인데 source_phrase 비면 위반.
    with pytest.raises(AppError):
        validate_owned_object_usage_coverage(
            [{"owned_token": "door", "usage_kind": "anchor", "source_phrase": ""}],
            ["door"],
        )
    # absent 면 빈 source_phrase 허용.
    validate_owned_object_usage_coverage(
        [{"owned_token": "door", "usage_kind": "absent", "source_phrase": ""}],
        ["door"],
    )


def test_g2_schema_usage_kind_enum():
    schema = json.loads((SCENE_DETAIL_V30_DIR / "detail_schema.json").read_text())
    oitem = (
        schema["properties"]["t2i_variations"]["items"]["properties"]
        ["owned_object_usage"]["items"]
    )
    assert set(oitem["properties"]["usage_kind"]["enum"]) == {"redraw", "anchor", "absent"}


# ── G3 — sentinel v2 + owned_usage_hash tamper mismatch → AppError ──
def test_g3_sentinel_v2_schema_version_and_fields():
    assert OWNED_SENTINEL_SCHEMA_VERSION == 2
    assert "owned_usage_hash" in _REQUIRED_SENTINEL_FIELDS
    assert len(_REQUIRED_SENTINEL_FIELDS) == 7


def test_g3_build_sentinel_has_owned_usage_hash():
    sentinel = build_owned_sentinel(
        owned=[], camera_direction="", t2i_prompt="x",
        is_close_framing=False, violations=[], owned_object_usage=[],
    )
    assert "owned_usage_hash" in sentinel
    assert_owned_sentinel_shape(sentinel, where="g3")


def test_g3_owned_usage_hash_tamper_mismatch_raises():
    usage = [_entry("door", "anchor", "near the door")]
    sentinel = build_owned_sentinel(
        owned=["door"], camera_direction="medium", t2i_prompt="a room near the door",
        is_close_framing=False, violations=[], owned_object_usage=usage,
    )
    # unmodified echo → match (no raise).
    assert_owned_usage_hash_matches(sentinel, usage)
    # tampered echo → owned_usage_hash mismatch → AppError.
    tampered = [_entry("door", "redraw", "draw a new door")]
    with pytest.raises(AppError) as exc:
        assert_owned_usage_hash_matches(sentinel, tampered)
    assert exc.value.code == "step.contract_violation"


def test_g3_owned_usage_hash_deterministic_order_independent():
    a = [_entry("door"), _entry("window")]
    b = [_entry("window"), _entry("door")]
    assert compute_owned_usage_hash(a) == compute_owned_usage_hash(b)


# ── G4 — judge v4 no closed-list residue + echo-verify prompt-pack ──
def test_g4_judge_v4_no_closed_list_residue():
    sys_md = (OWNED_JUDGE_V4_DIR / "system.md").read_text()
    for residue in _V3_CLOSED_LIST_RESIDUE:
        assert residue not in sys_md, f"judge v4 closed-list residue leak: {residue!r}"


def test_g4_judge_v4_is_echo_verify():
    sys_md = (OWNED_JUDGE_V4_DIR / "system.md").read_text()
    assert sys_md.strip(), "judge v4 system.md empty"
    # echo-verify-only prompt 은 owned_object_usage 를 cross-check 대상으로 명시.
    assert "owned_object_usage" in sys_md


def test_g4_v4_user_template_placeholders():
    tmpl = (OWNED_JUDGE_V4_DIR / "user_template.md").read_text()
    assert "{owned_object_usage}" in tmpl
    for ph in ("{t2i_prompt}", "{owned_list_block}", "{camera_direction}"):
        assert ph in tmpl, f"judge v4 user_template missing placeholder {ph}"


# ── G5 — verdict enum + has_redraw_violation preserve ──
def test_g5_v4_schema_byte_identical_v3():
    v3 = (OWNED_JUDGE_V3_DIR / "schema.json").read_bytes()
    v4 = (OWNED_JUDGE_V4_DIR / "schema.json").read_bytes()
    assert v3 == v4, "v4 schema.json must be byte-identical to v3 (verdict enum preserve)"


def test_g5_v4_schema_verdict_enum():
    sch = json.loads((OWNED_JUDGE_V4_DIR / "schema.json").read_text())
    verdict = sch["properties"]["violations"]["items"]["properties"]["verdict"]
    assert set(verdict["enum"]) == {"redraw_violation", "anchor_reference"}


def test_g5_has_redraw_violation_preserve():
    assert has_redraw_violation([]) is False
    redraw = [{"owned_object": "door", "violating_phrase": "p", "reason": "r",
               "verdict": "redraw_violation"}]
    anchor = [{"owned_object": "door", "violating_phrase": "p", "reason": "r",
               "verdict": "anchor_reference"}]
    assert has_redraw_violation(redraw) is True
    assert has_redraw_violation(anchor) is False


# ── G6 — boundary preserve (t2i_prompt body + C1 + Area C) ──
def test_g6_t2i_prompt_body_shape_unchanged_v29_to_v30():
    v29 = json.loads((SCENE_DETAIL_V29_DIR / "detail_schema.json").read_text())
    v30 = json.loads((SCENE_DETAIL_V30_DIR / "detail_schema.json").read_text())
    i29 = v29["properties"]["t2i_variations"]["items"]["properties"]
    i30 = v30["properties"]["t2i_variations"]["items"]["properties"]
    assert i29["t2i_prompt"] == i30["t2i_prompt"], "t2i_prompt body shape drift"
    # v30 = v29 + owned_object_usage 만 추가 (sidecar evidence, 다른 field 변경 0).
    assert set(i30) - set(i29) == {"owned_object_usage"}
    assert set(i29) - set(i30) == set()


def test_g6_c1_perception_mode_boundary_preserve():
    from app.core.perception_mode import PERCEPTION_MODES
    assert "mirror" in PERCEPTION_MODES
    assert len(PERCEPTION_MODES) == 8
    # v30 = v29 copy — C1 perception_mode cross-reference 보존.
    sys_md = (SCENE_DETAIL_V30_DIR / "system.md").read_text()
    assert "REPRODUCTION_PERCEPTION_MODES" in sys_md or "perception_mode.py" in sys_md


def test_g6_area_c_boundary_preserve():
    # C2 는 render_prompt_card 를 건드리지 않음 — Area C directionality anchor 보존.
    src = (REPO_ROOT / "backend/app/core/steps/render_prompt_card.py").read_text()
    assert "_ALLOWED_DIRECTIONALITY_CLASSES" in src


# ── G7 — NO VLM (AST) + no silent fallback + close-framing absent echo ──
def test_g7_no_vlm_owned_modules():
    for mod_path in (
        "backend/app/core/steps/_owned_judge.py",
        "backend/app/core/steps/_owned_helpers.py",
    ):
        tree = ast.parse((REPO_ROOT / mod_path).read_text())
        _assert_no_forbidden_model_refs(tree, mod_path)


def test_g7_no_vlm_test_module():
    _assert_no_forbidden_model_refs(
        ast.parse(pathlib.Path(__file__).read_text()), __file__
    )


def test_g7_close_framing_absent_echo_ok():
    ok = [_entry("door", "absent"), _entry("window", "absent")]
    assert_close_framing_absent_echo(ok, where="g7")  # no raise


def test_g7_close_framing_non_absent_raises():
    bad = [_entry("door", "anchor", "near the door"), _entry("window", "absent")]
    with pytest.raises(AppError) as exc:
        assert_close_framing_absent_echo(bad, where="g7")
    assert exc.value.code == "step.contract_violation"


def test_g7_coverage_mismatch_raises_no_silent_fallback():
    # owned_object_usage missing / non-list / coverage mismatch → AppError fail-fast.
    with pytest.raises(AppError):
        validate_owned_object_usage_coverage("not a list", ["door"])
    with pytest.raises(AppError):
        validate_owned_object_usage_coverage([_entry("window")], ["door"])
