On main: stash-all-pre-IMP38-commit-20260521

index on main: dceb101 feat(#63 ): IMP-34 R1 donor capacity measured bound (u1+u2)
untracked files on main: dceb101 feat(#63 ): IMP-34 R1 donor capacity measured bound (u1+u2)
2026-05-21 22:07:41 +09:00 · 2026-05-21 22:07:41 +09:00 · 2026-05-21 22:07:41 +09:00
16 changed files with 0 additions and 2777 deletions
@@ -1,91 +0,0 @@
 """IMP-46 u1 — Frame transformation cache signature builder.
 Deterministic SHA256 over the 8 declared structural axes:
  frame_id, v4_label, cardinality, source_shape,
  h3_count, char_count_bucket, layout_preset, zone_position
 Guardrails:
  * No sample/section identifiers in the signature surface (no-hardcoding lock).
  * source_shape constrained to the bullet/paragraph/table/mixed enum.
  * char_count_bucket is the *bucket label*; numeric counts must be projected
    via :func:`bucket_char_count` before being fed to :func:`build_signature`.
  * Schema version is embedded in the hashed payload so a future axis change
    breaks the digest by design (cache invalidation on schema bump).
 """
 from __future__ import annotations
 import hashlib
 import json
 from enum import Enum
 SCHEMA_VERSION = 1
 class SourceShape(str, Enum):
    BULLET = "bullet"
    PARAGRAPH = "paragraph"
    TABLE = "table"
    MIXED = "mixed"
 _CHAR_COUNT_BUCKETS: tuple[tuple[int, str], ...] = (
    (50, "0-50"),
    (150, "51-150"),
    (400, "151-400"),
    (1000, "401-1000"),
 )
 _CHAR_COUNT_BUCKET_OVERFLOW = "1001+"
 CHAR_COUNT_BUCKET_LABELS: tuple[str, ...] = tuple(
    label for _, label in _CHAR_COUNT_BUCKETS
 ) + (_CHAR_COUNT_BUCKET_OVERFLOW,)
 def bucket_char_count(char_count: int) -> str:
    """Project a non-negative character count to its fixed bucket label."""
    if isinstance(char_count, bool) or not isinstance(char_count, int):
        raise TypeError("char_count must be a non-negative int")
    if char_count < 0:
        raise ValueError("char_count must be non-negative")
    for upper, label in _CHAR_COUNT_BUCKETS:
        if char_count <= upper:
            return label
    return _CHAR_COUNT_BUCKET_OVERFLOW
 def build_signature(
    *,
    frame_id: str,
    v4_label: str,
    cardinality: int | None,
    source_shape: SourceShape | str,
    h3_count: int,
    char_count_bucket: str,
    layout_preset: str,
    zone_position: str,
 ) -> str:
    """Return a deterministic SHA256 hex digest over the 8 declared axes."""
    if isinstance(source_shape, SourceShape):
        source_shape_value = source_shape.value
    elif isinstance(source_shape, str):
        source_shape_value = SourceShape(source_shape).value
    else:
        raise TypeError("source_shape must be SourceShape or str")
    if char_count_bucket not in CHAR_COUNT_BUCKET_LABELS:
        raise ValueError(
            f"char_count_bucket={char_count_bucket!r} is not a known bucket "
            f"label (expected one of {CHAR_COUNT_BUCKET_LABELS})"
        )
    payload = {
        "schema_version": SCHEMA_VERSION,
        "frame_id": frame_id,
        "v4_label": v4_label,
        "cardinality": cardinality,
        "source_shape": source_shape_value,
        "h3_count": h3_count,
        "char_count_bucket": char_count_bucket,
        "layout_preset": layout_preset,
        "zone_position": zone_position,
    }
    encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()
@@ -1,45 +0,0 @@
 # IMP-38 V4 max_rank 정책 — separate yaml (catalog 오염 방지)
 #
 # 도입 배경:
 #   기존 `lookup_v4_match_with_fallback(max_rank=3)` hardcoded → rank 4~32 의 등록 frame 도달 못함
 #   mdx05-2 같이 V4 rank 1~9 가 catalog 미등록 + rank 10~ 등록 case → chain_exhausted → unit 생성 X
 #
 # 4 round 합의 (IMP-38 #67):
 #   - Codex #1: frame_contracts.yaml 오염 회피 → 별 yaml 파일 (이 파일)
 #   - Codex #2: 3 변수 분리 (configured / judgments / catalog count)
 #   - Codex #3: effective_extended_ceiling = min(configured, len(judgments_full32))
 #
 # 적용 path: src/phase_z2_mapper.py 의 load_v4_fallback_policy() loader
 #           + src/phase_z2_pipeline.py 의 lookup_v4_match_with_fallback() 동적 max_rank logic
 policy_type: dynamic_usable_count_based
 # usable_threshold N:
 #   rank 1~default_max_rank 중 "usable" predicate 충족 frame 수 >= N → default_max_rank 유지
 #   < N → extended_max_rank 로 확장
 usable_threshold: 1
 # default_max_rank:
 #   normal case (usable_count >= threshold) 의 fallback chain 길이
 #   mdx03 같이 rank 1 use_as_is 매칭 잘 되는 case 보호
 default_max_rank: 3
 # extended_max_rank:
 #   usable_count < threshold case 의 확장 ceiling
 #   mdx05-2 같이 rank 1~9 미등록 case 처리
 #   ★ 실제 effective_extended_ceiling = min(extended_max_rank, len(judgments_full32))
 #     (Codex #2 정정: yaml ceiling 무력화 방지 + V4 schema 범위 초과 방지)
 extended_max_rank: 32
 # usable predicate (3-tier):
 #   (a) phase_z_status in MVP1_ALLOWED_STATUSES (matched_zone / adapt_matched_zone)
 #   (b) get_contract(template_id) is not None (catalog 등록)
 #   (c) capacity_fit ok (raw_content 제공 시만 — optional)
 # 의미 신뢰 vs catalog presence trade-off:
 #   N=1 = 가장 보수 (rank 1 usable 시 확장 X — mdx03 정상 case 보호)
 #   default_max_rank=3 = 의미 신뢰 범위 (V4 rank 1~3)
 #   extended_max_rank=32 = catalog presence fallback (rank 4~32)
 # graceful fallback (yaml 없을 시):
 #   loader 가 default {default_max_rank: 3, extended_max_rank: 3} 로 fall through (backward compat)
@@ -1,347 +0,0 @@
 """IMP-46 u3 — Fingerprint-based cache invalidation tests.
 Scope (Stage 2 plan, u3):
 * ``save_proposal`` persists ``fingerprints`` verbatim (u2 already covers
  the round-trip; this suite re-asserts the read-side comparator).
 * ``read_proposal`` accepts an optional ``fingerprints`` kwarg. When
  supplied, the stored dict must equal the supplied dict EXACTLY (strict
  equality). Mismatch — including missing keys, extra keys, or value
  drift — returns ``None``.
 * Default ``fingerprints=None`` performs no comparison (back-compat for
  legacy callers).
 * Fingerprint *computation* stays outside ``cache.py`` — these tests
  treat the three declared shas (``contract_sha`` / ``partial_sha`` /
  ``catalog_sha``) as opaque hex strings, never recomputing them. The
  cache layer is a content-addressed *comparator*, not a content
  *hasher*.
 All filesystem writes are scoped to ``tmp_path`` via
 ``monkeypatch.setattr`` on the module-level :data:`CACHE_ROOT`.
 """
 from __future__ import annotations
 import json
 import pathlib
 import pytest
 from src.phase_z2_ai_fallback import cache as cache_mod
 from src.phase_z2_ai_fallback.cache import (
    KEY_DELIMITER,
    read_proposal,
    save_proposal,
 )
 from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind
 _FRAME_ID = "1171281190"
 _SIG_HASH = "f" * 64
 _KEY = f"{_FRAME_ID}{KEY_DELIMITER}{_SIG_HASH}"
 _FINGERPRINTS_BASELINE: dict[str, str] = {
    "contract_sha": "c" * 64,
    "partial_sha": "p" * 64,
    "catalog_sha": "x" * 64,
 }
 def _proposal(payload: dict | None = None) -> AiFallbackProposal:
    return AiFallbackProposal(
        proposal_kind=ProposalKind.BUILDER_OPTIONS_PATCH,
        payload=payload if payload is not None else {"item_parser": "bullet_v2"},
        rationale="u3-test",
    )
@pytest.fixture(autouse=True)
 def _isolated_cache_root(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setattr(cache_mod, "CACHE_ROOT", tmp_path / "frame_cache")
    yield tmp_path / "frame_cache"
 # -- save side: fingerprints persisted verbatim ---------------------------
 def test_save_persists_fingerprints_verbatim(
    _isolated_cache_root: pathlib.Path,
 ):
    path = save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=_FINGERPRINTS_BASELINE,
    )
    stored = json.loads(path.read_text(encoding="utf-8"))["fingerprints"]
    assert stored == _FINGERPRINTS_BASELINE
 # -- read side: back-compat (no fingerprints kwarg) -----------------------
 def test_read_without_fingerprints_kwarg_returns_proposal(
    _isolated_cache_root: pathlib.Path,
 ):
    """Legacy read path (no kwarg) skips invalidation — round-trip succeeds."""
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=_FINGERPRINTS_BASELINE,
    )
    loaded = read_proposal(_KEY)
    assert loaded is not None
    assert loaded.payload == {"item_parser": "bullet_v2"}
 def test_read_without_fingerprints_kwarg_ignores_stored_mismatch(
    _isolated_cache_root: pathlib.Path,
 ):
    """A caller that has not adopted fingerprint-aware lookup must still
    see the proposal — invalidation only kicks in when explicitly asked."""
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints={"contract_sha": "old"},
    )
    loaded = read_proposal(_KEY)
    assert loaded is not None
 # -- read side: matching fingerprints -------------------------------------
 def test_read_with_matching_fingerprints_returns_proposal(
    _isolated_cache_root: pathlib.Path,
 ):
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=_FINGERPRINTS_BASELINE,
    )
    loaded = read_proposal(_KEY, fingerprints=dict(_FINGERPRINTS_BASELINE))
    assert loaded is not None
    assert loaded.proposal_kind is ProposalKind.BUILDER_OPTIONS_PATCH
 def test_read_with_empty_fingerprints_matches_empty_stored(
    _isolated_cache_root: pathlib.Path,
 ):
    """Both sides empty is an exact match, not a special-case None."""
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        # default fingerprints=None → stored as {}
    )
    loaded = read_proposal(_KEY, fingerprints={})
    assert loaded is not None
 # -- read side: invalidation on mismatch ----------------------------------
