"""form_reference → seed 소비자 계약 (2026-08-01, Codex 리뷰 BLOCKING 3).

## 무엇이 잘못됐었나

소비자가 필수 그룹을 이렇게 정했다:

    mandatory = (cp or {}).get("data", {}).get("mandatory_group_ids", [])
                or collect_lane2_groups(lane_data, all_groups=False)

이 `or` 폴백이 **두 상황을 구분하지 못한다**:

1. 선행 스텝을 아예 안 돌렸다 → 전부 필수로 세우는 것이 맞다(fail-closed)
2. 선행이 정상으로 돌았고 필수 대상이 0건이다 → 폴백이 그 판정을 뒤집어
   제외된 그룹까지 되살린다

게다가 폴백은 `all_groups=False` 로 **고정**인데 seed·form_reference 의
대상 집합은 `outdoor_seed_all_groups_enabled` 플래그를 따른다. 플래그가
켜지면 두 스텝이 **서로 다른 세상**을 보고, 구 CP 가 남아 있으면 15그룹이
조용히 순수 T2I 로 내려간다.

## 새 계약

선행 CP 는 **완결**이어야 하고(status=completed), 그 CP 가 본 우주
(`mandatory_group_ids ∪ no_spec_group_ids`)가 **지금 seed 가 보는 대상과
같아야** 한다. 어긋나면 조용히 내려가지 않고 그 자리에서 선다.
"""
from __future__ import annotations

import pytest

from app.core.errors import AppError
from app.core.steps.outdoor_structure_seed_step import (
    resolve_form_reference_contract,
)

TARGET = ["g_alpha", "g_beta"]


def _meta(**over):
    from app.modules.pipeline.search_grounded_ref import (
        TARGET_POLICY_VERSION,
        resolve_ref_pack_version,
    )

    base = {"policy_version": TARGET_POLICY_VERSION,
            "pack_version": resolve_ref_pack_version()}
    base.update(over)
    return base


def _cp(*, mandatory, no_spec=(), status="completed", universe=None,
        groups=None):
    ids = list(mandatory)
    return {
        "status": status,
        "data": {
            "groups": ({g: {"status": "ok"} for g in ids}
                       if groups is None else groups),
            "mandatory_group_ids": ids,
            "target": _meta(
                no_spec_group_ids=list(no_spec),
                universe_group_ids=(list(universe) if universe is not None
                                    else ids + list(no_spec))),
        },
    }


def _specs(*, with_items=(), without_items=()):
    """현재 outdoor_place_spec 산출 모양 — items 유무가 분할을 정한다."""
    out = {g: {"spec": {"items": [{"name": "sample-fixture-item"}]}}
           for g in with_items}
    out.update({g: {"spec": {"items": []}} for g in without_items})
    return out


def _specs_from_cp(cp, targets):
    """CP 가 선언한 분할과 **일치하는** 현재 spec = 드리프트 없음 기준선."""
    no_spec = set()
    if isinstance(cp, dict):
        meta = ((cp.get("data") or {}) if isinstance(cp.get("data"), dict)
                else {}).get("target")
        if isinstance(meta, dict):
            raw = meta.get("no_spec_group_ids")
            if isinstance(raw, list):
                no_spec = {g for g in raw if isinstance(g, str)}
    return _specs(with_items=[g for g in targets if g not in no_spec],
                  without_items=[g for g in targets if g in no_spec])


def _resolve(*, formref_cp, target_gids, spec_groups=None):
    """spec_groups 미지정 = CP 분할과 일치하는 현재 spec (드리프트 없음)."""
    return resolve_form_reference_contract(
        formref_cp=formref_cp, target_gids=target_gids,
        spec_groups=(_specs_from_cp(formref_cp, target_gids)
                     if spec_groups is None else spec_groups))


def test_missing_checkpoint_is_fail_closed():
    """선행 미실행 + 대상 있음 = 조용히 순수 T2I 로 내려가지 않는다."""
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=None, target_gids=TARGET)
    assert "outdoor_structure_form_reference" in str(exc.value.message)


def test_missing_checkpoint_with_no_target_is_noop():
    """대상 자체가 없으면 선행도 필요 없다 — 빈 집합."""
    assert _resolve(
        formref_cp=None, target_gids=[]) == set()


def test_all_targets_mandatory_is_accepted():
    assert _resolve(
        formref_cp=_cp(mandatory=TARGET), target_gids=TARGET) == set(TARGET)


