"""W21B-w5 — bg_space_partition pure module tests.

Locks the anchor-centered constrained clustering against the REAL pass-2 edge
judgements observed on production cp (manifest.json, 05-30) plus synthetic
guards (transitive trap, adjacency, hub non-bridge). Deterministic only — the
LLM/VLM edge judgements are fixtures here, not re-derived (TDD valid domain).
"""
from app.modules.pipeline.bg_space_partition import (
    build_candidate_edges,
    build_space_partition_plan,
    classify_edge,
    STRONG_CONF_THRESHOLD,
)


def _marker(number, decision="base_structural_unit", label="unit"):
    return {"number": number, "base_layer_decision": decision, "label": label}


def _facts(target_units):
    return {"target_unit_marker_numbers": list(target_units)}


# ── real L05 edge judgements (current cp), compact fixture ──────────────
L05_EDGES = [
    ("L05B02", "L05B03", "adjacent_related", 0.78, False),
    ("L05B02", "L05B05", "same_space", 0.86, True),
    ("L05B02", "L05B06", "adjacent_related", 0.78, False),
    ("L05B02", "L05B09", "adjacent_related", 0.74, False),
    ("L05B02", "L05B10", "different_space", 0.91, False),
    ("L05B03", "L05B05", "adjacent_related", 0.82, False),
    ("L05B03", "L05B06", "adjacent_related", 0.73, False),
    ("L05B03", "L05B09", "same_space", 0.88, True),
    ("L05B03", "L05B10", "different_space", 0.84, False),
    ("L05B05", "L05B06", "adjacent_related", 0.82, False),
    ("L05B05", "L05B09", "adjacent_related", 0.78, False),
    ("L05B05", "L05B10", "different_space", 0.90, False),
    ("L05B06", "L05B09", "adjacent_related", 0.78, False),
    ("L05B06", "L05B10", "different_space", 0.88, False),
    ("L05B09", "L05B10", "adjacent_related", 0.82, False),
]
L05_BGS = ["L05B02", "L05B03", "L05B05", "L05B06", "L05B09", "L05B10"]
L05_HUBS = frozenset({"L05B03", "L05B05", "L05B10"})  # >=5 target structural units

L20_EDGES = [
    ("L20B04", "L20B05", "same_space", 0.90, True),
    ("L20B04", "L20B06", "same_space", 0.90, True),
    ("L20B05", "L20B06", "same_space", 0.94, True),
]
L20_BGS = ["L20B04", "L20B05", "L20B06"]


def _as_dicts(tuples):
    return [{"bg_a": a, "bg_b": b, "edge_state": s, "confidence": c,
             "strong_parent_allowed": sp} for (a, b, s, c, sp) in tuples]


def _groups_as_sets(plan):
    return {frozenset([g["anchor_bg_id"], *g["member_bg_ids"]]) for g in plan["plate_groups"]}


# ── classify_edge ──────────────────────────────────────────────────────
def test_classify_edge_strong_requires_all_three():
    assert classify_edge({"edge_state": "same_space", "confidence": 0.88,
                          "strong_parent_allowed": True}) == "strong"
    # same_space but below conf → not strong
    assert classify_edge({"edge_state": "same_space", "confidence": 0.70,
                          "strong_parent_allowed": True}) == "weak"
    # strong_parent_allowed False → not strong
    assert classify_edge({"edge_state": "same_space", "confidence": 0.90,
                          "strong_parent_allowed": False}) == "weak"
    assert classify_edge({"edge_state": "different_space", "confidence": 0.9,
                          "strong_parent_allowed": False}) == "different"
    assert classify_edge({"edge_state": "adjacent_related", "confidence": 0.8,
                          "strong_parent_allowed": False}) == "weak"


# ── L05 exact partition lock ───────────────────────────────────────────
def test_l05_partition_matches_visual_review():
    plan = build_space_partition_plan(
        bg_ids=L05_BGS, edge_judgements=_as_dicts(L05_EDGES), hub_bg_ids=L05_HUBS)
    assert _groups_as_sets(plan) == {
        frozenset({"L05B03", "L05B09"}),   # living/kitchen
        frozenset({"L05B02", "L05B05"}),   # small bedroom
        frozenset({"L05B06"}),             # transitional, own plate
        frozenset({"L05B10"}),             # bathroom, own plate
    }