@pytest.mark.parametrize(
    "drifted_axis",
    ["contract_sha", "partial_sha", "catalog_sha"],
 )
 def test_read_invalidates_on_single_axis_drift(
    drifted_axis: str, _isolated_cache_root: pathlib.Path
 ):
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=_FINGERPRINTS_BASELINE,
    )
    supplied = dict(_FINGERPRINTS_BASELINE)
    supplied[drifted_axis] = "deadbeef" * 8  # 64-char distinct value
    assert read_proposal(_KEY, fingerprints=supplied) is None
 def test_read_invalidates_when_caller_supplies_extra_key(
    _isolated_cache_root: pathlib.Path,
 ):
    """Strict equality — extra key on caller side is a mismatch."""
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=_FINGERPRINTS_BASELINE,
    )
    supplied = dict(_FINGERPRINTS_BASELINE)
    supplied["future_axis_sha"] = "z" * 64
    assert read_proposal(_KEY, fingerprints=supplied) is None
 def test_read_invalidates_when_caller_supplies_subset(
    _isolated_cache_root: pathlib.Path,
 ):
    """Strict equality — subset on caller side is a mismatch."""
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=_FINGERPRINTS_BASELINE,
    )
    subset = {"contract_sha": _FINGERPRINTS_BASELINE["contract_sha"]}
    assert read_proposal(_KEY, fingerprints=subset) is None
 def test_read_invalidates_when_entry_saved_without_fingerprints(
    _isolated_cache_root: pathlib.Path,
 ):
    """A pre-invalidation cache entry (empty stored fingerprints) MUST NOT
    satisfy a fingerprint-aware lookup — caller demands proof of freshness."""
    save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        # default fingerprints=None → stored as {}
    )
    assert read_proposal(_KEY, fingerprints=_FINGERPRINTS_BASELINE) is None
 def test_read_invalidates_when_stored_fingerprints_not_dict(
    _isolated_cache_root: pathlib.Path,
 ):
    """Hand-corrupted payload (fingerprints serialized as non-dict) → None."""
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "proposal": _proposal().model_dump(mode="json"),
                "slide_css": None,
                "fingerprints": ["contract_sha", "c" * 64],
            }
        ),
        encoding="utf-8",
    )
    assert read_proposal(_KEY, fingerprints=_FINGERPRINTS_BASELINE) is None
 def test_read_invalidates_when_stored_fingerprints_field_missing(
    _isolated_cache_root: pathlib.Path,
 ):
    """Legacy payload (no ``fingerprints`` field at all) → None when caller
    demands fingerprint comparison."""
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "proposal": _proposal().model_dump(mode="json"),
                "slide_css": None,
                # fingerprints field deliberately omitted
            }
        ),
        encoding="utf-8",
    )
    assert read_proposal(_KEY, fingerprints={"contract_sha": "c" * 64}) is None
 def test_read_with_matching_fingerprints_still_loses_to_missing_file():
    """File missing takes precedence over fingerprint check — no false hit."""
    assert read_proposal(_KEY, fingerprints=_FINGERPRINTS_BASELINE) is None
 def test_read_with_matching_fingerprints_still_loses_to_corrupt_json(
    _isolated_cache_root: pathlib.Path,
 ):
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("{not valid json", encoding="utf-8")
    assert read_proposal(_KEY, fingerprints=_FINGERPRINTS_BASELINE) is None
 # -- read side: input validation symmetry with save -----------------------
 def test_read_rejects_non_dict_fingerprints():
    with pytest.raises(TypeError):
        read_proposal(_KEY, fingerprints=["contract_sha", "c" * 64])  # type: ignore[arg-type]
 def test_read_rejects_non_dict_fingerprints_string():
    with pytest.raises(TypeError):
        read_proposal(_KEY, fingerprints="contract_sha=c" * 8)  # type: ignore[arg-type]
 def test_read_rejects_non_dict_fingerprints_int():
    with pytest.raises(TypeError):
        read_proposal(_KEY, fingerprints=42)  # type: ignore[arg-type]
 # -- isolation: cache.py never computes fingerprints ----------------------
 def test_cache_module_has_no_fingerprint_computer():
    """Guardrail: cache.py is a *comparator*, not a *hasher*. The three
    declared shas are computed outside this module (step 12 / pipeline
    glue). Adding a fingerprint computer here would leak Phase Z runtime
    knowledge into the cache layer and violate AI isolation."""
    public_surface = [
        name
        for name in dir(cache_mod)
        if not name.startswith("_") and callable(getattr(cache_mod, name))
    ]
    forbidden_substrings = ("hash", "sha", "fingerprint")
    leaks = [
        name
        for name in public_surface
        if any(sub in name.lower() for sub in forbidden_substrings)
    ]
    assert leaks == [], (
        f"cache.py public surface leaks fingerprint computation: {leaks}; "
        "computation must live outside cache.py per IMP-46 u3 contract."
    )
 # -- isolation across distinct fingerprint sets ---------------------------
 def test_distinct_fingerprint_sets_isolated_per_signature(
    _isolated_cache_root: pathlib.Path,
 ):
    """Two entries under different signature hashes keep their own
    fingerprints; reading one with the other's fingerprints misses."""
    key_a = f"{_FRAME_ID}{KEY_DELIMITER}{'a' * 64}"
    key_b = f"{_FRAME_ID}{KEY_DELIMITER}{'b' * 64}"
    fps_a = {"contract_sha": "a" * 64}
    fps_b = {"contract_sha": "b" * 64}
    save_proposal(
        key_a,
        _proposal(payload={"sig": "a"}),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=fps_a,
    )
    save_proposal(
        key_b,
        _proposal(payload={"sig": "b"}),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=fps_b,
    )
    # Crossed lookups miss.
    assert read_proposal(key_a, fingerprints=fps_b) is None
    assert read_proposal(key_b, fingerprints=fps_a) is None
    # Aligned lookups hit.
    a_hit = read_proposal(key_a, fingerprints=fps_a)
    b_hit = read_proposal(key_b, fingerprints=fps_b)
    assert a_hit is not None and a_hit.payload == {"sig": "a"}
    assert b_hit is not None and b_hit.payload == {"sig": "b"}
@@ -1,93 +0,0 @@
 """IMP-46 u6 — repository layout coverage for the persistent frame cache.
 This module is a *layout* contract test, not a runtime test. It asserts the
 files committed to source control that make ``data/frame_cache/`` exist on a
 fresh checkout while keeping cached JSON payloads ignored by git:
 * ``data/frame_cache/.gitkeep`` is tracked (so the cache root exists for a
  fresh clone before any AI fallback run materialises payloads).
 * ``.gitignore`` ignores ``data/*`` broadly, re-includes the
  ``data/frame_cache/`` directory, ignores its contents, and re-includes
  ``data/frame_cache/.gitkeep`` so cache payloads under
  ``data/frame_cache/{frame_id}/{signature_hash}.json`` remain ignored.
 If somebody removes the ``.gitkeep`` marker, drops the negation lines from
 ``.gitignore``, or commits a real cache payload, this test fails. The cache
 module surface (cache.py) is exercised by ``test_cache.py`` /
 ``test_cache_invalidation.py`` and is intentionally *not* re-asserted here —
 this file is the layout-only lock that Stage 2 u6 declared.
 """
 from __future__ import annotations
 from pathlib import Path
 import pytest
 REPO_ROOT = Path(__file__).resolve().parents[2]
 GITIGNORE_PATH = REPO_ROOT / ".gitignore"
 CACHE_ROOT = REPO_ROOT / "data" / "frame_cache"
 GITKEEP_PATH = CACHE_ROOT / ".gitkeep"
 def _gitignore_lines() -> list[str]:
    assert GITIGNORE_PATH.is_file(), f".gitignore missing at {GITIGNORE_PATH}"
    text = GITIGNORE_PATH.read_text(encoding="utf-8")
    return [line.strip() for line in text.splitlines()]
 def test_frame_cache_root_directory_exists() -> None:
    """``data/frame_cache/`` must exist on disk as the cache root."""
    assert CACHE_ROOT.is_dir(), (
        f"frame cache root missing: {CACHE_ROOT}. The directory must exist "
        "for save_proposal to write JSON payloads without first conjuring a "
        "parent on demand from outside the cache module."
    )
 def test_gitkeep_marker_is_tracked_file() -> None:
    """``data/frame_cache/.gitkeep`` is the marker that keeps the dir tracked."""
    assert GITKEEP_PATH.is_file(), (
        f".gitkeep marker missing: {GITKEEP_PATH}. Without it the cache root "
        "would disappear on a fresh clone (everything under data/ is "
        "ignored by default)."
    )
@pytest.mark.parametrize(
    "rule",
    [
        # Broad ignore for everything under data/ (cache payloads, runs/, etc.).
        "data/*",
        # Re-include the frame_cache directory itself so child negations work.
        "!data/frame_cache/",
        # Ignore everything inside frame_cache/ (cached JSON payloads).
        "data/frame_cache/*",
        # Re-include the .gitkeep marker only.
        "!data/frame_cache/.gitkeep",
    ],
 )
 def test_gitignore_contains_frame_cache_exception(rule: str) -> None:
    """The four ignore rules together pin the 'track marker only' contract."""
    lines = _gitignore_lines()
    assert rule in lines, (
        f".gitignore missing IMP-46 u6 rule: {rule!r}. The four-line block "
        "(data/*, !data/frame_cache/, data/frame_cache/*, "
        "!data/frame_cache/.gitkeep) together ensure the cache root is "
        "tracked while cached payloads remain ignored."
    )
 def test_gitignore_rule_order_keeps_payloads_ignored() -> None:
    """Rule order matters: the ``data/frame_cache/*`` re-ignore must come
    AFTER the ``!data/frame_cache/`` directory re-include, otherwise the
    re-include would shadow it and cached JSON payloads would be tracked."""
    lines = _gitignore_lines()
    reinclude_dir = lines.index("!data/frame_cache/")
    reignore_contents = lines.index("data/frame_cache/*")
    reinclude_marker = lines.index("!data/frame_cache/.gitkeep")
    assert reinclude_dir < reignore_contents < reinclude_marker, (
        "gitignore IMP-46 u6 block out of order: expected "
        "'!data/frame_cache/' < 'data/frame_cache/*' < "
        "'!data/frame_cache/.gitkeep' so cached payloads stay ignored while "
        "only the marker is tracked."
    )
@@ -1,184 +0,0 @@
 """IMP-46 u1 — Frame cache signature builder tests.
 Verifies:
  * Determinism — identical inputs yield the same SHA256 digest.
  * Axis-change sensitivity — every one of the 8 declared axes mutates the
    digest when changed in isolation.
  * Public surface — only the 8 declared axes are accepted (no
    sample/section identifier leakage).
  * char_count bucket boundaries (0-50, 51-150, 151-400, 401-1000, 1001+).
  * source_shape enum equivalence (string and SourceShape inputs match).
  * schema_version is part of the hashed payload (digest stable for fixture).
 """
 from __future__ import annotations
 import inspect
 import pytest
 from src.phase_z2_ai_fallback.signature import (
    CHAR_COUNT_BUCKET_LABELS,
    SCHEMA_VERSION,
    SourceShape,
    bucket_char_count,
    build_signature,
 )
 def _base_kwargs() -> dict:
    return dict(
        frame_id="frame_03",
        v4_label="light_edit",
        cardinality=3,
        source_shape=SourceShape.BULLET,
        h3_count=2,
        char_count_bucket="51-150",
        layout_preset="sidebar-right",
        zone_position="top",
    )
 def test_schema_version_is_one() -> None:
    assert SCHEMA_VERSION == 1
 def test_bucket_labels_match_spec() -> None:
    assert CHAR_COUNT_BUCKET_LABELS == (
        "0-50",
        "51-150",
        "151-400",
        "401-1000",
        "1001+",
    )
 def test_signature_is_deterministic() -> None:
    a = build_signature(**_base_kwargs())
    b = build_signature(**_base_kwargs())
    assert a == b
    assert len(a) == 64
