On main: B4 role reframe + F29 expansion experiments — held during Option E (phase_z2_pipeline/classifier/placement_planner/content_extractor)

src/phase_z2_classifier.py

@@ -313,8 +313,6 @@ def _build_placement_diagnostic_for_zone(
             "zone_position": zone_position,
             "mapper_frame_template_id": mapper_template_id,
             "b4_selected_template_id": None,
-            "frame_selection_matches_mapper": None,
-            "frame_selection_match_note": "no placement_trace recorded",
             "region_count": 0,
             "slot_assignment_count": 0,
             "rejection_count": 0,
@@ -325,8 +323,6 @@ def _build_placement_diagnostic_for_zone(
             placement_trace.get("mapper_frame_template_id") or mapper_template_id
         ),
         "b4_selected_template_id": placement_trace.get("selected_template_id"),
-        "frame_selection_matches_mapper": placement_trace.get("frame_selection_matches_mapper"),
-        "frame_selection_match_note": placement_trace.get("frame_selection_match_note"),
         "region_count": len(placement_trace.get("internal_regions") or []),
         "slot_assignment_count": len(placement_trace.get("slot_assignments") or []),
         "rejection_count": len(placement_trace.get("rejection") or []),

src/phase_z2_content_extractor.py

@@ -216,25 +216,29 @@ def extract_content_objects(section, source_shape: Optional[str] = None) -> list
     content = section.raw_content
     section_id = section.section_id
 
-    if source_shape == "top_bullets":
+    if source_shape in ("top_bullets", "h3_subsections"):
         from phase_z2_mapper import split_source
-        units = split_source("top_bullets", content)
+        units = split_source(source_shape, content)
         objects: list[ContentObject] = []
         for i, unit in enumerate(units):
-            unit_text = unit if isinstance(unit, str) else str(unit)
-            if not unit_text.strip():
+            if source_shape == "h3_subsections" and isinstance(unit, tuple) and len(unit) == 2:
+                title, body = unit
+                raw_payload = f"### {title}\n{body}".strip()
+            else:
+                raw_payload = (unit if isinstance(unit, str) else str(unit)).strip()
+            if not raw_payload:
                 continue
-            text_specific, line_count = _detect_text_block_specific(unit_text)
+            text_specific, line_count = _detect_text_block_specific(raw_payload)
             objects.append(
                 ContentObject(
                     id=f"{section_id}.text-{i + 1}",
                     type="text_block",
                     role="summary",
-                    raw_payload=unit_text.strip(),
+                    raw_payload=raw_payload,
                     size_estimate={"line_count": line_count},
                     type_specific=text_specific,
                     source_shape_index=i,
-                    source_shape_kind="top_bullets",
+                    source_shape_kind=source_shape,
                 )
             )
         return objects

src/phase_z2_pipeline.py

@@ -1063,66 +1063,60 @@ def run_phase_z2_mvp1(mdx_path: Path, run_id: Optional[str] = None) -> Path:
         # 결과 (PlacementPlan) = debug_zones[i].placement_trace 로 *기록만*.
         # render path / mapper output / final.html 모두 미변경 — B5 baseline SHA 유지.
         # B4 frame selection = catalog declaration order (V4 evidence 미사용 — 별 axis).
-        # Option 1 (PHASE_Z_B4_SOURCE_SHAPE_ENABLED, default OFF) : pilot = F13 top_bullets only.
-        b4_source_shape_enabled = (
+        # Option 1 source_shape-aware path (default OFF) :
+        #   PHASE_Z_B4_SOURCE_SHAPE_ENABLED  → top_bullets (F13/F16 path, Option 1 pilot)
+        #   PHASE_Z_B4_H3_SUBSECTIONS_ENABLED → h3_subsections (F29 path, Option 1 expansion)
+        # 두 flag 독립 — backward compat 보호 (기존 top_bullets flag 의미 = top_bullets only 보존).
+        top_bullets_enabled = (
             os.environ.get("PHASE_Z_B4_SOURCE_SHAPE_ENABLED", "").strip().lower()
             in {"1", "true", "yes"}
         )
-        b1_source_shape = (
-            contract.get("source_shape")
-            if b4_source_shape_enabled and contract.get("source_shape") == "top_bullets"
-            else None
+        h3_subsections_enabled = (
+            os.environ.get("PHASE_Z_B4_H3_SUBSECTIONS_ENABLED", "").strip().lower()
+            in {"1", "true", "yes"}
         )
+        contract_source_shape = contract.get("source_shape")
+        if contract_source_shape == "top_bullets" and top_bullets_enabled:
+            b1_source_shape = "top_bullets"
+        elif contract_source_shape == "h3_subsections" and h3_subsections_enabled:
+            b1_source_shape = "h3_subsections"
+        else:
+            b1_source_shape = None
         content_objects = extract_content_objects(synth_section, source_shape=b1_source_shape)
         placement_plan = plan_placement(
             content_objects=content_objects,
-            frame_contracts=list(load_frame_contracts().values()),
+            contract=contract,
             section_id=synth_section.section_id,
         )
         mapper_frame_template_id = unit.frame_template_id
-        matches_mapper = (
-            placement_plan.selected_template_id == mapper_frame_template_id
-        )
-        match_note: Optional[str] = None
-        if not matches_mapper:
-            if placement_plan.selected_template_id is None:
-                match_note = "no_frame_covers_content_types"
-            else:
-                match_note = (
-                    f"B4 selected '{placement_plan.selected_template_id}' but "
-                    f"mapper uses '{mapper_frame_template_id}' (composition V4 rank-1)"
-                )
         placement_trace = {
             **asdict(placement_plan),
             "mapper_frame_template_id": mapper_frame_template_id,
-            "frame_selection_matches_mapper": matches_mapper,
-            "frame_selection_match_note": match_note,
         }
         # ─── end trace-only runtime 연결 v0 ───
 
-        # ─── B4 gatekeeper (Q-V4B4 / PHASE_Z_B4_GATEKEEPER, default OFF) ───
+        # ─── Q-V4B4 reframe : slot-level validation gatekeeper (default OFF) ───
+        # B4 role reframe : V4/composition = frame authority / B4 = contract validation +
+        # slot assignment + rejection generation. rejection 발생 시 (contract content_type
+        # mismatch / cardinality 위반 / no compatible sub_zone 등) adapter_needed.
         if (
             os.environ.get("PHASE_Z_B4_GATEKEEPER", "").strip().lower()
             in {"1", "true", "yes"}
-            and not matches_mapper
+            and placement_plan.rejection
         ):
             adapter_record = {
                 "position": position,
                 "source_section_ids": unit.source_section_ids,
                 "merge_type": unit.merge_type,
                 "template_id": unit.frame_template_id,
-                "reason": "v4_b4_mismatch",
-                "mismatch_detail": {
-                    "v4_template_id": mapper_frame_template_id,
-                    "b4_selected_template_id": placement_plan.selected_template_id,
-                    "match_note": match_note,
-                },
+                "reason": "slot_validation_failure",
+                "rejection_detail": list(placement_plan.rejection),
             }
             adapter_needed_units.append(adapter_record)
             print(f"  adapter : zone--{position} {unit.source_section_ids} → "
-                  f"{unit.frame_template_id} v4_b4_mismatch → adapter_needed (skip render)")
+                  f"{unit.frame_template_id} slot validation rejection → adapter_needed (skip render)")
             continue
-        # ─── end B4 gatekeeper ───
+        # ─── end Q-V4B4 reframe ───
 
         # mapper 시도 — 실패 (FitError) 시 zone 을 adapter_needed 로 표시하고 skip
         try:

src/phase_z2_placement_planner.py

@@ -82,31 +82,6 @@ class PlacementPlan:
     rejection: list[dict] = field(default_factory=list)
 
 
-# ─── Frame selection ─────────────────────────────────────────────
-
-
-def _select_frame(
-    content_objects: list[ContentObject],
-    frame_contracts: list[dict[str, Any]],
-) -> Optional[dict[str, Any]]:
-    """frame_contracts 중 *content_type_set 을 모두 cover* 하는 첫 frame.
-
-    rule (B4 v0 lock) :
-      1. content_type_set = {obj.type for obj in content_objects}
-      2. frame_contract.accepted_content_types ⊇ content_type_set 인 후보 모음
-      3. frame_contracts 입력 순서 (= YAML declaration order) 첫 entry 선택
-
-    Returns :
-      frame_contract dict 또는 None (cover 가능 frame 없음)
-    """
-    content_type_set = {obj.type for obj in content_objects}
-    for fc in frame_contracts:
-        accepted = set(fc.get("accepted_content_types") or [])
-        if content_type_set <= accepted:               # ⊇ check
-            return fc
-    return None
-
-
 # ─── Sub_zone assignment (Stage B) ───────────────────────────────
 
 
@@ -163,28 +138,26 @@ def _assign_region_to_sub_zone(
 
 def plan_placement(
     content_objects: list[ContentObject],
-    frame_contracts: list[dict[str, Any]],
+    contract: dict[str, Any],
     section_id: str = "",
 ) -> PlacementPlan:
-    """ContentObject[] + frame_contracts → PlacementPlan (Stage A + Stage B 통합).
+    """ContentObject[] + selected contract → PlacementPlan.
+
+    B4 role (post role-reframe) :
+      - frame selection authority = V4/composition (caller); B4 receives the selected contract
+      - B4 책임 = contract validation + slot assignment + rejection generation
 
     v0 algorithm :
-      1. Stage A = B2 plan_internal_regions() 호출 → internal_regions 획득
-      2. frame 선택 : accepted_content_types cover + 입력 순서 first
-         - cover 실패 시 → rejection + early return
-      3. selected_frame 의 sub_zones 읽음 (B3 catalog)
-      4. Stage B (region 1:1 sub_zone 매핑) :
-         - 각 region 마다 narrowest-accepts first + declaration order sub_zone 선택
-         - region.content_unit_ids 를 sub_zone.cardinality 와 비교
-           - count > strict → rejection 추가 / SlotAssignment 미생성
-           - count ≤ strict → SlotAssignment 생성 (under-fill 허용)
-         - 매칭 sub_zone 없는 region → rejection 추가
-      5. display_strategy = inline_full 모두 / overflow_buffer = [] (v0)
+      1. Stage A = plan_internal_regions() → internal_regions
+      2. Contract validation : accepted_content_types ⊇ content_type_set
+         - mismatch → rejection (selected_contract_content_type_mismatch) + early return
+      3. contract 의 sub_zones 읽음 (B3 catalog 의 Frame Slot 선언)
+      4. Stage B (region → sub_zone 매핑) — Option 1 positional or legacy narrowest-first
+      5. display_strategy / overflow_buffer = v0 default
 
     Args :
       content_objects : list[ContentObject] — B1 v0 extractor 출력
-      frame_contracts : list[dict] — frame_contracts.yaml 의 contract dict list
-                         (YAML declaration order = list 순서로 입력 권고)
+      contract        : dict — V4/composition 결정 frame 의 contract (frame_contracts.yaml entry)
       section_id      : region_id / 결과 식별자 prefix
 
     Returns :
@@ -195,29 +168,31 @@ def plan_placement(
     if not content_objects:
         return plan
 
-    # 1. Stage A — B2 호출 (logic 중복 X)
+    # 1. Stage A — B2 호출
     zone_plan: ZoneRegionPlan = plan_internal_regions(
         content_objects=content_objects,
-        frame_contracts=frame_contracts,
         section_id=section_id,
     )
     plan.internal_regions = list(zone_plan.internal_regions)
 
-    # 2. frame 선택
-    selected_frame = _select_frame(content_objects, frame_contracts)
-    if selected_frame is None:
+    # 2. Contract validation — selected contract 가 content_type_set 을 수용?
+    content_type_set = {o.type for o in content_objects}
+    accepted = set(contract.get("accepted_content_types") or [])
+    if not (content_type_set <= accepted):
         plan.rejection.append({
-            "reason": "no_frame_covers_content_types",
-            "content_types": sorted({o.type for o in content_objects}),
+            "reason": "selected_contract_content_type_mismatch",
+            "content_types": sorted(content_type_set),
+            "accepted_content_types": sorted(accepted),
+            "template_id": contract.get("template_id"),
         })
         return plan
 
-    plan.selected_template_id = selected_frame.get("template_id")
-    fid = selected_frame.get("frame_id")
+    plan.selected_template_id = contract.get("template_id")
+    fid = contract.get("frame_id")
     plan.selected_frame_id = str(fid) if fid is not None else None
 
-    # 3. selected_frame 의 sub_zones (B3 catalog 의 Frame Slot 선언)
-    sub_zones = list(selected_frame.get("sub_zones") or [])
+    # 3. contract 의 sub_zones (B3 catalog 의 Frame Slot 선언)
+    sub_zones = list(contract.get("sub_zones") or [])
 
     # 4. Stage B — region 1:1 sub_zone 매핑
     assigned_sub_zone_ids: set[str] = set()
@@ -279,8 +254,9 @@ def _run_self_test():
     PROJECT_ROOT = Path(__file__).parent.parent
     catalog_path = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "frame_contracts.yaml"
     catalog = yaml.safe_load(catalog_path.read_text(encoding="utf-8"))
-    # YAML 의 top-level dict — Python 3.7+ insertion-order 보존. declaration order = F13/F29/F16.
-    frame_contracts = list(catalog.values())
+    # B4 role reframe : caller (V4/composition) 가 frame 선택, B4 는 contract 받음.
+    f13_contract = catalog["three_parallel_requirements"]
+    f29_contract = catalog["process_product_two_way"]
 
     # ─── Test 1 : 1 text_block → F13 → pillar_1 ─────────────────
     text_obj = ContentObject(
@@ -294,7 +270,7 @@ def _run_self_test():
             "max_indent_level": 0, "has_emphasis": False,
         },
     )
-    plan1 = plan_placement([text_obj], frame_contracts, section_id="t1")
+    plan1 = plan_placement([text_obj], f13_contract, section_id="t1")
 
     # frame 선택 = F13 (declaration order first, content_type_set={text_block} cover)
     assert plan1.selected_template_id == "three_parallel_requirements", \
@@ -346,7 +322,7 @@ def _run_self_test():
             "rows": [{"from": "a", "arrow": "➜", "to": "b"}],
         },
     )
-    plan2 = plan_placement([text_obj2, transform_obj], frame_contracts, section_id="t2")
+    plan2 = plan_placement([text_obj2, transform_obj], f29_contract, section_id="t2")
 
     # frame 선택 = F29 (transform_table 수용 유일)
     assert plan2.selected_template_id == "process_product_two_way", \