C.E.L_Slide_test2/tests/phase_z2_ai_fallback/test_step17.py

"""IMP-33 u9 — Step 17 AI repair wiring tests (BLOCKED until IMP-34 + IMP-35).

Covers:
  * :data:`OVERFLOW_CASCADE_ORDER` canonical order (4 stages).
  * :class:`OverflowCascadeStage` member values.
  * :data:`STEP17_AI_REPAIR_BLOCKED_REASON` constant value.
  * :func:`gather_step17_ai_repair_proposals` BLOCKED contract — every unit
    returns ``ai_called=False`` + ``skip_reason=STEP17_AI_REPAIR_BLOCKED_REASON``
    + ``proposal=None`` regardless of provisional / label / route_hint.
  * Structural guarantee — the u9 module does NOT import
    :func:`src.phase_z2_ai_fallback.router.route_ai_fallback` or the
    ``anthropic`` SDK. Step 17 AI repair stays structurally blocked.
"""
from __future__ import annotations

import ast
from dataclasses import dataclass, field
from pathlib import Path

from src.phase_z2_ai_fallback import step17 as step17_mod
from src.phase_z2_ai_fallback.step17 import (
    OVERFLOW_CASCADE_ORDER,
    STEP17_AI_REPAIR_BLOCKED_REASON,
    STEP17_POPUP_SPLIT_DECISION_API_GATED_REASON,
    OverflowCascadeStage,
    gather_step17_ai_repair_proposals,
    gather_step17_popup_split_decisions,
)


@dataclass
class FakeUnit:
    label: str | None
    provisional: bool
    frame_template_id: str = "tmpl"
    frame_id: str = "fid"
    source_section_ids: list[str] = field(default_factory=lambda: ["s1"])
    raw_content: str = "raw"
    v4_rank: int | None = 1


_ROUTE_HINTS: dict[str | None, str | None] = {
    "use_as_is": "direct_render",
    "light_edit": "deterministic_minor_adjustment",
    "restructure": "ai_adaptation_required",
    "reject": "design_reference_only",
    None: None,
}


def _route_for_label(label: str | None) -> str | None:
    return _ROUTE_HINTS.get(label)


# ─── Stage / order constants ─────────────────────────────────────────


def test_overflow_cascade_order_is_canonical():
    assert OVERFLOW_CASCADE_ORDER == (
        OverflowCascadeStage.DETERMINISTIC,
        OverflowCascadeStage.POPUP,
        OverflowCascadeStage.AI_REPAIR,
        OverflowCascadeStage.USER_OVERRIDE,
    )


def test_overflow_cascade_stage_string_values():
    assert OverflowCascadeStage.DETERMINISTIC.value == "deterministic"
    assert OverflowCascadeStage.POPUP.value == "popup"
    assert OverflowCascadeStage.AI_REPAIR.value == "ai_repair"
    assert OverflowCascadeStage.USER_OVERRIDE.value == "user_override"


def test_step17_blocked_reason_constant_value():
    assert (
        STEP17_AI_REPAIR_BLOCKED_REASON
        == "step17_ai_blocked_imp_34_35_prerequisites_missing"
    )


# ─── BLOCKED contract: every unit returns blocked record ─────────────


def test_gather_returns_one_record_per_unit():
    units = [
        FakeUnit(label="restructure", provisional=True),
        FakeUnit(label="reject", provisional=False),
        FakeUnit(label="use_as_is", provisional=True),
    ]
    records = gather_step17_ai_repair_proposals(units, route_for_label=_route_for_label)
    assert len(records) == 3


def test_gather_records_blocked_skip_reason():
    """Every record must carry the IMP-34/IMP-35 prerequisite block reason."""
    units = [FakeUnit(label="restructure", provisional=True)]
    records = gather_step17_ai_repair_proposals(units, route_for_label=_route_for_label)
    assert records[0]["skip_reason"] == STEP17_AI_REPAIR_BLOCKED_REASON


def test_gather_blocks_even_when_route_is_ai_adaptation_required():
    """Provisional + ai_adaptation_required must NOT bypass the u9 block.

    Stage 2 contract: AI repair at Step 17 is blocked behind IMP-34 + IMP-35
    regardless of V4 route hint. Only u8 (Step 12) is allowed to invoke AI today.
    """
    units = [FakeUnit(label="restructure", provisional=True)]
    record = gather_step17_ai_repair_proposals(
        units, route_for_label=_route_for_label
    )[0]
    assert record["route_hint"] == "ai_adaptation_required"
    assert record["ai_called"] is False
    assert record["proposal"] is None
    assert record["skip_reason"] == STEP17_AI_REPAIR_BLOCKED_REASON