@pytest.mark.parametrize(
    "axis, new_value",
    [
        ("frame_id", "frame_04"),
        ("v4_label", "restructure"),
        ("cardinality", 5),
        ("source_shape", SourceShape.PARAGRAPH),
        ("h3_count", 3),
        ("char_count_bucket", "151-400"),
        ("layout_preset", "two-column"),
        ("zone_position", "bottom_l"),
    ],
 )
 def test_signature_changes_for_each_axis(axis: str, new_value: object) -> None:
    base = build_signature(**_base_kwargs())
    kwargs = _base_kwargs()
    kwargs[axis] = new_value
    assert build_signature(**kwargs) != base
 def test_signature_accepts_string_source_shape() -> None:
    enum_sig = build_signature(**_base_kwargs())
    kwargs = _base_kwargs()
    kwargs["source_shape"] = "bullet"
    assert build_signature(**kwargs) == enum_sig
 def test_signature_rejects_unknown_source_shape() -> None:
    kwargs = _base_kwargs()
    kwargs["source_shape"] = "nonsense"
    with pytest.raises(ValueError):
        build_signature(**kwargs)
 def test_signature_rejects_unknown_char_count_bucket() -> None:
    kwargs = _base_kwargs()
    kwargs["char_count_bucket"] = "999-1234"
    with pytest.raises(ValueError):
        build_signature(**kwargs)
 def test_signature_handles_none_cardinality() -> None:
    kwargs = _base_kwargs()
    kwargs["cardinality"] = None
    sig = build_signature(**kwargs)
    assert len(sig) == 64
    kwargs2 = _base_kwargs()
    kwargs2["cardinality"] = 0
    assert build_signature(**kwargs2) != sig
 def test_signature_surface_only_8_declared_axes() -> None:
    params = set(inspect.signature(build_signature).parameters)
    expected = {
        "frame_id",
        "v4_label",
        "cardinality",
        "source_shape",
        "h3_count",
        "char_count_bucket",
        "layout_preset",
        "zone_position",
    }
    assert params == expected
 def test_bucket_boundaries() -> None:
    assert bucket_char_count(0) == "0-50"
    assert bucket_char_count(50) == "0-50"
    assert bucket_char_count(51) == "51-150"
    assert bucket_char_count(150) == "51-150"
    assert bucket_char_count(151) == "151-400"
    assert bucket_char_count(400) == "151-400"
    assert bucket_char_count(401) == "401-1000"
    assert bucket_char_count(1000) == "401-1000"
    assert bucket_char_count(1001) == "1001+"
    assert bucket_char_count(10_000) == "1001+"
 def test_bucket_rejects_negative() -> None:
    with pytest.raises(ValueError):
        bucket_char_count(-1)
 def test_bucket_rejects_non_int() -> None:
    with pytest.raises(TypeError):
        bucket_char_count(3.14)  # type: ignore[arg-type]
    with pytest.raises(TypeError):
        bucket_char_count(True)  # type: ignore[arg-type]
 def test_signature_stable_known_fixture() -> None:
    """Lock the digest for a known fixture so a silent payload-shape change
    (e.g. a new axis sneaks in, or schema_version drifts) breaks this test.
    """
    sig = build_signature(
        frame_id="frame_03",
        v4_label="light_edit",
        cardinality=3,
        source_shape=SourceShape.BULLET,
        h3_count=2,
        char_count_bucket="51-150",
        layout_preset="sidebar-right",
        zone_position="top",
    )
    import hashlib
    import json
    expected_payload = {
        "schema_version": 1,
        "frame_id": "frame_03",
        "v4_label": "light_edit",
        "cardinality": 3,
        "source_shape": "bullet",
        "h3_count": 2,
        "char_count_bucket": "51-150",
        "layout_preset": "sidebar-right",
        "zone_position": "top",
    }
    expected = hashlib.sha256(
        json.dumps(expected_payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
    ).hexdigest()
    assert sig == expected
@@ -1,159 +0,0 @@
 """IMP-38 U2 — dynamic effective max_rank + trace 8-field + 3-tier usable predicate.
 Verify:
  - max_rank=None (default) → policy applied (usable_count + effective_max_rank 결정)
  - max_rank=int (caller override) → that value used as-is (backward compat)
  - trace contains 8 IMP-38 fields + legacy "max_rank" alias
  - usable_count >= threshold → default_max_rank (mdx03 정상 case)
  - usable_count < threshold → effective_extended_ceiling (mdx05-2 확장 case)
  - effective_extended_ceiling = min(configured, len(judgments_full32)) (Codex #2)
  - IMP-30 allow_provisional byte-identical (chain_exhausted 후 provisional 합성)
 4 round 합의 (#67):
  - Codex #1: 별 yaml (catalog 오염 방지)
  - Codex #2: min(configured, len(judgments)) 정정
  - Codex #3: load_frame_contracts() shape 무변
 """
 from __future__ import annotations
 import pytest
@pytest.fixture(autouse=True)
 def _reset_policy_cache():
    """Reset module-level _V4_FALLBACK_POLICY_CACHE for test isolation."""
    import src.phase_z2_mapper as mapper
    mapper._V4_FALLBACK_POLICY_CACHE = None
    yield
    mapper._V4_FALLBACK_POLICY_CACHE = None
 def _make_v4_section(judgments: list[dict]) -> dict:
    """Helper — V4 fixture with mdx_sections[section_id].judgments_full32."""
    return {
        "mdx_sections": {
            "sec-1": {
                "judgments_full32": judgments,
            }
        }
    }
 def _judgment(template_id: str, label: str, confidence: float = 0.5, frame_id: int = 0) -> dict:
    """Helper — V4 judgment entry shape."""
    return {
        "template_id": template_id,
        "frame_id": frame_id or hash(template_id) % 10000,
        "frame_number": 0,
        "confidence": confidence,
        "label": label,
    }
 # ─── U2 Test: caller override (backward compat) ────────────────────
 def test_caller_override_uses_explicit_max_rank():
    """max_rank=3 explicit → effective_max_rank=3, policy_applied=caller_override."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    judgments = [_judgment(f"t{i}", "reject") for i in range(5)]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1", max_rank=3)
    assert trace["policy_applied"] == "caller_override"
    assert trace["effective_max_rank"] == 3
    assert trace["max_rank"] == 3  # legacy alias
 def test_caller_override_max_rank_5_used_directly():
    """max_rank=5 explicit → effective_max_rank=5 (policy 무시)."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    judgments = [_judgment(f"t{i}", "reject") for i in range(10)]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1", max_rank=5)
    assert trace["policy_applied"] == "caller_override"
    assert trace["effective_max_rank"] == 5
 # ─── U2 Test: 8 trace fields presence ──────────────────────────────
 def test_trace_contains_8_imp38_fields():
    """trace dict must contain all 8 IMP-38 fields + legacy max_rank alias."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    judgments = [_judgment(f"t{i}", "reject") for i in range(3)]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1")
    expected = {
        "requested_max_rank",
        "default_max_rank",
        "configured_extended_max_rank",
        "judgments_count",
        "effective_extended_ceiling",
        "effective_max_rank",
        "usable_count",
        "policy_applied",
        "max_rank",  # legacy alias
    }
    missing = expected - set(trace.keys())
    assert not missing, f"missing IMP-38 trace fields: {missing}"
 # ─── U2 Test: Codex #2 정정 — min(configured, len(judgments_full32)) ──
 def test_effective_extended_ceiling_is_min_of_configured_and_judgments_count():
    """Codex #2 LOCK — judgments_count < configured 일 때 ceiling = judgments_count."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    # 5 judgments only — configured extended (32) 보다 작음
    judgments = [_judgment(f"t{i}", "reject") for i in range(5)]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1")
    assert trace["judgments_count"] == 5
    assert trace["effective_extended_ceiling"] == 5  # min(32, 5) = 5
 # ─── U2 Test: no_judgments path ──────────────────────────────────
 def test_no_judgments_path():
    """judgments_count=0 → policy_applied=no_judgments, effective_max_rank=default."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    v4 = _make_v4_section([])
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1")
    assert trace["policy_applied"] == "no_judgments"
    assert trace["judgments_count"] == 0
    assert trace["effective_max_rank"] == trace["default_max_rank"]
    assert trace["fallback_reason"] == "empty_v4_judgments"
 # ─── U2 Test: no_v4_section ─────────────────────────────────────
 def test_no_v4_section_path():
    """unknown section_id → fallback_reason=no_v4_section + trace still has 8 IMP-38 fields."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    v4 = {"mdx_sections": {}}
    _match, trace = lookup_v4_match_with_fallback(v4, "unknown-sec")
    assert trace["fallback_reason"] == "no_v4_section"
    # 8 fields still present even when no section found
    assert "policy_applied" in trace
    assert "effective_max_rank" in trace
 # ─── U2 Test: chain_exhausted message reflects effective_max_rank ──
 def test_chain_exhausted_message_includes_effective_max_rank():
    """fallback_reason 메시지가 동적 effective_max_rank 반영 (hardcoded "1_to_3" X)."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    # 3 judgments all reject (catalog 등록 X 가정 — t1/t2/t3 는 catalog 에 없음)
    judgments = [_judgment(f"unregistered_t{i}", "reject") for i in range(3)]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1", max_rank=3)
    # chain exhausted — 메시지 가 effective_max_rank=3 반영
    if trace["selection_path"] == "chain_exhausted":
        # first_skip_reason 가 있으면 그게 우선, 없으면 default 메시지
        assert (
            trace["fallback_reason"] is not None
            and ("no_auto_renderable" in trace["fallback_reason"] or "phase_z_status" in trace["fallback_reason"] or "no_contract" in trace["fallback_reason"])
        )
@@ -1,213 +0,0 @@
 """IMP-47B u13 — Persist validated proposals through ``save_proposal`` after gates.
 Scope (this slice):
  Verify the new ``_persist_ai_repair_proposals_to_cache`` helper in
  ``src/phase_z2_pipeline.py`` honours the IMP-46 dual-gate truth table
  on the post-Step-14 cache-save seam. The helper is exercised in
  isolation (no Selenium, no full pipeline) with synthetic AI repair
  records that mirror the gather → apply → coverage chain shape
  produced by IMP-47B u4 / u5 / u7.
 Guardrails proven by this test (IMP-46 + IMP-47B policy bullets):
  * ``visual_check_passed=False`` always blocks — never bypassable, even
    when ``auto_cache=True`` (IMP-46 u5 truth table cell).
  * ``user_approved=False`` AND ``auto_cache=False`` → gate blocked
    (default pipeline path has no UX approval gate; ``--auto-cache`` is
    the documented bypass).
  * ``visual_check_passed=True`` AND ``auto_cache=True`` → proposal
    persisted on disk under ``data/frame_cache/{frame_id}/{hash}.json``
    via ``cache.save_proposal``.
  * Non-applied records (no_proposal / no_zone_match / unsupported /
    error) → ``cache_save_status='not_applied'`` and NEVER reach
    ``save_proposal`` (no filesystem touch).
  * Settings axis — ``settings.ai_fallback_auto_cache`` sourced through
    the helper kwargs, never inlined (hardcoding ban).
 """
 from __future__ import annotations
 import pathlib
 import pytest
 from src.phase_z2_ai_fallback import cache as cache_mod
 from src.phase_z2_ai_fallback.cache import AiFallbackCacheGateError
 from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind
 from src.phase_z2_pipeline import _persist_ai_repair_proposals_to_cache
 def _applied_record(
    *,
    cache_key: str = "MOCK_FRAME::deadbeef" + "0" * 56,
    fingerprints: dict | None = None,
    slots: dict | None = None,
 ) -> dict:
    """Build an IMP-47B u4/u5 shaped record marked ``applied:partial_overrides``."""
    if fingerprints is None:
        fingerprints = {"contract_sha": "c1", "partial_sha": "p1", "catalog_sha": "k1"}
    if slots is None:
        slots = {"title": "AI repaired", "bullets": ["b1", "b2"]}
    proposal = AiFallbackProposal(
        proposal_kind=ProposalKind.PARTIAL_OVERRIDES,
        payload={"slots": slots},
        rationale="cache save gate test",
    )
    return {
        "unit_index": 0,
        "source_section_ids": ["MOCK_S1"],
        "frame_template_id": "MOCK_FRAME",
        "label": "reject",
        "route_hint": "ai_adaptation_required",
        "provisional": True,
        "ai_called": True,
        "skip_reason": None,
        "proposal": proposal.model_dump(),
        "error": None,
        "cache_key": cache_key,
        "fingerprints": fingerprints,
        "apply_status": "applied:partial_overrides",
    }
@pytest.fixture(autouse=True)
 def _isolate_cache_root(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch):
    """Redirect ``cache.CACHE_ROOT`` to a per-test tmp dir so save_proposal
    writes never touch the real ``data/frame_cache/`` tree."""
    monkeypatch.setattr(cache_mod, "CACHE_ROOT", tmp_path / "frame_cache")
    yield tmp_path / "frame_cache"
 def test_visual_check_failed_blocks_save_even_with_auto_cache(_isolate_cache_root):
    """visual_check_passed=False is never bypassable — auto_cache cannot override."""
    record = _applied_record()
    records = [record]
    _persist_ai_repair_proposals_to_cache(
        records,
        visual_check_passed=False,
        user_approved=True,
        auto_cache=True,
    )
    assert record["cache_save_status"].startswith("gate_blocked:")
    assert "visual_check_passed=False" in record["cache_save_status"]
    # No filesystem write occurred.
    assert not _isolate_cache_root.exists() or not any(_isolate_cache_root.rglob("*.json"))
 def test_user_not_approved_and_no_auto_cache_blocks_save(_isolate_cache_root):
    """Default pipeline path (user_approved=False, auto_cache=False) → gate blocked."""
    record = _applied_record()
    records = [record]
    _persist_ai_repair_proposals_to_cache(
        records,
        visual_check_passed=True,
        user_approved=False,
        auto_cache=False,
    )
    assert record["cache_save_status"].startswith("gate_blocked:")
    assert "user_approved=False" in record["cache_save_status"]
    assert not _isolate_cache_root.exists() or not any(_isolate_cache_root.rglob("*.json"))
 def test_visual_passed_and_auto_cache_persists_proposal(_isolate_cache_root):
    """Happy path — visual_check_passed=True + auto_cache=True persists JSON."""
    record = _applied_record()
    records = [record]
    _persist_ai_repair_proposals_to_cache(
        records,
        visual_check_passed=True,
        user_approved=False,
        auto_cache=True,
    )
    assert record["cache_save_status"] == "saved"
    written = list(_isolate_cache_root.rglob("*.json"))
    assert len(written) == 1
    # Layout = {CACHE_ROOT}/{frame_id}/{signature_hash}.json.
    written_path = written[0]
    assert written_path.parent.name == "MOCK_FRAME"
 def test_non_applied_records_are_skipped_without_filesystem_touch(_isolate_cache_root):
    """no_proposal / no_zone_match / unsupported_kind / error → never reach save_proposal."""
    no_proposal_record = {
        "unit_index": 0,
        "apply_status": "no_proposal",
        "proposal": None,
        "cache_key": None,
        "fingerprints": None,
    }
    no_zone_record = {
        "unit_index": 1,
        "apply_status": "no_zone_match",
        "proposal": {"proposal_kind": "partial_overrides", "payload": {"slots": {}}, "rationale": ""},
        "cache_key": "MOCK::abc",
        "fingerprints": {"contract_sha": "c", "partial_sha": "p", "catalog_sha": "k"},
    }
    unsupported_record = {
        "unit_index": 2,
        "apply_status": "unsupported_kind_for_reject_route:builder_options_patch",
        "proposal": {"proposal_kind": "builder_options_patch", "payload": {}, "rationale": ""},
        "cache_key": "MOCK::def",
        "fingerprints": {"contract_sha": "c", "partial_sha": "p", "catalog_sha": "k"},
    }
    error_record = {
        "unit_index": 3,
        "apply_status": None,
        "proposal": None,
        "cache_key": "MOCK::ghi",
        "fingerprints": {"contract_sha": "c", "partial_sha": "p", "catalog_sha": "k"},
        "error": "RuntimeError: boom",
    }
    records = [no_proposal_record, no_zone_record, unsupported_record, error_record]
    _persist_ai_repair_proposals_to_cache(
        records,
        visual_check_passed=True,
        user_approved=True,
        auto_cache=True,
    )
    for r in records:
        assert r["cache_save_status"] == "not_applied"
    # Zero JSON files written because none of the records were applied.
    assert not _isolate_cache_root.exists() or not any(_isolate_cache_root.rglob("*.json"))
 def test_mixed_records_only_persist_applied_ones(_isolate_cache_root):
    """Mixed batch — only the ``applied:`` record is persisted."""
    applied = _applied_record(cache_key="MOCK_FRAME::aaaaaaaa" + "0" * 56)
    not_applied = {
        "unit_index": 1,
        "apply_status": "no_proposal",
        "proposal": None,
        "cache_key": None,
        "fingerprints": None,
    }
    records = [applied, not_applied]
    _persist_ai_repair_proposals_to_cache(
        records,
        visual_check_passed=True,
        user_approved=False,
        auto_cache=True,
    )
    assert applied["cache_save_status"] == "saved"
    assert not_applied["cache_save_status"] == "not_applied"
    written = list(_isolate_cache_root.rglob("*.json"))
    assert len(written) == 1
 def test_invalid_proposal_payload_surfaces_without_raising(_isolate_cache_root):
    """Malformed ``proposal`` dict → ``cache_save_status='invalid_proposal:...'``,
    no filesystem write, no exception bubbling into the pipeline runtime."""
    bad_record = {
        "unit_index": 0,
        "apply_status": "applied:partial_overrides",
        "proposal": {"proposal_kind": "not_a_valid_enum_value", "payload": {}, "rationale": ""},
        "cache_key": "MOCK::bad",
        "fingerprints": {"contract_sha": "c", "partial_sha": "p", "catalog_sha": "k"},
    }
    records = [bad_record]
    _persist_ai_repair_proposals_to_cache(
        records,
        visual_check_passed=True,
        user_approved=True,
        auto_cache=True,
    )
    assert bad_record["cache_save_status"].startswith("invalid_proposal:")
    assert not _isolate_cache_root.exists() or not any(_isolate_cache_root.rglob("*.json"))
@@ -1,95 +0,0 @@
 """IMP-47B u7 — Post-AI source_section_ids coverage invariant tests.
 Scope (this slice):
  * Helper ``_check_post_ai_coverage_invariant(units, ai_repair_records)``
    (src/phase_z2_pipeline.py) compares the pre-AI superset (unit
    ``source_section_ids``) to the post-apply superset present on
    gather records. Per the AI isolation contract + dropped 절대 룰
    (``feedback_ai_isolation_contract``), AI repair must not silently
    drop a section.
  * The helper returns a structured dict (``pre_ai_section_ids``,
    ``post_ai_section_ids``, ``dropped_section_ids``, ``status``) so u8
    can surface ``status`` through ``slide_status.ai_repair_status``.
 u8 slide_status surfacing and u10 E2E no-text-loss assertion are out
 of scope for this unit. The helper is pure (no AI call, no IO) so a
 synthetic stub-unit / stub-record fixture exercises it directly.
 """
 from __future__ import annotations
 from dataclasses import dataclass, field
 from src.phase_z2_pipeline import _check_post_ai_coverage_invariant
@dataclass
 class _StubUnit:
    source_section_ids: list[str] = field(default_factory=list)
 def _record(source_section_ids: list[str]) -> dict:
    """Minimal gather-record stub — only the field u7 reads."""
    return {"source_section_ids": list(source_section_ids)}
 # ─── Case 1 : matched coverage → status='ok' ────────────────────────
 def test_coverage_invariant_ok_when_records_match_units():
    """Records carry every unit's source_section_ids → no drop, status='ok'."""
    units = [_StubUnit(["MOCK_S1", "MOCK_S2"]), _StubUnit(["MOCK_S3"])]
    records = [_record(["MOCK_S1", "MOCK_S2"]), _record(["MOCK_S3"])]
    result = _check_post_ai_coverage_invariant(units, records)
    assert result["status"] == "ok"
    assert result["dropped_section_ids"] == []
    assert result["pre_ai_section_ids"] == ["MOCK_S1", "MOCK_S2", "MOCK_S3"]
    assert result["post_ai_section_ids"] == ["MOCK_S1", "MOCK_S2", "MOCK_S3"]
 # ─── Case 2 : record drops a section → status='violated' ────────────
 def test_coverage_invariant_violated_when_record_drops_section():
    """If a record loses a unit's section_id (e.g., apply mutation bug),
    the invariant reports status='violated' + dropped list (dropped 절대 룰).
    """
    units = [_StubUnit(["MOCK_S1", "MOCK_S2"]), _StubUnit(["MOCK_S3"])]
    records = [_record(["MOCK_S1"]), _record(["MOCK_S3"])]  # MOCK_S2 dropped
    result = _check_post_ai_coverage_invariant(units, records)
    assert result["status"] == "violated"
    assert result["dropped_section_ids"] == ["MOCK_S2"]
    assert "MOCK_S2" in result["pre_ai_section_ids"]
    assert "MOCK_S2" not in result["post_ai_section_ids"]
 # ─── Case 3 : empty inputs → status='ok' (no false positive) ────────
 def test_coverage_invariant_ok_on_empty_units_and_records():
    """Empty pipeline (no units / no records) is a vacuous pass —
    avoids false-positive 'violated' on edge-case shapes (no AI work).
    """
    result = _check_post_ai_coverage_invariant([], [])
    assert result["status"] == "ok"
    assert result["dropped_section_ids"] == []
    assert result["pre_ai_section_ids"] == []
    assert result["post_ai_section_ids"] == []
 # ─── Case 4 : multiple drops + dedup ────────────────────────────────
 def test_coverage_invariant_lists_all_dropped_sections_sorted_and_deduped():
    """Multiple missing sections → dropped_section_ids is sorted + deduped.
    Duplicate ids across units / records collapse to a set comparison.
    """
    units = [
        _StubUnit(["MOCK_S3", "MOCK_S1"]),
        _StubUnit(["MOCK_S2", "MOCK_S1"]),  # MOCK_S1 duplicate
    ]
    records: list[dict] = []  # full drop — every unit section missing
    result = _check_post_ai_coverage_invariant(units, records)
    assert result["status"] == "violated"
    assert result["dropped_section_ids"] == ["MOCK_S1", "MOCK_S2", "MOCK_S3"]
    assert result["pre_ai_section_ids"] == ["MOCK_S1", "MOCK_S2", "MOCK_S3"]
    assert result["post_ai_section_ids"] == []