def test_spec_missing_group_is_excluded_not_revived():
    """★폴백 제거의 핵심 — spec 결손으로 제외된 그룹을 되살리지 않는다.

    form_reference 가 근거 부족으로 뺀 그룹을 소비자가 필수로 되돌리면,
    통과할 수 없는 게이트를 세우고 메시지도 원인을 가린다.
    """
    got = _resolve(
        formref_cp=_cp(mandatory=["g_alpha"], no_spec=["g_beta"]),
        target_gids=TARGET)
    assert got == {"g_alpha"}


def test_zero_mandatory_does_not_trigger_fallback():
    """필수 0건 + 나머지가 전부 no_spec = 정상. 폴백이 뒤집으면 안 된다."""
    got = _resolve(
        formref_cp=_cp(mandatory=[], no_spec=TARGET), target_gids=TARGET)
    assert got == set()


def test_universe_mismatch_is_fail_closed():
    """CP 가 본 우주와 지금 대상이 다르면 선다 — 구 CP 재사용 차단."""
    with pytest.raises(AppError) as exc:
        _resolve(
            formref_cp=_cp(mandatory=["g_alpha"]), target_gids=TARGET)
    assert "g_beta" in str(exc.value.message)


def test_unknown_group_in_checkpoint_is_fail_closed():
    """CP 에 지금 대상에 없는 그룹이 있어도 어긋난 것이다."""
    with pytest.raises(AppError):
        _resolve(
            formref_cp=_cp(mandatory=TARGET + ["g_gone"]), target_gids=TARGET)


def test_incomplete_checkpoint_is_fail_closed():
    """running/failed CP 를 완료로 취급하지 않는다."""
    with pytest.raises(AppError) as exc:
        _resolve(
            formref_cp=_cp(mandatory=TARGET, status="running"),
            target_gids=TARGET)
    assert "running" in str(exc.value.message)


# ── 구 shape CP 는 default-deny (Codex 지적 10 을 A1 에 흡수) ────────
# 구 CP fallback 을 관대하게 두면, 계약이 생기기 전에 만들어진 체크포인트가
# 현재 지문으로 "승격"되어 입력이 바뀐 것까지 재사용된다. 구 shape 는
# 조용히 통과시키지 않고 재실행을 요구한다.

def test_checkpoint_without_mandatory_field_is_rejected():
    """`mandatory_group_ids` 키 자체가 없는 구 CP = 계약 이전 산출."""
    cp = {"status": "completed",
          "data": {"groups": {}, "target": _meta(no_spec_group_ids=[])}}
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET)
    assert "mandatory_group_ids" in str(exc.value.message)


def test_checkpoint_without_no_spec_field_is_rejected():
    """`no_spec_group_ids` 가 없으면 우주를 복원할 수 없다 — 빈 값과 다르다."""
    cp = {"status": "completed",
          "data": {"groups": {}, "mandatory_group_ids": list(TARGET),
                   "target": _meta()}}
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET)
    assert "no_spec_group_ids" in str(exc.value.message)


# ── Codex 재리뷰 BLOCKING 2 — 키 존재가 아니라 **의미**를 검증한다 ──

def test_mandatory_and_no_spec_must_be_disjoint():
    """★union 만 보면 겹쳐도 통과한다(실측: accepted={'g'}).

    한 그룹이 "참조 필수"이면서 동시에 "근거 없어 제외"일 수는 없다.
    """
    with pytest.raises(AppError) as exc:
        _resolve(
            formref_cp=_cp(mandatory=["g_alpha"], no_spec=["g_alpha"],
                           universe=["g_alpha"]),
            target_gids=["g_alpha"])
    assert "동시에" in str(exc.value.message)


@pytest.mark.parametrize("bad", ["bad", 42, [], 3.5])
def test_malformed_root_is_app_error_not_attribute_error(bad):
    """malformed 입력이 AttributeError 로 새면 계약 위반이 생성 실패로 오해된다."""
    with pytest.raises(AppError):
        _resolve(formref_cp=bad, target_gids=TARGET)


def test_stale_policy_version_is_rejected():
    """같은 그룹 목록이어도 다른 근거로 뽑힌 집합은 재사용할 수 없다."""
    cp = _cp(mandatory=TARGET)
    cp["data"]["target"]["policy_version"] = "2-old-policy"
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET)
    assert "policy_version" in str(exc.value.message)


def test_stale_pack_version_is_rejected():
    cp = _cp(mandatory=TARGET)
    cp["data"]["target"]["pack_version"] = "1.202600000000"
    with pytest.raises(AppError):
        _resolve(formref_cp=cp, target_gids=TARGET)