def test_gather_blocks_reject_units_too():
    """Reject units (design_reference_only) are also blocked at u9 — same reason."""
    units = [FakeUnit(label="reject", provisional=False)]
    record = gather_step17_ai_repair_proposals(
        units, route_for_label=_route_for_label
    )[0]
    assert record["ai_called"] is False
    assert record["skip_reason"] == STEP17_AI_REPAIR_BLOCKED_REASON


def test_gather_records_proposal_none_and_no_error():
    units = [FakeUnit(label="restructure", provisional=True)]
    record = gather_step17_ai_repair_proposals(
        units, route_for_label=_route_for_label
    )[0]
    assert record["proposal"] is None
    assert record["error"] is None


def test_gather_records_cascade_stage_is_ai_repair():
    units = [FakeUnit(label="restructure", provisional=True)]
    record = gather_step17_ai_repair_proposals(
        units, route_for_label=_route_for_label
    )[0]
    assert record["cascade_stage"] == OverflowCascadeStage.AI_REPAIR.value


def test_gather_preserves_unit_metadata():
    units = [
        FakeUnit(
            label="restructure",
            provisional=True,
            frame_template_id="frame_05_overview",
            source_section_ids=["s1", "s2"],
        )
    ]
    record = gather_step17_ai_repair_proposals(
        units, route_for_label=_route_for_label
    )[0]
    assert record["unit_index"] == 0
    assert record["frame_template_id"] == "frame_05_overview"
    assert record["source_section_ids"] == ["s1", "s2"]
    assert record["label"] == "restructure"
    assert record["provisional"] is True


def test_gather_with_empty_units_returns_empty_list():
    records = gather_step17_ai_repair_proposals([], route_for_label=_route_for_label)
    assert records == []


# ─── IMP-35 u4: POPUP cascade AI split-decision contract (API gated) ─────


def test_popup_split_decision_api_gated_reason_constant_value():
    """u4 binding contract — API-gated skip_reason is a stable, machine-readable
    constant that downstream consumers can distinguish from the AI_REPAIR
    block reason. Never collide with STEP17_AI_REPAIR_BLOCKED_REASON."""
    assert (
        STEP17_POPUP_SPLIT_DECISION_API_GATED_REASON
        == "step17_popup_split_decision_api_gated"
    )
    assert (
        STEP17_POPUP_SPLIT_DECISION_API_GATED_REASON
        != STEP17_AI_REPAIR_BLOCKED_REASON
    )


def test_popup_split_decision_returns_one_record_per_unit():
    units = [
        FakeUnit(label="restructure", provisional=True),
        FakeUnit(label="reject", provisional=False),
        FakeUnit(label="use_as_is", provisional=True),
    ]
    records = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )
    assert len(records) == 3


def test_popup_split_decision_cascade_stage_is_popup():
    """u4 — cascade_stage must mark these records as the POPUP stage, NOT
    AI_REPAIR. This lets consumers multiplex POPUP and AI_REPAIR records on
    the same retry trace without ambiguity."""
    units = [FakeUnit(label="restructure", provisional=True)]
    record = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )[0]
    assert record["cascade_stage"] == OverflowCascadeStage.POPUP.value
    assert record["cascade_stage"] != OverflowCascadeStage.AI_REPAIR.value


def test_popup_split_decision_api_gated_flag_true():
    """u4 — api_gated=True everywhere. The flag is the primary state signal
    consumers read to decide whether the AI hook is active."""
    units = [FakeUnit(label="restructure", provisional=True)]
    record = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )[0]
    assert record["api_gated"] is True


def test_popup_split_decision_ai_called_is_false_and_no_proposal():
    """u4 — ai_called=False, split_decision=None, error=None. The hook is the
    contract surface only; the Anthropic API is NOT invoked at u4."""
    units = [FakeUnit(label="restructure", provisional=True)]
    record = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )[0]
    assert record["ai_called"] is False
    assert record["split_decision"] is None
    assert record["error"] is None