@@ -1,269 +0,0 @@
 """IMP-47B u10 — End-to-end reject smoke (mocked client + full chain + render).
 Scope (this slice):
  E2E chain proving the IMP-47B reject route activates, preserves
  full coverage, and propagates the AI-repaired ``slot_payload``
  into the rendered ``final.html`` artifact when the AI fallback
  client returns a deterministic PARTIAL_OVERRIDES proposal. Wires
  together the four pipeline helpers introduced by u4 / u5 / u7 / u8
  plus the Step 13 render step:
    gather → apply → coverage_invariant → ai_repair_status surfacing
                                       → render_slide → final.html
  The chain mirrors the ``run_phase_z2_mvp1`` call sequence between
  the Step 12 slot_payload write and the Step 20 ``slide_status``
  attach (src/phase_z2_pipeline.py — u4 call site, u5 apply, u6
  artifact, u7 invariant, u8 surface). The Step 13 render path
  (``render_slide`` at src/phase_z2_pipeline.py:2319, called from the
  production write site at src/phase_z2_pipeline.py:5107-5111)
  consumes ``zones_data[i]["slot_payload"]`` verbatim, so this test
  drives that exact production seam: it calls ``render_slide`` on
  the post-apply ``zones_data`` and writes the resulting HTML to a
  ``final.html`` file inside ``tmp_path``, then asserts the AI
  proposal text appears in the on-disk artifact. A heavy
  ``run_phase_z2_mvp1`` integration variant with Selenium overflow
  check remains deferred — this smoke test stops at the rendered
  HTML.
 Guardrails proven by this test (IMP-47B policy bullets):
  * AI 호출 = fallback path only  → master flag default OFF preserved
    (test enables for itself only, restores after).
  * MDX 원문 100% 보존              → coverage_invariant.status == "ok",
    source_section_ids identical before/after AI.
  * 자동 frame swap 금지            → frame_template_id unchanged.
  * frame visual 임의 변경 금지    → frame_contract / partial untouched
    (apply only merges proposal.payload.slots into slot_payload).
  * dropped 절대 룰                 → slot_payload AI keys merged on top
    of deterministic keys; pre-existing meta keys survive.
 """
 from __future__ import annotations
 from dataclasses import dataclass, field
 from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind
 from src.phase_z2_pipeline import (
    _apply_ai_repair_proposals_to_zones,
    _check_post_ai_coverage_invariant,
    _run_step12_ai_repair,
    _summarize_ai_repair_status,
 )
@dataclass
 class _StubUnit:
    """Synthetic CompositionUnit stand-in (subset of fields gather reads)."""
    label: str | None = "reject"
    provisional: bool = True
    frame_template_id: str = "MOCK_T_reject"
    frame_id: str = "MOCK_F_reject"
    source_section_ids: list[str] = field(default_factory=lambda: ["MOCK_S1"])
    raw_content: str = "MOCK MDX paragraph that must survive AI repair."
    v4_rank: int | None = 1
    cardinality: int | None = None
    layout_preset: str = "two_zone_vertical"
    zone_position: str = "top"
    source_shape: str = "paragraph"
    h3_count: int = 0
    char_count: int = 48
 def _patched_route_ai_fallback(**kwargs):
    """Deterministic stand-in for ``route_ai_fallback`` — returns a
    PARTIAL_OVERRIDES proposal that mirrors the declared frame slots.
    The validator (src/phase_z2_ai_fallback/validate.py:61-74) is not
    re-invoked here because this helper bypasses the router; the
    structural slot completeness is asserted by the apply step + the
    coverage invariant downstream.
    """
    return AiFallbackProposal(
        proposal_kind=ProposalKind.PARTIAL_OVERRIDES,
        payload={
            "slots": {
                "title": "AI repaired title",
                "bullets": ["AI repaired bullet 1", "AI repaired bullet 2"],
            }
        },
        rationale="E2E smoke proposal — deterministic.",
    )
 def test_e2e_reject_chain_applies_proposal_and_preserves_coverage(monkeypatch):
    """End-to-end reject smoke (synthetic chain, mocked client).
    Drives the four IMP-47B u4/u5/u7/u8 helpers in pipeline order with
    a single reject+provisional unit. Asserts every guardrail listed
    in the module docstring + the four E2E invariants
    (final.html-bound slot_payload / full coverage / no text loss /
    human_review NOT required on the success path).
    """
    # IMP-47B u4 wiring — patch the router seam in src/phase_z2_ai_fallback/step12.py
    # so the gather call returns a deterministic PARTIAL_OVERRIDES proposal
    # without touching the master flag / network / cache layers.
    import src.phase_z2_ai_fallback.step12 as step12_mod
    monkeypatch.setattr(step12_mod, "route_ai_fallback", _patched_route_ai_fallback)
    unit = _StubUnit()
    units = [unit]
    # Step 12 gather (u4) — eligible reject reaches the patched router.
    records = _run_step12_ai_repair(units)
    assert len(records) == 1
    assert records[0]["route_hint"] == "ai_adaptation_required"
    assert records[0]["ai_called"] is True
    assert records[0]["skip_reason"] is None
    assert records[0]["proposal"]["proposal_kind"] == "partial_overrides"
    assert records[0]["source_section_ids"] == ["MOCK_S1"]
    # Step 12 apply (u5) — PARTIAL_OVERRIDES merged into the matching zone.
    # zones_data[0]["slot_payload"] is exactly what render_slide consumes
    # to emit final.html (src/phase_z2_pipeline.py:5107) — asserting it
    # here proves the reject route now flows into the rendered HTML.
    zones = [{
        "position": "top",
        "template_id": "MOCK_T_reject",
        "slot_payload": {
            "title": "deterministic title",
            "bullets": ["deterministic bullet"],
            "_truncated_count": 0,
        },
    }]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    assert records[0]["apply_status"] == "applied:partial_overrides"
    # final.html-bound slot_payload carries AI proposal values
    assert zones[0]["slot_payload"]["title"] == "AI repaired title"
    assert zones[0]["slot_payload"]["bullets"] == [
        "AI repaired bullet 1",
        "AI repaired bullet 2",
    ]
    # frame visual / pre-existing meta keys survive (no silent shrink).
    assert zones[0]["template_id"] == "MOCK_T_reject"
    assert zones[0]["slot_payload"]["_truncated_count"] == 0
    # frame_template_id on the unit is byte-identical (no auto frame swap).
    assert unit.frame_template_id == "MOCK_T_reject"
    # Step 12 coverage invariant (u7) — full coverage, no text loss.
    coverage = _check_post_ai_coverage_invariant(units, records)
    assert coverage["status"] == "ok"
    assert coverage["pre_ai_section_ids"] == ["MOCK_S1"]
    assert coverage["post_ai_section_ids"] == ["MOCK_S1"]
    assert coverage["dropped_section_ids"] == []
    # Step 20 ai_repair_status surfacing (u8) — applied without human review.
    status = _summarize_ai_repair_status(records, coverage)
    assert status["status"] == "applied"
    assert status["counts"]["applied"] == 1
    assert status["counts"]["error"] == 0
    assert status["counts"]["unsupported_kind"] == 0
    assert status["coverage_status"] == "ok"
    assert status.get("human_review_required") is not True
 def test_e2e_reject_chain_writes_final_html_with_ai_repaired_slot(monkeypatch, tmp_path):
    """End-to-end reject smoke (real render path → final.html on disk).
    Drives the full Stage-2 u10 chain INCLUDING ``render_slide``: the
    AI-repaired ``slot_payload`` is fed through the same Jinja2
    rendering seam the production pipeline uses
    (src/phase_z2_pipeline.py:5107-5111), the resulting HTML is
    written to ``tmp_path / "final.html"``, and the on-disk artifact
    is then asserted to carry the AI proposal value. Uses
    ``bim_dx_comparison_table`` — a real registered frame partial
    (templates/phase_z2/families/bim_dx_comparison_table.html) whose
    template emits ``{{ slot_payload.title }}`` verbatim, so a
    proposal-overridden title surfaces literally in the HTML output.
    """
    import src.phase_z2_ai_fallback.step12 as step12_mod
    monkeypatch.setattr(step12_mod, "route_ai_fallback", _patched_route_ai_fallback)
    from src.phase_z2_pipeline import build_layout_css, render_slide
    unit = _StubUnit(
        frame_template_id="bim_dx_comparison_table",
        zone_position="primary",
        layout_preset="single",
    )
    # Step 12 gather + apply. Deterministic non-overridden slots
    # (col_a_label, col_b_label, rows[*]) are seeded BEFORE apply so the
    # post-render assertions below can prove u5 merge semantics
    # (dict.update — not dict-replace) survive the render seam. The
    # router proposal only carries ``{title, bullets}`` — every other
    # slot must reach final.html untouched.
    records = _run_step12_ai_repair([unit])
    zones = [{
        "position": "primary",
        "template_id": "bim_dx_comparison_table",
        "slot_payload": {
            "title": "deterministic frame title",
            "col_a_label": "DETERMINISTIC_COL_A_LABEL",
            "col_b_label": "DETERMINISTIC_COL_B_LABEL",
            "rows": [
                {"label": "DET_ROW_LABEL", "col_a": "DET_ROW_A", "col_b": "DET_ROW_B"},
            ],
        },
    }]
    _apply_ai_repair_proposals_to_zones(records, ["primary"], zones)
    assert records[0]["apply_status"] == "applied:partial_overrides"
    # Step 13 render — production seam (src/phase_z2_pipeline.py:5107-5111).
    layout_css = build_layout_css("single", zones)
    html = render_slide("IMP-47B E2E reject smoke", None, zones, "single", layout_css)
    final_html_path = tmp_path / "final.html"
    final_html_path.write_text(html, encoding="utf-8")
    # final.html artifact exists on disk and is non-empty.
    assert final_html_path.is_file()
    assert final_html_path.stat().st_size > 0
    rendered = final_html_path.read_text(encoding="utf-8")
    # AI-repaired slot content appears in the rendered HTML.
    assert "AI repaired title" in rendered
    # Deterministic pre-apply title was overridden in the HTML output
    # (no silent merge that leaves both values visible).
    assert "deterministic frame title" not in rendered
    # Non-overridden deterministic slots survive merge → render (u5
    # dict.update semantics, not dict-replace; dropped 절대 룰 honoured
    # at the render seam, not just in slot_payload memory).
    assert "DETERMINISTIC_COL_A_LABEL" in rendered
    assert "DETERMINISTIC_COL_B_LABEL" in rendered
    assert "DET_ROW_LABEL" in rendered
    assert "DET_ROW_A" in rendered
    assert "DET_ROW_B" in rendered
    # Frame template id is preserved end-to-end (no auto frame swap).
    assert 'data-template-id="bim_dx_comparison_table"' in rendered
    assert unit.frame_template_id == "bim_dx_comparison_table"
    # MDX 원문 100% 보존 — coverage invariant + status surfacing.
    coverage = _check_post_ai_coverage_invariant([unit], records)
    assert coverage["status"] == "ok"
    assert coverage["dropped_section_ids"] == []
    status = _summarize_ai_repair_status(records, coverage)
    assert status["status"] == "applied"
    assert status.get("human_review_required") is not True
 def test_e2e_reject_chain_no_text_loss_on_multi_section_unit(monkeypatch):
    """Multi-section reject unit — every section id flows through gather,
    apply, coverage invariant, and ai_repair_status surfacing without a
    drop. Locks the 'MDX 원문 100% 보존' guardrail at unit-multiplicity
    granularity (gather copies the list via ``list(...)`` at
    src/phase_z2_ai_fallback/step12.py:124 so apply mutations cannot
    silently drop it)."""
    import src.phase_z2_ai_fallback.step12 as step12_mod
    monkeypatch.setattr(step12_mod, "route_ai_fallback", _patched_route_ai_fallback)
    unit = _StubUnit(source_section_ids=["MOCK_S1", "MOCK_S2", "MOCK_S3"])
    records = _run_step12_ai_repair([unit])
    zones = [{
        "position": "top",
        "template_id": "MOCK_T_reject",
        "slot_payload": {"title": "det", "bullets": ["det"]},
    }]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    coverage = _check_post_ai_coverage_invariant([unit], records)
    assert coverage["pre_ai_section_ids"] == ["MOCK_S1", "MOCK_S2", "MOCK_S3"]
    assert coverage["post_ai_section_ids"] == ["MOCK_S1", "MOCK_S2", "MOCK_S3"]
    assert coverage["dropped_section_ids"] == []
    status = _summarize_ai_repair_status(records, coverage)
    assert status["status"] == "applied"
    assert status.get("human_review_required") is not True