def test_l05_bathroom_is_independent_no_parent():
    plan = build_space_partition_plan(
        bg_ids=L05_BGS, edge_judgements=_as_dicts(L05_EDGES), hub_bg_ids=L05_HUBS)
    # bathroom must be its own render_new_plate with no spatial parent.
    assert plan["render_actions"]["L05B10"]["render_action"] == "render_new_plate"
    assert plan["ref_tree_parents"]["L05B10"] == []
    # and it must never share a group with a living/bedroom bg.
    grp = plan["node_assignments"]["L05B10"]
    members = [bg for bg, g in plan["node_assignments"].items() if g == grp]
    assert members == ["L05B10"]


def test_l05_hub_does_not_anchor_over_nonhub():
    # B03 is a hub; the {B03,B09} group should anchor on the NON-hub B09.
    plan = build_space_partition_plan(
        bg_ids=L05_BGS, edge_judgements=_as_dicts(L05_EDGES), hub_bg_ids=L05_HUBS)
    living = next(g for g in plan["plate_groups"]
                  if {g["anchor_bg_id"], *g["member_bg_ids"]} == {"L05B03", "L05B09"})
    assert living["anchor_bg_id"] == "L05B09"
    assert living["member_bg_ids"] == ["L05B03"]
    assert plan["render_actions"]["L05B03"]["render_action"] == "reuse_existing_plate"
    assert plan["render_actions"]["L05B03"]["reuse_target_bg_id"] == "L05B09"
    assert plan["ref_tree_parents"]["L05B03"] == ["L05B09"]


# ── L20 all-one-group (small fp, all same_space) ───────────────────────
def test_l20_all_one_group():
    plan = build_space_partition_plan(
        bg_ids=L20_BGS, edge_judgements=_as_dicts(L20_EDGES), hub_bg_ids=frozenset())
    assert _groups_as_sets(plan) == {frozenset({"L20B04", "L20B05", "L20B06"})}
    # exactly one anchor renders new; the other two reuse it.
    actions = {bg: plan["render_actions"][bg]["render_action"] for bg in L20_BGS}
    assert list(actions.values()).count("render_new_plate") == 1
    assert list(actions.values()).count("reuse_existing_plate") == 2


# ── synthetic transitive trap (Codex-required) ────────────────────────
def test_transitive_trap_a_c_different_prevents_single_group():
    # A~B strong, B~C strong, but A~C different_space → must NOT be one group.
    edges = _as_dicts([
        ("A", "B", "same_space", 0.90, True),
        ("B", "C", "same_space", 0.90, True),
        ("A", "C", "different_space", 0.95, False),
    ])
    plan = build_space_partition_plan(bg_ids=["A", "B", "C"], edge_judgements=edges,
                                      hub_bg_ids=frozenset())
    groups = _groups_as_sets(plan)
    assert frozenset({"A", "B", "C"}) not in groups
    # A and C must be in different groups (the different_space pair is honored).
    assert plan["node_assignments"]["A"] != plan["node_assignments"]["C"]


def test_adjacent_only_never_groups():
    edges = _as_dicts([("X", "Y", "adjacent_related", 0.85, False)])
    plan = build_space_partition_plan(bg_ids=["X", "Y"], edge_judgements=edges,
                                      hub_bg_ids=frozenset())
    assert _groups_as_sets(plan) == {frozenset({"X"}), frozenset({"Y"})}
    # adjacency may surface as weak cross-group context, never a strong parent.
    assert plan["ref_tree_parents"]["X"] == []
    assert plan["ref_tree_parents"]["Y"] == []


def test_max_refs_cap_respected():
    # a member never gets more than max_refs strong parents.
    edges = _as_dicts([
        ("P", "Q", "same_space", 0.9, True),
        ("P", "R", "same_space", 0.9, True),
        ("P", "S", "same_space", 0.9, True),
        ("Q", "R", "same_space", 0.9, True),
        ("Q", "S", "same_space", 0.9, True),
        ("R", "S", "same_space", 0.9, True),
    ])
    plan = build_space_partition_plan(bg_ids=["P", "Q", "R", "S"], edge_judgements=edges,
                                      hub_bg_ids=frozenset(), max_refs=2)
    for bg, parents in plan["ref_tree_parents"].items():
        assert len(parents) <= 2


def test_threshold_constant_is_conservative():
    assert STRONG_CONF_THRESHOLD == 0.75


# ── candidate edge generator (deterministic upstream floor) ────────────

