"""IMP-33 u8 — Step 12 AI repair wiring tests. Covers the two structural gates layered on top of the u7 router: * IMP-30 provisional gate (only provisional units may invoke AI repair) * Reject gate (route_hint=design_reference_only NEVER calls AI) Plus the record-shape contract returned for downstream Step 12 artifacts. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any from unittest.mock import MagicMock from src.phase_z2_ai_fallback import step12 as step12_mod from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind @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) def _get_contract(_tid: str) -> dict[str, Any]: return {"frame_id": "fid", "payload": {"builder_options": {}}, "sub_zones": []} def _frame_visual(_tid: str) -> str: return "" def _call( units: list[FakeUnit], *, route_ai_fallback: Any | None = None, **overrides: Any, ) -> list[dict]: if route_ai_fallback is not None: step12_mod.route_ai_fallback = route_ai_fallback # type: ignore[assignment] kwargs: dict[str, Any] = dict( route_for_label=_route_for_label, get_contract_fn=_get_contract, frame_visual_loader=_frame_visual, ) kwargs.update(overrides) return step12_mod.gather_step12_ai_repair_proposals(units, **kwargs) def test_non_provisional_unit_is_skipped_without_ai_call(monkeypatch): router = MagicMock() monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [FakeUnit(label="restructure", provisional=False)] records = _call(units) assert records[0]["ai_called"] is False assert records[0]["skip_reason"] == "not_provisional" assert records[0]["provisional"] is False router.assert_not_called() def test_reject_route_is_skipped_without_ai_call(monkeypatch): router = MagicMock() monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [FakeUnit(label="reject", provisional=True)] records = _call(units) assert records[0]["ai_called"] is False assert records[0]["skip_reason"] == "design_reference_only_no_ai" assert records[0]["route_hint"] == "design_reference_only" router.assert_not_called() def test_non_ai_route_is_skipped_with_reason(monkeypatch): router = MagicMock() monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [FakeUnit(label="light_edit", provisional=True)] records = _call(units) assert records[0]["ai_called"] is False assert records[0]["skip_reason"] == ( "route_not_ai_adaptation:deterministic_minor_adjustment" ) router.assert_not_called() def test_router_short_circuit_returns_none_skip_reason(monkeypatch): router = MagicMock(return_value=None) monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [FakeUnit(label="restructure", provisional=True)] records = _call(units) assert records[0]["ai_called"] is False assert records[0]["skip_reason"] == "router_short_circuit" assert records[0]["proposal"] is None router.assert_called_once() def test_ai_adaptation_call_records_proposal(monkeypatch): proposal = AiFallbackProposal( proposal_kind=ProposalKind.PARTIAL_OVERRIDES, payload={"slots": {"s_text": "x"}}, rationale="r", ) router = MagicMock(return_value=proposal) monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [FakeUnit(label="restructure", provisional=True)] records = _call(units) rec = records[0] assert rec["ai_called"] is True assert rec["skip_reason"] is None assert rec["proposal"]["proposal_kind"] == "partial_overrides" router.assert_called_once() kwargs = router.call_args.kwargs assert kwargs["v4_result"]["route"] == "ai_adaptation_required" assert kwargs["v4_result"]["label"] == "restructure" def test_router_exception_is_captured_per_record(monkeypatch): router = MagicMock(side_effect=RuntimeError("transient_boom")) monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [FakeUnit(label="restructure", provisional=True)] records = _call(units) rec = records[0] assert rec["ai_called"] is True assert rec["proposal"] is None assert rec["error"] == "RuntimeError: transient_boom" router.assert_called_once() def test_mixed_units_each_independently_classified(monkeypatch): router = MagicMock(return_value=None) monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [ FakeUnit(label="use_as_is", provisional=False), FakeUnit(label="reject", provisional=True), FakeUnit(label="restructure", provisional=True), FakeUnit(label="restructure", provisional=False), ] records = _call(units) assert [r["skip_reason"] for r in records] == [ "not_provisional", "design_reference_only_no_ai", "router_short_circuit", "not_provisional", ] assert router.call_count == 1 def test_cache_key_includes_template_and_section_ids(monkeypatch): router = MagicMock(return_value=None) monkeypatch.setattr(step12_mod, "route_ai_fallback", router) units = [ FakeUnit( label="restructure", provisional=True, frame_template_id="tmpl_abc", source_section_ids=["02-1", "02-2"], ) ] _call(units) assert router.call_args.kwargs["cache_key"] == "tmpl_abc::02-1,02-2" def test_record_shape_contract_is_stable(monkeypatch): monkeypatch.setattr(step12_mod, "route_ai_fallback", MagicMock(return_value=None)) units = [FakeUnit(label="reject", provisional=True)] rec = _call(units)[0] assert set(rec.keys()) == { "unit_index", "source_section_ids", "frame_template_id", "label", "route_hint", "provisional", "ai_called", "skip_reason", "proposal", "error", }