@@ -1,174 +0,0 @@
 """IMP-47B u8 — slide_status.ai_repair_status surfacing tests.
 Scope (this slice):
  Helper ``_summarize_ai_repair_status(ai_repair_records, coverage_invariant)``
  (src/phase_z2_pipeline.py) composes u4 gather ``error`` + u5
  ``apply_status`` + u7 ``coverage_invariant`` into a single
  ``ai_repair_status`` axis attached to ``slide_status``. Failure-axis
  priority (highest → lowest): ``error`` > ``coverage_violated`` >
  ``unsupported_kind`` > ``applied`` > ``ok``. ``human_review_required``
  flips True on the three failure axes for u11 frontend surfacing.
 The frontend reads ``slide_status.ai_repair_status`` to render a
 notification per the IMP-47B policy ("AI 호출 실패 / proposal validation
 실패 / coverage 미달 → frontend notification"). u9~u13 are out of scope.
 The helper is pure (no IO, no AI call) so synthetic record / invariant
 dicts exercise every branch directly.
 """
 from __future__ import annotations
 from src.phase_z2_pipeline import _summarize_ai_repair_status
 def _record(
    *,
    unit_index: int = 0,
    apply_status: str | None = None,
    error: str | None = None,
    source_section_ids: list[str] | None = None,
 ) -> dict:
    """Minimal Step 12 AI repair record stub — fields u8 reads."""
    return {
        "unit_index": unit_index,
        "source_section_ids": source_section_ids or [f"MOCK_S{unit_index}"],
        "apply_status": apply_status,
        "error": error,
    }
 _OK_COVERAGE = {"status": "ok", "dropped_section_ids": []}
 _VIOLATED_COVERAGE = {"status": "violated", "dropped_section_ids": ["MOCK_S2"]}
 # ─── Case 1 : empty pipeline → status='ok' ──────────────────────────
 def test_empty_records_returns_ok_no_human_review():
    """No AI work executed → status='ok', human_review_required=False.
    The flag-off default (no provisional units) lands here."""
    result = _summarize_ai_repair_status([], _OK_COVERAGE)
    assert result["status"] == "ok"
    assert result["human_review_required"] is False
    assert result["counts"]["total"] == 0
    assert result["unsupported_kind_records"] == []
    assert result["error_records"] == []
    assert result["dropped_section_ids"] == []
 # ─── Case 2 : applied → status='applied', no human_review ───────────
 def test_applied_partial_overrides_marks_applied_no_human_review():
    """Successful AI repair (PARTIAL_OVERRIDES applied) is the happy
    path. status='applied', no human_review surfacing."""
    records = [_record(apply_status="applied:partial_overrides")]
    result = _summarize_ai_repair_status(records, _OK_COVERAGE)
    assert result["status"] == "applied"
    assert result["human_review_required"] is False
    assert result["counts"]["applied"] == 1
    assert result["counts"]["error"] == 0
 # ─── Case 3 : unsupported kind → status='unsupported_kind' ──────────
 def test_unsupported_kind_marks_human_review_required():
    """u5 surfaces ``unsupported_kind_for_reject_route:<kind>`` for
    builder_options_patch / slot_mapping_proposal. u8 must classify as
    human_review_required so the frontend renders a notification."""
    records = [
        _record(
            unit_index=1,
            apply_status="unsupported_kind_for_reject_route:builder_options_patch",
            source_section_ids=["MOCK_S1"],
        ),
    ]
    result = _summarize_ai_repair_status(records, _OK_COVERAGE)
    assert result["status"] == "unsupported_kind"
    assert result["human_review_required"] is True
    assert result["counts"]["unsupported_kind"] == 1
    assert result["unsupported_kind_records"] == [
        {
            "unit_index": 1,
            "source_section_ids": ["MOCK_S1"],
            "apply_status": "unsupported_kind_for_reject_route:builder_options_patch",
        }
    ]
 # ─── Case 4 : gather error → status='error' (highest priority) ──────
 def test_gather_error_marks_status_error_with_records():
    """``record['error']`` set means ``gather_step12_ai_repair_proposals``
    caught a router exception (AI call / validator). status='error'
    is the highest-priority failure axis."""
    records = [_record(
        unit_index=2,
        error="ValueError: missing slot 'title'",
        source_section_ids=["MOCK_S2"],
    )]
    result = _summarize_ai_repair_status(records, _OK_COVERAGE)
    assert result["status"] == "error"
    assert result["human_review_required"] is True
    assert result["counts"]["error"] == 1
    assert result["error_records"] == [
        {
            "unit_index": 2,
            "source_section_ids": ["MOCK_S2"],
            "error": "ValueError: missing slot 'title'",
        }
    ]
 # ─── Case 5 : coverage violated → status='coverage_violated' ────────
 def test_coverage_violation_surfaces_dropped_sections():
    """u7 coverage_invariant 'violated' means the AI repair dropped a
    section_id from the post-AI superset. dropped 절대 룰 — surface as
    human_review_required."""
    records = [_record(apply_status="applied:partial_overrides")]
    result = _summarize_ai_repair_status(records, _VIOLATED_COVERAGE)
    assert result["status"] == "coverage_violated"
    assert result["human_review_required"] is True
    assert result["coverage_status"] == "violated"
    assert result["dropped_section_ids"] == ["MOCK_S2"]
 # ─── Case 6 : priority order — error > coverage > unsupported ───────
 def test_error_dominates_over_coverage_and_unsupported():
    """When multiple failure axes coexist, priority order is
    error > coverage_violated > unsupported_kind > applied > ok."""
    records = [
        _record(unit_index=0, error="RuntimeError"),
        _record(unit_index=1,
                apply_status="unsupported_kind_for_reject_route:slot_mapping_proposal"),
        _record(unit_index=2, apply_status="applied:partial_overrides"),
    ]
    result = _summarize_ai_repair_status(records, _VIOLATED_COVERAGE)
    assert result["status"] == "error"
    assert result["human_review_required"] is True
    assert result["counts"]["error"] == 1
    assert result["counts"]["unsupported_kind"] == 1
    assert result["counts"]["applied"] == 1
 # ─── Case 7 : no_proposal + no_zone_match counted, not failure ──────
 def test_no_proposal_and_no_zone_match_do_not_trigger_human_review():
    """Flag-off short-circuit, not_provisional, route_not_ai_adaptation,
    and B4-mismatch (no_zone_match) are structural skips — not AI
    failures. They count but do not flip human_review_required."""
    records = [
        _record(unit_index=0, apply_status="no_proposal"),
        _record(unit_index=1, apply_status="no_zone_match"),
    ]
    result = _summarize_ai_repair_status(records, _OK_COVERAGE)
    assert result["status"] == "ok"
    assert result["human_review_required"] is False
    assert result["counts"]["no_proposal"] == 1
    assert result["counts"]["no_zone_match"] == 1
@@ -1,304 +0,0 @@
 """IMP-47B u12 — Initial plan_composition allow_provisional_fill for mixed direct+reject.
 Scope (this slice):
  The u12 glue inserted in ``run_phase_z2_mvp1`` (src/phase_z2_pipeline.py,
  right after the initial plan_composition + telemetry build, before the
  Step 7-A layout override block) detects the mixed direct+reject case
  (initial plan_composition returns a viable layout but some sections
  remain uncovered) and re-runs plan_composition with:
    * a lookup_fn that passes ``allow_provisional=True`` (so chain_exhausted
      sections synthesize a provisional rank-1 V4Match), and
    * ``allow_provisional_fill=True`` (so uncovered sections receive a
      last-resort provisional candidate fill in select_composition_units).
  This admits the mixed direct+reject case to the AI repair path
  (IMP-47B u4/u5) on first render — the reject section becomes a
  provisional unit (``provisional=True`` + ``label="reject"``) which Step
  12's reject route gather (u4) routes to AI fallback.
 Gate predicates (mirrored from src/phase_z2_pipeline.py u12 block):
  * units non-empty  (all-reject case is handled by IMP-30 u4 retry below)
  * layout_preset is not None
  * not override_section_assignments  (operator override bypasses the gate)
  * at least one section_id is uncovered after initial pass
 Guardrails proven by these tests:
  * MDX 원문 100% 보존 — every section_id covered after mixed admission
    (no silent drop).
  * 자동 frame swap 금지 — mixed admission only re-runs plan_composition
    with provisional flags; rank-1 reject judgment is preserved as the
    provisional V4Match (no template_id swap to a different rank).
  * Normal-path AI=0 — the mixed admission still emits the reject label;
    AI activation is gated separately in router (config.py:19 default OFF).
  * All-direct slides are a no-op — gate skips when no uncovered sections.
 This test file exercises ``plan_composition`` directly with synthetic
 stub V4 matches + a stub lookup_fn that mirrors the u12 retry seam.
 Stub naming follows the IMP-30 u3 convention (MOCK_ prefix mandatory,
 no real catalog template_id / frame_id leakage).
 """
 from __future__ import annotations
 from dataclasses import dataclass
 from typing import Optional
 from src.phase_z2_composition import plan_composition
 # ─── Synthetic V4Match duck-type (mirrors IMP-30 _StubV4Match) ───────────
@dataclass
 class _StubV4Match:
    template_id: str
    frame_id: str
    frame_number: int
    confidence: float
    label: str
    v4_rank: Optional[int] = None
    selection_path: str = "rank_1"
    fallback_reason: Optional[str] = None
    provisional: bool = False