def test_candidate_edges_only_use_structural_units_and_ignore_labels():
    # non-structural markers and label text must be ignored entirely.
    inv = [
        _marker(1, label="anything"),
        _marker(2, label="other"),
        _marker(99, decision="base_persistent_furniture", label="couch"),
    ]
    facts = {
        "B1": _facts([1, 2, 99]),  # 99 is furniture, dropped
        "B2": _facts([1, 2]),
    }
    out = build_candidate_edges(
        base_marker_inventory=inv, per_bg_render_facts_by_bg_id=facts
    )
    assert out["structural_units"] == [1, 2]
    assert out["signatures"]["B1"] == [1, 2]  # 99 excluded
    assert out["signatures"]["B2"] == [1, 2]


def test_candidate_edges_drop_hard_negative_no_shared_unit():
    inv = [_marker(1), _marker(2), _marker(3), _marker(4)]
    # two bgs that share nothing => hard negative => not a candidate.
    facts = {"B1": _facts([1, 2]), "B2": _facts([3, 4])}
    out = build_candidate_edges(
        base_marker_inventory=inv, per_bg_render_facts_by_bg_id=facts
    )
    assert out["candidate_edges"] == []


def test_candidate_edges_obvious_positive_is_same_space():
    inv = [_marker(1), _marker(2), _marker(3)]
    # B1,B2 identical distinctive floor (unit 2,3 not ubiquitous since other bg
    # exists targeting only 1) => obvious positive.
    facts = {
        "B1": _facts([2, 3]),
        "B2": _facts([2, 3]),
        "B3": _facts([1]),
    }
    out = build_candidate_edges(
        base_marker_inventory=inv, per_bg_render_facts_by_bg_id=facts
    )
    edge = next(e for e in out["candidate_edges"] if e["bg_a"] == "B1" and e["bg_b"] == "B2")
    assert edge["state"] == "same_space"
    assert edge["distinctive_units"] == [2, 3]


def test_candidate_edges_shared_only_ubiquitous_is_uncertain():
    inv = [_marker(1), _marker(2), _marker(3), _marker(4)]
    # unit 1 ubiquitous (all 3 bgs); B1,B2 share only unit 1 => uncertain.
    facts = {
        "B1": _facts([1, 2]),
        "B2": _facts([1, 3]),
        "B3": _facts([1, 4]),
    }
    out = build_candidate_edges(
        base_marker_inventory=inv, per_bg_render_facts_by_bg_id=facts
    )
    assert 1 in out["ubiquitous_units"]
    edge = next(e for e in out["candidate_edges"] if {e["bg_a"], e["bg_b"]} == {"B1", "B2"})
    assert edge["state"] == "uncertain"
    assert edge["distinctive_units"] == []  # only ubiquitous shared


def test_candidate_edges_hub_node_never_obvious_positive():
    inv = [_marker(i) for i in range(1, 7)]
    # B_hub targets 5 units => hub. Any edge it touches is hub_uncertain.
    facts = {
        "Bhub": _facts([1, 2, 3, 4, 5]),
        "B2": _facts([2, 3]),
    }
    out = build_candidate_edges(
        base_marker_inventory=inv, per_bg_render_facts_by_bg_id=facts
    )
    assert "Bhub" in out["hub_bg_ids"]
    edge = out["candidate_edges"][0]
    assert edge["state"] == "hub_uncertain"


def test_candidate_edges_feed_partition_with_recall():
    # candidate generator output must, after judging, be consumable by the
    # partition assembler — over-generation OK, every shared-signal pair present.
    inv = [_marker(1), _marker(2), _marker(3)]
    facts = {"A": _facts([1, 2]), "B": _facts([1, 2]), "C": _facts([3])}
    out = build_candidate_edges(
        base_marker_inventory=inv, per_bg_render_facts_by_bg_id=facts
    )
    pairs = {frozenset((e["bg_a"], e["bg_b"])) for e in out["candidate_edges"]}
    # A~B share structural floor -> candidate; A~C / B~C share nothing -> dropped
    assert frozenset(("A", "B")) in pairs
    assert frozenset(("A", "C")) not in pairs


# ── pass-2 edge judge: R3 precondition / prompt / R2 validation ────────

