diff --git a/src/phase_z2_classifier.py b/src/phase_z2_classifier.py index a4c3160..707795c 100644 --- a/src/phase_z2_classifier.py +++ b/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 []), diff --git a/src/phase_z2_content_extractor.py b/src/phase_z2_content_extractor.py index 8c579f7..cee5720 100644 --- a/src/phase_z2_content_extractor.py +++ b/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 diff --git a/src/phase_z2_pipeline.py b/src/phase_z2_pipeline.py index 1e6e2a7..323a6e3 100644 --- a/src/phase_z2_pipeline.py +++ b/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: diff --git a/src/phase_z2_placement_planner.py b/src/phase_z2_placement_planner.py index 2213e91..f1a1b93 100644 --- a/src/phase_z2_placement_planner.py +++ b/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", \