@dataclass
 class _StubSection:
    section_id: str
    title: str = ""
    raw_content: str = ""
 _LABEL_TO_STATUS = {
    "use_as_is": "matched_zone",
    "light_edit": "adapt_matched_zone",
    "restructure": "extract_matched_zone",
    "reject": "fallback_candidate",
 }
 _ALLOWED_STATUSES = {"matched_zone", "adapt_matched_zone"}
 def _make_normal_lookup(matches_by_section: dict[str, _StubV4Match]):
    """Lookup_fn that returns the synthetic rank-1 match (no provisional path).
    Mirrors the pipeline initial ``lookup_fn`` at
    src/phase_z2_pipeline.py:3456-3465 (no ``allow_provisional`` kwarg).
    """
    def _fn(section_id: str):
        return matches_by_section.get(section_id)
    return _fn
 def _make_provisional_lookup(matches_by_section: dict[str, _StubV4Match]):
    """Lookup_fn that flags reject rank-1 matches provisional.
    Mirrors the pipeline u12 retry ``_lookup_fn_mixed_admission`` at the
    inserted block — for reject judgments, returns a provisional=True
    rank-1 V4Match-shaped stub so plan_composition's last-resort fill
    pool can see it (provisional candidates are otherwise filtered out
    of the normal greedy pass).
    """
    def _fn(section_id: str):
        m = matches_by_section.get(section_id)
        if m is not None and m.label == "reject":
            # Synthesize the provisional shape that
            # lookup_v4_match_with_fallback returns when allow_provisional
            # is True: provisional=True + selection_path="provisional_rank_1".
            return _StubV4Match(
                template_id=m.template_id,
                frame_id=m.frame_id,
                frame_number=m.frame_number,
                confidence=m.confidence,
                label=m.label,
                v4_rank=1,
                selection_path="provisional_rank_1",
                provisional=True,
            )
        return m
    return _fn
 def _make_candidates_lookup_empty():
    def _fn(section_id: str):
        return []
    return _fn
 # ─── u12 case 1 : mechanic — mixed admission via provisional lookup + fill ────
 def test_u12_mechanic_mixed_admission_covers_reject_section_via_provisional_fill():
    """Positive proof. Mixed direct+reject (S1=use_as_is, S2=reject).
    Without u12 (initial path: normal lookup + allow_provisional_fill=False),
    plan_composition returns only the S1 unit and S2 is silently dropped.
    With u12 (retry: provisional lookup + allow_provisional_fill=True),
    plan_composition returns both units; S2 is a provisional unit with
    label="reject" — ready to be picked up by Step 12's reject route
    gather (IMP-47B u4).
    """
    sections = [_StubSection("S1"), _StubSection("S2")]
    matches = {
        "S1": _StubV4Match(
            template_id="MOCK_template_direct_a",
            frame_id="MOCK_frame_001",
            frame_number=1,
            confidence=0.92,
            label="use_as_is",
            v4_rank=1,
        ),
        "S2": _StubV4Match(
            template_id="MOCK_template_reject_a",
            frame_id="MOCK_frame_002",
            frame_number=2,
            confidence=0.30,
            label="reject",
            v4_rank=1,
        ),
    }
    # Pre-u12 baseline — normal lookup, no provisional fill.
    units_pre, preset_pre, _ = plan_composition(
        sections,
        _make_normal_lookup(matches),
        _LABEL_TO_STATUS,
        _ALLOWED_STATUSES,
        v4_candidates_lookup_fn=_make_candidates_lookup_empty(),
    )
    covered_pre = {sid for u in units_pre for sid in u.source_section_ids}
    assert "S1" in covered_pre, "S1 (use_as_is) must cover pre-u12"
    assert "S2" not in covered_pre, (
        "Pre-u12 baseline regression: reject S2 should be uncovered (no provisional fill)"
    )
    # u12 mixed-admission retry — provisional lookup + allow_provisional_fill=True.
    units_post, preset_post, _ = plan_composition(
        sections,
        _make_provisional_lookup(matches),
        _LABEL_TO_STATUS,
        _ALLOWED_STATUSES,
        v4_candidates_lookup_fn=_make_candidates_lookup_empty(),
        allow_provisional_fill=True,
    )
    covered_post = {sid for u in units_post for sid in u.source_section_ids}
    assert covered_post == {"S1", "S2"}, (
        "u12 mixed admission must cover every section (no text loss)"
    )
    assert preset_post is not None
    # The S2 unit must be marked provisional so the reject route gather
    # (src/phase_z2_ai_fallback/step12.py:133-136) admits it.
    s2_unit = next(u for u in units_post if "S2" in u.source_section_ids)
    assert s2_unit.provisional is True, (
        "Reject S2 unit must be provisional so Step 12 reject route admits it"
    )
    assert s2_unit.label == "reject"
    # Frame template id is preserved — no auto frame swap.
    assert s2_unit.frame_template_id == "MOCK_template_reject_a"
 # ─── u12 case 2 : gate — all-direct slides are a no-op ──────────────────────
 def test_u12_gate_all_direct_yields_no_uncovered_sections():
    """No-op proof. When every section is auto-renderable (use_as_is or
    light_edit), the initial plan_composition covers everything — the
    u12 mixed-admission gate's ``_u12_uncovered_ids`` list is empty and
    the retry is skipped.
    """
    sections = [_StubSection("S1"), _StubSection("S2")]
    matches = {
        "S1": _StubV4Match(
            template_id="MOCK_template_direct_a",
            frame_id="MOCK_frame_001",
            frame_number=1,
            confidence=0.92,
            label="use_as_is",
            v4_rank=1,
        ),
        "S2": _StubV4Match(
            template_id="MOCK_template_direct_b",
            frame_id="MOCK_frame_002",
            frame_number=2,
            confidence=0.81,
            label="light_edit",
            v4_rank=1,
        ),
    }
    units, preset, _ = plan_composition(
        sections,
        _make_normal_lookup(matches),
        _LABEL_TO_STATUS,
        _ALLOWED_STATUSES,
        v4_candidates_lookup_fn=_make_candidates_lookup_empty(),
    )
    covered = {sid for u in units for sid in u.source_section_ids}
    assert covered == {"S1", "S2"}, "All-direct must cover every section pre-u12"
    # Predicate from src/phase_z2_pipeline.py u12 block:
    uncovered = [s.section_id for s in sections if s.section_id not in covered]
    assert uncovered == [], (
        "u12 gate must classify all-direct as no-op (uncovered list empty)"
    )
    assert preset is not None
 # ─── u12 case 3 : gate — initial empty units bypass u12 (IMP-30 retry owns it) ──
 def test_u12_gate_skips_when_initial_units_empty():
    """All-reject case is owned by IMP-30 u4 retry (units=[] guard at
    src/phase_z2_pipeline.py:3646). u12 mixed-admission must NOT compete
    with that path; the gate ``units and layout_preset is not None``
    short-circuits when the initial plan_composition returns nothing.
    """
    sections = [_StubSection("S1")]
    matches = {
        "S1": _StubV4Match(
            template_id="MOCK_template_reject_a",
            frame_id="MOCK_frame_002",
            frame_number=2,
            confidence=0.30,
            label="reject",
            v4_rank=1,
        ),
    }
    units, preset, _ = plan_composition(
        sections,
        _make_normal_lookup(matches),
        _LABEL_TO_STATUS,
        _ALLOWED_STATUSES,
        v4_candidates_lookup_fn=_make_candidates_lookup_empty(),
    )
    # All-reject initial pass: no auto-renderable units, no layout preset.
    assert units == [] and preset is None
    # u12 gate predicate would short-circuit on `units` truthiness:
    gate_active = bool(units) and preset is not None
    assert gate_active is False, (
        "u12 mixed-admission gate must skip the all-reject case (IMP-30 u4 owns it)"
    )
 # ─── u12 case 4 : code-path anchor — pipeline source contains u12 marker ────
 def test_u12_pipeline_source_contains_mixed_admission_marker():
    """Anchor test. Ensures the inserted u12 block in src/phase_z2_pipeline.py
    is reachable (not silently removed by a future refactor).
    Asserts on the marker comment + ``imp47b_u12_mixed_admission`` debug key
    + ``allow_provisional_fill=True`` invocation co-located in the file.
    Cheap structural guard — does not run the heavy pipeline.
    """
    from pathlib import Path
    src_path = Path(__file__).resolve().parent.parent / "src" / "phase_z2_pipeline.py"
    text = src_path.read_text(encoding="utf-8")
    assert "IMP-47B u12 — mixed direct+reject first-render admission" in text, (
        "u12 marker comment missing from pipeline — block may have been removed"
    )
    assert "imp47b_u12_mixed_admission" in text, (
        "u12 comp_debug telemetry key missing"
    )
    # The mixed-admission retry must pass allow_provisional_fill=True.
    # Anchor against the helper function name + the kwarg co-occurrence.
    assert "_lookup_fn_mixed_admission" in text
    assert "allow_provisional_fill=True" in text
@@ -1,180 +0,0 @@
 """IMP-47B u3 — override-selected reject frames are admitted as provisional.
 Scope (this slice):
  Helper `_apply_frame_override_to_unit` (src/phase_z2_pipeline.py) covers
  the three probe layers used by the `--override-frame` path:
    1. ``v4_candidates`` exact match (non-reject; existing behaviour).
    2. Full 32 V4 judgments probe (reject inclusive) — when the user
       picks a reject frame, the unit is promoted to
       ``provisional=True`` with ``label="reject"`` so Step 12
       (IMP-47B u4) admits the AI repair path.
    3. Raw fall-through (template_id only) — no provisional promotion,
       no label mutation.
 Frame visual / contract stay untouched per the AI isolation contract
 (frame auto-swap forbidden — AI re-places content into the existing
 frame only). Sibling test confirms a non-reject override still goes
 through the v4_candidates path without provisional promotion.
 Synthetic naming convention mirrors tests/test_phase_z2_imp30_first_render.py
 (MOCK_ prefix mandatory, no real catalog template_id / frame_id leakage).
 """
 from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import Optional
 from src.phase_z2_pipeline import _apply_frame_override_to_unit
@dataclass
 class _StubCandidate:
    template_id: str
    frame_id: str
    frame_number: int
    confidence: float
    label: str
@dataclass
 class _StubUnit:
    source_section_ids: list[str]
    frame_template_id: Optional[str] = None
    frame_id: Optional[str] = None
    frame_number: int = 0
    confidence: float = 0.0
    label: Optional[str] = None
    provisional: bool = False
    v4_candidates: list = field(default_factory=list)
 def _v4_with_reject(section_id: str, target_tid: str) -> dict:
    """Synthetic V4 dict with target_tid mapped to a reject judgment.
    Mirrors the production V4 schema surface (``mdx_sections`` →
    ``judgments_full32`` → list of judgment dicts with template_id /
    frame_id / frame_number / confidence / label). Two judgments so we
    can also assert that the helper picks the reject entry rather than
    the first non-reject one when the template_ids differ.
    """
    return {
        "mdx_sections": {
            section_id: {
                "judgments_full32": [
                    {
                        "template_id": "MOCK_T_other",
                        "frame_id": "F_other",
                        "frame_number": 1,
                        "confidence": 0.85,
                        "label": "use_as_is",
                    },
                    {
                        "template_id": target_tid,
                        "frame_id": "F_reject",
                        "frame_number": 32,
                        "confidence": 0.40,
                        "label": "reject",
                    },
                ],
            },
        },
    }
 # ─── Case 1 : reject override → provisional promotion ────────────
 def test_override_to_reject_judgment_marks_unit_provisional():
    """User picks a reject frame → unit.label=reject, provisional=True.
    Frame metadata is sourced from the reject judgment (frame_id /
    frame_number / confidence) so Step 9 metadata stays consistent.
    """
    unit = _StubUnit(
        source_section_ids=["MOCK_S1"],
        frame_template_id="MOCK_T_auto",
        frame_id="F_auto",
        frame_number=5,
        confidence=0.90,
        label="use_as_is",
        provisional=False,
    )
    v4 = _v4_with_reject("MOCK_S1", "MOCK_T_reject")
    meta = _apply_frame_override_to_unit(unit, "MOCK_T_reject", v4)
    assert meta == "v4_reject_judgment_provisional"
    assert unit.frame_template_id == "MOCK_T_reject"
    assert unit.frame_id == "F_reject"
    assert unit.frame_number == 32
    assert unit.confidence == 0.40
    assert unit.label == "reject"
    assert unit.provisional is True
 # ─── Case 2 : non-reject override → existing v4_candidates path ───
 def test_override_to_v4_candidate_keeps_non_provisional():
    """User picks a non-reject candidate → existing v4_candidates path.
    Helper takes the early v4_candidates branch without consulting the
    full 32 judgments. provisional remains False (normal-path AI=0
    contract — IMP-30 / IMP-47B router gate intact for this unit).
    """
    unit = _StubUnit(
        source_section_ids=["MOCK_S2"],
        frame_template_id="MOCK_T_auto",
        frame_id="F_auto",
        frame_number=3,
        confidence=0.95,
        label="use_as_is",
        provisional=False,
        v4_candidates=[
            _StubCandidate(
                template_id="MOCK_T_pick",
                frame_id="F_pick",
                frame_number=2,
                confidence=0.85,
                label="light_edit",
            ),
        ],
    )
    v4 = {"mdx_sections": {}}  # full-judgment probe must NOT be reached
    meta = _apply_frame_override_to_unit(unit, "MOCK_T_pick", v4)
    assert meta == "v4_candidates"
    assert unit.frame_template_id == "MOCK_T_pick"
    assert unit.frame_id == "F_pick"
    assert unit.label == "light_edit"
    assert unit.provisional is False
 # ─── Case 3 : unknown template → raw fall-through (no provisional) ─
 def test_override_unknown_template_falls_through_without_provisional():
    """Template ID absent from v4_candidates AND from judgments_full32 →
    raw_template_id_only path. No provisional flag, no label change.
    """
    unit = _StubUnit(
        source_section_ids=["MOCK_S3"],
        frame_template_id="MOCK_T_auto",
        frame_id="F_auto",
        frame_number=4,
        confidence=0.92,
        label="use_as_is",
        provisional=False,
    )
    v4 = {"mdx_sections": {}}
    meta = _apply_frame_override_to_unit(unit, "MOCK_T_unknown", v4)
    assert meta == "raw_template_id_only"
    assert unit.frame_template_id == "MOCK_T_unknown"
    # frame_id / label unchanged — caller's print path warns on this case.
    assert unit.frame_id == "F_auto"
    assert unit.label == "use_as_is"
    assert unit.provisional is False
