"""Step Catalog — manifest + class + applicability 통합 단일 계약. Phase 1.1 (architecture-refactor-final/02-final-roadmap.md §Phase 1). 원칙 (01-principles-revised.md §5): - `STEP_MANIFEST` (step_manifest.py)는 **데이터 소스**. 수정은 여기서. - `STEP_CLASSES` (steps/__init__.py)는 runner 바인딩. - `STEP_CATALOG` (본 파일)는 **소비자 전용 View**. manifest + class + 파생 정보 병합. 소비자는 STEP_CATALOG만 본다. manifest/classes를 직접 참조하지 않음. 순환 import 주의 (Claude Phase 1 리뷰 M1): - 본 모듈은 `app.core.steps` 전체를 지연 import (`_build()` 내부). - `app.core.steps` 하위 개별 step 파일은 `app.core.step_catalog`를 import하면 안 됨. 필요 시 step 파일 내부 함수에서 지연 import로 회피. - 부팅 시점 `STEP_CATALOG = _build()`가 즉시 실행되어 `STEP_CLASSES`가 한 번 빌드되므로, ImportError/부수효과는 서버 기동 시점에 조기 노출됨 (late crash 방지). """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Set, Type from app.core.step_manifest import ( STEP_MANIFEST, contains as _manifest_contains, get_all_downstream_recursive as _manifest_get_downstream, get_depends_on as _manifest_get_depends_on, get_downstream_steps as _manifest_get_direct_downstream, get_manifest_dict as _manifest_get_manifest_dict, get_ordered_steps as _manifest_get_ordered, ) @dataclass(frozen=True) class StepEntry: """Step의 모든 메타데이터와 runner 클래스 바인딩을 담은 단일 레코드. manifest에서 가져온 필드 + STEP_CLASSES에서 바인딩한 runner_cls + 파생 필드. """ step_id: str label: str category: str # analysis | image | auxiliary order: float default_model: str provider: str depends_on: List[str] fan_out: bool applicability: str # always | disabled | on_demand | if_* step_type: str # transform | projection | asset | editorial lifecycle: str # active | deprecated | removed resume_sensitive: bool = False modifies_checkpoints: Optional[List[str]] = None invalidate_downstream_on_edit: bool = False # editorial만: True면 타 체크포인트 수정 후 cascade # 2-pass 의존성(two-pass dep) 선언 — 상류(이 step) 입장에서 어떤 하류를 역참조하는지 명시. # step_manifest의 depends_on은 단방향 DAG이나, 일부 상류 step의 코드가 하류 step # 체크포인트를 역참조로 소비하는 구조가 존재한다. 이때 상류 force로 인한 cascade # invalidation이 해당 하류 파일을 지우면 역참조는 dead code가 된다. # 예: scene_detail.consumes_downstream = ["shot_dependency_t2i"] # scene_detail이 scene_context_loader를 통해 shot_dependency_t2i의 refined ref_usage # (zoom_in_detail 등)를 읽어 user_prompt에 주입. scene_detail force 시에도 이 파일은 # 보존되어야 함 (DB step_run은 stale로 표시). # 주의: 파일은 살아있어도 DB는 stale. drift 방지를 위해 상류 force 완료 후 **선언된 # 하류 step도 수동 재실행** 권장. 다른 상류(예: 이 하류의 직접 depends_on 상위)가 # force하는 경로에서는 일반 삭제 동작이므로, 의미가 암묵적이지 않고 명시적이다. consumes_downstream: List[str] = field(default_factory=list) replaced_by: Optional[str] = None sub_steps: Optional[List[Dict[str, Any]]] = None runner_cls: Optional[Type[Any]] = None # ── step_manifest helper re-export (원형은 step_manifest, drift 방지) ─── # Codex W1 Low 2: 기존엔 이 모듈에도 동일 구현이 있었으나 한쪽만 수정될 위험이 있어 # re-export만 남긴다. 소비자는 `step_catalog.contains` 또는 `step_manifest.contains` # 어느 쪽을 써도 동일 구현을 참조. contains = _manifest_contains get_manifest_dict = _manifest_get_manifest_dict def get_step_ids() -> List[str]: """등록된 전체 step_id 목록 — STEP_CATALOG 구축 이전에도 안전.""" return list(STEP_MANIFEST.keys()) def _build() -> Dict[str, StepEntry]: """STEP_MANIFEST + STEP_CLASSES 병합하여 StepCatalog 구축.""" # 지연 import: step_classes 로딩 시점에 순환 의존 회피 from app.core.steps import STEP_CLASSES catalog: Dict[str, StepEntry] = {} for sid, info in STEP_MANIFEST.items(): catalog[sid] = StepEntry( step_id=sid, label=info["label"], category=info["category"], order=info["order"], default_model=info.get("default_model", "-"), provider=info.get("provider", "-"), depends_on=list(info.get("depends_on", [])), fan_out=bool(info.get("fan_out", False)), applicability=info.get("applicability", "always"), step_type=info["step_type"], lifecycle=info["lifecycle"], resume_sensitive=bool(info.get("resume_sensitive", False)), modifies_checkpoints=info.get("modifies_checkpoints"), invalidate_downstream_on_edit=bool(info.get("invalidate_downstream_on_edit", False)), consumes_downstream=list(info.get("consumes_downstream", [])), replaced_by=info.get("replaced_by"), sub_steps=info.get("sub_steps"), runner_cls=STEP_CLASSES.get(sid), ) # ★★★활성화 안전문 — **조립된 것**을 보고, 반쪽이면 여기서 선다. # 여기가 `STEP_MANIFEST` 와 `STEP_CLASSES` 가 **처음 실제로 합쳐지는** # 자리다(Codex 2026-09-01). 계약 모듈은 새 목록을 만들지 않고 이 # 조립물과 **실제 집합**을 인자로 받는다. from app.core import grounding_activation_contract as _act from app.core.grounding_mode import (GROUNDING_MODE_V2_CHUNK, GROUNDING_MODES, PLANNED_MODES, buys_v2_research) from app.modules.pipeline.grounding_entity_contract import ( MATERIALIZABLE_OWNER_TYPES, REFERENCE_SUPPORTED_OWNERS) _act.assert_activation( catalog, accepted_modes=set(GROUNDING_MODES), planned_modes=set(PLANNED_MODES), chunk_mode=GROUNDING_MODE_V2_CHUNK, reference_owners=REFERENCE_SUPPORTED_OWNERS, ledger_owners=MATERIALIZABLE_OWNER_TYPES, buys_legacy_research=buys_v2_research) return catalog STEP_CATALOG: Dict[str, StepEntry] = _build() # ── 조회 헬퍼 ────────────────────────────────────────────────────────── def get_entry(step_id: str) -> Optional[StepEntry]: return STEP_CATALOG.get(step_id) def get_depends_on(step_id: str) -> List[str]: entry = STEP_CATALOG.get(step_id) return list(entry.depends_on) if entry else [] def get_downstream_steps(step_id: str) -> List[str]: """직접 의존 하위 step 목록.""" return [sid for sid, e in STEP_CATALOG.items() if step_id in e.depends_on] def get_all_downstream_recursive(step_id: str) -> List[str]: """재귀적 downstream (BFS). 결과는 정렬.""" visited: Set[str] = set() queue = get_downstream_steps(step_id) while queue: current = queue.pop(0) if current not in visited: visited.add(current) queue.extend(get_downstream_steps(current)) return sorted(visited) def get_ordered_entries() -> List[StepEntry]: """order 오름차순 StepEntry 리스트.""" return sorted(STEP_CATALOG.values(), key=lambda e: e.order) def get_resume_sensitive_step_ids() -> List[str]: """resume_sensitive=True인 step_id 목록 (Phase 1.3 steps.py 소비).""" return [sid for sid, e in STEP_CATALOG.items() if e.resume_sensitive] def get_active_entries(category: Optional[str] = None) -> List[StepEntry]: """lifecycle=active이며 (옵션) 특정 category 일치. applicability는 여기서 해석하지 않음 — 그건 StepRunner 런타임에서 app.core.applicability.resolve_applicability가 처리. """ return [ e for e in get_ordered_entries() if e.lifecycle == "active" and (category is None or e.category == category) ] def get_modifiers_of(target_step_id: str) -> List[str]: """target_step_id의 체크포인트를 수정하는 editorial step 목록. t2i_review가 entity_t2i/scene_detail을 덮어쓰는 것과 같은 관계 조회용. Phase 2/3에서 editorial invalidation 설계 시 사용. """ return [ sid for sid, e in STEP_CATALOG.items() if e.step_type == "editorial" and e.modifies_checkpoints and target_step_id in e.modifies_checkpoints ] def get_consumes_downstream(step_id: str) -> List[str]: """step_id가 역참조로 소비하는 하류 step_id 목록. 2-pass 의존성 선언. 상류 force 시 cascade invalidation에서 이 목록의 체크포인트 파일은 보존된다 (DB는 stale). 기본 대응: 상류 force 완료 후 선언된 하류도 수동 재실행 권장. """ entry = STEP_CATALOG.get(step_id) return list(entry.consumes_downstream) if entry else [] def get_consumers_of(step_id: str) -> List[str]: """step_id를 consumes_downstream으로 역참조 소비하는 상류 step_id 목록. `get_consumes_downstream`의 역방향 조회. UI가 "이 step이 stale인 이유가 상류의 2-pass 소비 때문인가?"를 판단해 drift stale을 일반 stale과 구분 표시하는 데 사용. """ return sorted([ sid for sid, e in STEP_CATALOG.items() if step_id in e.consumes_downstream ]) # ── manifest 호환 re-export ──────────────────────────────────────────── # 기존 코드(step_manifest.get_*)는 당분간 그대로 유지. 신규 코드는 step_catalog 사용. __all__ = [ "StepEntry", "STEP_CATALOG", "get_entry", "get_depends_on", "get_downstream_steps", "get_all_downstream_recursive", "get_ordered_entries", "get_resume_sensitive_step_ids", "get_active_entries", "get_modifiers_of", "get_consumes_downstream", "get_consumers_of", "contains", "get_manifest_dict", "get_step_ids", ]