def test_popup_split_decision_skip_reason_is_api_gated():
    """u4 — every record carries the API-gated skip_reason regardless of
    label / provisional / route_hint."""
    units = [
        FakeUnit(label="restructure", provisional=True),
        FakeUnit(label="reject", provisional=False),
        FakeUnit(label="use_as_is", provisional=True),
        FakeUnit(label=None, provisional=False),
    ]
    records = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )
    for record in records:
        assert (
            record["skip_reason"]
            == STEP17_POPUP_SPLIT_DECISION_API_GATED_REASON
        )


def test_popup_split_decision_honors_route_for_label():
    """u4 — route_for_label callable is applied per unit. Verifies the hook
    surface accepts the same label→route mapping as the AI_REPAIR path."""
    units = [
        FakeUnit(label="restructure", provisional=True),
        FakeUnit(label="reject", provisional=False),
        FakeUnit(label="use_as_is", provisional=True),
        FakeUnit(label="light_edit", provisional=False),
        FakeUnit(label=None, provisional=False),
    ]
    records = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )
    assert [r["route_hint"] for r in records] == [
        "ai_adaptation_required",
        "design_reference_only",
        "direct_render",
        "deterministic_minor_adjustment",
        None,
    ]


def test_popup_split_decision_preserves_unit_metadata():
    """u4 — schema mirrors gather_step17_ai_repair_proposals (unit_index,
    source_section_ids, frame_template_id, label, provisional)."""
    units = [
        FakeUnit(
            label="restructure",
            provisional=True,
            frame_template_id="frame_05_overview",
            source_section_ids=["s1", "s2"],
        )
    ]
    record = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )[0]
    assert record["unit_index"] == 0
    assert record["frame_template_id"] == "frame_05_overview"
    assert record["source_section_ids"] == ["s1", "s2"]
    assert record["label"] == "restructure"
    assert record["provisional"] is True


def test_popup_split_decision_with_empty_units_returns_empty_list():
    records = gather_step17_popup_split_decisions(
        [], route_for_label=_route_for_label
    )
    assert records == []


def test_popup_split_decision_record_schema_disjoint_from_ai_repair_extras():
    """u4 — POPUP record must carry api_gated + split_decision keys; the
    AI_REPAIR record carries proposal (not split_decision). This lock keeps
    the two contract surfaces machine-distinguishable on the retry trace."""
    units = [FakeUnit(label="restructure", provisional=True)]
    popup_rec = gather_step17_popup_split_decisions(
        units, route_for_label=_route_for_label
    )[0]
    ai_repair_rec = gather_step17_ai_repair_proposals(
        units, route_for_label=_route_for_label
    )[0]
    # POPUP-specific keys
    assert "api_gated" in popup_rec
    assert "split_decision" in popup_rec
    # AI_REPAIR-specific key
    assert "proposal" in ai_repair_rec
    # Disjoint payload keys (the two contracts must NOT cross-leak):
    assert "proposal" not in popup_rec
    assert "split_decision" not in ai_repair_rec
    assert "api_gated" not in ai_repair_rec


# ─── Structural guarantee: u9 must NOT import route_ai_fallback / anthropic ─


def _u9_imports() -> list[str]:
    src_path = Path(step17_mod.__file__)
    tree = ast.parse(src_path.read_text(encoding="utf-8"))
    imports: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            imports.extend(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            module = node.module or ""
            for alias in node.names:
                imports.append(f"{module}.{alias.name}")
    return imports


def test_step17_module_does_not_import_route_ai_fallback():
    """u9 must not be able to reach the u7 router — structural block."""
    imports = _u9_imports()
    forbidden = {
        "src.phase_z2_ai_fallback.router.route_ai_fallback",
        "src.phase_z2_ai_fallback.router",
    }
    assert not any(imp in forbidden for imp in imports), imports
    assert not hasattr(step17_mod, "route_ai_fallback")


def test_step17_module_does_not_import_anthropic():
    """u9 must not reach the Anthropic SDK directly — AI=0 in this layer."""
    imports = _u9_imports()
    leaked = [imp for imp in imports if imp.split(".", 1)[0] == "anthropic"]
    assert leaked == [], leaked


def test_step17_module_does_not_import_ai_fallback_client():
    """u9 must not instantiate the u4 client either."""
    imports = _u9_imports()
    forbidden_prefixes = ("src.phase_z2_ai_fallback.client",)
    leaked = [
        imp for imp in imports if imp.startswith(forbidden_prefixes)
    ]
    assert leaked == [], leaked