@@ -1,223 +0,0 @@
 """IMP-47B u5 — PARTIAL_OVERRIDES apply tests.
 Scope (this slice):
  Helper ``_apply_ai_repair_proposals_to_zones`` (src/phase_z2_pipeline.py)
  merges ``proposal.payload.slots`` into ``zones_data[k]["slot_payload"]``
  for PARTIAL_OVERRIDES proposals only, and loud-fails out-of-scope
  proposal kinds (builder_options_patch, slot_mapping_proposal) with an
  explicit ``apply_status`` marker.
 The IMP-33 u5 validator inside ``route_ai_fallback`` already enforces
 declared-slot completeness — the apply helper is therefore a structural
 merge over the validator's contract, not a per-slot guard re-implementation.
 u6 (step12_ai_repair.json audit), u7 (coverage invariant), and u8
 (slide_status surfacing) are out of scope for this unit.
 """
 from __future__ import annotations
 from src.phase_z2_pipeline import _apply_ai_repair_proposals_to_zones
 def _record(
    *,
    unit_index: int,
    proposal: dict | None,
    source_section_ids: list[str] | None = None,
 ) -> dict:
    """Synthetic gather_step12_ai_repair_proposals record."""
    return {
        "unit_index": unit_index,
        "source_section_ids": source_section_ids or [f"MOCK_S{unit_index}"],
        "frame_template_id": "MOCK_T",
        "label": "reject",
        "route_hint": "ai_adaptation_required",
        "provisional": True,
        "ai_called": proposal is not None,
        "skip_reason": None,
        "proposal": proposal,
        "error": None,
        "cache_key": "MOCK_F::abc" if proposal is not None else None,
        "fingerprints": {"contract_sha": "x", "partial_sha": "y", "catalog_sha": ""}
        if proposal is not None
        else None,
    }
 def _zone(*, position: str, slot_payload: dict | None = None) -> dict:
    """Synthetic zones_data entry — only fields the apply helper touches."""
    return {
        "position": position,
        "template_id": "MOCK_T",
        "slot_payload": slot_payload if slot_payload is not None else {},
    }
 # ─── Case 1 : PARTIAL_OVERRIDES → merged + applied marker ──────────
 def test_partial_overrides_merges_slots_into_zone_slot_payload():
    """The validator already guarantees declared-slot completeness, so
    apply is a structural ``dict.update``. Pre-existing meta keys
    (``_truncated_count``) survive; declared slot values are replaced
    by the AI proposal values."""
    proposal = {
        "proposal_kind": "partial_overrides",
        "payload": {
            "slots": {
                "title": "AI title",
                "bullets": ["AI bullet 1", "AI bullet 2"],
            }
        },
        "rationale": "MOCK",
    }
    records = [_record(unit_index=0, proposal=proposal)]
    zones = [
        _zone(
            position="top",
            slot_payload={
                "title": "deterministic title",
                "bullets": ["det bullet"],
                "_truncated_count": 0,
            },
        )
    ]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    assert records[0]["apply_status"] == "applied:partial_overrides"
    assert zones[0]["slot_payload"]["title"] == "AI title"
    assert zones[0]["slot_payload"]["bullets"] == ["AI bullet 1", "AI bullet 2"]
    # meta keys not in proposal must survive the merge
    assert zones[0]["slot_payload"]["_truncated_count"] == 0
 # ─── Case 2 : BUILDER_OPTIONS_PATCH → loud-fail unsupported_kind ───
 def test_builder_options_patch_is_unsupported_for_reject_route():
    """Builder-options application is out-of-scope for IMP-47B reject
    route (see Stage 2 plan). u5 must mark, not apply — the zone
    slot_payload stays byte-identical and the record carries the
    ``unsupported_kind_for_reject_route:<kind>`` marker so u8 can
    surface human_review downstream."""
    proposal = {
        "proposal_kind": "builder_options_patch",
        "payload": {"font_size_px": 14},
        "rationale": "MOCK",
    }
    records = [_record(unit_index=0, proposal=proposal)]
    original_slot_payload = {"title": "deterministic"}
    zones = [_zone(position="top", slot_payload=dict(original_slot_payload))]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    assert (
        records[0]["apply_status"]
        == "unsupported_kind_for_reject_route:builder_options_patch"
    )
    assert zones[0]["slot_payload"] == original_slot_payload
 # ─── Case 3 : SLOT_MAPPING_PROPOSAL → loud-fail unsupported_kind ───
 def test_slot_mapping_proposal_is_unsupported_for_reject_route():
    """Slot-mapping (restructuring) application is also out-of-scope —
    builder-options + slot-mapping share the same marker path."""
    proposal = {
        "proposal_kind": "slot_mapping_proposal",
        "payload": {"slots": {"title": "x"}},
        "rationale": "MOCK",
    }
    records = [_record(unit_index=0, proposal=proposal)]
    zones = [_zone(position="top", slot_payload={"title": "deterministic"})]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    assert (
        records[0]["apply_status"]
        == "unsupported_kind_for_reject_route:slot_mapping_proposal"
    )
    assert zones[0]["slot_payload"] == {"title": "deterministic"}
 # ─── Case 4 : no proposal (router short-circuit / not_provisional) ──
 def test_record_without_proposal_marked_no_proposal_and_zone_untouched():
    """Flag-off short-circuit and non-AI-route units carry
    ``proposal=None``. apply_status must distinguish "no proposal to
    apply" from real apply outcomes so u8 can categorise the per-unit
    status without re-reading skip_reason."""
    records = [_record(unit_index=0, proposal=None)]
    zones = [_zone(position="top", slot_payload={"title": "deterministic"})]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    assert records[0]["apply_status"] == "no_proposal"
    assert zones[0]["slot_payload"] == {"title": "deterministic"}
 # ─── Case 5 : proposal exists but no matching zone (B4 mismatch) ────
 def test_proposal_for_unit_without_zone_match_marked_no_zone_match():
    """When a unit is dropped from zones_data (B4 mismatch or FitError
    in the Step 12 render loop) but still gathered an AI proposal,
    apply must surface the mismatch via ``no_zone_match`` rather than
    silently dropping the proposal or writing into a wrong zone."""
    proposal = {
        "proposal_kind": "partial_overrides",
        "payload": {"slots": {"title": "AI title"}},
        "rationale": "MOCK",
    }
    records = [_record(unit_index=0, proposal=proposal)]
    # unit_positions[0]="top" but zones_data has only the bottom zone
    # → no match for the dropped unit's position.
    zones = [_zone(position="bottom", slot_payload={"title": "other zone"})]
    _apply_ai_repair_proposals_to_zones(records, ["top"], zones)
    assert records[0]["apply_status"] == "no_zone_match"
    # untouched zone — apply must not bleed into a different position
    assert zones[0]["slot_payload"] == {"title": "other zone"}
 # ─── Case 6 : mixed records — independent per-record classification ──
 def test_mixed_records_classified_independently():
    """All five apply_status branches coexist in one batch — confirms
    the helper does not short-circuit on the first non-applied record."""
    records = [
        _record(unit_index=0, proposal={
            "proposal_kind": "partial_overrides",
            "payload": {"slots": {"title": "AI"}},
            "rationale": "",
        }),
        _record(unit_index=1, proposal={
            "proposal_kind": "builder_options_patch",
            "payload": {"font_size_px": 14},
            "rationale": "",
        }),
        _record(unit_index=2, proposal=None),
    ]
    zones = [
        _zone(position="top", slot_payload={"title": "det"}),
        _zone(position="middle", slot_payload={"title": "det"}),
        _zone(position="bottom", slot_payload={"title": "det"}),
    ]
    _apply_ai_repair_proposals_to_zones(
        records, ["top", "middle", "bottom"], zones,
    )
    assert [r["apply_status"] for r in records] == [
        "applied:partial_overrides",
        "unsupported_kind_for_reject_route:builder_options_patch",
        "no_proposal",
    ]
    assert zones[0]["slot_payload"]["title"] == "AI"
    assert zones[1]["slot_payload"]["title"] == "det"
    assert zones[2]["slot_payload"]["title"] == "det"
@@ -1,154 +0,0 @@
 """IMP-47B u4 + u6 — Step 12 AI repair wiring + audit artifact tests.
 Scope (this slice):
  * u4 — Helper ``_run_step12_ai_repair`` (src/phase_z2_pipeline.py)
    wires the pipeline's local route-hint helper (``_imp05_route_hint``),
    the frame contract loader (``get_contract``), and a
    templates/phase_z2/families partial reader
    (``_load_frame_partial_html``) into
    ``gather_step12_ai_repair_proposals``.
  * u6 — The gather records flow into ``_write_step_artifact`` under
    ``step12_ai_repair.json``. The audit shape must stay
    JSON-serialisable (no Pydantic / dataclass leakage) so the artifact
    write never raises on real runs.
 The router short-circuits when ``settings.ai_fallback_enabled`` is
 False (default), so AI=0 for non-AI-route units stays a structural
 guarantee. Synthetic naming mirrors tests/test_imp47b_override_provisional.py
 (MOCK_ prefix; no real catalog template_id / frame_id leakage).
 u5 (PARTIAL_OVERRIDES apply), u7 (coverage invariant), and u8
 (slide_status surfacing) are out of scope for this unit.
 """
 from __future__ import annotations
 import json
 from dataclasses import dataclass, field
 from src.phase_z2_pipeline import (
    _load_frame_partial_html,
    _run_step12_ai_repair,
    _write_step_artifact,
 )
@dataclass
 class _StubUnit:
    label: str | None
    provisional: bool
    frame_template_id: str = "MOCK_T_x"
    frame_id: str = "MOCK_F_x"
    source_section_ids: list[str] = field(default_factory=lambda: ["MOCK_S1"])
    raw_content: str = "MOCK_raw"
    v4_rank: int | None = 1
    cardinality: int | None = None
    layout_preset: str = ""
    zone_position: str = ""
    source_shape: str = "paragraph"
    h3_count: int = 0
    char_count: int = 0
 # ─── Case 1 : mixed units → per-unit skip_reason classification ─────
 def test_mixed_units_classified_by_route_and_provisional_flag():
    """Reject + restructure provisional both route to ai_adaptation;
    use_as_is / light_edit / non-provisional skip without router call.
    With ai_fallback_enabled=False (default) the router returns None,
    so the two ai_adaptation provisional units record
    ``skip_reason='router_short_circuit'``; the rest record their
    structural skip_reason (not_provisional / route_not_ai_adaptation).
    """
    units = [
        _StubUnit(label="use_as_is", provisional=False),
        _StubUnit(label="light_edit", provisional=True),
        _StubUnit(label="restructure", provisional=True),
        _StubUnit(label="reject", provisional=True),
        _StubUnit(label="restructure", provisional=False),
    ]
    records = _run_step12_ai_repair(units)
    assert [r["skip_reason"] for r in records] == [
        "not_provisional",
        "route_not_ai_adaptation:deterministic_minor_adjustment",
        "router_short_circuit",
        "router_short_circuit",
        "not_provisional",
    ]
    assert [r["route_hint"] for r in records] == [
        "direct_render",
        "deterministic_minor_adjustment",
        "ai_adaptation_required",
        "ai_adaptation_required",
        "ai_adaptation_required",
    ]
    assert all(r["ai_called"] is False for r in records)
 # ─── Case 2 : reject provisional unit reaches AI gate ───────────────
 def test_reject_provisional_unit_reaches_router_short_circuit():
    """Reject + provisional → route_hint=ai_adaptation_required.
    Router short-circuit (flag-off default) is the only thing keeping
    AI from firing; the wiring proves reject is no longer blocked by
    Step 12's bespoke design_reference_only skip (removed by u2).
    """
    records = _run_step12_ai_repair([_StubUnit(label="reject", provisional=True)])
    assert records[0]["route_hint"] == "ai_adaptation_required"
    assert records[0]["skip_reason"] == "router_short_circuit"
    assert records[0]["ai_called"] is False
    # cache_key / fingerprints populated only after the route + provisional
    # gates pass — confirms gather reached the AI-eligible code path.
    assert records[0]["cache_key"] is not None
    assert records[0]["fingerprints"] is not None
 # ─── Case 3 : frame visual loader degrades on missing partial ──────
 def test_load_frame_partial_html_returns_empty_for_missing_file():
    """__empty__ shell (IMP-30) and any unknown template_id → "".
    Keeps gather() crash-free for the IMP-30 first-render-invariant
    path where the synthesized empty-shell unit has no families partial.
    """
    assert _load_frame_partial_html("__empty__") == ""
    assert _load_frame_partial_html("MOCK_T_does_not_exist") == ""
 # ─── Case 4 (u6) : audit artifact write is JSON-serialisable ────────
 def test_step12_ai_repair_artifact_writes_json_serialisable_records(tmp_path):
    """IMP-47B u6 — gather records feed ``_write_step_artifact`` as the
    ``step12_ai_repair.json`` audit. Confirms the gather schema contains
    only JSON-native primitives (str / int / None / bool / list / dict)
    so the artifact write never raises on real runs and the audit
    payload preserves per-unit ``route_hint`` / ``skip_reason`` /
    ``ai_called`` for reviewers.
    """
    records = _run_step12_ai_repair([
        _StubUnit(label="reject", provisional=True),
        _StubUnit(label="use_as_is", provisional=False),
    ])
    fpath = _write_step_artifact(
        tmp_path, 12, "ai_repair",
        data={"per_unit": records},
        outputs=["step12_ai_repair.json"],
    )
    assert fpath.is_file()
    assert fpath.name == "step12_ai_repair.json"
    payload = json.loads(fpath.read_text(encoding="utf-8"))
    assert payload["step_num"] == 12
    assert payload["step_name"] == "ai_repair"
    assert payload["step_status"] == "done"
    per_unit = payload["data"]["per_unit"]
    assert len(per_unit) == 2
    assert per_unit[0]["route_hint"] == "ai_adaptation_required"
    assert per_unit[0]["skip_reason"] == "router_short_circuit"
    assert per_unit[0]["ai_called"] is False
    assert per_unit[1]["route_hint"] == "direct_render"
    assert per_unit[1]["skip_reason"] == "not_provisional"