def test_card_is_judgeable_requires_pass_and_visible_items():
    from app.modules.pipeline.bg_space_partition import card_is_judgeable
    assert card_is_judgeable({"card_state": "pass", "visible_items": ["sink"]})
    assert not card_is_judgeable({"card_state": "needs_review", "visible_items": ["x"]})
    assert not card_is_judgeable({"card_state": "pass", "visible_items": ["x"], "synthetic": True})
    assert not card_is_judgeable({"card_state": "pass", "visible_items": []})
    assert not card_is_judgeable(None)


def test_build_edge_judge_prompt_carries_visible_items_and_hint():
    from app.modules.pipeline.bg_space_partition import build_edge_judge_prompt
    edge = {"bg_a": "A", "bg_b": "B", "state": "uncertain",
            "shared_units": [1], "distinctive_units": []}
    bundle = build_edge_judge_prompt(
        edge=edge,
        card_a={"visible_items": ["a tiled wall"]},
        card_b={"visible_items": ["a sink"]},
    )
    assert set(bundle) == {"system", "user", "schema"}
    assert "tiled wall" in bundle["user"]
    assert "uncertain" in bundle["user"]
    assert bundle["schema"]["properties"]["edge_state"]["enum"][0] == "same_space"


def test_validate_edge_judge_output_strong_parent_gate():
    from app.modules.pipeline.bg_space_partition import validate_edge_judge_output
    edge = {"bg_a": "A", "bg_b": "B", "state": "uncertain"}
    good = validate_edge_judge_output(edge=edge, raw={
        "edge_state": "same_space", "confidence": 0.88,
        "evidence": "same window and couch", "shared_distinctive_features": ["bay window"]})
    assert good["strong_parent_allowed"] is True
    assert good["validation_ok"] is True
    no_feat = validate_edge_judge_output(edge=edge, raw={
        "edge_state": "same_space", "confidence": 0.9,
        "evidence": "looks similar", "shared_distinctive_features": []})
    assert no_feat["strong_parent_allowed"] is False
    low = validate_edge_judge_output(edge=edge, raw={
        "edge_state": "same_space", "confidence": 0.6,
        "evidence": "x", "shared_distinctive_features": ["window"]})
    assert low["strong_parent_allowed"] is False


def test_validate_edge_judge_output_rejects_garbage():
    from app.modules.pipeline.bg_space_partition import validate_edge_judge_output
    edge = {"bg_a": "A", "bg_b": "B", "state": "uncertain"}
    bad = validate_edge_judge_output(edge=edge, raw="not a dict")
    assert bad["edge_state"] == "uncertain"
    assert bad["strong_parent_allowed"] is False
    assert bad["validation_ok"] is False
    oov = validate_edge_judge_output(edge=edge, raw={
        "edge_state": "teleported", "confidence": 2.0,
        "evidence": "", "shared_distinctive_features": ["x"]})
    assert oov["edge_state"] == "uncertain"
    assert oov["confidence"] == 1.0


def test_validate_edge_judge_output_feeds_classify_edge():
    from app.modules.pipeline.bg_space_partition import (
        validate_edge_judge_output, classify_edge,
    )
    edge = {"bg_a": "A", "bg_b": "B", "state": "uncertain"}
    j = validate_edge_judge_output(edge=edge, raw={
        "edge_state": "same_space", "confidence": 0.9,
        "evidence": "shared bay window and same couch",
        "shared_distinctive_features": ["bay window", "green couch"]})
    assert classify_edge(j) == "strong"


def test_edge_judge_provider_failclose_without_api_key(monkeypatch):
    # provider must fail-closed (no network) when env/preflight is not satisfied.
    from app.modules.pipeline.bg_space_partition_provider import (
        litellm_edge_judge_provider, EdgeJudgeProviderError,
    )
    # [2026-08-01] 키 유무의 권위가 환경변수에서 슬롯 브로커로 옮겼다 —
    # 보조 슬롯만 있어도 '있음'이므로 env 만 지우면 fail-closed 가 아니다.
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    monkeypatch.setattr(
        "app.core.config.settings.openai_api_key", "", raising=False)
    monkeypatch.setattr(
        "app.core.config.settings.openai_api_key_secondary", "",
        raising=False)
    bundle = {"system": "s", "user": "u", "schema": {"type": "object"}}
    try:
        litellm_edge_judge_provider(prompt_bundle=bundle)
        assert False, "expected EdgeJudgeProviderError"
    except EdgeJudgeProviderError as e:
        assert "OPENAI_API_KEY" in str(e)