def test_universe_must_equal_the_partition():
    """producer 가 남긴 우주와 분할이 어긋나면 CP 가 스스로 모순이다."""
    with pytest.raises(AppError):
        _resolve(
            formref_cp=_cp(mandatory=TARGET, universe=["g_alpha"]),
            target_gids=TARGET)


def test_mandatory_group_without_ok_output_is_rejected():
    """필수인데 참조 산출이 실패했으면 세운다."""
    with pytest.raises(AppError) as exc:
        _resolve(
            formref_cp=_cp(mandatory=TARGET,
                           groups={"g_alpha": {"status": "ok"},
                                   "g_beta": {"status": "failed"}}),
            target_gids=TARGET)
    assert "g_beta" in str(exc.value.message)


@pytest.mark.parametrize("bad_list", [
    "not-a-list", [1, 2], ["", "g"], ["g", "g"]])
def test_malformed_group_lists_are_rejected(bad_list):
    """타입·빈 문자열·중복 — 전부 계약 위반이다."""
    cp = _cp(mandatory=TARGET)
    cp["data"]["mandatory_group_ids"] = bad_list
    with pytest.raises(AppError):
        _resolve(formref_cp=cp, target_gids=TARGET)


def test_empty_no_spec_list_is_valid():
    """빈 리스트는 정상 — '제외된 그룹이 없다'는 판정이다."""
    assert _resolve(
        formref_cp=_cp(mandatory=TARGET, no_spec=[]),
        target_gids=TARGET) == set(TARGET)


# ── NARROW 7 — 그래프 계약을 직접 잠근다 ──────────────────────────

def test_manifest_declares_producer_dependency():
    """소비자가 CP 를 직접 읽는데 그래프상 남남이면, producer 의 partial·
    재실행이 seed 를 막지도 무효화하지도 못한다."""
    from app.core.step_manifest import STEP_MANIFEST

    seed = STEP_MANIFEST["outdoor_structure_seed"]
    assert "outdoor_structure_form_reference" in seed["depends_on"]


def test_producer_does_not_allow_partial_downstream():
    """부분 산출로 소비를 열어 두면 claim 후 늦게 실패한다."""
    from app.core.step_manifest import STEP_MANIFEST

    producer = STEP_MANIFEST["outdoor_structure_form_reference"]
    assert producer.get("allow_partial_downstream") is not True


def test_consumer_contract_version_is_in_config_hash():
    """소비 계약이 바뀌면 완료된 CP 가 stale 되어야 한다.

    ★이것이 없으면 이 wave 의 수정이 실행에 도달하지 않는다 — 실측으로
    관할절을 통째로 바꿔도 config_hash 가 동일했다.
    """
    from unittest.mock import patch

    from app.core.steps.outdoor_structure_seed_step import (
        OutdoorStructureSeedStep,
    )

    step = OutdoorStructureSeedStep.__new__(OutdoorStructureSeedStep)
    step.project_id, step.episode_id, step.project_config = "P", "E", {}
    before = step._config_hash()
    with patch("app.modules.pipeline.search_grounded_ref"
               ".build_form_only_clause", return_value="DIFFERENT CONTRACT"):
        after = step._config_hash()
    assert before != after


# ── Codex 2차 재리뷰 BLOCKING-1 ───────────────────────────────────────
# 지금까지 소비자는 **producer 가 남긴 분할을 그대로 믿었다.** 그래서 구 CP
# 의 `no_spec` 에 든 그룹은, 그 뒤 spec 이 생겨 지금은 참조를 만들 수 있는
# 상태여도 "근거 없음"으로 남아 mandatory 에서 빠지고 → `_resolve_form_ref`
# 가 비필수로 보고 참조 없이 통과시켜 → 조용히 순수 T2I 로 내려갔다.
# Codex 직접 재현: 현재 spec 정상인 그룹을 구 CP no_spec 에 넣으니
# completed_count=1, labeled_refs=[] 로 씨드가 생성됐다.
#
# 새 계약: 분할은 producer·consumer 가 **같은 결정론 helper** 로 계산하고,
# 소비 직전에 **현재 spec 기준으로 다시 계산해 exact 비교**한다.

def test_stale_no_spec_with_current_valid_spec_is_rejected():
    """★핵심 재현 — 구 CP 가 "근거 없음"이라 한 그룹에 지금은 spec 이 있다."""
    cp = _cp(mandatory=["g_alpha"], no_spec=["g_beta"])
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET,
                 spec_groups=_specs(with_items=TARGET))
    assert "g_beta" in str(exc.value.message)