@@ -1,137 +0,0 @@
 """IMP-38 U3 regression — call site cleanup (max_rank=3 제거) 후 policy 활성 검증.
 Scenarios:
  (A) normal case: rank 1~default_max_rank window 에 usable candidate 충분
      → effective_max_rank=default_max_rank (rank-3-preserved)
      → mdx03 식: rank 1 use_as_is 매칭 정상 case 보호 확인
  (B) extended case: rank 1~default_max_rank window 에 usable candidate 0
      → effective_max_rank=effective_extended_ceiling (rank-extended)
      → mdx05-2 식: rank 1~9 미등록/reject + rank 10+ 등록 frame case 처리
 4 round 합의 (#67):
  - Codex #1: 별 yaml + loader (catalog 오염 방지)
  - Codex #2: min(configured, len(judgments)) 정정
  - Codex #6: 2 call site cleanup (HEAD 기준 — IMP-47B 가 추가한 3 번째는 별 axis)
  - Codex #7: U3 execute ready
 """
 from __future__ import annotations
 import pytest
@pytest.fixture(autouse=True)
 def _reset_policy_cache():
    """Reset module-level _V4_FALLBACK_POLICY_CACHE for test isolation."""
    import src.phase_z2_mapper as mapper
    mapper._V4_FALLBACK_POLICY_CACHE = None
    yield
    mapper._V4_FALLBACK_POLICY_CACHE = None
 def _make_v4_section(judgments: list[dict]) -> dict:
    return {"mdx_sections": {"sec-1": {"judgments_full32": judgments}}}
 def _judgment(template_id: str, label: str, confidence: float = 0.5, frame_id: int = 0) -> dict:
    return {
        "template_id": template_id,
        "frame_id": frame_id or (hash(template_id) % 10000),
        "frame_number": 0,
        "confidence": confidence,
        "label": label,
    }
 # ─── Scenario A — normal case (rank-3-preserved) ──────────────────
 def test_normal_case_with_usable_candidates_preserves_default_max_rank():
    """rank 1~3 window 에 usable >= threshold(1) 시 effective_max_rank=default_max_rank(3)."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    from src.phase_z2_mapper import load_frame_contracts
    # mdx03 식 — 첫 rank 가 catalog 등록 + use_as_is/light_edit/restructure(allowed)
    # 실제 catalog 등록 frame 사용 (catalog hardcode 의존 — 단 frame 32 중 어느 게 등록인지는 yaml 기반)
    catalog = load_frame_contracts()
    registered_template_ids = [k for k, v in catalog.items() if isinstance(v, dict)]
    assert len(registered_template_ids) >= 1, "catalog 등록 frame 1+ 필요 (mdx03 식 fixture)"
    # rank 1 = registered frame + use_as_is (auto-renderable)
    # rank 2~3 = reject (catalog 등록 무관)
    first_registered = registered_template_ids[0]
    judgments = [
        _judgment(first_registered, "use_as_is", 0.95),
        _judgment("dummy_rank2", "reject", 0.3),
        _judgment("dummy_rank3", "reject", 0.2),
    ]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1")  # no explicit max_rank → policy
    assert trace["policy_applied"] == "default_max_rank", (
        f"normal case 에서 default 유지 기대, got {trace['policy_applied']}"
    )
    assert trace["effective_max_rank"] == trace["default_max_rank"]
    assert trace["usable_count"] >= 1
 # ─── Scenario B — extended case (rank-extended) ────────────────────
 def test_extended_case_with_no_usable_in_default_window_expands_to_ceiling():
    """rank 1~3 window 에 0 usable 시 effective_max_rank=effective_extended_ceiling."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    # mdx05-2 식 — rank 1~3 미등록 (template_id 가 catalog 에 없음) + reject 라벨
    # rank 4~ 도 등록 안 됨 (fixture 단순화)
    # 다만 judgments_count=10 으로 충분 → effective_extended_ceiling = min(extended, 10) = 10
    judgments = [
        _judgment(f"unregistered_t{i}", "reject", 0.1 + i * 0.01) for i in range(10)
    ]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1")
    assert trace["policy_applied"] == "extended_max_rank", (
        f"extended case 기대, got {trace['policy_applied']}"
    )
    assert trace["usable_count"] == 0
    assert trace["judgments_count"] == 10
    # Codex #2 정정: min(configured, 10) — configured 32 면 10, 5 면 5
    assert trace["effective_extended_ceiling"] == min(
        trace["configured_extended_max_rank"], 10
    )
    assert trace["effective_max_rank"] == trace["effective_extended_ceiling"]
 # ─── Scenario C — call site cleanup byte-identical (caller_override 제거 후 policy 활성) ─
 def test_default_call_site_now_uses_policy_after_cleanup():
    """U3 cleanup 후 call site = no explicit max_rank → policy path 자동 활성.
    이전: caller 가 max_rank=3 명시 → policy_applied=caller_override
    U3 후: caller 가 명시 X → policy_applied=default_max_rank (usable >= 1 시) or extended_max_rank
    """
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    judgments = [_judgment(f"unregistered_t{i}", "reject") for i in range(5)]
    v4 = _make_v4_section(judgments)
    # caller 가 max_rank 명시 X (U3 cleanup 후 production caller 의 새 동작)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1")
    assert trace["policy_applied"] in {"default_max_rank", "extended_max_rank"}
    assert trace["policy_applied"] != "caller_override", (
        "U3 cleanup 후 production caller = no explicit, policy path 활성 기대"
    )
 # ─── Scenario D — explicit caller_override 여전히 동작 (test path 보호) ────
 def test_explicit_caller_override_still_works_for_tests():
    """test 에서 explicit max_rank=N 보낼 시 caller_override 그대로 동작 (backward compat)."""
    from src.phase_z2_pipeline import lookup_v4_match_with_fallback
    judgments = [_judgment(f"unregistered_t{i}", "reject") for i in range(10)]
    v4 = _make_v4_section(judgments)
    _match, trace = lookup_v4_match_with_fallback(v4, "sec-1", max_rank=5)
    assert trace["policy_applied"] == "caller_override"
    assert trace["effective_max_rank"] == 5
@@ -1,109 +0,0 @@
 """IMP-38 U1 — v4_fallback_policy.yaml loader test.
 Verify:
  - load_v4_fallback_policy() returns dict with expected keys
  - yaml parsed correctly (usable_threshold, default_max_rank, extended_max_rank, policy_type)
  - graceful fallback when yaml missing → _V4_FALLBACK_POLICY_DEFAULT
  - _V4_FALLBACK_POLICY_CACHE pattern (lazy load, mirror of _CATALOG_CACHE)
  - load_frame_contracts() shape unchanged (separate yaml, catalog 오염 X)
 4 round 합의 (#67):
  - Codex #1: separate yaml (not frame_contracts.yaml top-level)
  - Codex #3: load_frame_contracts() shape 변경 X
 """
 from __future__ import annotations
 import importlib
 from pathlib import Path
 from unittest.mock import patch
 import pytest
 PROJECT_ROOT = Path(__file__).parent.parent
 V4_POLICY_PATH = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "v4_fallback_policy.yaml"
 CATALOG_PATH = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "frame_contracts.yaml"
 def _reset_caches():
    """Reset module-level caches for test isolation."""
    import src.phase_z2_mapper as mapper
    mapper._V4_FALLBACK_POLICY_CACHE = None
    mapper._CATALOG_CACHE = None
@pytest.fixture(autouse=True)
 def clean_caches():
    _reset_caches()
    yield
    _reset_caches()
 def test_v4_fallback_policy_yaml_exists():
    """IMP-38 U1 — separate yaml file must exist."""
    assert V4_POLICY_PATH.exists(), (
        f"v4_fallback_policy.yaml not found at {V4_POLICY_PATH}. "
        "IMP-38 U1 expects separate yaml (Codex #1 corr — not frame_contracts.yaml top-level)."
    )
 def test_load_v4_fallback_policy_returns_dict_with_expected_keys():
    """load_v4_fallback_policy() must return dict with policy keys."""
    from src.phase_z2_mapper import load_v4_fallback_policy
    policy = load_v4_fallback_policy()
    assert isinstance(policy, dict)
    expected_keys = {"policy_type", "usable_threshold", "default_max_rank", "extended_max_rank"}
    missing = expected_keys - set(policy.keys())
    assert not missing, f"missing keys in v4_fallback_policy: {missing}"
 def test_load_v4_fallback_policy_values_match_yaml():
    """Loaded policy values must match v4_fallback_policy.yaml (initial commit)."""
    from src.phase_z2_mapper import load_v4_fallback_policy
    policy = load_v4_fallback_policy()
    assert policy["policy_type"] == "dynamic_usable_count_based"
    assert policy["usable_threshold"] == 1
    assert policy["default_max_rank"] == 3
    assert policy["extended_max_rank"] == 32
 def test_load_v4_fallback_policy_cache_pattern():
    """_V4_FALLBACK_POLICY_CACHE pattern — second call returns same dict (lazy load)."""
    from src.phase_z2_mapper import load_v4_fallback_policy
    policy_a = load_v4_fallback_policy()
    policy_b = load_v4_fallback_policy()
    assert policy_a is policy_b, "cache pattern violated (should return same dict instance)"
 def test_load_v4_fallback_policy_graceful_when_yaml_missing():
    """yaml 파일 없을 시 → _V4_FALLBACK_POLICY_DEFAULT (extended_max_rank=3, byte-identical pre-IMP-38)."""
    import src.phase_z2_mapper as mapper
    with patch.object(mapper, "V4_FALLBACK_POLICY_PATH", PROJECT_ROOT / "tests" / "__nonexistent_policy.yaml"):
        # reset cache to force reload via patched path
        mapper._V4_FALLBACK_POLICY_CACHE = None
        policy = mapper.load_v4_fallback_policy()
        assert policy["default_max_rank"] == 3
        assert policy["extended_max_rank"] == 3, (
            "graceful fallback must keep extended==default (byte-identical pre-IMP-38)"
        )
 def test_load_frame_contracts_shape_unchanged():
    """Codex #3 LOCK — load_frame_contracts() must still return template_id → entry dict."""
    from src.phase_z2_mapper import load_frame_contracts, load_v4_fallback_policy
    catalog = load_frame_contracts()
    policy = load_v4_fallback_policy()
    # catalog 의 key 가 모두 frame entry (dict with template_id/frame_id) 여야 함
    for key, entry in catalog.items():
        assert isinstance(entry, dict), f"catalog entry {key} should be dict"
        assert "template_id" in entry, f"catalog entry {key} missing template_id (policy bleed?)"
    # policy keys 는 catalog 에 안 들어감
    policy_keys = {"policy_type", "usable_threshold", "default_max_rank", "extended_max_rank"}
    catalog_top_keys = set(catalog.keys())
    bleed = policy_keys & catalog_top_keys
    assert not bleed, (
        f"policy keys leaked into frame_contracts.yaml: {bleed}. "
        "Codex #1 corr violated — policy must stay in separate v4_fallback_policy.yaml."
    )
Author	SHA1	Message	Date
Kyeongmin	d777315a81	On main: stash-all-pre-IMP38-commit-20260521	2026-05-21 22:07:41 +09:00
Kyeongmin	921c8f6884	index on main: `dceb101` feat(#63 ): IMP-34 R1 donor capacity measured bound (u1+u2)	2026-05-21 22:07:41 +09:00
Kyeongmin	8f085a28d3	untracked files on main: `dceb101` feat(#63 ): IMP-34 R1 donor capacity measured bound (u1+u2)	2026-05-21 22:07:41 +09:00