refactor(#41): IMP-32 Step 9 application_plan helper extraction (u1~u5)

Pure refactor — extract inline Step 9 per-unit application_plan dict
assembly into module-level private helpers for testability. Replaces
IMP-05 Case 7 inspect.getsource() literal guard with direct helper-call
shape test. Behavior preserved: key set/order, candidate_evidence +
fallback_chain compat alias identity, IMP-06 additive plan fields,
IMP-11 D-2 markers (single _contract = get_contract(c.template_id)
bind + catalog_registered + min_height_px chain).

- u1 _application_candidates_for_unit(unit) at src/phase_z2_pipeline.py
  :2829-2853 — APPLICATION_MODE_BY_V4_LABEL mapping (pure extraction)
- u2 _v4_all_judgments_for_unit(v4_all_for_unit) at :2855-2882 —
  IMP-11 D-2 chain preserved literally
- u3 _build_application_plan_unit(unit, zone_plan, selection_trace,
  plan_record, v4_all_for_unit, layout_preset, layout_candidates_list)
  at :2885-2995 — byte-identical per-unit dict (key set + order +
  value identity), candidate_evidence / fallback_chain compat alias,
  v4_candidates list, v4_all_judgments, application_candidates, IMP-06
  additive plan fields
- u4 Step 9 inline loop body at :4620-4658 replaced with helper call;
  per-index/per-id lookups (zone_region_plans[i], v4_fallback_traces
  .get(...), plan_record_by_unit_id.get(id(unit)), section_alias_by_id,
  lookup_v4_all_judgments(...)) stay at call-site
- u5 tests/test_phase_z2_v4_fallback.py Case 7 rewritten to
  test_build_application_plan_unit_emits_candidate_evidence_and_alias
  — direct helper call with SimpleNamespace duck-typed input; asserts
  candidate_evidence list identity (is), fallback_chain compat-alias
  identity (is), key order (candidate_evidence before fallback_chain),
  and compat-alias comment scoped to inspect.getsource(_build_
  application_plan_unit)

Verification: targeted 22 passed, full pytest 408 passed (0 fail/skip),
smoke 11/11 PASS (2 pre-existing baseline SKIPs unchanged).

Cross-ref: IMP-05 (#5) commit 23d1b25 Case 7 temporary source guard
(replaced) / Codex #20 + #21 / IMP-11 D-2 marker preserved.
This commit is contained in:
2026-05-21 03:17:27 +09:00
parent 182aa7c47f
commit c412f1ea75
2 changed files with 253 additions and 112 deletions

View File

@@ -2826,6 +2826,176 @@ def write_debug_json(run_dir: Path, layout_preset: str,
return debug_path return debug_path
# ─── Step 9 application-plan helpers (IMP-32 u1) ───────────────
def _application_candidates_for_unit(unit) -> list[dict]:
"""Step 9 (IMP-32 u1) — application candidate dicts from unit.v4_candidates.
Pure extraction of inline block at src/phase_z2_pipeline.py:4487-4501.
Behavior preserved: key set/order, APPLICATION_MODE_BY_V4_LABEL lookup,
required_changes placeholder = [] (v0 = trace-only).
"""
app_candidates = []
for c in unit.v4_candidates:
mode, auto_app, delegated = APPLICATION_MODE_BY_V4_LABEL.get(
c.label, ("exclude", False, None)
)
app_candidates.append({
"template_id": c.template_id,
"frame_id": c.frame_id,
"v4_label": c.label,
"application_mode": mode,
"auto_applicable": auto_app,
"required_changes": [], # v0 = trace-only
"delegated_to": delegated,
})
return app_candidates
def _v4_all_judgments_for_unit(v4_all_for_unit) -> list[dict]:
"""Step 9 (IMP-32 u2) — V4 all-judgment dicts (reject 포함) for a unit.
Pure extraction of inline block at src/phase_z2_pipeline.py:4529-4545
(post-u1 line numbers). IMP-11 D-2 markers preserved in this helper:
single `_contract = get_contract(c.template_id)` bind, `catalog_registered`
boolean, and `min_height_px` chain `(_contract or {}).get("visual_hints", {}).get("min_height_px")`.
Key set/order unchanged: template_id, frame_id, frame_number, v4_rank,
confidence, label, catalog_registered, min_height_px.
"""
# IMP-11 D-2 (u1) — per-candidate min_height_px source = catalog
# frame_contracts[template_id].visual_hints.min_height_px (logical 1280×720 px).
# None when contract unregistered (frontend tolerates undefined).
# Single get_contract lookup binds both catalog_registered and min_height_px.
v4_all_judgments_list = []
for c in v4_all_for_unit:
_contract = get_contract(c.template_id)
v4_all_judgments_list.append({
"template_id": c.template_id,
"frame_id": c.frame_id,
"frame_number": c.frame_number,
"v4_rank": c.v4_rank,
"confidence": c.confidence,
"label": c.label,
"catalog_registered": _contract is not None,
"min_height_px": (_contract or {}).get("visual_hints", {}).get("min_height_px"),
})
return v4_all_judgments_list
def _build_application_plan_unit(
unit,
zone_plan,
selection_trace,
plan_record,
v4_all_for_unit,
layout_preset,
layout_candidates_list,
) -> dict:
"""Step 9 (IMP-32 u3) — per-unit application_plan dict assembly.
Pure extraction of the inline `application_plan_units.append({...})` block
currently at src/phase_z2_pipeline.py:4577-4623 (post-u1/u2 line numbers).
Byte-identical output (key set + key order + value identity) when called
with the same per-unit inputs:
- unit : Step 6 unit (source_section_ids, v4_candidates,
v4_rank, selection_path, fallback_reason,
frame_template_id).
- zone_plan : Step 8 per-unit zone_plan dict (region_layout_
candidates, display_strategy_candidates).
- selection_trace : v4_fallback_traces[unit.source_section_ids[0]]
(candidates list for candidate_evidence /
fallback_chain compat alias).
- plan_record : plan_record_by_unit_id[id(unit)] or None
(IMP-06 plan-aware additive fields).
- v4_all_for_unit : lookup_v4_all_judgments(...) result (Step 7-A
axis trace — reject 포함 모든 V4 후보).
- layout_preset : Step 7 preset name (e.g., "Type A").
- layout_candidates_list : Step 7 candidate list.
Per-index/per-id lookups (zone_region_plans[i], v4_fallback_traces.get(...),
plan_record_by_unit_id.get(id(unit)), section_alias_by_id, lookup_v4_all_
judgments(...)) stay at the call-site (u4).
Invariants preserved:
- candidate_evidence = selection_trace.get("candidates", []) — primary field.
- fallback_chain = same list — compat alias for pre-IMP-05 readers.
- v4_candidates list comprehension fields + order unchanged.
- IMP-06 additive plan fields (position / assignment_source / section_
assignment_override / replaced_auto_unit / skipped_collided_auto_units /
skipped_reason) — None / False / [] when no override CLI used.
"""
unit_id = "+".join(unit.source_section_ids)
has_v4 = bool(unit.v4_candidates)
candidate_status = "ok" if has_v4 else "no_non_reject_v4_candidate"
application_status = "ok" if has_v4 else "no_v4_candidate"
current_default = unit.frame_template_id if has_v4 else None
# IMP-06 blocker-fix (Codex #13 Blocker 3 / #16) — plan-aware additive
# fields. additive = pre-IMP-06 readers (no override CLI used) see
# position=None / assignment_source=None / section_assignment_override
# =False / replaced_auto_unit=None / skipped_collided_auto_units=[] /
# skipped_reason=None — i.e. byte-identical absent overrides.
plan_position = plan_record.get("position") if plan_record else None
plan_assignment_source = plan_record.get("assignment_source") if plan_record else None
plan_section_override = bool(plan_record.get("section_assignment_override")) if plan_record else False
plan_replaced_auto = plan_record.get("replaced_auto_unit") if plan_record else None
plan_skipped_collided = list(plan_record.get("skipped_collided_auto_units") or []) if plan_record else []
plan_skipped_reason = plan_record.get("skipped_reason") if plan_record else None
app_candidates = _application_candidates_for_unit(unit)
v4_all_judgments_list = _v4_all_judgments_for_unit(v4_all_for_unit)
return {
"unit_id": unit_id,
"layout_preset": layout_preset,
"layout_candidates": layout_candidates_list,
"region_layout_candidates": zone_plan.get("region_layout_candidates", []),
"display_strategy_candidates": zone_plan.get("display_strategy_candidates", []),
"candidate_status": candidate_status,
"application_status": application_status,
"current_default_candidate": current_default,
"selected_v4_rank": unit.v4_rank,
"selection_path": unit.selection_path,
"fallback_used": bool(unit.selection_path and "fallback" in unit.selection_path),
"fallback_reason": unit.fallback_reason,
# IMP-05 L2 (Codex #10 D4 / #16 idea A) — Step 9 per-unit candidate evidence.
# candidate_evidence is the primary field for future frontend / AI consumers.
# fallback_chain is kept as a compat alias for any pre-IMP-05 reader.
"candidate_evidence": selection_trace.get("candidates", []),
"fallback_chain": selection_trace.get("candidates", []), # compat alias; prefer candidate_evidence
"v4_candidates": [
{
"template_id": c.template_id,
"frame_id": c.frame_id,
"frame_number": c.frame_number,
"v4_rank": c.v4_rank,
"confidence": c.confidence,
"label": c.label,
}
for c in unit.v4_candidates
],
# Step 7-A axis 보강 (사용자 lock 2026-05-08) — frontend UI 가 reject
# 포함 모든 V4 후보를 시각 차별 (회색) 로 보여줄 수 있도록 trace.
# length = 0~32. label 별 count : v4_candidates 는 non-reject only,
# v4_all_judgments 는 reject 포함.
# catalog_registered = frame_contracts.yaml 에 contract 있는지 여부.
# false 면 사용자가 override 시도해도 Step 7-A 가 skip (render path 미연결).
# IMP-11 D-2 (u1) : per-candidate min_height_px added (None when unregistered).
"v4_all_judgments": v4_all_judgments_list,
"application_candidates": app_candidates,
# IMP-06 blocker-fix (Codex #13 Blocker 3 / #16) — plan-aware
# additive fields. None / False / [] when no override CLI used.
"position": plan_position,
"assignment_source": plan_assignment_source,
"section_assignment_override": plan_section_override,
"replaced_auto_unit": plan_replaced_auto,
"skipped_collided_auto_units": plan_skipped_collided,
"skipped_reason": plan_skipped_reason,
}
# ─── Main entry ──────────────────────────────────────────────── # ─── Main entry ────────────────────────────────────────────────
def run_phase_z2_mvp1( def run_phase_z2_mvp1(
@@ -4450,28 +4620,10 @@ def run_phase_z2_mvp1(
application_plan_units = [] application_plan_units = []
for i, unit in enumerate(units): for i, unit in enumerate(units):
unit_id = "+".join(unit.source_section_ids)
# zone_region_plans 는 unit i 와 1:1 (Step 6 unit → Step 8 zone_plan). # zone_region_plans 는 unit i 와 1:1 (Step 6 unit → Step 8 zone_plan).
zone_plan = zone_region_plans[i] if i < len(zone_region_plans) else {} zone_plan = zone_region_plans[i] if i < len(zone_region_plans) else {}
has_v4 = bool(unit.v4_candidates)
candidate_status = "ok" if has_v4 else "no_non_reject_v4_candidate"
application_status = "ok" if has_v4 else "no_v4_candidate"
current_default = unit.frame_template_id if has_v4 else None
selection_trace = v4_fallback_traces.get(unit.source_section_ids[0], {}) selection_trace = v4_fallback_traces.get(unit.source_section_ids[0], {})
# IMP-06 blocker-fix (Codex #13 Blocker 3 / #16) — plan-aware additive
# fields. additive = pre-IMP-06 readers (no override CLI used) see
# position=None / assignment_source=None / section_assignment_override
# =False / replaced_auto_unit=None / skipped_collided_auto_units=[] /
# skipped_reason=None — i.e. byte-identical absent overrides.
plan_record = plan_record_by_unit_id.get(id(unit)) plan_record = plan_record_by_unit_id.get(id(unit))
plan_position = plan_record.get("position") if plan_record else None
plan_assignment_source = plan_record.get("assignment_source") if plan_record else None
plan_section_override = bool(plan_record.get("section_assignment_override")) if plan_record else False
plan_replaced_auto = plan_record.get("replaced_auto_unit") if plan_record else None
plan_skipped_collided = list(plan_record.get("skipped_collided_auto_units") or []) if plan_record else []
plan_skipped_reason = plan_record.get("skipped_reason") if plan_record else None
# Step 7-A axis 보강 — reject 포함 모든 V4 judgments (frontend UI 가 # Step 7-A axis 보강 — reject 포함 모든 V4 judgments (frontend UI 가
# 모든 frame 의 png 를 카드로 보여주기 위함). # 모든 frame 의 png 를 카드로 보여주기 위함).
@@ -4484,87 +4636,22 @@ def run_phase_z2_mvp1(
v4, _first_sid, alias_keys=section_alias_by_id.get(_first_sid) v4, _first_sid, alias_keys=section_alias_by_id.get(_first_sid)
) )
# application_candidates : V4 후보 zip 으로 application_mode 변환 # IMP-32 u4 — per-unit application_plan dict assembly extracted into
app_candidates = [] # _build_application_plan_unit(...). Per-index/per-id lookups
for c in unit.v4_candidates: # (zone_region_plans[i], v4_fallback_traces.get(...),
mode, auto_app, delegated = APPLICATION_MODE_BY_V4_LABEL.get( # plan_record_by_unit_id.get(id(unit)), section_alias_by_id,
c.label, ("exclude", False, None) # lookup_v4_all_judgments(...)) stay at the call-site.
application_plan_units.append(
_build_application_plan_unit(
unit,
zone_plan,
selection_trace,
plan_record,
v4_all_for_unit,
layout_preset,
layout_candidates_list,
) )
app_candidates.append({ )
"template_id": c.template_id,
"frame_id": c.frame_id,
"v4_label": c.label,
"application_mode": mode,
"auto_applicable": auto_app,
"required_changes": [], # v0 = trace-only
"delegated_to": delegated,
})
# IMP-11 D-2 (u1) — per-candidate min_height_px source = catalog
# frame_contracts[template_id].visual_hints.min_height_px (logical 1280×720 px).
# None when contract unregistered (frontend tolerates undefined).
# Single get_contract lookup binds both catalog_registered and min_height_px.
v4_all_judgments_list = []
for c in v4_all_for_unit:
_contract = get_contract(c.template_id)
v4_all_judgments_list.append({
"template_id": c.template_id,
"frame_id": c.frame_id,
"frame_number": c.frame_number,
"v4_rank": c.v4_rank,
"confidence": c.confidence,
"label": c.label,
"catalog_registered": _contract is not None,
"min_height_px": (_contract or {}).get("visual_hints", {}).get("min_height_px"),
})
application_plan_units.append({
"unit_id": unit_id,
"layout_preset": layout_preset,
"layout_candidates": layout_candidates_list,
"region_layout_candidates": zone_plan.get("region_layout_candidates", []),
"display_strategy_candidates": zone_plan.get("display_strategy_candidates", []),
"candidate_status": candidate_status,
"application_status": application_status,
"current_default_candidate": current_default,
"selected_v4_rank": unit.v4_rank,
"selection_path": unit.selection_path,
"fallback_used": bool(unit.selection_path and "fallback" in unit.selection_path),
"fallback_reason": unit.fallback_reason,
# IMP-05 L2 (Codex #10 D4 / #16 idea A) — Step 9 per-unit candidate evidence.
# candidate_evidence is the primary field for future frontend / AI consumers.
# fallback_chain is kept as a compat alias for any pre-IMP-05 reader.
"candidate_evidence": selection_trace.get("candidates", []),
"fallback_chain": selection_trace.get("candidates", []), # compat alias; prefer candidate_evidence
"v4_candidates": [
{
"template_id": c.template_id,
"frame_id": c.frame_id,
"frame_number": c.frame_number,
"v4_rank": c.v4_rank,
"confidence": c.confidence,
"label": c.label,
}
for c in unit.v4_candidates
],
# Step 7-A axis 보강 (사용자 lock 2026-05-08) — frontend UI 가 reject
# 포함 모든 V4 후보를 시각 차별 (회색) 로 보여줄 수 있도록 trace.
# length = 0~32. label 별 count : v4_candidates 는 non-reject only,
# v4_all_judgments 는 reject 포함.
# catalog_registered = frame_contracts.yaml 에 contract 있는지 여부.
# false 면 사용자가 override 시도해도 Step 7-A 가 skip (render path 미연결).
# IMP-11 D-2 (u1) : per-candidate min_height_px added (None when unregistered).
"v4_all_judgments": v4_all_judgments_list,
"application_candidates": app_candidates,
# IMP-06 blocker-fix (Codex #13 Blocker 3 / #16) — plan-aware
# additive fields. None / False / [] when no override CLI used.
"position": plan_position,
"assignment_source": plan_assignment_source,
"section_assignment_override": plan_section_override,
"replaced_auto_unit": plan_replaced_auto,
"skipped_collided_auto_units": plan_skipped_collided,
"skipped_reason": plan_skipped_reason,
})
units_with_no_v4 = [ units_with_no_v4 = [
u["unit_id"] for u in application_plan_units u["unit_id"] for u in application_plan_units

View File

@@ -295,25 +295,79 @@ def test_existing_trace_shape_does_not_regress(patch_selector_deps):
assert trace["selection_path"] == "rank_1" assert trace["selection_path"] == "rank_1"
# ─── Case 7 : Step 9 production-source guard (Codex #20 blocker fix) ─── # ─── Case 7 : Step 9 helper-call shape test (IMP-32 u5 — replaces source guard) ───
def test_step9_production_emits_candidate_evidence_and_alias(): def test_build_application_plan_unit_emits_candidate_evidence_and_alias():
"""Temporary production-source guard for IMP-05 Step 9 evidence fields. """IMP-32 u5 — direct helper-call shape test for Step 9 evidence fields.
Step 9 application-plan unit assembly is currently inline, so this test Replaces the IMP-05 Case 7 `inspect.getsource(phase_z2_pipeline)` literal
checks the exact production assignments until IMP-32 extracts a helper. guard (introduced at commit `23d1b25` while Step 9 unit assembly was
Once that helper exists, replace this source-string guard with a direct inline) with a direct call to `_build_application_plan_unit`, the helper
helper-call test. extracted in IMP-32 u3. Verification axes preserved:
- candidate_evidence list identity sourced from `selection_trace["candidates"]`
- fallback_chain compat-alias identity (same list object as candidate_evidence)
- key order: candidate_evidence before fallback_chain
- compat-alias comment preserved on the helper's fallback_chain line
""" """
source = inspect.getsource(phase_z2_pipeline) from types import SimpleNamespace
candidate_line = '"candidate_evidence": selection_trace.get("candidates", [])'
alias_line = '"fallback_chain": selection_trace.get("candidates", [])'
assert candidate_line in source from src.phase_z2_pipeline import _build_application_plan_unit
assert alias_line in source
assert source.index(candidate_line) < source.index(alias_line) candidates_list = [
assert "compat alias; prefer candidate_evidence" in source {"rank": 1, "template_id": "MOCK_template_direct_a", "label": "use_as_is"},
]
selection_trace = {"candidates": candidates_list}
# Synthetic CompositionUnit-shape duck-typed input — matches V4Match attrs
# used inside the helper (template_id / frame_id / frame_number / v4_rank /
# confidence / label per src/phase_z2_pipeline.py V4Match dataclass).
v4_candidate = SimpleNamespace(
template_id="MOCK_template_direct_a",
frame_id="MOCK_frame_001",
frame_number=1,
v4_rank=1,
confidence=0.9,
label="use_as_is",
)
unit = SimpleNamespace(
source_section_ids=["S1"],
v4_candidates=[v4_candidate],
v4_rank=1,
selection_path="rank_1",
fallback_reason=None,
frame_template_id="MOCK_template_direct_a",
)
result = _build_application_plan_unit(
unit=unit,
zone_plan={},
selection_trace=selection_trace,
plan_record=None,
v4_all_for_unit=[],
layout_preset="Type A",
layout_candidates_list=[],
)
# IMP-05 L2 — candidate_evidence is the primary field, identity-bound to
# selection_trace["candidates"] (not a copy).
assert "candidate_evidence" in result
assert result["candidate_evidence"] is candidates_list
# compat alias — fallback_chain references the SAME list object as
# candidate_evidence (verified by `is` identity, not equality).
assert "fallback_chain" in result
assert result["fallback_chain"] is candidates_list
# key order — candidate_evidence MUST precede fallback_chain in the
# returned dict to preserve documented L2 ordering.
keys = list(result.keys())
assert keys.index("candidate_evidence") < keys.index("fallback_chain")
# compat-alias comment preserved on the helper's fallback_chain line.
helper_source = inspect.getsource(_build_application_plan_unit)
assert "compat alias; prefer candidate_evidence" in helper_source
# ─── Case 8 : Step 20 slide-status qualifier fields presence + defensive default # ─── Case 8 : Step 20 slide-status qualifier fields presence + defensive default