def test_stale_mandatory_with_current_missing_spec_is_rejected():
    """반대 방향 — 구 CP 는 필수로 봤는데 지금 spec 이 사라졌다."""
    cp = _cp(mandatory=TARGET)
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET,
                 spec_groups=_specs(with_items=["g_alpha"],
                                    without_items=["g_beta"]))
    assert "g_beta" in str(exc.value.message)


def test_partition_matching_current_spec_is_accepted():
    """드리프트가 없으면 그대로 통과한다 — 게이트가 과엄격하지 않다."""
    got = _resolve(formref_cp=_cp(mandatory=["g_alpha"], no_spec=["g_beta"]),
                   target_gids=TARGET,
                   spec_groups=_specs(with_items=["g_alpha"],
                                      without_items=["g_beta"]))
    assert got == {"g_alpha"}


def test_universe_key_absence_is_rejected():
    """★universe 키가 없으면 복원해 통과시키던 경로 — default-deny 로 바꾼다."""
    cp = _cp(mandatory=TARGET)
    cp["data"]["target"].pop("universe_group_ids")
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET)
    assert "universe_group_ids" in str(exc.value.message)


def test_extra_group_entry_is_rejected():
    """★`mandatory <= ok_groups` 는 잉여 stale entry 를 허용했다."""
    cp = _cp(mandatory=TARGET)
    cp["data"]["groups"]["g_stale"] = {"status": "ok"}
    with pytest.raises(AppError) as exc:
        _resolve(formref_cp=cp, target_gids=TARGET)
    assert "g_stale" in str(exc.value.message)


def test_non_dict_group_entry_is_rejected():
    """entry 가 dict 가 아니면 status 판정 자체가 성립하지 않는다."""
    with pytest.raises(AppError):
        _resolve(formref_cp=_cp(mandatory=TARGET,
                                groups={"g_alpha": {"status": "ok"},
                                        "g_beta": "ok"}),
                 target_gids=TARGET)


def test_spec_groups_is_required():
    """소비자가 현재 SOT 를 안 보면 계약이 다시 producer 신뢰로 돌아간다."""
    with pytest.raises(TypeError):
        resolve_form_reference_contract(
            formref_cp=_cp(mandatory=TARGET), target_gids=TARGET)


# ── 분할 helper 자체 = producer·consumer 의 단일 소유자 ────────────────

def test_partition_helper_splits_on_spec_items():
    from app.core.steps.outdoor_structure_seed_step import (
        partition_form_reference_targets,
    )

    mandatory, no_spec = partition_form_reference_targets(
        universe=["g_beta", "g_alpha", "g_gamma"],
        spec_groups=_specs(with_items=["g_alpha", "g_gamma"],
                           without_items=["g_beta"]))
    assert (mandatory, no_spec) == (["g_alpha", "g_gamma"], ["g_beta"])


def test_partition_helper_treats_absent_spec_as_no_spec():
    from app.core.steps.outdoor_structure_seed_step import (
        partition_form_reference_targets,
    )

    assert partition_form_reference_targets(
        universe=["g_alpha"], spec_groups={}) == ([], ["g_alpha"])


def test_producer_uses_the_shared_partition_helper():
    """producer 가 제 계산을 따로 가지면 두 분할이 다시 갈라진다."""
    import inspect

    from app.core.steps import outdoor_structure_form_reference_step as mod

    src = inspect.getsource(mod)
    assert "partition_form_reference_targets" in src


def test_contract_version_moves_the_config_hash():
    """소비 계약 버전이 hash 축에 실려 있어야 완료 CP 가 stale 된다.

    ★지난 wave 의 실측 교훈 — 관할절 배선을 고쳤는데 config_hash 가 그대로라
    완료 CP 가 재사용됐다. 즉 "고쳤다"가 실행에 도달하지 않았다.
    """
    from unittest.mock import patch

    from app.core.steps import outdoor_structure_seed_step as mod

    step = mod.OutdoorStructureSeedStep.__new__(mod.OutdoorStructureSeedStep)
    step.project_id, step.episode_id, step.project_config = "P", "E", {}
    before = step._config_hash()
    with patch.object(mod, "FORM_REFERENCE_CONSUMER_CONTRACT_VERSION", "999"):
        after = step._config_hash()
    assert before != after